📊 项目2：自动化办公 Agent

难度: ⭐⭐⭐⭐ Claude Code ⏱ 预计 4-5 小时

1. 需求分析

业务场景

小华每天上班要处理大量重复办公任务：整理会议纪要、批量重命名文件、汇总 Excel 数据、整理邮件内容、生成周报。这些任务占用了大量时间，但本质上都是"读取→分析→处理→输出"的模式化工作。

核心功能

• 📄 文本总结：长文档/会议纪要一键摘要
• 📊 表格解析：读取 CSV/Excel，提取关键数据，生成分析
• 📁 文件批量处理：批量重命名、格式转换、内容替换
• 📅 日程整理：从文本中提取时间安排，生成日程表
• 📝 报告生成：按要求格式自动生成工作报告

2. 架构设计

3. 完整代码

#!/usr/bin/env python3
"""
OfficeAgent — 自动化办公 Agent
================================
功能：文本总结 | 表格解析 | 文件批量处理 | 日程整理 | 报告生成

额外依赖：pip install openpyxl pandas

运行：python3 office_agent.py
"""

import csv
import json
import os
import re
import shutil
from datetime import datetime, timedelta
from pathlib import Path
from typing import Any

from dotenv import load_dotenv
from anthropic import Anthropic

load_dotenv()


# ================================================================
# Prompt 模板库
# ================================================================

PROMPTS = {
    "summarize": """请用中文总结以下文本的核心要点：
- 保持简洁（原文的 10-20% 长度）
- 突出关键数据和结论
- 使用要点列表格式

原文：
{text}""",

    "extract_schedule": """从以下文本中提取所有日程安排信息，按时间顺序列出：
- 时间: 日期+具体时间
- 事件: 做什么
- 地点: 在哪里（如有）
- 参与者: 参与人（如有）

如果没有明确的日程信息，返回"未找到日程安排"。

文本：
{text}""",

    "generate_report": """根据以下数据和模板，生成一份专业的工作报告。
格式要求：
1. 标题清晰
2. 分章节（概述 → 详细内容 → 数据 → 结论/下一步）
3. 数据部分用表格展示（如有）
4. 语言专业但易读

数据：
{data}

模板要求：
{template}""",

    "analyze_table": """分析以下表格数据，提供洞察：
1. 数据概览（行数、列数、列含义）
2. 关键统计（总数、均值、最大/最小值等，根据数据类型智能选择）
3. 异常数据点（如有）
4. 趋势或模式（如有）
5. 简要建议

表格数据：
{table_data}""",
}


# ================================================================
# 文件处理工具
# ================================================================

class FileHandler:
    """文件批量处理"""

    @staticmethod
    def batch_rename(directory: str, pattern: str, replacement: str, dry_run: bool = True) -> str:
        """批量重命名文件"""
        dir_path = Path(directory)
        if not dir_path.exists():
            return f"❌ 目录不存在: {directory}"

        results = []
        count = 0
        for filepath in sorted(dir_path.iterdir()):
            if filepath.is_file() and not filepath.name.startswith("."):
                new_name = filepath.name.replace(pattern, replacement)
                if new_name != filepath.name:
                    new_path = filepath.parent / new_name
                    if not dry_run:
                        filepath.rename(new_path)
                    results.append(f"  {filepath.name} → {new_name}")
                    count += 1

        prefix = "🔄 [预览]" if dry_run else "✅ [已执行]"
        if count == 0:
            return f"{prefix} 没有匹配的文件 (模式: '{pattern}' → '{replacement}')"
        return f"{prefix} 共{count}个文件:\n" + "\n".join(results)

    @staticmethod
    def convert_encoding(directory: str, from_enc: str = "gbk", to_enc: str = "utf-8") -> str:
        """批量转换文件编码（GBK → UTF-8 很常见）"""
        dir_path = Path(directory)
        results = []
        for filepath in dir_path.rglob("*.txt"):
            try:
                content = filepath.read_bytes()
                text = content.decode(from_enc)
                filepath.write_text(text, encoding=to_enc)
                results.append(f"  ✅ {filepath.name}: {from_enc} → {to_enc}")
            except Exception as e:
                results.append(f"  ⚠️ {filepath.name}: {e}")
        return "编码转换结果:\n" + "\n".join(results[:20])

    @staticmethod
    def organize_by_date(directory: str) -> str:
        """按文件修改日期整理到月份文件夹"""
        dir_path = Path(directory)
        moved = 0
        for filepath in dir_path.iterdir():
            if filepath.is_file() and not filepath.name.startswith("."):
                mtime = datetime.fromtimestamp(filepath.stat().st_mtime)
                month_dir = dir_path / mtime.strftime("%Y-%m")
                month_dir.mkdir(exist_ok=True)
                dest = month_dir / filepath.name
                if dest.exists():
                    dest = month_dir / f"{filepath.stem}_dup{filepath.suffix}"
                shutil.move(str(filepath), str(dest))
                moved += 1
        return f"✅ 已按日期整理 {moved} 个文件"


