第 8 章：综合项目实战

┌─────────────────────────────────────────────────────────┐
│                      智能办公助手                         │
├─────────────────────────────────────────────────────────┤
│  ┌──────────┐   ┌──────────┐   ┌──────────┐            │
│  │ 邮件处理 │   │ 报表生成 │   │ AI 问答  │            │
│  │  (IMAP)  │   │(pandas)  │   │  (LLM)   │            │
│  └────┬─────┘   └────┬─────┘   └────┬─────┘            │
│       │              │              │                   │
│       └──────────────┼──────────────┘                   │
│                      │                                   │
│              ┌───────┴───────┐                          │
│              │   任务调度器   │                          │
│              │ (APScheduler) │                          │
│              └───────┬───────┘                          │
│                      │                                   │
│              ┌───────┴───────┐                          │
│              │   日志/配置   │                          │
│              │(logging/env)  │                          │
│              └───────────────┘                          │
└─────────────────────────────────────────────────────────┘

smart-office-assistant/
├── pyproject.toml           # 项目依赖与元数据
├── .env                     # 环境变量（不提交到 Git）
├── .env.example             # 环境变量模板
├── README.md
├── src/
│   ├── __init__.py
│   ├── config.py            # 配置管理
│   ├── logger.py            # 日志配置
│   ├── scheduler.py         # 任务调度器
│   ├── email_handler.py     # 邮件处理模块
│   ├── report_generator.py  # 报表生成模块
│   ├── ai_assistant.py      # AI 问答模块
│   └── main.py              # 入口文件
├── tests/
│   ├── test_email.py
│   ├── test_report.py
│   └── conftest.py
├── reports/                 # 生成的报表
└── logs/                    # 日志文件

[project]
name = "smart-office-assistant"
version = "0.1.0"
description = "智能办公助手 - 邮件处理、报表生成、AI 问答"
requires-python = ">=3.10"
dependencies = [
    "openai >=1.0",
    "httpx >=0.27",
    "openpyxl >=3.1",
    "APScheduler >=3.10",
    "python-dotenv >=1.0",
    "pandas >=2.0",
]

[project.optional-dependencies]
dev = [
    "pytest >=8.0",
    "pytest-asyncio >=0.23",
    "ruff >=0.4",
    "mypy >=1.9",
]

[tool.ruff]
line-length = 100
target-version = "py310"

[tool.ruff.lint]
select = ["E", "F", "W", "I", "N", "UP"]

[tool.mypy]
python_version = "3.10"
warn_return_any = true
warn_unused_configs = true

# .env.example - 提交到 Git，作为模板
# 复制为 .env 后填入真实值

# LLM 配置
OPENAI_API_KEY=your-api-key
OPENAI_BASE_URL=https://api.openai.com/v1
LLM_MODEL=gpt-4o-mini

# 邮件配置
EMAIL_HOST=imap.example.com
EMAIL_USER=your-email@example.com
EMAIL_PASSWORD=your-app-password
EMAIL_FOLDER=INBOX

# 报表配置
REPORT_OUTPUT_DIR=./reports
DAILY_REPORT_TIME=09:00

# 日志配置
LOG_LEVEL=INFO
LOG_DIR=./logs

# src/config.py
"""配置管理模块"""

import os
from pathlib import Path
from dotenv import load_dotenv

# 加载 .env 文件
load_dotenv()

