Python教程30:综合项目-任务管理器

“纸上得来终觉浅,绝知此事要躬行。”

经过第三部分13课的学习,我们掌握了函数、模块、文件操作、异常处理、测试等核心知识。今天我们通过一个完整项目——命令行任务管理器,把这些知识综合应用起来。

1. 项目需求

功能需求

开发一个命令行任务管理器(Todo List),支持:

  1. 基本功能

    • 添加任务
    • 查看任务列表
    • 标记任务完成
    • 删除任务
  2. 进阶功能

    • 任务优先级(高、中、低)
    • 任务分类(工作、生活、学习)
    • 截止日期
    • 数据持久化(保存到文件)
  3. 额外功能

    • 搜索任务
    • 统计信息
    • 导入/导出

技术要求

  • 使用函数和类组织代码
  • 模块化设计
  • 完善的异常处理
  • 文件操作(JSON格式存储)
  • 单元测试
  • 命令行交互

2. 项目结构

task_manager/
├── task_manager/          # 主包
│   ├── __init__.py
│   ├── task.py           # 任务类
│   ├── manager.py        # 管理器类
│   ├── storage.py        # 存储模块
│   └── utils.py          # 工具函数
├── tests/                 # 测试
│   ├── __init__.py
│   ├── test_task.py
│   ├── test_manager.py
│   └── test_storage.py
├── main.py               # 主程序入口
├── README.md
└── requirements.txt

3. 核心模块实现

3.1 任务类(task.py)

  1"""
  2任务类模块
  3- 定义Task类表示单个任务
  4- 应用:类、属性、方法
  5"""
  6
  7from datetime import datetime
  8from enum import Enum
  9
 10class Priority(Enum):
 11    """
 12    任务优先级枚举
 13    - 使用Enum定义常量
 14    - 避免魔法数字
 15    """
 16    LOW = 1
 17    MEDIUM = 2
 18    HIGH = 3
 19
 20class Category(Enum):
 21    """任务分类"""
 22    WORK = "工作"
 23    LIFE = "生活"
 24    STUDY = "学习"
 25    OTHER = "其他"
 26
 27class Task:
 28    """
 29    任务类
 30    
 31    Attributes:
 32        title: 任务标题
 33        description: 任务描述
 34        priority: 优先级
 35        category: 分类
 36        due_date: 截止日期
 37        completed: 是否完成
 38        created_at: 创建时间
 39    """
 40    
 41    def __init__(self, title, description="", 
 42                 priority=Priority.MEDIUM,
 43                 category=Category.OTHER,
 44                 due_date=None):
 45        """
 46        初始化任务
 47        
 48        Args:
 49            title: 任务标题(必需)
 50            description: 任务描述
 51            priority: 优先级(默认中等)
 52            category: 分类(默认其他)
 53            due_date: 截止日期(datetime对象)
 54        
 55        Raises:
 56            ValueError: 标题为空时
 57        """
 58        if not title:
 59            raise ValueError("任务标题不能为空")
 60        
 61        self.title = title
 62        self.description = description
 63        self.priority = priority
 64        self.category = category
 65        self.due_date = due_date
 66        self.completed = False
 67        self.created_at = datetime.now()
 68    
 69    def mark_complete(self):
 70        """标记任务完成"""
 71        self.completed = True
 72    
 73    def mark_incomplete(self):
 74        """标记任务未完成"""
 75        self.completed = False
 76    
 77    def to_dict(self):
 78        """
 79        转换为字典(用于JSON序列化)
 80        
 81        Returns:
 82            dict: 任务的字典表示
 83        """
 84        return {
 85            "title": self.title,
 86            "description": self.description,
 87            "priority": self.priority.value,
 88            "category": self.category.value,
 89            "due_date": self.due_date.isoformat() if self.due_date else None,
 90            "completed": self.completed,
 91            "created_at": self.created_at.isoformat()
 92        }
 93    
 94    @classmethod
 95    def from_dict(cls, data):
 96        """
 97        从字典创建任务(用于JSON反序列化)
 98        
 99        Args:
100            data: 任务字典
101        
102        Returns:
103            Task: 任务对象
104        """
105        task = cls(
106            title=data["title"],
107            description=data.get("description", ""),
108            priority=Priority(data.get("priority", 2)),
109            category=Category(data.get("category", "其他"))
110        )
111        
112        if data.get("due_date"):
113            task.due_date = datetime.fromisoformat(data["due_date"])
114        
115        task.completed = data.get("completed", False)
116        task.created_at = datetime.fromisoformat(data["created_at"])
117        
118        return task
119    
120    def __str__(self):
121        """字符串表示"""
122        status = "✓" if self.completed else " "
123        priority_symbols = {
124            Priority.LOW: "▽",
125            Priority.MEDIUM: "◇",
126            Priority.HIGH: "▲"
127        }
128        
129        return f"[{status}] {priority_symbols[self.priority]} {self.title}"
130    
131    def __repr__(self):
132        """开发者表示"""
133        return f"Task(title='{self.title}', completed={self.completed})"