# ================================================================
# 表格处理工具
# ================================================================

class TableHandler:
    """CSV/Excel 表格解析"""

    @staticmethod
    def read_csv(filepath: str, max_rows: int = 50) -> tuple[list[str], list[list]]:
        """读取CSV，返回(列名, 数据行)"""
        path = Path(filepath)
        if not path.exists():
            return [], []
        with open(path, "r", encoding="utf-8-sig") as f:
            reader = csv.reader(f)
            headers = next(reader, [])
            rows = [row for i, row in enumerate(reader) if i < max_rows]
        return headers, rows

    @staticmethod
    def read_excel(filepath: str, max_rows: int = 50) -> dict[str, list]:
        """读取 Excel（需要 openpyxl/pandas）"""
        try:
            import pandas as pd
            df = pd.read_excel(filepath, nrows=max_rows)
            return {"headers": df.columns.tolist(), "data": df.values.tolist()}
        except ImportError:
            return {"error": "请安装 pandas 和 openpyxl: pip install pandas openpyxl"}
        except Exception as e:
            return {"error": str(e)}

    @staticmethod
    def quick_stats(headers: list, rows: list) -> dict:
        """快速统计表格数据"""
        if not headers or not rows:
            return {"error": "表格为空"}

        stats = {"row_count": len(rows), "col_count": len(headers), "columns": {}}

        for col_idx, col_name in enumerate(headers):
            values = [row[col_idx] for row in rows if col_idx < len(row)]
            # 尝试解析数字
            num_values = []
            for v in values:
                try:
                    num_values.append(float(v))
                except (ValueError, TypeError):
                    pass

            col_stats = {"non_empty": len([v for v in values if v])}
            if num_values:
                col_stats["type"] = "数值"
                col_stats["sum"] = sum(num_values)
                col_stats["avg"] = sum(num_values) / len(num_values)
                col_stats["max"] = max(num_values)
                col_stats["min"] = min(num_values)
            else:
                col_stats["type"] = "文本"
                # 最常见的前3个值
                from collections import Counter
                top = Counter(values).most_common(3)
                col_stats["top_values"] = top

            stats["columns"][col_name] = col_stats

        return stats


# ================================================================
# 办公 Agent 主类
# ================================================================