class Config:
    """应用配置类"""

    # LLM
    OPENAI_API_KEY: str = os.getenv("OPENAI_API_KEY", "")
    OPENAI_BASE_URL: str = os.getenv("OPENAI_BASE_URL", "https://api.openai.com/v1")
    LLM_MODEL: str = os.getenv("LLM_MODEL", "gpt-4o-mini")

    # 邮件
    EMAIL_HOST: str = os.getenv("EMAIL_HOST", "")
    EMAIL_USER: str = os.getenv("EMAIL_USER", "")
    EMAIL_PASSWORD: str = os.getenv("EMAIL_PASSWORD", "")
    EMAIL_FOLDER: str = os.getenv("EMAIL_FOLDER", "INBOX")

    # 报表
    REPORT_OUTPUT_DIR: Path = Path(os.getenv("REPORT_OUTPUT_DIR", "./reports"))
    DAILY_REPORT_TIME: str = os.getenv("DAILY_REPORT_TIME", "09:00")

    # 日志
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")
    LOG_DIR: Path = Path(os.getenv("LOG_DIR", "./logs"))

    @classmethod
    def validate(cls) -> list[str]:
        """验证必填配置，返回缺失项列表"""
        required = ["OPENAI_API_KEY", "EMAIL_HOST", "EMAIL_USER", "EMAIL_PASSWORD"]
        missing = [key for key in required if not getattr(cls, key)]
        return missing

    @classmethod
    def ensure_dirs(cls):
        """确保必要的目录存在"""
        cls.REPORT_OUTPUT_DIR.mkdir(parents=True, exist_ok=True)
        cls.LOG_DIR.mkdir(parents=True, exist_ok=True)

# 全局配置实例
config = Config()

# src/email_handler.py
"""邮件处理模块：收取、解析、分类"""

import imaplib
import email
from email.header import decode_header
from dataclasses import dataclass
from typing import Optional
import logging

from config import config

logger = logging.getLogger(__name__)

@dataclass
class EmailMessage:
    """邮件数据结构"""
    subject: str
    sender: str
    date: str
    body: str
    category: Optional[str] = None

class EmailHandler:
    """邮件处理器"""

    def __init__(self):
        self.host = config.EMAIL_HOST
        self.user = config.EMAIL_USER
        self.password = config.EMAIL_PASSWORD
        self.folder = config.EMAIL_FOLDER

    def connect(self) -> imaplib.IMAP4_SSL:
        """连接 IMAP 服务器"""
        mail = imaplib.IMAP4_SSL(self.host)
        mail.login(self.user, self.password)
        logger.info(f"已登录邮箱: {self.user}")
        return mail

    def fetch_unread(self, limit: int = 10) -> list[EmailMessage]:
        """获取未读邮件"""
        emails = []
        mail = self.connect()

        try:
            mail.select(self.folder)
            # 搜索未读邮件
            _, data = mail.search(None, "UNSEEN")
            msg_ids = data[0].split()

            for msg_id in msg_ids[-limit:]:  # 最近 N 封
                _, msg_data = mail.fetch(msg_id, "(RFC822)")
                raw_email = msg_data[0][1]
                msg = email.message_from_bytes(raw_email)

                email_msg = self._parse_message(msg)
                emails.append(email_msg)
                logger.info(f"获取邮件: {email_msg.subject}")

        finally:
            mail.logout()

        return emails

    def _parse_message(self, msg) -> EmailMessage:
        """解析原始邮件为结构化数据"""
        # 解析主题
        subject = self._decode_header(msg["Subject"])
        sender = self._decode_header(msg["From"])
        date = msg["Date"] or ""

        # 解析正文
        body = ""
        if msg.is_multipart():
            for part in msg.walk():
                content_type = part.get_content_type()
                if content_type == "text/plain":
                    body = part.get_payload(decode=True).decode("utf-8", errors="ignore")
                    break
        else:
            body = msg.get_payload(decode=True).decode("utf-8", errors="ignore")

        return EmailMessage(
            subject=subject,
            sender=sender,
            date=date,
            body=body[:1000]  # 限制长度
        )

    def _decode_header(self, value: Optional[str]) -> str:
        """解码邮件头中的编码文本"""
        if not value:
            return ""
        decoded = decode_header(value)
        result = ""
        for part, charset in decoded:
            if isinstance(part, bytes):
                result += part.decode(charset or "utf-8", errors="ignore")
            else:
                result += part
        return result

    def classify_with_llm(self, email_msg: EmailMessage) -> str:
        """使用 LLM 对邮件分类"""
        from openai import OpenAI

        client = OpenAI(
            api_key=config.OPENAI_API_KEY,
            base_url=config.OPENAI_BASE_URL
        )

        prompt = f"""请对以下邮件进行分类，只输出类别名称：
可选类别：工作/会议、技术问题、通知公告、垃圾邮件、其他

主题: {email_msg.subject}
发件人: {email_msg.sender}
正文: {email_msg.body[:500]}

类别:"""

        response = client.chat.completions.create(
            model=config.LLM_MODEL,
            messages=[{"role": "user", "content": prompt}],
            max_tokens=20,
            temperature=0.3
        )

        category = response.choices[0].message.content.strip()
        email_msg.category = category
        logger.info(f"邮件分类: {email_msg.subject} -> {category}")
        return category