3.2 存储模块(storage.py)

 1"""
 2存储模块
 3- 负责任务数据的持久化
 4- 应用:文件操作、JSON、异常处理
 5"""
 6
 7import json
 8import os
 9from pathlib import Path
10from typing import List
11from task import Task
12
13class TaskStorage:
14    """
15    任务存储类
16    - 使用JSON格式存储
17    - 包含完整的异常处理
18    """
19    
20    def __init__(self, file_path="tasks.json"):
21        """
22        初始化存储
23        
24        Args:
25            file_path: 存储文件路径
26        """
27        self.file_path = Path(file_path)
28        self._ensure_file_exists()
29    
30    def _ensure_file_exists(self):
31        """确保存储文件存在"""
32        if not self.file_path.exists():
33            self.file_path.write_text("[]", encoding="utf-8")
34    
35    def save_tasks(self, tasks: List[Task]):
36        """
37        保存任务列表
38        
39        Args:
40            tasks: 任务列表
41        
42        Raises:
43            IOError: 文件写入失败时
44        """
45        try:
46            data = [task.to_dict() for task in tasks]
47            
48            # 使用with确保文件正确关闭
49            with open(self.file_path, "w", encoding="utf-8") as f:
50                json.dump(data, f, ensure_ascii=False, indent=2)
51        
52        except Exception as e:
53            raise IOError(f"保存任务失败:{e}")
54    
55    def load_tasks(self) -> List[Task]:
56        """
57        加载任务列表
58        
59        Returns:
60            List[Task]: 任务列表
61        
62        Raises:
63            IOError: 文件读取失败时
64        """
65        try:
66            with open(self.file_path, encoding="utf-8") as f:
67                data = json.load(f)
68            
69            return [Task.from_dict(item) for item in data]
70        
71        except json.JSONDecodeError as e:
72            raise IOError(f"JSON解析失败:{e}")
73        except Exception as e:
74            raise IOError(f"加载任务失败:{e}")
75    
76    def backup(self, backup_path=None):
77        """
78        备份任务数据
79        
80        Args:
81            backup_path: 备份文件路径
82        """
83        if backup_path is None:
84            backup_path = f"{self.file_path}.backup"
85        
86        import shutil
87        shutil.copy(self.file_path, backup_path)

3.3 任务管理器(manager.py)

  1"""
  2任务管理器
  3- 核心业务逻辑
  4- 应用:类、方法、列表操作
  5"""
  6
  7from typing import List, Optional
  8from task import Task, Priority, Category
  9from storage import TaskStorage
 10
 11class TaskManager:
 12    """
 13    任务管理器
 14    - 管理任务的CRUD操作
 15    - 提供搜索、统计等功能
 16    """
 17    
 18    def __init__(self, storage: TaskStorage):
 19        """
 20        初始化管理器
 21        
 22        Args:
 23            storage: 存储对象
 24        """
 25        self.storage = storage
 26        self.tasks: List[Task] = []
 27        self.load()
 28    
 29    def load(self):
 30        """加载任务"""
 31        try:
 32            self.tasks = self.storage.load_tasks()
 33        except IOError as e:
 34            print(f"警告:{e}")
 35            self.tasks = []
 36    
 37    def save(self):
 38        """保存任务"""
 39        try:
 40            self.storage.save_tasks(self.tasks)
 41        except IOError as e:
 42            print(f"错误:{e}")
 43    
 44    def add_task(self, task: Task):
 45        """添加任务"""
 46        self.tasks.append(task)
 47        self.save()
 48    
 49    def get_all_tasks(self) -> List[Task]:
 50        """获取所有任务"""
 51        return self.tasks
 52    
 53    def get_task(self, index: int) -> Optional[Task]:
 54        """
 55        获取指定任务
 56        
 57        Args:
 58            index: 任务索引
 59        
 60        Returns:
 61            Task或None
 62        """
 63        if 0 <= index < len(self.tasks):
 64            return self.tasks[index]
 65        return None
 66    
 67    def complete_task(self, index: int) -> bool:
 68        """标记任务完成"""
 69        task = self.get_task(index)
 70        if task:
 71            task.mark_complete()
 72            self.save()
 73            return True
 74        return False
 75    
 76    def delete_task(self, index: int) -> bool:
 77        """删除任务"""
 78        if 0 <= index < len(self.tasks):
 79            self.tasks.pop(index)
 80            self.save()
 81            return True
 82        return False
 83    
 84    def search_tasks(self, keyword: str) -> List[Task]:
 85        """
 86        搜索任务
 87        
 88        Args:
 89            keyword: 关键词
 90        
 91        Returns:
 92            匹配的任务列表
 93        """
 94        keyword = keyword.lower()
 95        return [
 96            task for task in self.tasks
 97            if keyword in task.title.lower() 
 98            or keyword in task.description.lower()
 99        ]
