# Daily Todo Implementation Plan > **For agentic workers:** REQUIRED SUB-SKILL: Use superpowers:subagent-driven-development (recommended) or superpowers:executing-plans to implement this plan task-by-task. Steps use checkbox (`- [ ]`) syntax for tracking. **Goal:** 实现每日待办功能——AI 每天早上自动预生成今日待办(看板未完成任务+前一天对话分析),用户可手动增删改。 **Architecture:** - 后端:FastAPI Router + SQLAlchemy Model + LLM Service 分析对话 - 前端:Vue 3 Composition API,TodoView 独立页面 - 数据库:新表 `daily_todos`,按日期过滤查询 **Tech Stack:** FastAPI + SQLAlchemy + APScheduler + LLM Service / Vue 3 + Motion + axios --- ## 文件总览 ``` backend/ app/ models/ __init__.py # 修改:导出 DailyTodo todo.py # 新建:DailyTodo 模型 schemas/ todo.py # 新建:Pydantic Schema routers/ todo.py # 新建:API Router services/ todo_service.py # 新建:AI 分析逻辑 main.py # 修改:注册 todo router frontend/ src/ api/ todo.ts # 新建:API 客户端 views/ TodoView.vue # 新建:待办页面 router/ index.ts # 修改:添加 /todo 路由 components/ SidebarNav.vue # 修改:添加待办菜单 ``` --- ## Task 1: 后端 - 数据模型 **Files:** - Create: `backend/app/models/todo.py` - Modify: `backend/app/models/__init__.py` - [ ] **Step 1: 创建 `app/models/todo.py`** 参考现有 `app/models/task.py` 的写法: ```python from sqlalchemy import Column, String, Boolean, DateTime, Enum as SQLEnum from datetime import datetime from enum import Enum as PyEnum from app.models.base import BaseModel class TodoSource(str, PyEnum): AI_KANBAN = "ai_kanban" AI_CHAT = "ai_chat" MANUAL = "manual" class DailyTodo(BaseModel): __tablename__ = "daily_todos" user_id = Column(String(36), nullable=False, index=True) title = Column(String(500), nullable=False) is_completed = Column(Boolean, default=False, nullable=False) source = Column(SQLEnum(TodoSource), default=TodoSource.MANUAL, nullable=False) source_detail = Column(String(500), nullable=True) source_ref_id = Column(String(36), nullable=True) todo_date = Column(String(10), nullable=False) # YYYY-MM-DD completed_at = Column(DateTime, nullable=True) ``` - [ ] **Step 2: 修改 `app/models/__init__.py`,添加导出** 在文件末尾添加: ```python from app.models.todo import DailyTodo __all__ = [..., "DailyTodo"] ``` - [ ] **Step 3: 提交** ```bash git add backend/app/models/todo.py backend/app/models/__init__.py git commit -m "feat(todo): add DailyTodo model" ``` --- ## Task 2: 后端 - Schema 定义 **Files:** - Create: `backend/app/schemas/todo.py` - [ ] **Step 1: 创建 `app/schemas/todo.py`** ```python from pydantic import BaseModel from datetime import datetime from app.models.todo import TodoSource class TodoCreate(BaseModel): title: str class TodoUpdate(BaseModel): title: str | None = None is_completed: bool | None = None class TodoOut(BaseModel): id: str title: str is_completed: bool source: TodoSource source_detail: str | None todo_date: str completed_at: datetime | None created_at: datetime updated_at: datetime model_config = {"from_attributes": True} class TodoListOut(BaseModel): items: list[TodoOut] total: int page: int page_size: int class TodoSummaryOut(BaseModel): date: str total: int completed: int pending: int ``` - [ ] **Step 2: 提交** ```bash git add backend/app/schemas/todo.py git commit -m "feat(todo): add Pydantic schemas" ``` --- ## Task 3: 后端 - API Router **Files:** - Create: `backend/app/routers/todo.py` - [ ] **Step 1: 创建 `app/routers/todo.py`** 完整实现以下端点: ```python from fastapi import APIRouter, Depends, HTTPException, Query from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, func from datetime import date from app.database import get_db from app.models.todo import DailyTodo, TodoSource from app.models.user import User from app.routers.auth import get_current_user from app.schemas.todo import ( TodoCreate, TodoUpdate, TodoOut, TodoListOut, TodoSummaryOut ) from app.services.todo_service import generate_daily_todos router = APIRouter(prefix="/api/todos", tags=["待办"]) @router.get("", response_model=TodoListOut) async def list_todos( date_str: str = Query(default=None), # YYYY-MM-DD,默认当天 page: int = Query(default=1, ge=1), page_size: int = Query(default=50, ge=1, le=100), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): target_date = date_str or date.today().isoformat() offset = (page - 1) * page_size # 查询总数 count_q = select(func.count()).select_from(DailyTodo).where( DailyTodo.user_id == current_user.id, DailyTodo.todo_date == target_date, ) total = (await db.execute(count_q)).scalar() # 查询列表 q = select(DailyTodo).where( DailyTodo.user_id == current_user.id, DailyTodo.todo_date == target_date, ).order_by(DailyTodo.created_at.desc()).offset(offset).limit(page_size) items = (await db.execute(q)).scalars().all() return TodoListOut(items=items, total=total, page=page, page_size=page_size) @router.post("", response_model=TodoOut, status_code=201) async def create_todo( data: TodoCreate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): todo = DailyTodo( user_id=current_user.id, title=data.title, source=TodoSource.MANUAL, todo_date=date.today().isoformat(), ) db.add(todo) await db.commit() await db.refresh(todo) return todo @router.patch("/{todo_id}", response_model=TodoOut) async def update_todo( todo_id: str, data: TodoUpdate, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute( select(DailyTodo).where(DailyTodo.id == todo_id, DailyTodo.user_id == current_user.id) ) todo = result.scalar_one_or_none() if not todo: raise HTTPException(status_code=404, detail="待办不存在") # 历史日期不允许修改 if todo.todo_date != date.today().isoformat(): raise HTTPException(status_code=403, detail="历史待办不可修改") if data.title is not None: todo.title = data.title if data.is_completed is not None: from datetime import datetime todo.is_completed = data.is_completed todo.completed_at = datetime.utcnow() if data.is_completed else None await db.commit() await db.refresh(todo) return todo @router.delete("/{todo_id}", status_code=204) async def delete_todo( todo_id: str, current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): result = await db.execute( select(DailyTodo).where(DailyTodo.id == todo_id, DailyTodo.user_id == current_user.id) ) todo = result.scalar_one_or_none() if not todo: raise HTTPException(status_code=404, detail="待办不存在") if todo.todo_date != date.today().isoformat(): raise HTTPException(status_code=403, detail="历史待办不可删除") await db.delete(todo) await db.commit() @router.post("/ai-generate", response_model=TodoListOut) async def ai_generate_todos( current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): target_date = date.today().isoformat() # 幂等检查:是否已有AI生成记录 check_q = select(func.count()).select_from(DailyTodo).where( DailyTodo.user_id == current_user.id, DailyTodo.todo_date == target_date, DailyTodo.source.in_([TodoSource.AI_KANBAN, TodoSource.AI_CHAT]), ) count = (await db.execute(check_q)).scalar() if count > 0: # 已生成,返回现有记录 q = select(DailyTodo).where( DailyTodo.user_id == current_user.id, DailyTodo.todo_date == target_date, ).order_by(DailyTodo.created_at.desc()) items = (await db.execute(q)).scalars().all() return TodoListOut(items=items, total=len(items), page=1, page_size=50) # 执行AI生成 todos = await generate_daily_todos(current_user.id, db) return TodoListOut(items=todos, total=len(todos), page=1, page_size=50) @router.get("/summary", response_model=TodoSummaryOut) async def get_summary( date_str: str = Query(default=None), current_user: User = Depends(get_current_user), db: AsyncSession = Depends(get_db), ): target_date = date_str or date.today().isoformat() q = select(DailyTodo).where( DailyTodo.user_id == current_user.id, DailyTodo.todo_date == target_date, ) todos = (await db.execute(q)).scalars().all() completed = sum(1 for t in todos if t.is_completed) return TodoSummaryOut(date=target_date, total=len(todos), completed=completed, pending=len(todos) - completed) ``` - [ ] **Step 2: 提交** ```bash git add backend/app/routers/todo.py git commit -m "feat(todo): add API router with CRUD + ai-generate" ``` --- ## Task 4: 后端 - AI 生成逻辑 Service **Files:** - Create: `backend/app/services/todo_service.py` - [ ] **Step 1: 创建 `app/services/todo_service.py`** ```python import json import logging from datetime import date, datetime, timedelta from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select from app.models.todo import DailyTodo, TodoSource from app.models.task import Task, TaskStatus from app.models.conversation import Conversation, Message from app.services.llm_service import llm_service logger = logging.getLogger(__name__) async def generate_daily_todos(user_id: str, db: AsyncSession) -> list[DailyTodo]: """ 为用户生成今日待办: 1. 来自前一天未完成的看板任务(最多20条) 2. 来自前一天对话记录分析(最多3条) """ today = date.today() yesterday = (today - timedelta(days=1)).isoformat() todos: list[DailyTodo] = [] # 1. 从看板任务导入 kanban_todos = await _import_kanban_tasks(user_id, yesterday, db) todos.extend(kanban_todos) # 2. 从对话记录分析 chat_todos = await _analyze_chat_history(user_id, yesterday, db) todos.extend(chat_todos) return todos async def _import_kanban_tasks(user_id: str, date_str: str, db: AsyncSession) -> list[DailyTodo]: """导入前一天创建的、未完成的看板任务""" q = select(Task).where( Task.user_id == user_id, Task.status != TaskStatus.DONE, ).order_by(Task.created_at.desc()).limit(20) tasks = (await db.execute(q)).scalars().all() todos = [] for task in tasks: todo = DailyTodo( user_id=user_id, title=task.title, source=TodoSource.AI_KANBAN, source_detail=f"看板:{task.title}", source_ref_id=task.id, todo_date=date.today().isoformat(), ) db.add(todo) todos.append(todo) if todos: await db.commit() for todo in todos: await db.refresh(todo) return todos async def _analyze_chat_history(user_id: str, date_str: str, db: AsyncSession) -> list[DailyTodo]: """分析前一天对话,提取待办事项""" try: # 查询前一天创建的对话 conv_q = select(Conversation).where( Conversation.user_id == user_id, ).order_by(Conversation.created_at.desc()).limit(10) convs = (await db.execute(conv_q)).scalars().all() # 过滤出昨天的对话 yesterday_convs = [] for conv in convs: created = conv.created_at if hasattr(created, 'date'): created_date = created.date() if hasattr(created, 'date') else created else: created_date = datetime.fromisoformat(str(created)).date() if str(created_date) == date_str or (created + timedelta(hours=8)).strftime('%Y-%m-%d') == date_str: yesterday_convs.append(conv) if not yesterday_convs: return [] # 收集消息内容(限制2000字) messages_content = [] for conv in yesterday_convs: msg_q = select(Message).where( Message.conversation_id == conv.id ).order_by(Message.created_at.asc()).limit(50) msgs = (await db.execute(msg_q)).scalars().all() for msg in msgs: if msg.content: messages_content.append(f"[{msg.role}]: {msg.content[:500]}") if not messages_content: return [] full_text = "\n".join(messages_content)[:2000] # 调用 LLM 分析 prompt = f"""你是一个任务规划助手。请分析以下对话记录,提取其中用户想要完成但尚未明确完成的事项。 要求: - 最多提取 3 条 - 每条格式:{{"title": "事项描述(50字以内)", "reason": "来源说明(60字以内)"}} - 只提取用户明确表达过需求但还未完成的事项 - 如果没有可提取的内容,返回空数组 [] 对话记录: {full_text} 返回 JSON 数组:""" response = await llm_service.chat(prompt=prompt, system=None) content = response if isinstance(response, str) else (response.get("content") or "") # 尝试解析 JSON try: # 提取 JSON 数组 start = content.find('[') end = content.rfind(']') + 1 if start != -1 and end > start: items = json.loads(content[start:end]) else: items = [] except (json.JSONDecodeError, ValueError): logger.warning(f"LLM 返回格式异常,跳过对话分析: {content[:200]}") items = [] if not items: return [] todos = [] for item in items[:3]: todo = DailyTodo( user_id=user_id, title=item.get("title", "")[:500], source=TodoSource.AI_CHAT, source_detail=f"对话:{item.get('reason', '')[:60]}", todo_date=date.today().isoformat(), ) db.add(todo) todos.append(todo) if todos: await db.commit() for todo in todos: await db.refresh(todo) return todos except Exception as e: logger.error(f"对话分析失败: {e}") return [] ``` - [ ] **Step 2: 提交** ```bash git add backend/app/services/todo_service.py git commit -m "feat(todo): add AI generation service" ``` --- ## Task 5: 后端 - 注册路由 **Files:** - Modify: `backend/app/main.py` - [ ] **Step 1: 在 `main.py` 中添加 todo router** 找到其他 router 注册的位置(如 `app/routers/conversation` 等),添加: ```python from app.routers.todo import router as todo_router app.include_router(todo_router) ``` - [ ] **Step 2: 提交** ```bash git add backend/app/main.py git commit -m "feat(todo): register todo router in main" ``` --- ## Task 6: 后端 - 定时任务(APScheduler) **Files:** - Modify: `backend/app/routers/scheduler.py` - Create: `backend/app/services/scheduler_service.py` (或修改现有) - [ ] **Step 1: 查看现有的 scheduler.py 结构** 读取 `backend/app/routers/scheduler.py` 和 `backend/app/services/scheduler_service.py`,了解现有定时任务模式。 - [ ] **Step 2: 添加每日 AI 生成定时任务** 在 scheduler_service 中添加: ```python async def daily_todo_generation(): """每天早上8点为所有活跃用户生成待办""" from app.database import async_session from app.models.user import User from sqlalchemy import select from app.services.todo_service import generate_daily_todos async with async_session() as db: result = await db.execute(select(User).where(User.is_active == True)) users = result.scalars().all() for user in users: try: await generate_daily_todos(user.id, db) except Exception as e: logger.error(f"用户 {user.id} 定时生成待办失败: {e}") ``` 在 APScheduler 中注册: ```python scheduler.add_job( daily_todo_generation, "cron", hour=8, minute=0, id="daily_todo_generation", replace_existing=True, ) ``` - [ ] **Step 3: 提交** ```bash git add backend/app/services/scheduler_service.py git commit -m "feat(todo): add daily AI todo generation scheduler" ``` --- ## Task 7: 前端 - API 客户端 **Files:** - Create: `frontend/src/api/todo.ts` - [ ] **Step 1: 创建 `src/api/todo.ts`** 参考 `src/api/task.ts` 的模式: ```typescript import api from './index' export type TodoSource = 'ai_kanban' | 'ai_chat' | 'manual' export interface Todo { id: string title: string is_completed: boolean source: TodoSource source_detail: string | null todo_date: string completed_at: string | null created_at: string updated_at: string } export interface TodoListResponse { items: Todo[] total: number page: number page_size: number } export interface TodoSummary { date: string total: number completed: number pending: number } export const todoApi = { list(date?: string, page = 1, pageSize = 50) { return api.get('/api/todos', { params: date ? { date_str: date, page, page_size: pageSize } : { page, page_size: pageSize }, }) }, create(title: string) { return api.post('/api/todos', { title }) }, update(id: string, data: { title?: string; is_completed?: boolean }) { return api.patch(`/api/todos/${id}`, data) }, delete(id: string) { return api.delete(`/api/todos/${id}`) }, aiGenerate() { return api.post('/api/todos/ai-generate') }, summary(date?: string) { return api.get('/api/todos/summary', { params: date ? { date_str: date } : {}, }) }, } ``` - [ ] **Step 2: 提交** ```bash git add frontend/src/api/todo.ts git commit -m "feat(todo): add todo API client" ``` --- ## Task 8: 前端 - 待办页面 TodoView.vue **Files:** - Create: `frontend/src/views/TodoView.vue` - [ ] **Step 1: 创建 TodoView.vue** 页面布局:顶部日期导航 + 待办列表,参考 AgentView 的 sci-fi 风格 核心结构: ```vue ``` - [ ] **Step 2: 提交** ```bash git add frontend/src/views/TodoView.vue git commit -m "feat(todo): add TodoView page" ``` --- ## Task 9: 前端 - 路由和侧边栏 **Files:** - Modify: `frontend/src/router/index.ts` - Modify: `frontend/src/components/SidebarNav.vue` - [ ] **Step 1: 在 `router/index.ts` 中添加 /todo 路由** 在 children 数组中添加: ```typescript { path: 'todo', name: 'todo', component: () => import('@/views/TodoView.vue'), }, ``` - [ ] **Step 2: 在 `SidebarNav.vue` 中添加待办菜单项** 在 navItems 数组中添加: ```typescript { name: '待办', path: '/todo', icon: CheckSquare }, ``` 导入 CheckSquare: ```typescript import { CheckSquare } from 'lucide-vue-next' ``` - [ ] **Step 3: 提交** ```bash git add frontend/src/router/index.ts frontend/src/components/SidebarNav.vue git commit -m "feat(todo): add route and sidebar menu" ``` --- ## Task 10: 数据库迁移 **Files:** - 创建迁移脚本(手动 SQL 或 Alembic) - [ ] **Step 1: 创建数据库表** 直接执行 SQL(参考 `app/models/todo.py` 的定义): ```sql CREATE TABLE IF NOT EXISTS daily_todos ( id VARCHAR(36) PRIMARY KEY, user_id VARCHAR(36) NOT NULL, title VARCHAR(500) NOT NULL, is_completed BOOLEAN NOT NULL DEFAULT FALSE, source VARCHAR(20) NOT NULL DEFAULT 'manual', source_detail VARCHAR(500), source_ref_id VARCHAR(36), todo_date VARCHAR(10) NOT NULL, completed_at TIMESTAMP NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, INDEX idx_user_date (user_id, todo_date), INDEX idx_user_id (user_id) ); ``` 使用 sqlite3 执行(假设数据库文件在 backend 目录): ```bash cd backend sqlite3 myagents.db < create_daily_todos.sql ``` 或通过 Python 直接创建表: ```python # 在 backend 目录执行 python -c " import asyncio from app.database import engine, BaseModel from app.models.todo import DailyTodo from sqlalchemy import text async def create_table(): async with engine.begin() as conn: await conn.execute(text(''' CREATE TABLE IF NOT EXISTS daily_todos ( id VARCHAR(36) PRIMARY KEY, user_id VARCHAR(36) NOT NULL, title VARCHAR(500) NOT NULL, is_completed BOOLEAN NOT NULL DEFAULT 0, source VARCHAR(20) NOT NULL DEFAULT 'manual', source_detail VARCHAR(500), source_ref_id VARCHAR(36), todo_date VARCHAR(10) NOT NULL, completed_at TIMESTAMP NULL, created_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ) ''')) await conn.execute(text('CREATE INDEX IF NOT EXISTS idx_daily_todos_user_date ON daily_todos(user_id, todo_date)')) await conn.execute(text('CREATE INDEX IF NOT EXISTS idx_daily_todos_user_id ON daily_todos(user_id)')) print('Table created successfully') asyncio.run(create_table()) " ``` - [ ] **Step 2: 提交迁移脚本** ```bash git add docs/superpowers/plans/2026-03-20-daily-todo-migration.sql git commit -m "feat(todo): add database migration" ``` --- ## 验证清单 完成所有 Task 后,验证以下内容: 1. **后端 API 可访问**:`GET /api/todos` 返回 200 2. **CRUD 正常**:新增、修改、删除待办均正常 3. **AI 生成正常**:`POST /api/todos/ai-generate` 返回待办列表 4. **前端页面正常**:`/todo` 页面可访问 5. **侧边栏菜单**:待办菜单项显示 6. **样式正常**:页面风格与 AgentView 一致(sci-fi 全息终端) 7. **日期切换正常**:昨天/前天数据正确加载 8. **历史只读**:修改历史日期的待办返回 403 --- ## 实现顺序建议 1. Task 1 → 2 → 3 → 5(后端核心) 2. Task 4(AI Service) 3. Task 6(定时任务) 4. Task 10(数据库) 5. Task 7 → 8 → 9(前端)