# src/report_generator.py
"""报表生成模块：定时统计、生成 Excel"""

import random
from datetime import datetime, timedelta
from pathlib import Path
from openpyxl import Workbook
from openpyxl.styles import Font, PatternFill, Alignment, Border, Side
import logging

from config import config

logger = logging.getLogger(__name__)

class ReportGenerator:
    """日报生成器"""

    def __init__(self):
        self.output_dir = config.REPORT_OUTPUT_DIR

    def generate_daily_report(self, date: datetime | None = None) -> Path:
        """生成日报"""
        if date is None:
            date = datetime.now()

        # 模拟统计数据（实际项目中从数据库/API 获取）
        data = self._fetch_daily_data(date)

        # 创建 Excel
        wb = Workbook()
        ws = wb.active
        ws.title = "日报"

        # 标题样式
        title_font = Font(name="微软雅黑", size=16, bold=True, color="FFFFFF")
        title_fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
        header_font = Font(name="微软雅黑", size=11, bold=True, color="FFFFFF")
        header_fill = PatternFill(start_color="5B9BD5", end_color="5B9BD5", fill_type="solid")

        # 标题行
        ws.merge_cells("A1:D1")
        ws["A1"] = f"{date.strftime('%Y-%m-%d')} 工作日报"
        ws["A1"].font = title_font
        ws["A1"].fill = title_fill
        ws["A1"].alignment = Alignment(horizontal="center", vertical="center")
        ws.row_dimensions[1].height = 30

        # 表头
        headers = ["指标", "数值", "环比", "备注"]
        for col, header in enumerate(headers, 1):
            cell = ws.cell(row=3, column=col, value=header)
            cell.font = header_font
            cell.fill = header_fill
            cell.alignment = Alignment(horizontal="center")

        # 数据行
        for row_idx, row_data in enumerate(data, 4):
            for col_idx, value in enumerate(row_data, 1):
                cell = ws.cell(row=row_idx, column=col_idx, value=value)
                cell.alignment = Alignment(horizontal="center" if col_idx <= 3 else "left")

        # 调整列宽
        ws.column_dimensions["A"].width = 20
        ws.column_dimensions["B"].width = 15
        ws.column_dimensions["C"].width = 15
        ws.column_dimensions["D"].width = 30

        # 保存
        filename = f"daily_report_{date.strftime('%Y%m%d')}.xlsx"
        filepath = self.output_dir / filename
        wb.save(filepath)
        logger.info(f"日报已生成: {filepath}")
        return filepath

    def _fetch_daily_data(self, date: datetime) -> list[list]:
        """获取日报数据（模拟数据）"""
        return [
            ["处理邮件数", random.randint(20, 50), "+5%", "含 3 封紧急邮件"],
            ["完成任务数", random.randint(5, 12), "-2%", "包含报表生成任务"],
            ["代码提交数", random.randint(3, 15), "+10%", "修复 2 个 Bug"],
            ["会议时长(小时)", round(random.uniform(1, 4), 1), "持平", "技术评审会议"],
            ["AI 问答次数", random.randint(10, 30), "+20%", "主要用于代码审查"],
        ]

    def generate_weekly_summary(self) -> Path:
        """生成周报摘要"""
        today = datetime.now()
        wb = Workbook()
        ws = wb.active
        ws.title = "周报"

        # 标题
        ws.merge_cells("A1:E1")
        ws["A1"] = f"周报 {today.strftime('%Y-W%W')}"
        ws["A1"].font = Font(name="微软雅黑", size=16, bold=True, color="FFFFFF")
        ws["A1"].fill = PatternFill(start_color="366092", end_color="366092", fill_type="solid")
        ws["A1"].alignment = Alignment(horizontal="center")
        ws.row_dimensions[1].height = 30

        # 表头
        headers = ["日期", "处理邮件数", "完成任务数", "代码提交数", "总体评价"]
        for col, header in enumerate(headers, 1):
            cell = ws.cell(row=3, column=col, value=header)
            cell.font = Font(bold=True, color="FFFFFF")
            cell.fill = PatternFill(start_color="5B9BD5", end_color="5B9BD5", fill_type="solid")
            cell.alignment = Alignment(horizontal="center")

        # 生成 7 天数据
        total_tasks = 0
        total_commits = 0
        for day_offset in range(7):
            date = today - timedelta(days=6 - day_offset)
            tasks = random.randint(5, 12)
            commits = random.randint(3, 15)
            total_tasks += tasks
            total_commits += commits
            ws.cell(row=4 + day_offset, column=1, value=date.strftime("%m/%d (%a)"))
            ws.cell(row=4 + day_offset, column=2, value=random.randint(20, 50))
            ws.cell(row=4 + day_offset, column=3, value=tasks)
            ws.cell(row=4 + day_offset, column=4, value=commits)
            ws.cell(row=4 + day_offset, column=5, value="优秀" if tasks > 7 else "正常")

        # 汇总行
        summary_row = 11
        ws.cell(row=summary_row, column=1, value="本周合计").font = Font(bold=True)
        ws.cell(row=summary_row, column=3, value=total_tasks)
        ws.cell(row=summary_row, column=4, value=total_commits)

        # 列宽
        for col in ["A", "B", "C", "D", "E"]:
            ws.column_dimensions[col].width = 18

        filename = f"weekly_summary_{today.strftime('%Y%m%d')}.xlsx"
        filepath = self.output_dir / filename
        wb.save(filepath)
        logger.info(f"周报已生成: {filepath}")
        return filepath