class OfficeAgent:
    """自动化办公 Agent"""

    def __init__(self):
        self.client = Anthropic(api_key=os.getenv("ANTHROPIC_API_KEY"))
        self.model = "claude-sonnet-4-6"
        self.file_handler = FileHandler()
        self.table_handler = TableHandler()
        self.task_log: list[dict] = []

    def _call_claude(self, prompt: str, max_tokens: int = 1500) -> str:
        """调用 Claude API"""
        try:
            response = self.client.messages.create(
                model=self.model,
                max_tokens=max_tokens,
                messages=[{"role": "user", "content": prompt}],
                temperature=0.3,
            )
            return response.content[0].text.strip()
        except Exception as e:
            return f"❌ API 调用失败: {e}"

    def summarize_text(self, filepath: str) -> str:
        """文本总结"""
        path = Path(filepath)
        if not path.exists():
            return f"❌ 文件不存在: {filepath}"

        try:
            text = path.read_text("utf-8")
        except Exception as e:
            return f"❌ 读取失败: {e}"

        if len(text) < 50:
            return f"📝 文本太短（{len(text)}字符），无需总结: {text}"

        print(f"📝 正在总结 {path.name} ({len(text)} 字符)...")
        prompt = PROMPTS["summarize"].format(text=text[:8000])
        summary = self._call_claude(prompt)
        self._log("summarize", filepath)
        return f"📝 **{path.name} 总结**:\n{summary}"

    def extract_schedule(self, text_or_file: str) -> str:
        """从文本中提取日程"""
        # 判断是文件路径还是直接文本
        path = Path(text_or_file)
        if path.exists():
            text = path.read_text("utf-8")
            source = f"文件 {path.name}"
        else:
            text = text_or_file
            source = "输入文本"

        print(f"📅 正在从{source}提取日程...")
        prompt = PROMPTS["extract_schedule"].format(text=text[:5000])
        schedule = self._call_claude(prompt)
        self._log("extract_schedule", source)
        return f"📅 **日程提取结果** ({source}):\n{schedule}"

    def analyze_table_file(self, filepath: str) -> str:
        """分析表格文件"""
        path = Path(filepath)
        if not path.exists():
            return f"❌ 文件不存在: {filepath}"

        # 读取表格
        if path.suffix.lower() == ".csv":
            headers, rows = self.table_handler.read_csv(filepath)
        elif path.suffix.lower() in (".xlsx", ".xls"):
            result = self.table_handler.read_excel(filepath)
            if "error" in result:
                return f"❌ {result['error']}"
            headers, rows = result["headers"], result["data"]
        else:
            return f"❌ 不支持的表格格式: {path.suffix}"

        # 快速统计
        stats = self.table_handler.quick_stats(headers, rows)

        # 格式化数据让 AI 分析
        table_text = f"列名: {', '.join(headers)}\n"
        table_text += f"行数: {len(rows)}\n"
        table_text += "前10行数据:\n"
        for i, row in enumerate(rows[:10], 1):
            table_text += f"  {i}. {row}\n"

        print(f"📊 正在分析表格 {path.name}...")
        prompt = PROMPTS["analyze_table"].format(table_data=table_text)
        analysis = self._call_claude(prompt)

        # 附加统计信息
        stats_text = f"\n\n📊 快速统计: {json.dumps(stats, ensure_ascii=False, indent=2)[:500]}"
        self._log("analyze_table", filepath)
        return f"📊 **{path.name} 分析**:\n{analysis}{stats_text}"

    def batch_process(self, directory: str, action: str, **kwargs) -> str:
        """批量文件处理"""
        actions = {
            "rename": lambda: self.file_handler.batch_rename(
                directory, kwargs.get("pattern", ""), kwargs.get("replacement", ""),
                dry_run=kwargs.get("dry_run", True)
            ),
            "encoding": lambda: self.file_handler.convert_encoding(directory),
            "organize": lambda: self.file_handler.organize_by_date(directory),
        }

        func = actions.get(action)
        if not func:
            return f"❌ 未知操作: {action}。可选: {list(actions.keys())}"

        result = func()
        self._log(f"batch_{action}", directory)
        return result

    def generate_report(self, data: str, template: str = "标准工作报告") -> str:
        """生成工作报告"""
        print("📝 正在生成报告...")
        prompt = PROMPTS["generate_report"].format(data=data[:5000], template=template)
        report = self._call_claude(prompt, max_tokens=2000)

        # 保存报告
        report_path = Path(f"report_{datetime.now().strftime('%Y%m%d_%H%M')}.md")
        report_path.write_text(report, encoding="utf-8")
        self._log("generate_report", str(report_path))
        return f"📝 **报告已生成**: {report_path.name}\n\n{report[:1000]}..."

    def _log(self, action: str, target: str):
        self.task_log.append({
            "action": action,
            "target": target,
            "time": datetime.now().isoformat(),
        })

    def show_log(self):
        print(f"\n📋 任务日志 ({len(self.task_log)} 条):")
        for entry in self.task_log[-10:]:
            print(f"  [{entry['time'][:19]}] {entry['action']}: {entry['target']}")


# ================================================================
# 交互入口
# ================================================================

