综合实战：构建命令行工具

mkdir -p task-cli/src/task_cli
mkdir -p task-cli/tests
touch task-cli/src/task_cli/__init__.py
touch task-cli/tests/__init__.py

# src/task_cli/models.py
from dataclasses import dataclass, field
from datetime import datetime, timezone
from enum import Enum
from typing import Optional


class TaskStatus(str, Enum):
    """任务状态枚举"""
    PENDING = "pending"
    COMPLETED = "completed"


class Priority(str, Enum):
    """优先级枚举"""
    LOW = "low"
    MEDIUM = "medium"
    HIGH = "high"
    CRITICAL = "critical"


@dataclass
class Task:
    """任务数据模型"""
    id: int
    title: str
    description: str = ""
    status: TaskStatus = TaskStatus.PENDING
    priority: Priority = Priority.MEDIUM
    created_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))
    updated_at: datetime = field(default_factory=lambda: datetime.now(timezone.utc))

    def complete(self) -> None:
        """将任务标记为已完成"""
        self.status = TaskStatus.COMPLETED
        self.updated_at = datetime.now(timezone.utc)

    def update(self, **kwargs) -> None:
        """更新任务属性"""
        for key, value in kwargs.items():
            if hasattr(self, key) and value is not None:
                setattr(self, key, value)
        self.updated_at = datetime.now(timezone.utc)

    def to_dict(self) -> dict:
        """转换为字典（用于 JSON 序列化）"""
        return {
            "id": self.id,
            "title": self.title,
            "description": self.description,
            "status": self.status.value,
            "priority": self.priority.value,
            "created_at": self.created_at.isoformat(),
            "updated_at": self.updated_at.isoformat(),
        }

    @classmethod
    def from_dict(cls, data: dict) -> "Task":
        """从字典创建 Task 实例（JSON 反序列化）"""
        return cls(
            id=data["id"],
            title=data["title"],
            description=data.get("description", ""),
            status=TaskStatus(data.get("status", "pending")),
            priority=Priority(data.get("priority", "medium")),
            created_at=datetime.fromisoformat(data["created_at"]),
            updated_at=datetime.fromisoformat(data["updated_at"]),
        )

    def __str__(self) -> str:
        status_mark = "✅" if self.status == TaskStatus.COMPLETED else "⬜"
        priority_tag = {
            Priority.LOW: "🟢",
            Priority.MEDIUM: "🟡",
            Priority.HIGH: "🔴",
            Priority.CRITICAL: "💀",
        }[self.priority]
        return f"{status_mark} [{self.id:>3}] {priority_tag} {self.title}"

# src/task_cli/storage.py
import json
import os
from pathlib import Path
from typing import Optional

from .models import Task


class TaskStorage:
    """任务数据持久化管理器"""

    def __init__(self, filepath: Optional[str] = None):
        if filepath:
            self.filepath = Path(filepath)
        else:
            # 默认保存在用户目录下的 .task-cli/tasks.json
            self.filepath = Path.home() / ".task-cli" / "tasks.json"

        self._ensure_directory()

    def _ensure_directory(self) -> None:
        """确保存储目录存在"""
        self.filepath.parent.mkdir(parents=True, exist_ok=True)

    def _read_all(self) -> list[dict]:
        """读取所有任务（原始字典格式）"""
        if not self.filepath.exists():
            return []

        try:
            with open(self.filepath, "r", encoding="utf-8") as f:
                return json.load(f)
        except (json.JSONDecodeError, OSError):
            return []

    def _write_all(self, tasks: list[dict]) -> None:
        """写入所有任务"""
        with open(self.filepath, "w", encoding="utf-8") as f:
            json.dump(tasks, f, ensure_ascii=False, indent=2)

    def add(self, task: Task) -> Task:
        """添加新任务"""
        tasks = self._read_all()
        tasks.append(task.to_dict())
        self._write_all(tasks)
        return task

    def get_all(self) -> list[Task]:
        """获取所有任务"""
        return [Task.from_dict(d) for d in self._read_all()]

    def get_by_id(self, task_id: int) -> Optional[Task]:
        """根据 ID 获取单个任务"""
        tasks = self._read_all()
        for data in tasks:
            if data["id"] == task_id:
                return Task.from_dict(data)
        return None

    def update(self, task_id: int, **kwargs) -> Optional[Task]:
        """更新任务"""
        tasks = self._read_all()
        for data in tasks:
            if data["id"] == task_id:
                task = Task.from_dict(data)
                task.update(**kwargs)
                tasks[tasks.index(data)] = task.to_dict()
                self._write_all(tasks)
                return task
        return None

    def delete(self, task_id: int) -> bool:
        """删除任务"""
        tasks = self._read_all()
        filtered = [t for t in tasks if t["id"] != task_id]
        if len(filtered) == len(tasks):
            return False  # 未找到
        self._write_all(filtered)
        return True

    def get_next_id(self) -> int:
        """生成下一个可用的任务 ID"""
        tasks = self._read_all()
        if not tasks:
            return 1
        return max(t["id"] for t in tasks) + 1

    def clear(self) -> None:
        """清空所有任务（用于测试）"""
        self._write_all([])