# src/ai_assistant.py
"""AI 问答模块：接入 LLM、维护上下文"""

import json
from dataclasses import dataclass, field
from typing import Optional
from openai import OpenAI
import logging

from config import config

logger = logging.getLogger(__name__)

@dataclass
class ChatSession:
    """对话会话"""
    session_id: str
    messages: list[dict] = field(default_factory=list)
    max_history: int = 10

    def __post_init__(self):
        if not self.messages:
            self.messages = [{
                "role": "system",
                "content": "你是一个智能办公助手，帮助用户处理工作问题。回答简洁专业。"
            }]

    def add_user(self, content: str):
        """添加用户消息"""
        self.messages.append({"role": "user", "content": content})
        self._trim_history()

    def add_assistant(self, content: str):
        """添加助手消息"""
        self.messages.append({"role": "assistant", "content": content})

    def _trim_history(self):
        """截断历史，保留最近 N 轮"""
        if len(self.messages) > self.max_history * 2 + 1:
            # 保留 system + 最近对话
            self.messages = [self.messages[0]] + self.messages[-(self.max_history * 2):]

class AIAssistant:
    """AI 助手"""

    def __init__(self):
        self.client = OpenAI(
            api_key=config.OPENAI_API_KEY,
            base_url=config.OPENAI_BASE_URL
        )
        self.model = config.LLM_MODEL
        self.sessions: dict[str, ChatSession] = {}

    def get_or_create_session(self, session_id: str) -> ChatSession:
        """获取或创建会话"""
        if session_id not in self.sessions:
            self.sessions[session_id] = ChatSession(session_id=session_id)
        return self.sessions[session_id]

    def chat(self, message: str, session_id: str = "default") -> str:
        """发送消息并获取回复"""
        session = self.get_or_create_session(session_id)
        session.add_user(message)

        try:
            response = self.client.chat.completions.create(
                model=self.model,
                messages=session.messages,
                temperature=0.7,
                max_tokens=1500
            )

            reply = response.choices[0].message.content
            session.add_assistant(reply)

            # 记录 Token 消耗
            usage = response.usage
            logger.info(
                f"会话 {session_id}: "
                f"prompt={usage.prompt_tokens}, "
                f"completion={usage.completion_tokens}"
            )

            return reply

        except Exception as e:
            logger.error(f"LLM 调用失败: {e}")
            return f"抱歉，服务暂时不可用: {str(e)}"

    def clear_session(self, session_id: str):
        """清空会话"""
        if session_id in self.sessions:
            del self.sessions[session_id]
            logger.info(f"会话已清空: {session_id}")

    def summarize_text(self, text: str, max_length: int = 200) -> str:
        """文本摘要"""
        prompt = f"请用 {max_length} 字以内总结以下内容：\n\n{text[:3000]}"
        return self.chat(prompt, session_id="summarize")