def main():
    ascii_art = """
  ___   __  __          _   _
 / _ \\ / _|/ _| ___ ___| |_(_) ___ _ __
| | | | |_| |_ / _ \\/ __| __| |/ _ \\ '_ \\
| |_| |  _|  _|  __/\\__ \\ |_| |  __/ | | |
 \\___/|_| |_|  \\___||___/\\__|_|\\___|_| |_|
    """
    print(ascii_art)
    print("=" * 55)
    print("📊 OfficeAgent — 自动化办公 Agent")
    print("=" * 55)
    print("命令:")
    print("  summarize <文件>     — 总结文档")
    print("  schedule <文件/文本>  — 提取日程")
    print("  table <csv/xlsx>     — 分析表格")
    print("  batch rename <目录> 旧模式 新模式 — 批量重命名(预览)")
    print("  batch rename! <目录> 旧模式 新模式 — 批量重命名(执行)")
    print("  batch organize <目录> — 按日期整理文件")
    print("  report <数据文本>     — 生成工作报告")
    print("  log                   — 查看任务日志")
    print("  quit                  — 退出")
    print("=" * 55)

    if not os.getenv("ANTHROPIC_API_KEY"):
        print("❌ 请先配置 ANTHROPIC_API_KEY！")
        return

    # 创建测试数据
    test_dir = Path("./office_test")
    test_dir.mkdir(exist_ok=True)

    (test_dir / "meeting_notes.txt").write_text(
        "项目周会纪要\n日期: 2024年6月12日\n\n"
        "议程:\n1. 上周进度回顾 — 后端API完成80%，前端完成60%\n"
        "2. 本周计划 — 完成API开发，开始联调\n"
        "3. 风险讨论 — 第三方API文档不完善，可能需要额外2天\n\n"
        "决议:\n- 张三分派到第三方API对接，预计6月15日完成\n"
        "- 李四负责前端联调，6月16日启动\n"
        "- 下周三进行集成测试\n",
        encoding="utf-8"
    )

    (test_dir / "sales_data.csv").write_text(
        "月份,产品,销量,收入,区域\n"
        "1月,产品A,120,36000,华东\n"
        "1月,产品B,85,42500,华东\n"
        "2月,产品A,150,45000,华南\n"
        "2月,产品B,95,47500,华南\n"
        "3月,产品A,200,60000,华东\n"
        "3月,产品B,110,55000,华南\n",
        encoding="utf-8"
    )

    print(f"📝 测试文件已创建: {test_dir.absolute()}")
    print(f"   试试: summarize {test_dir}/meeting_notes.txt")
    print(f"   试试: table {test_dir}/sales_data.csv\n")

    agent = OfficeAgent()

    while True:
        try:
            user_input = input("👤 你: ").strip()
            if not user_input:
                continue
            if user_input.lower() in ("quit", "exit", "q"):
                break
            if user_input.lower() == "log":
                agent.show_log()
                continue

            # 命令解析
            parts = user_input.split(maxsplit=2)

            if parts[0] == "summarize" and len(parts) >= 2:
                print(agent.summarize_text(parts[1]))

            elif parts[0] == "schedule" and len(parts) >= 2:
                print(agent.extract_schedule(parts[1]))

            elif parts[0] == "table" and len(parts) >= 2:
                print(agent.analyze_table_file(parts[1]))

            elif parts[0] == "batch" and len(parts) >= 2:
                if parts[1] == "rename!" and len(parts) >= 5:
                    # batch rename! dir old new
                    args = parts[2].split()
                    print(agent.batch_process(args[0], "rename", pattern=args[1], replacement=args[2], dry_run=False))
                elif parts[1] == "rename" and len(parts) >= 4:
                    args = parts[2].split()
                    if len(args) >= 3:
                        print(agent.batch_process(args[0], "rename", pattern=args[1], replacement=args[2], dry_run=True))
                    else:
                        print("用法: batch rename <目录> <旧模式> <新模式>")
                elif parts[1] == "organize" and len(parts) >= 3:
                    print(agent.batch_process(parts[2], "organize"))
                else:
                    print("用法: batch  <参数>")

            elif parts[0] == "report" and len(parts) >= 2:
                print(agent.generate_report(parts[1]))

            else:
                print("未知命令，请输入有效命令。输入空行查看帮助。")

        except KeyboardInterrupt:
            break

    agent.show_log()
    print(f"\n👋 OfficeAgent 已退出。")


if __name__ == "__main__":
    main()

4. 调试与常见问题