# src/task_cli/formatter.py
from typing import Optional

from .models import Task, TaskStatus, Priority


def format_task_detail(task: Task) -> str:
    """格式化任务详情"""
    status_text = "已完成" if task.status == TaskStatus.COMPLETED else "待办"
    priority_names = {
        Priority.LOW: "低",
        Priority.MEDIUM: "中",
        Priority.HIGH: "高",
        Priority.CRITICAL: "紧急",
    }

    lines = [
        f"任务 #{task.id}",
        "─" * 40,
        f"标题:     {task.title}",
        f"描述:     {task.description or '(无描述)'}",
        f"状态:     {status_text}",
        f"优先级:   {priority_names[task.priority]}",
        f"创建时间: {task.created_at.strftime('%Y-%m-%d %H:%M:%S')}",
        f"更新时间: {task.updated_at.strftime('%Y-%m-%d %H:%M:%S')}",
    ]
    return "\n".join(lines)


def format_task_list(
    tasks: list[Task],
    status: Optional[TaskStatus] = None,
) -> str:
    """格式化任务列表"""
    if not tasks:
        return "📭 还没有任务，使用 task-cli add 添加一个吧！"

    # 按状态筛选
    if status:
        tasks = [t for t in tasks if t.status == status]

    if not tasks:
        status_text = "已完成" if status == TaskStatus.COMPLETED else "待办"
        return f"📭 没有{status_text}任务。"

    lines = [f"📋 共 {len(tasks)} 个任务：", ""]
    for task in tasks:
        lines.append(f"  {task}")
    return "\n".join(lines)


def format_error(message: str) -> str:
    """格式化错误信息"""
    return f"❌ 错误: {message}"


def format_success(message: str) -> str:
    """格式化成功信息"""
    return f"✅ {message}"

# src/task_cli/cli.py
import argparse
import sys

from .models import Task, TaskStatus, Priority
from .storage import TaskStorage
from .formatter import (
    format_task_detail,
    format_task_list,
    format_error,
    format_success,
)


def create_parser() -> argparse.ArgumentParser:
    """创建命令行参数解析器"""
    parser = argparse.ArgumentParser(
        prog="task-cli",
        description="📋 任务管理器 - 在终端中管理你的任务",
        epilog="使用 task-cli <command> --help 查看子命令帮助",
    )

    subparsers = parser.add_subparsers(dest="command", help="可用命令")

    # add 子命令
    add_parser = subparsers.add_parser("add", help="添加新任务")
    add_parser.add_argument("title", help="任务标题")
    add_parser.add_argument("-d", "--description", default="", help="任务描述")
    add_parser.add_argument(
        "-p", "--priority",
        choices=["low", "medium", "high", "critical"],
        default="medium",
        help="优先级（默认: medium）",
    )

    # list 子命令
    list_parser = subparsers.add_parser("list", help="列出任务")
    list_parser.add_argument(
        "-s", "--status",
        choices=["pending", "completed"],
        help="按状态筛选",
    )

    # show 子命令
    show_parser = subparsers.add_parser("show", help="查看任务详情")
    show_parser.add_argument("id", type=int, help="任务 ID")

    # done 子命令
    done_parser = subparsers.add_parser("done", help="完成任务")
    done_parser.add_argument("id", type=int, help="任务 ID")

    # update 子命令
    update_parser = subparsers.add_parser("update", help="更新任务")
    update_parser.add_argument("id", type=int, help="任务 ID")
    update_parser.add_argument("-t", "--title", help="新标题")
    update_parser.add_argument("-d", "--description", help="新描述")
    update_parser.add_argument(
        "-p", "--priority",
        choices=["low", "medium", "high", "critical"],
        help="新优先级",
    )

    # delete 子命令
    delete_parser = subparsers.add_parser("delete", help="删除任务")
    delete_parser.add_argument("id", type=int, help="任务 ID")

    # clear 子命令
    subparsers.add_parser("clear", help="清空所有任务")

    return parser