# src/scheduler.py
"""任务调度模块：APScheduler 定时触发"""

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.cron import CronTrigger
from datetime import datetime
import logging

from config import config
from email_handler import EmailHandler
from report_generator import ReportGenerator

logger = logging.getLogger(__name__)

class TaskScheduler:
    """任务调度器"""

    def __init__(self):
        self.scheduler = BackgroundScheduler()
        self.email_handler = EmailHandler()
        self.report_generator = ReportGenerator()

    def start(self):
        """启动调度器"""
        # 每小时检查一次邮件
        self.scheduler.add_job(
            self._check_emails,
            trigger="interval",
            minutes=60,
            id="check_emails",
            replace_existing=True
        )

        # 每天生成日报
        hour, minute = config.DAILY_REPORT_TIME.split(":")
        self.scheduler.add_job(
            self._generate_daily_report,
            trigger=CronTrigger(hour=hour, minute=minute),
            id="daily_report",
            replace_existing=True
        )

        self.scheduler.start()
        logger.info("调度器已启动")
        logger.info(f"日报时间: {config.DAILY_REPORT_TIME}")

    def _check_emails(self):
        """检查邮件任务"""
        logger.info("开始检查邮件...")
        try:
            emails = self.email_handler.fetch_unread(limit=5)
            for email in emails:
                # 分类邮件
                self.email_handler.classify_with_llm(email)
                logger.info(f"新邮件: [{email.category}] {email.subject}")
        except Exception as e:
            logger.error(f"邮件检查失败: {e}")

    def _generate_daily_report(self):
        """生成日报任务"""
        logger.info("开始生成日报...")
        try:
            filepath = self.report_generator.generate_daily_report()
            logger.info(f"日报已生成: {filepath}")
        except Exception as e:
            logger.error(f"日报生成失败: {e}")

    def shutdown(self):
        """关闭调度器"""
        self.scheduler.shutdown()
        logger.info("调度器已关闭")

# src/logger.py
"""日志配置模块"""

import logging
import logging.handlers
from pathlib import Path
from config import config

