第 9 课：进阶扩展

第 1-2 课：了解全貌 + 环境搭建
第 3 课：query() 一键自动化
第 4 课：MCP 自定义工具
第 5 课：ClaudeSDKClient 多轮对话
第 6 课：Hook 安全管控
第 7 课：多 Agent 协作
第 8 课：MiniClaw 完整实战

"""日程管理工具"""

import json
import datetime
from claude_agent_sdk import tool
from storage.db import get_conn


def init_schedule_table():
    conn = get_conn()
    conn.execute("""
        CREATE TABLE IF NOT EXISTS schedule (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            title TEXT NOT NULL,
            start_time TEXT NOT NULL,
            end_time TEXT,
            description TEXT,
            created_at TEXT DEFAULT CURRENT_TIMESTAMP
        )
    """)
    conn.commit()
    conn.close()


@tool(
    "add_event",
    "添加日程事件",
    {
        "title": {"type": "string", "description": "事件标题"},
        "start_time": {"type": "string", "description": "开始时间，如 2025-03-20 14:00"},
        "end_time": {"type": "string", "description": "结束时间（可选）"},
        "description": {"type": "string", "description": "详细描述（可选）"},
    }
)
async def add_event(title: str, start_time: str,
                     end_time: str = "", description: str = "") -> str:
    conn = get_conn()
    cursor = conn.execute(
        "INSERT INTO schedule (title, start_time, end_time, description) VALUES (?,?,?,?)",
        (title, start_time, end_time or None, description or None),
    )
    conn.commit()
    event_id = cursor.lastrowid
    conn.close()

    msg = f"📅 已添加日程 #{event_id}: {title} @ {start_time}"
    if end_time:
        msg += f" ~ {end_time}"
    return msg


@tool(
    "list_schedule",
    "查看日程安排。可以查今天、明天、或指定日期",
    {"date": {"type": "string", "description": "日期，如 2025-03-20 或 today/tomorrow"}}
)
async def list_schedule(date: str = "today") -> str:
    today = datetime.date.today()
    if date == "today":
        target = today.isoformat()
    elif date == "tomorrow":
        target = (today + datetime.timedelta(days=1)).isoformat()
    else:
        target = date

    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM schedule WHERE start_time LIKE ? ORDER BY start_time",
        (f"{target}%",),
    ).fetchall()
    conn.close()

    if not rows:
        return f"📅 {target} 没有日程安排"

    lines = [f"📅 {target} 的日程（{len(rows)} 项）:"]
    for r in rows:
        time_part = r["start_time"].split(" ")[1] if " " in r["start_time"] else ""
        line = f"  {time_part} {r['title']}"
        if r.get("end_time"):
            end_time = r["end_time"].split(" ")[1] if " " in r["end_time"] else r["end_time"]
            line += f" ~ {end_time}"
        lines.append(line)
    return "\n".join(lines)


@tool(
    "upcoming_events",
    "查看未来 N 天的日程",
    {"days": {"type": "number", "description": "未来几天，默认 7"}}
)
async def upcoming_events(days: int = 7) -> str:
    today = datetime.date.today()
    end = today + datetime.timedelta(days=days)

    conn = get_conn()
    rows = conn.execute(
        "SELECT * FROM schedule WHERE start_time >= ? AND start_time < ? ORDER BY start_time",
        (today.isoformat(), end.isoformat()),
    ).fetchall()
    conn.close()

    if not rows:
        return f"📅 未来 {days} 天没有日程安排"

    lines = [f"📅 未来 {days} 天的日程（{len(rows)} 项）:"]
    current_date = ""
    for r in rows:
        date_part = r["start_time"][:10]
        if date_part != current_date:
            current_date = date_part
            lines.append(f"\n  📆 {current_date}:")
        time_part = r["start_time"][11:16] if len(r["start_time"]) > 10 else ""
        lines.append(f"    {time_part} {r['title']}")
    return "\n".join(lines)