100    
101    def get_statistics(self) -> dict:
102        """
103        获取统计信息
104        
105        Returns:
106            统计字典
107        """
108        total = len(self.tasks)
109        completed = sum(1 for task in self.tasks if task.completed)
110        pending = total - completed
111        
112        by_priority = {
113            Priority.HIGH: 0,
114            Priority.MEDIUM: 0,
115            Priority.LOW: 0
116        }
117        
118        for task in self.tasks:
119            if not task.completed:
120                by_priority[task.priority] += 1
121        
122        return {
123            "total": total,
124            "completed": completed,
125            "pending": pending,
126            "high_priority": by_priority[Priority.HIGH],
127            "medium_priority": by_priority[Priority.MEDIUM],
128            "low_priority": by_priority[Priority.LOW]
129        }

3.4 主程序(main.py)

  1"""
  2主程序入口
  3- 命令行交互界面
  4- 应用:函数、控制流程、异常处理
  5"""
  6
  7from task import Task, Priority, Category
  8from manager import TaskManager
  9from storage import TaskStorage
 10from datetime import datetime
 11
 12def print_menu():
 13    """打印菜单"""
 14    print("\n" + "=" * 50)
 15    print("任务管理器")
 16    print("=" * 50)
 17    print("1. 查看所有任务")
 18    print("2. 添加任务")
 19    print("3. 完成任务")
 20    print("4. 删除任务")
 21    print("5. 搜索任务")
 22    print("6. 统计信息")
 23    print("0. 退出")
 24    print("=" * 50)
 25
 26def display_tasks(tasks):
 27    """显示任务列表"""
 28    if not tasks:
 29        print("没有任务")
 30        return
 31    
 32    print("\n任务列表:")
 33    for i, task in enumerate(tasks):
 34        print(f"{i}. {task}")
 35        if task.due_date:
 36            print(f"   截止:{task.due_date.strftime('%Y-%m-%d')}")
 37
 38def add_task_interactive(manager: TaskManager):
 39    """交互式添加任务"""
 40    print("\n添加新任务")
 41    
 42    title = input("标题:").strip()
 43    if not title:
 44        print("标题不能为空")
 45        return
 46    
 47    description = input("描述(可选):").strip()
 48    
 49    print("优先级:1-低 2-中 3-高")
 50    priority_input = input("选择(默认2):").strip() or "2"
 51    priority = Priority(int(priority_input))
 52    
 53    try:
 54        task = Task(
 55            title=title,
 56            description=description,
 57            priority=priority
 58        )
 59        manager.add_task(task)
 60        print("✓ 任务已添加")
 61    
 62    except ValueError as e:
 63        print(f"错误:{e}")
 64
 65def main():
 66    """主函数"""
 67    print("欢迎使用任务管理器")
 68    
 69    # 初始化
 70    storage = TaskStorage()
 71    manager = TaskManager(storage)
 72    
 73    while True:
 74        print_menu()
 75        choice = input("\n请选择操作:").strip()
 76        
 77        try:
 78            if choice == "1":
 79                display_tasks(manager.get_all_tasks())
 80            
 81            elif choice == "2":
 82                add_task_interactive(manager)
 83            
 84            elif choice == "3":
 85                display_tasks(manager.get_all_tasks())
 86                index = int(input("请输入任务编号:"))
 87                if manager.complete_task(index):
 88                    print("✓ 任务已完成")
 89                else:
 90                    print("无效的任务编号")
 91            
 92            elif choice == "4":
 93                display_tasks(manager.get_all_tasks())
 94                index = int(input("请输入任务编号:"))
 95                if manager.delete_task(index):
 96                    print("✓ 任务已删除")
 97                else:
 98                    print("无效的任务编号")
 99            