def setup_logging():
    """配置应用日志"""

    # 确保日志目录存在
    config.LOG_DIR.mkdir(parents=True, exist_ok=True)

    # 日志格式
    formatter = logging.Formatter(
        "%(asctime)s | %(levelname)-8s | %(name)s | %(message)s",
        datefmt="%Y-%m-%d %H:%M:%S"
    )

    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    console_handler.setLevel(logging.INFO)

    # 文件处理器（按天轮转）
    file_handler = logging.handlers.TimedRotatingFileHandler(
        filename=config.LOG_DIR / "app.log",
        when="midnight",      # 每天午夜轮转
        interval=1,
        backupCount=30,       # 保留 30 天
        encoding="utf-8"
    )
    file_handler.setFormatter(formatter)
    file_handler.setLevel(logging.DEBUG)

    # 错误日志单独文件
    error_handler = logging.FileHandler(
        config.LOG_DIR / "error.log",
        encoding="utf-8"
    )
    error_handler.setFormatter(formatter)
    error_handler.setLevel(logging.ERROR)

    # 根日志器配置
    root_logger = logging.getLogger()
    root_logger.setLevel(getattr(logging, config.LOG_LEVEL))
    root_logger.addHandler(console_handler)
    root_logger.addHandler(file_handler)
    root_logger.addHandler(error_handler)

    # 降低第三方库日志级别
    logging.getLogger("apscheduler").setLevel(logging.WARNING)
    logging.getLogger("urllib3").setLevel(logging.WARNING)

import functools
import logging
import time

logger = logging.getLogger(__name__)

def retry_on_error(max_retries=3, delay=1, exceptions=(Exception,)):
    """重试装饰器"""
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except exceptions as e:
                    logger.warning(
                        f"{func.__name__} 第 {attempt + 1} 次失败: {e}"
                    )
                    if attempt < max_retries - 1:
                        time.sleep(delay * (attempt + 1))  # 指数退避
                    else:
                        logger.error(f"{func.__name__} 最终失败")
                        raise
        return wrapper
    return decorator

# 使用示例
@retry_on_error(max_retries=3, exceptions=(ConnectionError, TimeoutError))
def fetch_api_data():
    """可能失败的 API 调用"""
    pass