def handle_add(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 add 命令"""
    try:
        task = Task(
            id=storage.get_next_id(),
            title=args.title,
            description=args.description,
            priority=Priority(args.priority),
        )
        storage.add(task)
        print(format_success(f"已添加任务 #{task.id}: {task.title}"))
        return 0
    except (OSError, ValueError) as e:
        print(format_error(str(e)))
        return 1


def handle_list(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 list 命令"""
    try:
        tasks = storage.get_all()
        status_filter = TaskStatus(args.status) if args.status else None
        print(format_task_list(tasks, status_filter))
        return 0
    except (OSError, ValueError) as e:
        print(format_error(str(e)))
        return 1


def handle_show(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 show 命令"""
    try:
        task = storage.get_by_id(args.id)
        if not task:
            print(format_error(f"任务 #{args.id} 不存在"))
            return 1
        print(format_task_detail(task))
        return 0
    except OSError as e:
        print(format_error(str(e)))
        return 1


def handle_done(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 done 命令"""
    try:
        task = storage.update(args.id, status=TaskStatus.COMPLETED)
        if not task:
            print(format_error(f"任务 #{args.id} 不存在"))
            return 1
        print(format_success(f"任务 #{args.id} 已完成: {task.title}"))
        return 0
    except OSError as e:
        print(format_error(str(e)))
        return 1


def handle_update(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 update 命令"""
    try:
        kwargs = {}
        if args.title:
            kwargs["title"] = args.title
        if args.description is not None:
            kwargs["description"] = args.description
        if args.priority:
            kwargs["priority"] = Priority(args.priority)

        if not kwargs:
            print(format_error("请至少指定一个要更新的字段"))
            return 1

        task = storage.update(args.id, **kwargs)
        if not task:
            print(format_error(f"任务 #{args.id} 不存在"))
            return 1
        print(format_success(f"任务 #{args.id} 已更新: {task.title}"))
        return 0
    except (OSError, ValueError) as e:
        print(format_error(str(e)))
        return 1


def handle_delete(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 delete 命令"""
    try:
        if storage.delete(args.id):
            print(format_success(f"已删除任务 #{args.id}"))
            return 0
        else:
            print(format_error(f"任务 #{args.id} 不存在"))
            return 1
    except OSError as e:
        print(format_error(str(e)))
        return 1


def handle_clear(args: argparse.Namespace, storage: TaskStorage) -> int:
    """处理 clear 命令"""
    try:
        storage.clear()
        print(format_success("已清空所有任务"))
        return 0
    except OSError as e:
        print(format_error(str(e)))
        return 1


def main(argv: list[str] | None = None) -> int:
    """主入口函数"""
    parser = create_parser()
    args = parser.parse_args(argv)

    if not args.command:
        parser.print_help()
        return 0

    storage = TaskStorage()

    handlers = {
        "add": handle_add,
        "list": handle_list,
        "show": handle_show,
        "done": handle_done,
        "update": handle_update,
        "delete": handle_delete,
        "clear": handle_clear,
    }

    handler = handlers.get(args.command)
    if handler:
        return handler(args, storage)

    return 0


if __name__ == "__main__":
    sys.exit(main())

# src/task_cli/__main__.py
"""支持 python -m task_cli 运行"""
from .cli import main

if __name__ == "__main__":
    import sys
    sys.exit(main())

# tests/test_models.py
import pytest
from datetime import datetime, timezone
from task_cli.models import Task, TaskStatus, Priority


class TestTaskCreation:
    def test_create_minimal_task(self):
        """创建只有必要字段的任务"""
        task = Task(id=1, title="测试任务")
        assert task.id == 1
        assert task.title == "测试任务"
        assert task.status == TaskStatus.PENDING
        assert task.priority == Priority.MEDIUM
        assert task.description == ""

    def test_create_full_task(self):
        """创建包含所有字段的任务"""
        task = Task(
            id=1,
            title="重要任务",
            description="这是一个重要任务",
            status=TaskStatus.COMPLETED,
            priority=Priority.HIGH,
        )
        assert task.description == "这是一个重要任务"
        assert task.status == TaskStatus.COMPLETED
        assert task.priority == Priority.HIGH


class TestTaskMethods:
    def test_complete_task(self):
        """完成一个任务"""
        task = Task(id=1, title="测试")
        task.complete()
        assert task.status == TaskStatus.COMPLETED

    def test_update_task(self):
        """更新任务属性"""
        task = Task(id=1, title="原标题")
        task.update(title="新标题", priority=Priority.HIGH)
        assert task.title == "新标题"
        assert task.priority == Priority.HIGH


class TestTaskSerialization:
    def test_to_dict(self):
        """任务序列化为字典"""
        task = Task(id=1, title="测试", priority=Priority.HIGH)
        data = task.to_dict()
        assert data["id"] == 1
        assert data["title"] == "测试"
        assert data["priority"] == "high"
        assert data["status"] == "pending"

    def test_from_dict(self):
        """从字典反序列化任务"""
        data = {
            "id": 1,
            "title": "测试",
            "description": "描述",
            "status": "completed",
            "priority": "high",
            "created_at": "2026-05-30T10:00:00+00:00",
            "updated_at": "2026-05-30T12:00:00+00:00",
        }
        task = Task.from_dict(data)
        assert task.id == 1
        assert task.title == "测试"
        assert task.status == TaskStatus.COMPLETED
        assert task.priority == Priority.HIGH

    def test_roundtrip(self):
        """序列化后再反序列化，结果应一致"""
        original = Task(id=42, title="往返测试", priority=Priority.CRITICAL)
        data = original.to_dict()
        restored = Task.from_dict(data)
        assert restored.id == original.id
        assert restored.title == original.title
        assert restored.priority == original.priority
        assert restored.status == original.status

# tests/test_storage.py
import pytest
import json
from pathlib import Path
from task_cli.models import Task, TaskStatus, Priority
from task_cli.storage import TaskStorage


@pytest.fixture
def storage(tmp_path):
    """每个测试使用临时文件存储"""
    filepath = tmp_path / "tasks.json"
    return TaskStorage(str(filepath))


class TestTaskStorage:
    def test_add_task(self, storage):
        """添加任务后应能读取"""
        task = Task(id=1, title="测试任务")
        storage.add(task)
        tasks = storage.get_all()
        assert len(tasks) == 1
        assert tasks[0].title == "测试任务"

    def test_get_by_id(self, storage):
        """按 ID 获取任务"""
        task = Task(id=1, title="任务一")
        storage.add(task)

        found = storage.get_by_id(1)
        assert found is not None
        assert found.title == "任务一"

        missing = storage.get_by_id(999)
        assert missing is None

    def test_update_task(self, storage):
        """更新任务"""
        storage.add(Task(id=1, title="旧标题"))
        updated = storage.update(1, title="新标题", priority=Priority.HIGH)
        assert updated is not None
        assert updated.title == "新标题"

        # 验证持久化
        task = storage.get_by_id(1)
        assert task.title == "新标题"
        assert task.priority == Priority.HIGH

    def test_delete_task(self, storage):
        """删除任务"""
        storage.add(Task(id=1, title="将被删除"))
        assert storage.delete(1) is True
        assert storage.get_by_id(1) is None

    def test_delete_nonexistent(self, storage):
        """删除不存在的任务应返回 False"""
        assert storage.delete(999) is False

    def test_next_id(self, storage):
        """生成正确的下一个 ID"""
        assert storage.get_next_id() == 1
        storage.add(Task(id=5, title="ID 5"))
        assert storage.get_next_id() == 6
        storage.add(Task(id=10, title="ID 10"))
        assert storage.get_next_id() == 11

    def test_clear(self, storage):
        """清空所有任务"""
        storage.add(Task(id=1, title="任务一"))
        storage.add(Task(id=2, title="任务二"))
        storage.clear()
        assert len(storage.get_all()) == 0

    def test_persistence(self, storage):
        """数据应持久化到文件"""
        storage.add(Task(id=1, title="持久化测试"))

        # 创建新的 storage 实例，从同一个文件读取
        storage2 = TaskStorage(str(storage.filepath))
        tasks = storage2.get_all()
        assert len(tasks) == 1
        assert tasks[0].title == "持久化测试"

    def test_empty_storage(self, storage):
        """空存储返回空列表"""
        assert storage.get_all() == []

    def test_corrupted_file(self, storage):
        """损坏的文件应被优雅处理"""
        storage.filepath.write_text("这不是 JSON", encoding="utf-8")
        assert storage.get_all() == []

[project]
name = "task-cli"
version = "1.0.0"
description = "一个终端任务管理器"
readme = "README.md"
requires-python = ">=3.10"
authors = [
    {name = "Python 从入门到精通", email = "learner@example.com"},
]

[project.scripts]
task-cli = "task_cli.cli:main"

[build-system]
requires = ["setuptools>=68.0"]
build-backend = "setuptools.build_meta"

[tool.setuptools.packages.find]
where = ["src"]

[tool.pytest.ini_options]
testpaths = ["tests"]

# 开发模式安装（可编辑）
pip install -e .

# 安装后可以直接使用
task-cli --help
task-cli add "学习 Python 30 篇"

# 查看帮助
$ task-cli --help
usage: task-cli [-h] {add,list,show,done,update,delete,clear} ...

📋 任务管理器 - 在终端中管理你的任务

positional arguments:
  {add,list,show,done,update,delete,clear}
    add         添加新任务
    list        列出任务
    show        查看任务详情
    done        完成任务
    update      更新任务
    delete      删除任务
    clear       清空所有任务

# 添加任务
$ task-cli add "完成 Python 课程项目" -d "实现任务管理器 CLI" -p high
✅ 已添加任务 #1: 完成 Python 课程项目

$ task-cli add "学习 pytest 高级用法" -p medium
✅ 已添加任务 #2: 学习 pytest 高级用法

$ task-cli add "写一篇技术博客" -d "总结 Python 学习心得" -p low
✅ 已添加任务 #3: 写一篇技术博客

$ task-cli add "修复 bug #1234" -p critical
✅ 已添加任务 #4: 修复 bug #1234

# 列出所有任务
$ task-cli list
📋 共 4 个任务：

  ⬜ [  1] 🔴 完成 Python 课程项目
  ⬜ [  2] 🟡 学习 pytest 高级用法
  ⬜ [  3] 🟢 写一篇技术博客
  ⬜ [  4] 💀 修复 bug #1234

# 完成任务
$ task-cli done 4
✅ 任务 #4 已完成: 修复 bug #1234

$ task-cli done 1
✅ 任务 #1 已完成: 完成 Python 课程项目

# 只查看待办任务
$ task-cli list --status pending
📋 共 2 个任务：

  ⬜ [  2] 🟡 学习 pytest 高级用法
  ⬜ [  3] 🟢 写一篇技术博客

# 查看已完成任务
$ task-cli list --status completed
📋 共 2 个任务：

  ✅ [  1] 🔴 完成 Python 课程项目
  ✅ [  4] 💀 修复 bug #1234

# 查看任务详情
$ task-cli show 1
任务 #1
────────────────────────────────────────
标题:     完成 Python 课程项目
描述:     实现任务管理器 CLI
状态:     已完成
优先级:   高
创建时间: 2026-05-30 10:00:00
更新时间: 2026-05-30 10:30:00

# 更新任务
$ task-cli update 3 -t "写一篇 Python 技术博客" -p medium
✅ 任务 #3 已更新: 写一篇 Python 技术博客

# 删除任务
$ task-cli delete 3
✅ 已删除任务 #3

$ pytest -v
============================== test session starts ==============================
collected 15 items

tests/test_models.py::TestTaskCreation::test_create_minimal_task PASSED
tests/test_models.py::TestTaskCreation::test_create_full_task PASSED
tests/test_models.py::TestTaskMethods::test_complete_task PASSED
tests/test_models.py::TestTaskMethods::test_update_task PASSED
tests/test_models.py::TestTaskSerialization::test_to_dict PASSED
tests/test_models.py::TestTaskSerialization::test_from_dict PASSED
tests/test_models.py::TestTaskSerialization::test_roundtrip PASSED
tests/test_storage.py::TestTaskStorage::test_add_task PASSED
tests/test_storage.py::TestTaskStorage::test_get_by_id PASSED
tests/test_storage.py::TestTaskStorage::test_update_task PASSED
tests/test_storage.py::TestTaskStorage::test_delete_task PASSED
tests/test_storage.py::TestTaskStorage::test_delete_nonexistent PASSED
tests/test_storage.py::TestTaskStorage::test_next_id PASSED
tests/test_storage.py::TestTaskStorage::test_clear PASSED
tests/test_storage.py::TestTaskStorage::test_persistence PASSED
tests/test_storage.py::TestTaskStorage::test_empty_storage PASSED
tests/test_storage.py::TestTaskStorage::test_corrupted_file PASSED

============================== 17 passed in 0.15s ===============================