hello-agents/Co-creation-projects/YYHDBL-HelloCodeAgentCli/tools/builtin/todo_tool.py

"""TodoTool - 轻量级待办板

MVP 目标：
- 仅支持 add / list / update
- 状态枚举：pending | in_progress | completed（强约束：同时最多 1 个 in_progress）
- 存储：.helloagents/todos/todos.json，原子写入 + 简单备份
- 输出：按状态分组的要点列表，便于 LLM 消化
"""

from __future__ import annotations

import json
from dataclasses import dataclass, asdict
from datetime import datetime
from pathlib import Path
from typing import Any, Dict, List, Optional

from ..base import Tool, ToolParameter


STATUSES = ("pending", "in_progress", "completed")


@dataclass
class TodoItem:
    id: int
    title: str
    desc: str = ""
    status: str = "pending"
    created_at: str = ""
    updated_at: str = ""

    @classmethod
    def from_dict(cls, data: Dict[str, Any]) -> "TodoItem":
        return cls(
            id=int(data["id"]),
            title=data.get("title", ""),
            desc=data.get("desc", ""),
            status=data.get("status", "pending"),
            created_at=data.get("created_at", ""),
            updated_at=data.get("updated_at", ""),
        )


class TodoTool(Tool):
    def __init__(self, workspace: str):
        super().__init__(
            name="todo",
            description="待办工具：add/list/update；状态 pending|in_progress|completed（同时仅允许1个in_progress）",
        )
        self.workspace = Path(workspace)
        self.workspace.mkdir(parents=True, exist_ok=True)
        self.data_file = self.workspace / "todos.json"
        self.backup_file = self.workspace / "todos.json.bak"
        if not self.data_file.exists():
            self._save({"items": []})

    def get_parameters(self) -> List[ToolParameter]:
        return [
            ToolParameter(name="action", type="string", description="add | list | update", required=True),
            ToolParameter(name="title", type="string", description="待办标题（add必填，update可选）", required=False),
            ToolParameter(name="desc", type="string", description="待办描述（可选）", required=False),
            ToolParameter(name="status", type="string", description="pending|in_progress|completed（update可选）", required=False),
            ToolParameter(name="id", type="integer", description="要更新的待办ID（update必填）", required=False),
        ]

    # ---------------- core ops ----------------
    def run(self, parameters: Dict[str, Any]) -> str:
        if not self.validate_parameters(parameters):
            return "参数缺失，需包含 action（add/list/update）。"
        action = str(parameters.get("action", "")).strip().lower().rstrip("]")
        if action == "add":
            return self._add(title=parameters.get("title", ""), desc=parameters.get("desc", ""), status=parameters.get("status", "pending"))
        if action == "list":
            return self._list(status_filter=parameters.get("status"))
        if action == "update":
            return self._update(
                todo_id=parameters.get("id"),
                title=parameters.get("title"),
                desc=parameters.get("desc"),
                status=parameters.get("status"),
            )
        return "不支持的 action，应为 add/list/update。"

    # ---------------- storage ----------------
    def _load(self) -> Dict[str, Any]:
        with open(self.data_file, "r", encoding="utf-8") as f:
            return json.load(f)

    def _save(self, data: Dict[str, Any]) -> None:
        tmp = self.data_file.with_suffix(".tmp")
        if self.data_file.exists():
            try:
                self.backup_file.write_text(self.data_file.read_text(encoding="utf-8"), encoding="utf-8")
            except Exception:
                pass
        tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
        tmp.replace(self.data_file)

    # ---------------- helpers ----------------
    def _next_id(self, items: List[TodoItem]) -> int:
        return (max([i.id for i in items], default=0) + 1) if items else 1

    def _now(self) -> str:
        return datetime.now().isoformat(timespec="seconds")

    def _enforce_single_in_progress(self, items: List[TodoItem], incoming_status: str, incoming_id: Optional[int]) -> Optional[str]:
        if incoming_status != "in_progress":
            return None
        for it in items:
            if it.status == "in_progress" and (incoming_id is None or it.id != incoming_id):
                return f"已有进行中的任务 #{it.id}《{it.title}》。先完成/更新它后再切换。"
        return None

    # ---------------- actions ----------------
    def _add(self, title: str, desc: str, status: str) -> str:
        title = (title or "").strip()
        if not title:
            return "❌ add 失败：title 不能为空。"
        status = status if status in STATUSES else "pending"
        data = self._load()
        items = [TodoItem.from_dict(i) for i in data.get("items", [])]
        conflict = self._enforce_single_in_progress(items, status, None)
        if conflict:
            return f"❌ add 失败：{conflict}"
        now = self._now()
        new_item = TodoItem(id=self._next_id(items), title=title, desc=desc or "", status=status, created_at=now, updated_at=now)
        items.append(new_item)
        self._save({"items": [asdict(i) for i in items]})
        return f"✅ 已添加 #{new_item.id} [{new_item.status}] {new_item.title}"

    def _update(self, todo_id: Any, title: Optional[str], desc: Optional[str], status: Optional[str]) -> str:
        try:
            tid = int(todo_id)
        except Exception:
            return "❌ update 失败：缺少有效的 id。"
        if status and status not in STATUSES:
            return "❌ update 失败：status 必须是 pending|in_progress|completed。"

        data = self._load()
        items = [TodoItem.from_dict(i) for i in data.get("items", [])]
        target = next((i for i in items if i.id == tid), None)
        if not target:
            return f"❌ update 失败：未找到 id={tid} 的任务。"

        conflict = self._enforce_single_in_progress(items, status or target.status, tid)
        if conflict:
            return f"❌ update 失败：{conflict}"

        changed = []
        if title is not None:
            target.title = title.strip()
            changed.append("title")
        if desc is not None:
            target.desc = desc
            changed.append("desc")
        if status is not None:
            target.status = status
            changed.append("status")
        target.updated_at = self._now()

        self._save({"items": [asdict(i) for i in items]})
        if not changed:
            return f"⚠️ 未修改任何字段 #{tid}"
        return f"✅ 已更新 #{tid} ({', '.join(changed)}) -> [{target.status}] {target.title}"

    def _list(self, status_filter: Optional[str]) -> str:
        data = self._load()
        items = [TodoItem.from_dict(i) for i in data.get("items", [])]
        if status_filter and status_filter in STATUSES:
            items = [i for i in items if i.status == status_filter]

        groups = {"in_progress": [], "pending": [], "completed": []}
        for it in items:
            groups.setdefault(it.status, []).append(it)

        # ANSI colors similar to v2_todo_agent
        COLOR_PENDING = "\x1b[38;2;176;176;176m"
        COLOR_PROGRESS = "\x1b[38;2;120;200;255m"
        COLOR_DONE = "\x1b[38;2;34;139;34m"
        RESET = "\x1b[0m"

        def fmt(group_name: str, arr: List[TodoItem]) -> str:
            if not arr:
                return ""
            lines = [f"[{group_name.upper()}]"]
            for it in sorted(arr, key=lambda x: x.id):
                mark = "☒" if it.status == "completed" else "☐"
                color = COLOR_PENDING
                if it.status == "in_progress":
                    color = COLOR_PROGRESS
                elif it.status == "completed":
                    color = COLOR_DONE
                line_main = f"- {mark} #{it.id} {it.title} (updated {it.updated_at})"
                line_desc = f"    {it.desc}" if it.desc else None

                if it.status == "completed":
                    line_main = f"{color}\x1b[9m{line_main}{RESET}"
                    if line_desc:
                        line_desc = f"{color}\x1b[9m{line_desc}{RESET}"
                else:
                    line_main = f"{color}{line_main}{RESET}"
                    if line_desc:
                        line_desc = f"{color}{line_desc}{RESET}"

                lines.append(line_main)
                if line_desc:
                    lines.append(line_desc)
            return "\n".join(lines)

        parts = [fmt("in_progress", groups["in_progress"]), fmt("pending", groups["pending"]), fmt("completed", groups["completed"])]
        out = "\n\n".join([p for p in parts if p])
        return out or "暂无待办。"