# 全局异常处理
def handle_exceptions(func):
    """捕获并记录所有异常"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        try:
            return func(*args, **kwargs)
        except Exception as e:
            logger.exception(f"未捕获的异常 in {func.__name__}: {e}")
            # 可以在这里发送告警通知
            raise
    return wrapper

# tests/test_email.py
"""邮件模块测试"""

import pytest
from unittest.mock import Mock, patch, MagicMock
from email_handler import EmailHandler, EmailMessage

class TestEmailHandler:
    """EmailHandler 测试类"""

    @pytest.fixture
    def handler(self):
        return EmailHandler()

    def test_decode_header_plain(self, handler):
        """测试普通文本解码"""
        result = handler._decode_header("Hello World")
        assert result == "Hello World"

    def test_decode_header_encoded(self, handler):
        """测试编码文本解码"""
        # =?UTF-8?B?5L2g5aW9?= 是 "你好" 的 Base64 编码
        result = handler._decode_header("=?UTF-8?B?5L2g5aW9?=")
        assert result == "你好"

    @patch("imaplib.IMAP4_SSL")
    def test_fetch_unread(self, mock_imap, handler):
        """测试获取未读邮件"""
        # 模拟 IMAP 连接
        mock_mail = MagicMock()
        mock_imap.return_value = mock_mail
        mock_mail.search.return_value = ("OK", [b"1 2 3"])

        # 模拟邮件内容
        raw_email = b"""Subject: Test\r\nFrom: test@example.com\r\n\r\nHello"""
        mock_mail.fetch.return_value = ("OK", [(b"1", raw_email)])

        emails = handler.fetch_unread()

        assert len(emails) == 3
        assert emails[0].subject == "Test"
        mock_mail.login.assert_called_once()

    def test_email_message_dataclass(self):
        """测试邮件数据类"""
        email = EmailMessage(
            subject="测试",
            sender="a@b.com",
            date="2024-01-01",
            body="内容"
        )
        assert email.category is None
        email.category = "工作"
        assert email.category == "工作"

# tests/conftest.py
"""pytest 共享配置"""

import pytest

@pytest.fixture(scope="session")
def mock_config():
    """提供测试配置"""
    from config import Config
    Config.OPENAI_API_KEY = "test-key"
    Config.EMAIL_HOST = "test.example.com"
    return Config

# 快速调试技巧
from pprint import pprint

# 美化打印复杂数据结构
pprint(some_dict)

# 查看对象所有属性
print(dir(some_object))

# 查看对象类型和内存地址
print(type(obj), id(obj))

# 使用 rich 美化输出（pip install rich）
from rich import print as rprint
from rich.pretty import Pretty
rprint(Pretty(large_data_structure))

# Dockerfile
FROM python:3.11-slim

# 设置工作目录
WORKDIR /app

# 安装系统依赖
RUN apt-get update && apt-get install -y \
    gcc \
    && rm -rf /var/lib/apt/lists/*

# 复制依赖文件并安装
COPY pyproject.toml .
RUN pip install --no-cache-dir -e "."

# 复制应用代码
COPY src/ ./src/

# 创建必要的目录
RUN mkdir -p /app/reports /app/logs

# 设置环境变量
ENV PYTHONPATH=/app/src
ENV LOG_LEVEL=INFO

# 运行应用
CMD ["python", "-m", "src.main"]

# docker-compose.yml
version: "3.8"

services:
  app:
    build: .
    container_name: smart-office-assistant
    env_file:
      - .env
    volumes:
      - ./reports:/app/reports
      - ./logs:/app/logs
    restart: unless-stopped
    # 如果需要后台持续运行
    stdin_open: true
    tty: true

# src/health_check.py
"""健康检查模块"""

import time
import psutil
import logging

logger = logging.getLogger(__name__)

def check_system_health():
    """检查系统健康状态"""
    health = {
        "timestamp": time.time(),
        "cpu_percent": psutil.cpu_percent(interval=1),
        "memory_percent": psutil.virtual_memory().percent,
        "disk_percent": psutil.disk_usage("/").percent,
        "status": "healthy"
    }

    # 告警阈值
    if health["memory_percent"] > 90:
        health["status"] = "warning"
        logger.warning(f"内存使用率过高: {health['memory_percent']}%")

    if health["disk_percent"] > 85:
        health["status"] = "warning"
        logger.warning(f"磁盘使用率过高: {health['disk_percent']}%")

    return health

# 可以配合 APScheduler 定期执行
# scheduler.add_job(check_system_health, "interval", minutes=5)

# 高效的 Python 开发 Prompt 模板

## 1. 生成模块代码
"""
请为 Python 项目生成 [功能描述] 模块。

项目背景：
- 使用 Python 3.11+
- 依赖：openai, httpx, pydantic
- 代码风格：使用类型注解、dataclass、异常处理

要求：
1. 包含完整的类型注解
2. 使用 logging 而非 print
3. 关键函数添加 docstring
4. 包含错误处理和重试逻辑
5. 提供使用示例
"""

## 2. 代码审查
"""
请审查以下 Python 代码，关注：
1. 是否有潜在的异常未处理？
2. 类型注解是否完整？
3. 是否有性能优化空间？
4. 是否符合 PEP8 规范？
5. 日志记录是否充分？

代码：
```python
[paste code]
```
"""

## 3. 调试辅助
"""
我的 Python 程序出现了以下错误：
[粘贴错误堆栈]

相关代码：
```python
[paste code]
```