@tool(
    "generate_weekly_report",
    "根据本周的任务和笔记自动生成周报",
    {}
)
async def generate_weekly_report() -> str:
    """收集数据供 AI 生成周报"""
    today = datetime.date.today()
    week_start = today - datetime.timedelta(days=today.weekday())

    conn = get_conn()

    # 本周完成的任务
    done_tasks = conn.execute(
        "SELECT * FROM todos WHERE done = 1 AND created_at >= ?",
        (week_start.isoformat(),)
    ).fetchall()

    # 本周待办
    pending_tasks = conn.execute(
        "SELECT * FROM todos WHERE done = 0"
    ).fetchall()

    # 本周笔记
    notes = conn.execute(
        "SELECT * FROM notes WHERE created_at >= ? ORDER BY created_at",
        (week_start.isoformat(),)
    ).fetchall()

    conn.close()

    report = f"📊 周报数据（{week_start} ~ {today}）\n\n"

    report += f"✅ 本周完成（{len(done_tasks)} 项）:\n"
    for t in done_tasks:
        report += f"  - {t['content']}\n"

    report += f"\n☐ 进行中（{len(pending_tasks)} 项）:\n"
    for t in pending_tasks:
        report += f"  - {t['content']} [{t['priority']}]\n"

    report += f"\n📝 本周笔记（{len(notes)} 条）:\n"
    for n in notes:
        tags = json.loads(n["tags"]) if n["tags"] else []
        tags_str = " ".join(f"#{t}" for t in tags)
        report += f"  - {n['content'][:60]} {tags_str}\n"

    report += "\n请根据以上数据生成一份简洁的周报。"
    return report

@tool(
    "track_habit",
    "记录今天的习惯完成情况",
    {
        "habit": {"type": "string", "description": "习惯名称（如: 运动、读书、冥想）"},
        "done": {"type": "boolean", "description": "是否完成"},
    }
)
async def track_habit(habit: str, done: bool = True) -> str:
    conn = get_conn()
    today = datetime.date.today().isoformat()

    conn.execute("""
        CREATE TABLE IF NOT EXISTS habits (
            id INTEGER PRIMARY KEY AUTOINCREMENT,
            habit TEXT NOT NULL,
            date TEXT NOT NULL,
            done INTEGER DEFAULT 1,
            UNIQUE(habit, date)
        )
    """)

    try:
        conn.execute(
            "INSERT OR REPLACE INTO habits (habit, date, done) VALUES (?, ?, ?)",
            (habit, today, 1 if done else 0),
        )
        conn.commit()
    except Exception as e:
        conn.close()
        return f"记录失败: {e}"
    conn.close()

    status = "✅" if done else "❌"
    return f"{status} 已记录: {habit} ({today})"


@tool(
    "habit_streak",
    "查看某个习惯的连续完成天数",
    {"habit": {"type": "string", "description": "习惯名称"}}
)
async def habit_streak(habit: str) -> str:
    conn = get_conn()
    rows = conn.execute(
        "SELECT date FROM habits WHERE habit = ? AND done = 1 ORDER BY date DESC",
        (habit,)
    ).fetchall()
    conn.close()

    if not rows:
        return f"'{habit}' 还没有记录"

    # 计算连续天数
    streak = 0
    expected = datetime.date.today()
    for row in rows:
        row_date = datetime.date.fromisoformat(row["date"])
        if row_date == expected:
            streak += 1
            expected -= datetime.timedelta(days=1)
        elif row_date < expected:
            break

    total = len(rows)
    return f"🔥 {habit}: 连续 {streak} 天 | 总计 {total} 天"

options = ClaudeAgentOptions(
    system_prompt="你是 MiniClaw...",
    thinking={
        "type": "enabled",
        "budget_tokens": 10000,  # 给 AI 1万 token 的"思考空间"
    },
    effort="high",  # 高强度思考
)

from claude_agent_sdk import ThinkingBlock

async for msg in client.receive_response():
    if isinstance(msg, AssistantMessage):
        for block in msg.content:
            if isinstance(block, ThinkingBlock):
                # AI 的思考过程（可选择显示）
                print(f"  💭 [思考中...{len(block.thinking)}字]")
            elif isinstance(block, TextBlock):
                print(f"🐾: {block.text}")

options = ClaudeAgentOptions(
    max_budget_usd=0.50,  # 单次对话最多花 5 毛钱
    max_turns=15,          # 最多 15 轮
    model="claude-haiku-4-5",  # 默认用便宜模型
)

1. 数据安全
   ├── SQLite 文件定期备份
   ├── 敏感笔记加密存储
   └── API Key 用环境变量

2. 性能优化
   ├── 简单操作（加任务、查笔记）用 Haiku
   ├── 规划和总结用 Sonnet
   └── 数据库加索引

3. 用户体验
   ├── 快捷命令 /todo /notes /today
   ├── 操作反馈清晰（emoji + 状态）
   └── 错误信息友好

4. 扩展方向
   ├── Web 界面（FastAPI + React）
   ├── 桌面通知（系统级提醒）
   ├── 日历集成（Google Calendar API）
   └── 邮件集成（参考 email-agent demo）