100            elif choice == "5":
101                keyword = input("请输入搜索关键词:")
102                results = manager.search_tasks(keyword)
103                display_tasks(results)
104            
105            elif choice == "6":
106                stats = manager.get_statistics()
107                print(f"\n总任务数:{stats['total']}")
108                print(f"已完成:{stats['completed']}")
109                print(f"待完成:{stats['pending']}")
110                print(f"  高优先级:{stats['high_priority']}")
111                print(f"  中优先级:{stats['medium_priority']}")
112                print(f"  低优先级:{stats['low_priority']}")
113            
114            elif choice == "0":
115                print("再见!")
116                break
117            
118            else:
119                print("无效的选择")
120        
121        except Exception as e:
122            print(f"发生错误:{e}")
123            import traceback
124            traceback.print_exc()
125
126if __name__ == "__main__":
127    main()

4. 单元测试

 1# tests/test_task.py
 2import unittest
 3from datetime import datetime
 4from task import Task, Priority, Category
 5
 6class TestTask(unittest.TestCase):
 7    """测试Task类"""
 8    
 9    def test_task_creation(self):
10        """测试任务创建"""
11        task = Task("测试任务")
12        
13        self.assertEqual(task.title, "测试任务")
14        self.assertFalse(task.completed)
15        self.assertEqual(task.priority, Priority.MEDIUM)
16    
17    def test_empty_title_raises_error(self):
18        """测试空标题抛出异常"""
19        with self.assertRaises(ValueError):
20            Task("")
21    
22    def test_mark_complete(self):
23        """测试标记完成"""
24        task = Task("测试")
25        task.mark_complete()
26        
27        self.assertTrue(task.completed)
28    
29    def test_to_dict_and_from_dict(self):
30        """测试序列化/反序列化"""
31        task = Task("测试", priority=Priority.HIGH)
32        
33        data = task.to_dict()
34        restored = Task.from_dict(data)
35        
36        self.assertEqual(restored.title, task.title)
37        self.assertEqual(restored.priority, task.priority)
38
39if __name__ == "__main__":
40    unittest.main()

5. 知识点回顾

这个项目综合运用了:

第三部分知识点

  • 函数print_menu(), display_tasks()
  • Task, TaskManager, TaskStorage
  • 装饰器@classmethod
  • 模块:分文件组织代码
  • 文件操作:JSON读写
  • 异常处理try-except, 自定义异常
  • 上下文管理器with open()
  • 单元测试unittest.TestCase

其他知识点

  • 枚举Priority, Category
  • 类型提示List[Task], Optional[Task]
  • 列表推导式:搜索功能
  • datetime:时间处理

6. 运行项目

1# 运行程序
2python main.py
3
4# 运行测试
5python -m unittest discover tests
6
7# 测试覆盖率
8coverage run -m unittest discover
9coverage report

7. 扩展方向

可以继续改进:

  1. 功能扩展

    • 任务标签系统
    • 任务提醒
    • 定期任务
    • 子任务
  2. 技术改进

    • 使用SQLite数据库
    • 多用户支持
    • 命令行参数(argparse)
    • 配置文件
  3. 用户体验

    • 彩色输出(colorama)
    • 进度条
    • 表格显示(rich/tabulate)
    • 图形界面(tkinter)

8. 小结

通过这个项目,我们:

  • 综合运用了函数、类、模块等知识
  • 实践了文件操作和数据持久化
  • 完善了异常处理
  • 编写了单元测试
  • 体验了完整的开发流程

这标志着第三部分的完成。下一部分我们将学习面向对象编程,进一步提升代码组织能力。


练习题

  1. 为项目添加任务标签功能
  2. 实现任务导入/导出(CSV格式)
  3. 添加命令行参数支持(不进入交互模式直接执行操作)

本文代码示例

关注公众号:极客老墨

更多 AI 应用开发、工程实践和效率工具分享,欢迎扫码关注。

极客老墨微信公众号二维码

相关阅读