请分析可能的原因，并提供修复方案。
"""

# src/main.py
"""智能办公助手 —— 应用入口"""

import sys
import time
from pathlib import Path
import logging

# 将 src 目录加入 Python 路径
sys.path.insert(0, str(Path(__file__).parent))

from config import config
from logger import setup_logging
from scheduler import TaskScheduler
from ai_assistant import AIAssistant
from email_handler import EmailHandler
from report_generator import ReportGenerator

logger = logging.getLogger(__name__)

def start_scheduler_mode():
    """调度模式：后台自动运行定时任务"""
    setup_logging()
    config.ensure_dirs()

    scheduler = TaskScheduler()
    scheduler.start()

    logger.info("智能办公助手已启动（调度模式），按 Ctrl+C 退出")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        logger.info("正在关闭...")
        scheduler.shutdown()

def start_interactive_mode():
    """交互模式：命令行 AI 对话"""
    setup_logging()
    config.ensure_dirs()

    assistant = AIAssistant()
    print("\n智能办公助手 - AI 问答模式")
    print("输入 /quit 退出, /report 生成日报, /weekly 生成周报\n")

    report_gen = ReportGenerator()

    while True:
        try:
            user_input = input("你: ").strip()
        except (KeyboardInterrupt, EOFError):
            print("\n再见！")
            break

        if not user_input:
            continue
        if user_input == "/quit":
            break
        elif user_input == "/report":
            filepath = report_gen.generate_daily_report()
            print(f"日报已生成: {filepath}\n")
            continue
        elif user_input == "/weekly":
            filepath = report_gen.generate_weekly_summary()
            print(f"周报已生成: {filepath}\n")
            continue

        reply = assistant.chat(user_input)
        print(f"助手: {reply}\n")

def main():
    """主入口"""
    missing = config.validate()
    if missing:
        print(f"错误：缺少以下配置项: {', '.join(missing)}")
        print("请检查 .env 文件")
        sys.exit(1)

    if len(sys.argv) > 1 and sys.argv[1] == "--schedule":
        start_scheduler_mode()
    else:
        start_interactive_mode()

if __name__ == "__main__":
    main()

# .github/workflows/ci.yml
name: Smart Office CI

on:
  push:
    branches: [main]
  pull_request:
    branches: [main]

jobs:
  test:
    runs-on: ubuntu-latest

    steps:
      - uses: actions/checkout@v4

      - name: 设置 Python
        uses: actions/setup-python@v5
        with:
          python-version: "3.11"

      - name: 安装依赖
        run: |
          pip install --upgrade pip
          pip install -e ".[dev]"

      - name: Ruff 检查
        run: ruff check src/

      - name: pytest 测试
        run: pytest -v --cov=src --cov-report=term

      - name: 类型检查
        run: mypy src/ --ignore-missing-imports

# tests/test_ai_assistant.py
"""AI 模块测试 —— 使用 Mock 避免真实 API 调用"""
import pytest
from unittest.mock import patch, MagicMock
from ai_assistant import AIAssistant, ChatSession

class TestAIAssistant:
    @pytest.fixture
    def assistant(self):
        return AIAssistant()

    def test_chat_session(self):
        """测试会话管理"""
        session = ChatSession(session_id="test")
        assert len(session.messages) == 1  # system message
        session.add_user("Hello")
        assert len(session.messages) == 2
        session.add_assistant("Hi!")
        assert len(session.messages) == 3

    def test_session_trim_history(self):
        """测试对话历史截断"""
        session = ChatSession(session_id="test", max_history=2)
        for i in range(10):
            session.add_user(f"msg {i}")
            session.add_assistant(f"reply {i}")
        # 应该只保留 system + 最近 2*2 = 5 条消息
        assert len(session.messages) == 5

    @patch("ai_assistant.OpenAI")
    def test_chat_with_mock(self, mock_openai, assistant):
        """测试聊天（Mock OpenAI API）"""
        # 模拟 API 响应
        mock_response = MagicMock()
        mock_response.choices = [
            MagicMock(message=MagicMock(content="你好！有什么可以帮你的？"))
        ]
        mock_response.usage = MagicMock(
            prompt_tokens=50,
            completion_tokens=10
        )
        mock_openai.return_value.chat.completions.create.return_value = mock_response

        reply = assistant.chat("你好")
        assert "你好" in reply

    def test_clear_session(self, assistant):
        """测试清除会话"""
        assistant.chat("hello")
        assistant.chat("world")
        assert "default" in assistant.sessions
        assistant.clear_session("default")
        assert "default" not in assistant.sessions