一、缘起
窗外的梧桐树叶已经开始泛黄,秋意渐浓。
豆豆一进门,就把书包往椅子上一扔,搓了搓手,一脸跃跃欲试的样子。
“马老师,上次您说今天开始写代码!”
“记性不错。” 马老师笑着推了推眼镜,”上节课我们把需求分析和系统设计都做完了,今天就从第一行代码开始,一步步把这个GTD任务管理系统搭起来。”
“从哪里开始?” 豆豆问。
“从最底层开始——数据访问层。” 马老师说,”你想想,为什么要从底层开始,而不是先做界面?”
豆豆想了一下:”因为……界面要显示数据,数据要从数据库里取,所以得先把数据库这层做好,界面才有东西可以显示?”
“完全正确。” 马老师说,”而且数据访问层是最独立的,它不依赖其它任何层,可以单独开发和测试。先把地基打好,上面的楼才稳。”
“那我们开始吧!”
二、建立项目框架
“动手之前,先把项目的目录结构建好。” 马老师说,”上节课我们设计了这样的结构,现在来创建它。”
在VSCode里,马老师新建了一个文件夹 gtd_task_manager,然后在里面依次创建了几个子目录和文件:
gtd_task_manager/├── main.py├── dao/│ ├── task_dao.py│ └── reminder_dao.py├── service/│ ├── task_service.py│ └── reminder_service.py├── ui/│ ├── main_window.py│ ├── task_dialog.py│ ├── review_dialog.py│ └── add_reminder_dialog.py└── tests/ └── test_task_dao.py
“注意,每个子目录里都要有一个空的 __init__.py 文件。” 马老师说。
“为什么?” 豆豆问。
“这是Python的规定——一个目录里有 __init__.py,Python才会把它当成一个包(package),这样其它文件才能用 from dao.task_dao import TaskDAO 这样的方式来引用它。” 马老师解释道,”你可以把它理解成一个’入场证’,有了它,这个目录才算是正式的Python模块目录。”
“明白了,就是个标记文件。”
“对,内容可以是空的,有它就行。” 马老师说,”好,框架建好了,我们开始写第一个文件——task_dao.py。”
三、实现数据访问层
3.1 定义Task数据类
“在写数据库操作之前,我们先要定义一个 Task 类,用来表示一条任务数据。” 马老师说,”你觉得这个类应该有哪些属性?”
豆豆翻开上节课的笔记,对照数据库设计表格说道:”id、title、due_date、category、description、progress、is_completed、created_at、completed_at、is_important、is_urgent……还有提醒列表reminders。”
“很好,全都想到了。” 马老师说,”我们用Python的 dataclass 来定义这个类,这是一种专门用来存储数据的类,写起来比普通类简洁很多。”
“dataclass?我没用过。”
“你看,普通的类要在 __init__ 里一个个赋值,很繁琐。用 dataclass 的话,只要在类里声明属性和默认值,Python会自动帮你生成 __init__ 方法。” 马老师说,”来,我们先写 Task 类:
from dataclasses import dataclassfrom datetime import datetimefrom typing import List@dataclassclass Task: id: int = None title: str = "" due_date: datetime = None category: str = "" description: str = "" progress: int = 0 is_completed: bool = False created_at: datetime = None completed_at: datetime = None is_important: bool = False is_urgent: bool = False reminders: List[datetime] = None
“这样就定义好了一个任务对象。” 马老师说,”注意每个属性都有默认值,这样创建任务的时候可以只传入需要的字段,其它的用默认值就行。”
“那 reminders 是一个列表,存的是提醒时间?” 豆豆问。
“对,一个任务可以有多个提醒时间,所以用列表存储。不过这个列表不直接存在数据库的tasks表里——你还记得我们设计了一个单独的reminders表吗?”
“记得!一对多关系,一个任务对应多条提醒记录。”
“正是。所以 Task 对象里的 reminders 列表,是我们从数据库里查出来之后手动填进去的,不是直接从tasks表里读出来的。”
3.2 实现TaskDAO
“好,现在来写 TaskDAO 类。” 马老师说,”DAO是Data Access Object的缩写,就是专门负责数据库操作的对象。我们先把数据库初始化和最基本的增删改查写出来。”
豆豆打开 dao/task_dao.py,开始跟着马老师一起写。
“首先是初始化数据库,也就是在数据库里建表。” 马老师说,”如果表已经存在就不重复建,用 CREATE TABLE IF NOT EXISTS 语句。”
两人一起,把 TaskDAO 的完整代码写了出来:
import sqlite3from datetime import datetimefrom dataclasses import dataclassfrom typing import List@dataclassclass Task: id: int = None title: str = "" due_date: datetime = None category: str = "" description: str = "" progress: int = 0 is_completed: bool = False created_at: datetime = None completed_at: datetime = None is_important: bool = False is_urgent: bool = False reminders: List[datetime] = None@dataclassclass TaskReminder: id: int = None task_id: int = None reminder_time: datetime = Noneclass TaskDAO: def __init__(self, db_path='gtd.db'): self.db_path = db_path self.init_db() def init_db(self): """初始化数据库表结构""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, due_date TIMESTAMP, category TEXT, description TEXT, progress INTEGER NOT NULL, is_completed BOOLEAN NOT NULL DEFAULT 0, created_at TIMESTAMP NOT NULL, completed_at TIMESTAMP, is_important BOOLEAN NOT NULL, is_urgent BOOLEAN NOT NULL ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS reminders ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, reminder_time TIMESTAMP NOT NULL, is_sent BOOLEAN NOT NULL DEFAULT 0 )''') conn.commit() def add_task(self, task: Task) -> int: """添加任务,返回新任务的id""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO tasks (title, due_date, category, description, progress, is_completed, created_at, completed_at, is_important, is_urgent) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (task.title, task.due_date, task.category, task.description, task.progress, task.is_completed, task.created_at, task.completed_at, task.is_important, task.is_urgent)) task_id = cursor.lastrowid if task.reminders: for reminder in task.reminders: cursor.execute(''' INSERT INTO reminders (task_id, reminder_time) VALUES (?, ?)''', (task_id, reminder)) conn.commit() return task_id def update_task(self, task: Task) -> int: """更新任务,只更新非None的字段""" all_fields = ['title', 'due_date', 'category', 'description', 'progress', 'is_important', 'is_urgent', 'completed_at', 'is_completed'] update_fields = [f"{field} = ?" for field in all_fields if getattr(task, field) is not None] if not update_fields: raise ValueError("没有要更新的属性") update_values = [getattr(task, field) for field in all_fields if getattr(task, field) is not None] update_clause = ','.join(update_fields) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(f''' UPDATE tasks SET {update_clause} WHERE id = ?''', update_values + [task.id]) cnt = cursor.rowcount # 先删除旧的提醒,再插入新的 cursor.execute('DELETE FROM reminders WHERE task_id = ?', (task.id,)) if task.reminders: for reminder in task.reminders: cursor.execute(''' INSERT INTO reminders (task_id, reminder_time) VALUES (?, ?)''', (task.id, reminder)) conn.commit() return cnt def delete_task(self, task_id: int) -> int: """删除任务(先删提醒,再删任务)""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM reminders WHERE task_id = ?', (task_id,)) cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,)) conn.commit() return cursor.rowcount def get_task(self, task_id: int) -> Task: """获取单个任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('SELECT * FROM tasks WHERE id = ?', (task_id,)) row = cursor.fetchone() if row: return Task(*row) return None def get_all_tasks(self) -> List[Task]: """获取所有任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('SELECT * FROM tasks') rows = cursor.fetchall() return [Task(*row) for row in rows] def get_tasks_by_quadrant(self, is_important, is_urgent) -> List[Task]: """根据象限获取未完成任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' SELECT * FROM tasks WHERE is_completed = 0 AND is_important = ? AND is_urgent = ?''', (is_important, is_urgent)) rows = cursor.fetchall() tasks = [Task(*row) for row in rows] for task in tasks: cursor.execute( 'SELECT reminder_time FROM reminders WHERE task_id = ?', (task.id,)) reminders = cursor.fetchall() task.reminders = [ datetime.strptime(r[0], '%Y-%m-%d %H:%M:%S') for r in reminders] return tasks def get_completed_tasks(self, keyword: str = '', sort_option: str = None, page: int = 1, page_size: int = 10): """获取已完成任务,支持关键字搜索、排序和分页""" sort_dict = { "按完成时间降序排序": "completed_at DESC", "按完成时间升序排序": "completed_at ASC", "按创建时间降序排序": "created_at DESC", "按创建时间升序排序": "created_at ASC" } sort_clause = sort_dict.get(sort_option, "completed_at DESC") kw = keyword or "" sql = (f"SELECT * FROM tasks WHERE is_completed = 1 " f"AND title LIKE '%{kw}%' " f"ORDER BY {sort_clause} " f"LIMIT {page_size} OFFSET {(page - 1) * page_size}") with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( f"SELECT COUNT(*) FROM tasks WHERE is_completed = 1 " f"AND title LIKE '%{kw}%'") total_count = cursor.fetchone()[0] total_pages = max(1, (total_count + page_size - 1) // page_size) cursor.execute(sql) rows = cursor.fetchall() tasks = [Task(*row) for row in rows] return {"tasks": tasks, "total_pages": total_pages, "page": page, "total_count": total_count}
“写完了!” 豆豆说,”不过这个 update_task 里有个地方我没太看懂——为什么要用 is not None 来判断要不要更新这个字段?”
马老师点点头:”问得好。这是因为我们希望 update_task 能够只更新传入的字段,没有传入的字段保持原样。比如你只想修改任务标题,就只传 title,其它字段设为 None,这样就只更新标题,不影响其它数据。”
“哦,我明白了。” 豆豆说,”那如果我想把某个字段改成空字符串,但它的默认值也是空字符串,这样不就分不清楚了吗?”
“你观察得很仔细!” 马老师说,”这确实是这种写法的一个局限。不过对于我们这个程序来说,在UI层调用 update_task 时,会把任务的所有字段都填好再传进来,所以不会有这个问题。这是一个设计上的约定——调用者有责任传入完整的Task对象。”
“明白了。”
四、自动化测试
4.1 为什么要写测试?
“TaskDAO写好了,我们怎么知道它是不是真的能用?” 马老师问。
“运行一下试试呗。” 豆豆说。
“怎么运行?数据库操作的代码没有界面,你怎么’运行一下试试’?”
豆豆愣了一下:”那……写一段测试代码,创建一个任务,再查出来,看看对不对?”
“对,这就是测试的基本思路。” 马老师说,”不过,如果每次改了代码都要手动写一段测试代码,然后眼睛盯着输出看有没有问题,这样做既麻烦,又容易漏掉情况。我们有更好的方式——自动化测试。”
“自动化测试是什么?”
“就是把测试本身也写成代码。” 马老师说,”你把’预期结果’也写进代码里,让程序自动比对实际结果和预期结果,如果不一致就报错。这样不管你改了多少代码,只要运行一下测试,就能立刻知道有没有破坏原来的功能。”
“这个想法很妙!” 豆豆说,”就像考试一样,有标准答案,对完就知道对不对了。”
“正是。Python内置了一个自动化测试框架,叫做 unittest,我们就用它来测试 TaskDAO。”
4.2 认识unittest
“unittest里有几个核心概念。” 马老师说,”首先是测试用例(TestCase),就是一个继承自 unittest.TestCase 的类,里面的每个以 test_ 开头的方法就是一个测试。”
“为什么要以 test_ 开头?”
“这是unittest的约定,它会自动找到所有 test_ 开头的方法并运行它们。” 马老师说,”其次是断言(assert),就是’我断定这个结果应该是这样的’。unittest提供了很多断言方法,比如:
assertEqual(a, b) # 断定a等于bassertIsNotNone(a) # 断定a不是NoneassertIsNone(a) # 断定a是NoneassertTrue(a) # 断定a为真
如果断言失败,测试就不通过,会报错告诉你哪里出了问题。”
“还有两个特殊的方法。” 马老师继续说,”setUp 方法在每个测试方法运行之前自动调用,用来准备测试环境;tearDown 方法在每个测试方法运行之后自动调用,用来清理测试环境。比如我们测试数据库操作,每次测试前创建一个新的测试数据库,测试完再删掉,这样每个测试都在干净的环境里运行,互不干扰。”
豆豆若有所思:”就像每次做实验之前先把实验台清干净,做完再收拾干净,下一个同学来的时候不会受到影响。”
“这个比喻很好!” 马老师笑道,”好,我们来写这个数据访问对象的测试 tests/test_task_dao.py 。”
4.3 编写测试代码
两人一起,在 tests/ 目录下新建了 test_task_dao.py:
import sqlite3from datetime import datetimefrom dataclasses import dataclassfrom typing import List@dataclassclass Task: id: int = None title: str = "" due_date: datetime = None category: str = "" description: str = "" progress: int = 0 is_completed: bool = False created_at: datetime = None completed_at: datetime = None is_important: bool = False is_urgent: bool = False reminders: List[datetime] = None@dataclassclass TaskReminder: id: int = None task_id: int = None reminder_time: datetime = Noneclass TaskDAO: def __init__(self, db_path='gtd.db'): self.db_path = db_path self.init_db() def init_db(self): """初始化数据库表结构""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' CREATE TABLE IF NOT EXISTS tasks ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, due_date TIMESTAMP, category TEXT, description TEXT, progress INTEGER NOT NULL, is_completed BOOLEAN NOT NULL DEFAULT 0, created_at TIMESTAMP NOT NULL, completed_at TIMESTAMP, is_important BOOLEAN NOT NULL, is_urgent BOOLEAN NOT NULL ) ''') cursor.execute(''' CREATE TABLE IF NOT EXISTS reminders ( id INTEGER PRIMARY KEY AUTOINCREMENT, task_id INTEGER NOT NULL, reminder_time TIMESTAMP NOT NULL, is_sent BOOLEAN NOT NULL DEFAULT 0 )''') conn.commit() def add_task(self, task: Task) -> int: """添加任务,返回新任务的id""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' INSERT INTO tasks (title, due_date, category, description, progress, is_completed, created_at, completed_at, is_important, is_urgent) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)''', (task.title, task.due_date, task.category, task.description, task.progress, task.is_completed, task.created_at, task.completed_at, task.is_important, task.is_urgent)) task_id = cursor.lastrowid if task.reminders: for reminder in task.reminders: cursor.execute(''' INSERT INTO reminders (task_id, reminder_time) VALUES (?, ?)''', (task_id, reminder)) conn.commit() return task_id def update_task(self, task: Task) -> int: """更新任务,只更新非None的字段""" all_fields = ['title', 'due_date', 'category', 'description', 'progress', 'is_important', 'is_urgent', 'completed_at', 'is_completed'] update_fields = [f"{field} = ?" for field in all_fields if getattr(task, field) is not None] if not update_fields: raise ValueError("没有要更新的属性") update_values = [getattr(task, field) for field in all_fields if getattr(task, field) is not None] update_clause = ','.join(update_fields) with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(f''' UPDATE tasks SET {update_clause} WHERE id = ?''', update_values + [task.id]) cnt = cursor.rowcount # 先删除旧的提醒,再插入新的 cursor.execute('DELETE FROM reminders WHERE task_id = ?', (task.id,)) if task.reminders: for reminder in task.reminders: cursor.execute(''' INSERT INTO reminders (task_id, reminder_time) VALUES (?, ?)''', (task.id, reminder)) conn.commit() return cnt def delete_task(self, task_id: int) -> int: """删除任务(先删提醒,再删任务)""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('DELETE FROM reminders WHERE task_id = ?', (task_id,)) cursor.execute('DELETE FROM tasks WHERE id = ?', (task_id,)) conn.commit() return cursor.rowcount def get_task(self, task_id: int) -> Task: """获取单个任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('SELECT * FROM tasks WHERE id = ?', (task_id,)) row = cursor.fetchone() if row: return Task(*row) return None def get_all_tasks(self) -> List[Task]: """获取所有任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute('SELECT * FROM tasks') rows = cursor.fetchall() return [Task(*row) for row in rows] def get_tasks_by_quadrant(self, is_important, is_urgent) -> List[Task]: """根据象限获取未完成任务""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(''' SELECT * FROM tasks WHERE is_completed = 0 AND is_important = ? AND is_urgent = ?''', (is_important, is_urgent)) rows = cursor.fetchall() tasks = [Task(*row) for row in rows] for task in tasks: cursor.execute( 'SELECT reminder_time FROM reminders WHERE task_id = ?', (task.id,)) reminders = cursor.fetchall() task.reminders = [ datetime.strptime(r[0], '%Y-%m-%d %H:%M:%S') for r in reminders] return tasks def get_completed_tasks(self, keyword: str = '', sort_option: str = None, page: int = 1, page_size: int = 10): """获取已完成任务,支持关键字搜索、排序和分页""" sort_dict = { "按完成时间降序排序": "completed_at DESC", "按完成时间升序排序": "completed_at ASC", "按创建时间降序排序": "created_at DESC", "按创建时间升序排序": "created_at ASC" } sort_clause = sort_dict.get(sort_option, "completed_at DESC") kw = keyword or "" sql = (f"SELECT * FROM tasks WHERE is_completed = 1 " f"AND title LIKE '%{kw}%' " f"ORDER BY {sort_clause} " f"LIMIT {page_size} OFFSET {(page - 1) * page_size}") with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( f"SELECT COUNT(*) FROM tasks WHERE is_completed = 1 " f"AND title LIKE '%{kw}%'") total_count = cursor.fetchone()[0] total_pages = max(1, (total_count + page_size - 1) // page_size) cursor.execute(sql) rows = cursor.fetchall() tasks = [Task(*row) for row in rows] return {"tasks": tasks, "total_pages": total_pages, "page": page, "total_count": total_count}
写好了,怎么运行?” 豆豆问。
“在项目根目录下,打开终端,运行这条命令:” 马老师说。
python -m unittest tests/test_task_dao.py -v
“后面的 -v 是什么意思?”
“verbose,就是’详细模式’,会把每个测试的名字和结果都打印出来,方便你看清楚哪些通过了,哪些没通过。” 马老师说,”来,你来运行一下。”
豆豆在终端里敲下命令,回车。片刻后,屏幕上出现了测试的输出:

“全部通过!” 豆豆兴奋地说,”这个感觉真好,就像考试全对了一样。”
“以后每次修改了 TaskDAO 的代码,都可以跑一遍这个测试,确认没有破坏原有功能。” 马老师说,”这就是自动化测试的价值——它是你代码的’安全网’。”
五、实现业务逻辑层
“数据访问层测试通过了,现在来写业务逻辑层。” 马老师说,”新建 service/task_service.py,你还记得 TaskService 要提供哪些方法吗?”
豆豆翻开笔记:”增加任务、修改任务、删除任务、完成任务,还有按象限查询、查询已完成任务、获取统计信息。”
“对,还有一个获取单个任务的方法。” 马老师说,”业务逻辑层最重要的职责是什么?”
“检查规则!” 豆豆脱口而出,”标题不能为空,截止日期不能早于今天,进度必须在0到100之间。”
“完全正确。” 马老师说,”这些规则放在 TaskService 里,而不是放在 TaskDAO 里——数据访问层只管存取数据,不管数据合不合法;业务逻辑层负责在存取之前先检查。”
“那如果规则检查不通过怎么办?”
“抛出异常。” 马老师说,”用 raise ValueError('错误信息') 告诉调用者哪里出了问题,调用者(也就是UI层)捕获异常后,再弹出提示框告诉用户。”
“哦,就是把错误信息从底层传到顶层。”
“对。好,来写 service/task_service.py:
from datetime import datetimefrom dao.task_dao import TaskDAO, Taskfrom typing import Listclass TaskService: def __init__(self): self.task_dao = TaskDAO() def _validate_task(self, task: Task): """校验任务数据(内部方法)""" if not task.title: raise ValueError("任务标题不能为空") if not task.due_date: raise ValueError("截止日期不能为空") # due_date可能是字符串(从数据库读出来的),也可能是date对象 due = task.due_date if isinstance(due, str): due = datetime.strptime(due, '%Y-%m-%d').date() if hasattr(due, 'date'): due = due.date() if due < datetime.now().date(): raise ValueError("截止日期不能早于今天") if task.progress < 0 or task.progress > 100: raise ValueError("进度必须在0-100之间") def add_task(self, task: Task) -> int: self._validate_task(task) task.created_at = datetime.now().replace(microsecond=0) return self.task_dao.add_task(task) def update_task(self, task: Task) -> int: if not self.task_dao.get_task(task.id): raise ValueError("任务不存在") self._validate_task(task) return self.task_dao.update_task(task) def delete_task(self, task_id: int) -> int: if not self.task_dao.get_task(task_id): raise ValueError("任务不存在") return self.task_dao.delete_task(task_id) def complete_task(self, task: Task) -> int: if not self.task_dao.get_task(task.id): raise ValueError("任务不存在") task.completed_at = datetime.now().replace(microsecond=0) task.progress = 100 task.is_completed = True return self.task_dao.update_task(task) def get_task(self, task_id: int) -> Task: return self.task_dao.get_task(task_id) def get_all_tasks(self) -> List[Task]: return self.task_dao.get_all_tasks() def get_tasks_by_quadrant(self, quadrant: int) -> List[Task]: """根据四象限编号获取未完成任务 1: 重要且紧急 2: 重要不紧急 3: 紧急不重要 4: 不重要不紧急 """ quadrant_map = { 1: (True, True), 2: (True, False), 3: (False, True), 4: (False, False) } if quadrant not in quadrant_map: raise ValueError("无效的象限编号") is_important, is_urgent = quadrant_map[quadrant] return self.task_dao.get_tasks_by_quadrant(is_important, is_urgent) def get_completed_tasks(self, keyword: str = None, sort_option: str = None, page: int = 1, page_size: int = 10): return self.task_dao.get_completed_tasks(keyword, sort_option, page, page_size) def get_tasks_statistics(self) -> dict: """获取任务统计信息""" all_tasks = self.task_dao.get_all_tasks() total = len(all_tasks) completed = len([t for t in all_tasks if t.is_completed]) return { "total": total, "completed": completed, "pending": total - completed, "quadrant_distribution": { 1: len([t for t in all_tasks if t.is_important and t.is_urgent and not t.is_completed]), 2: len([t for t in all_tasks if t.is_important and not t.is_urgent and not t.is_completed]), 3: len([t for t in all_tasks if not t.is_important and t.is_urgent and not t.is_completed]), 4: len([t for t in all_tasks if not t.is_important and not t.is_urgent and not t.is_completed]), } }
“注意看 _validate_task 这个方法名前面有个下划线。” 马老师说,”Python里,以单下划线开头的方法表示这是一个内部方法,不打算给外部直接调用,只在类内部使用。这是一种约定,不是强制规定,但有助于让代码的意图更清晰。”
“就是’这个方法是内部用的,外面别乱调’的意思。” 豆豆说。
“正是。” 马老师说,”还有,你注意到 add_task 里我在哪里设置了 created_at 吗?”
“在 TaskService 里,task.created_at = datetime.now()。” 豆豆说,”不是在DAO层设置的。”
“对。created_at 是’任务创建时间’,这是一个业务概念——任务什么时候被创建的,这个逻辑属于业务规则,所以放在业务逻辑层设置,而不是在数据库层。”
豆豆点点头,在本子上记下:业务规则放在Service层,数据操作放在DAO层。
六、实现表示层——主窗口
“数据层和服务层都好了,现在来做界面。” 马老师说,”先来做主窗口,我们分两步来:第一步先把主窗口的骨架搭起来,能看到界面就行;第二步再逐步完善每个功能。”
“好的,从哪里开始?”
“从 QuadrantWidget 开始。” 马老师说,”这是可复用的象限组件,主窗口里会创建四个这样的对象。你还记得每个象限要显示什么吗?”
“一个标题标签,下面是一个表格,表格里显示任务的状态、标题、进度、截止日期,最右边是编辑和删除按钮。”
“对。” 马老师说,”我们来写 ui/main_window.py,这个文件包含 QuadrantWidget 和 MainWindow 两个类:
from PyQt5.QtWidgets import ( QMainWindow, QWidget, QPushButton, QVBoxLayout, QHBoxLayout, QLabel, QCheckBox, QMessageBox, QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView)from PyQt5.QtCore import Qtfrom service.task_service import TaskServicefrom ui.task_dialog import TaskDialogfrom ui.review_dialog import ReviewDialogfrom dao.task_dao import Taskclass QuadrantWidget(QWidget): """可复用的象限组件,传入象限编号(1-4)区分四个象限""" def __init__(self, service: TaskService, quadrant: int, parent=None): super().__init__(parent) self.service = service self.quadrant = quadrant self.parent_window = parent self.init_ui() def init_ui(self): layout = QVBoxLayout(self) titles = {1: "重要紧急", 2: "重要不紧急", 3: "不重要紧急", 4: "不重要不紧急"} title_label = QLabel(f"第{self.quadrant}象限 - {titles[self.quadrant]}") title_label.setStyleSheet("font-size: 16px; font-weight: bold;") layout.addWidget(title_label) self.task_table = QTableWidget() self.task_table.setEditTriggers(QTableWidget.NoEditTriggers) self.task_table.verticalHeader().setVisible(False) self.task_table.setAlternatingRowColors(True) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.SingleSelection) self.task_table.setColumnCount(5) self.task_table.setHorizontalHeaderLabels( ["状态", "任务标题", "完成进度", "截止日期", "操作"]) self.task_table.horizontalHeader().setSectionResizeMode( 0, QHeaderView.ResizeToContents) self.task_table.horizontalHeader().setSectionResizeMode( 1, QHeaderView.Stretch) layout.addWidget(self.task_table) self.refresh_tasks() def refresh_tasks(self): tasks = self.service.get_tasks_by_quadrant(self.quadrant) self.task_table.setRowCount(len(tasks)) for row, task in enumerate(tasks): # 完成状态复选框 state_widget = QWidget() state_layout = QHBoxLayout(state_widget) checkbox = QCheckBox() checkbox.setChecked(task.is_completed) checkbox.stateChanged.connect( lambda state, t=task: self.complete_task(t)) state_layout.addWidget(checkbox) state_layout.setContentsMargins(10, 0, 0, 0) self.task_table.setCellWidget(row, 0, state_widget) self.task_table.setItem(row, 1, QTableWidgetItem(task.title)) self.task_table.setItem(row, 2, QTableWidgetItem(str(task.progress))) self.task_table.setItem(row, 3, QTableWidgetItem(str(task.due_date or ""))) # 操作按钮 btn_widget = QWidget() btn_layout = QHBoxLayout(btn_widget) edit_btn = QPushButton("编辑") del_btn = QPushButton("删除") edit_btn.clicked.connect(lambda _, t=task: self.edit_task(t)) del_btn.clicked.connect(lambda _, t=task: self.delete_task(t)) btn_layout.addWidget(edit_btn) btn_layout.addWidget(del_btn) btn_layout.setContentsMargins(2, 2, 2, 2) self.task_table.setCellWidget(row, 4, btn_widget) def edit_task(self, task: Task): dialog = TaskDialog(self, task) if dialog.exec_(): try: self.service.update_task(dialog.get_task()) self.refresh_tasks() except ValueError as e: QMessageBox.warning(self, "保存出错", str(e)) def delete_task(self, task: Task): reply = QMessageBox.question( self, "确认删除", "您确定要删除该任务吗?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: self.service.delete_task(task.id) self.refresh_tasks() self.parent_window.update_statistics() def complete_task(self, task: Task): self.service.complete_task(task) self.refresh_tasks() self.parent_window.update_statistics()class MainWindow(QMainWindow): def __init__(self): super().__init__() self.service = TaskService() self.init_ui() def init_ui(self): self.setWindowTitle("GTD 任务管理系统") self.setGeometry(100, 100, 1200, 800) central_widget = QWidget() self.setCentralWidget(central_widget) layout = QVBoxLayout(central_widget) # 顶部工具栏 top_layout = QHBoxLayout() add_btn = QPushButton("新建任务") add_btn.setMaximumSize(100, 30) add_btn.clicked.connect(self.add_task) review_btn = QPushButton("任务回顾") review_btn.setMaximumSize(100, 30) review_btn.clicked.connect(self.task_review) self.stats_label = QLabel("统计信息:") top_layout.addWidget(add_btn) top_layout.addWidget(review_btn) top_layout.addWidget(self.stats_label) layout.addLayout(top_layout) # 四象限布局 self.quadrant_dict = {} for i in range(1, 5): self.quadrant_dict[i] = QuadrantWidget(self.service, i, self) row1 = QHBoxLayout() row1.addWidget(self.quadrant_dict[1]) row1.addWidget(self.quadrant_dict[2]) row2 = QHBoxLayout() row2.addWidget(self.quadrant_dict[3]) row2.addWidget(self.quadrant_dict[4]) layout.addLayout(row1) layout.addLayout(row2) self.update_statistics() def add_task(self): dialog = TaskDialog(self) if dialog.exec_(): try: self.service.add_task(dialog.get_task()) self.refresh_all() except ValueError as e: QMessageBox.warning(self, "保存出错", str(e)) def task_review(self): ReviewDialog(self, self.service).exec_() def update_statistics(self): stats = self.service.get_tasks_statistics() self.stats_label.setText( f'总任务: {stats["total"]} | 待办: {stats["pending"]} | ' f'已完成: {stats["completed"]} | ' f'第一象限: {stats["quadrant_distribution"][1]} | ' f'第二象限: {stats["quadrant_distribution"][2]} | ' f'第三象限: {stats["quadrant_distribution"][3]} | ' f'第四象限: {stats["quadrant_distribution"][4]}' ) def refresh_all(self): for q in self.quadrant_dict.values(): q.refresh_tasks() self.update_statistics()
“主窗口写好了。” 马老师说,”你注意到 QuadrantWidget 的构造函数里有一个 parent 参数,存在 self.parent_window 里?”
“对,然后在 delete_task 和 complete_task 里调用了 self.parent_window.update_statistics()。” 豆豆说,”这是为了让主窗口的统计信息也跟着更新。”
“正是。这里有个设计上的权衡——象限组件需要通知主窗口刷新统计,所以保存了主窗口的引用。这让两个类之间有了一定的耦合,但对于这个程序来说是可以接受的。”
七、实现任务对话框
“主窗口写好了,但现在点’新建任务’还没有对话框可以用。” 马老师说,”我们来写 ui/task_dialog.py。这个对话框新建和编辑任务共用,怎么区分两种模式?”
“传不传任务对象进来?” 豆豆说,”如果传了就是编辑模式,没传就是新建模式。”
“完全正确!” 马老师说,”这就是一个对话框两用的技巧——构造函数里有一个可选的 task 参数,如果传入了 task 就预先填入它的数据,标题也显示’编辑任务’;如果没传,就显示空表单,标题显示’新建任务’。”
两人一起写好了 ui/task_dialog.py,以及辅助的 ui/add_reminder_dialog.py:
ui/add_reminder_dialog.py(添加提醒对话框):
from PyQt5.QtWidgets import ( QDialog, QVBoxLayout, QLabel, QPushButton, QDateTimeEdit)from datetime import datetimeclass AddReminderDialog(QDialog): def __init__(self, parent=None): super().__init__(parent) self.setWindowTitle("添加提醒") self.init_ui() def init_ui(self): layout = QVBoxLayout() layout.addWidget(QLabel("提醒时间:")) self.reminder_edit = QDateTimeEdit() self.reminder_edit.setDateTime( datetime.now().replace(microsecond=0, second=0)) self.reminder_edit.setCalendarPopup(True) self.reminder_edit.setDisplayFormat("yyyy-MM-dd HH:mm") layout.addWidget(self.reminder_edit) save_btn = QPushButton("保存") save_btn.clicked.connect(self.accept) layout.addWidget(save_btn) self.setLayout(layout)
ui/task_dialog.py(新建/编辑任务对话框):from PyQt5.QtWidgets import ( QDialog, QVBoxLayout, QHBoxLayout, QLabel, QLineEdit, QPushButton, QDateEdit, QComboBox, QTextEdit, QCheckBox, QListWidget, QMessageBox)from dao.task_dao import Taskfrom datetime import datetimefrom ui.add_reminder_dialog import AddReminderDialogclass TaskDialog(QDialog): def __init__(self, parent=None, task: Task = None): super().__init__(parent) self.task = task self.init_ui() def init_ui(self): layout = QVBoxLayout(self) # 任务标题 layout.addWidget(QLabel("任务标题:")) self.title_edit = QLineEdit() self.title_edit.setPlaceholderText("请输入任务标题") if self.task: self.title_edit.setText(self.task.title) layout.addWidget(self.title_edit) # 截止日期 layout.addWidget(QLabel("截止日期:")) self.due_date_edit = QDateEdit() self.due_date_edit.setCalendarPopup(True) if self.task: self.due_date_edit.setDate( datetime.strptime(str(self.task.due_date), '%Y-%m-%d').date()) else: self.due_date_edit.setDate(datetime.now().date()) layout.addWidget(self.due_date_edit) # 分类 layout.addWidget(QLabel("分类:")) self.category_combo = QComboBox() self.category_combo.addItems(["工作", "学习", "生活", "其他"]) if self.task: self.category_combo.setCurrentText(self.task.category) layout.addWidget(self.category_combo) # 描述 layout.addWidget(QLabel("描述:")) self.description_edit = QTextEdit() self.description_edit.setPlaceholderText("请输入任务描述") self.description_edit.setFixedHeight(80) if self.task: self.description_edit.setText(self.task.description) layout.addWidget(self.description_edit) # 进度 + 重要 + 紧急(同一行) progress_layout = QHBoxLayout() progress_layout.addWidget(QLabel("进度:")) self.progress_edit = QLineEdit() self.progress_edit.setPlaceholderText("0-100") if self.task: self.progress_edit.setText(str(self.task.progress)) progress_layout.addWidget(self.progress_edit) progress_layout.addWidget(QLabel("%")) self.importance_check = QCheckBox("重要") if self.task: self.importance_check.setChecked(self.task.is_important) self.urgency_check = QCheckBox("紧急") if self.task: self.urgency_check.setChecked(self.task.is_urgent) progress_layout.addWidget(self.importance_check) progress_layout.addWidget(self.urgency_check) layout.addLayout(progress_layout) # 提醒列表 reminder_layout = QHBoxLayout() reminder_layout.addWidget(QLabel("提醒列表:")) add_reminder_btn = QPushButton("添加提醒") del_reminder_btn = QPushButton("删除提醒") reminder_layout.addWidget(add_reminder_btn) reminder_layout.addWidget(del_reminder_btn) layout.addLayout(reminder_layout) self.reminder_listbox = QListWidget() self.reminder_listbox.setFixedHeight(80) if self.task and self.task.reminders: for r in self.task.reminders: self.reminder_listbox.addItem(r.strftime("%Y-%m-%d %H:%M:%S")) layout.addWidget(self.reminder_listbox) # 保存/取消按钮 btn_layout = QHBoxLayout() save_btn = QPushButton("保存") cancel_btn = QPushButton("取消") btn_layout.addWidget(save_btn) btn_layout.addWidget(cancel_btn) layout.addLayout(btn_layout) self.setWindowTitle("编辑任务" if self.task else "新建任务") layout.setSpacing(8) layout.setContentsMargins(20, 20, 20, 20) add_reminder_btn.clicked.connect(self.add_reminder) del_reminder_btn.clicked.connect( lambda: self.reminder_listbox.takeItem( self.reminder_listbox.currentRow())) save_btn.clicked.connect(self.save) cancel_btn.clicked.connect(self.reject) def add_reminder(self): dialog = AddReminderDialog(self) if dialog.exec_(): reminder = dialog.reminder_edit.dateTime().toString( "yyyy-MM-dd HH:mm:ss") self.reminder_listbox.addItem(reminder) def save(self): if not self.title_edit.text(): QMessageBox.warning(self, "警告", "任务标题不能为空") return progress_text = self.progress_edit.text() if progress_text and (not progress_text.isdigit() or not (0 <= int(progress_text) <= 100)): QMessageBox.warning(self, "警告", "进度必须是0-100之间的整数") return if self.due_date_edit.date().toPyDate() < datetime.now().date(): QMessageBox.warning(self, "警告", "截止日期不能早于今天") return self.accept() def get_task(self) -> Task: task = Task() task.id = self.task.id if self.task else None task.title = self.title_edit.text() task.due_date = self.due_date_edit.date().toPyDate() task.category = self.category_combo.currentText() task.description = self.description_edit.toPlainText() task.progress = int(self.progress_edit.text()) if self.progress_edit.text() else 0 task.is_completed = False task.completed_at = None task.created_at = datetime.now().replace(microsecond=0) task.is_important = self.importance_check.isChecked() task.is_urgent = self.urgency_check.isChecked() task.reminders = [ self.reminder_listbox.item(i).text() for i in range(self.reminder_listbox.count())] return task
“写好了!” 豆豆说,”我有个问题——save 方法里也做了输入检查,TaskService 里也有检查,这是不是重复了?”
“问得好!” 马老师说,”这叫做双重验证。UI层的检查是为了给用户即时的友好提示——比如弹出提示框告诉用户’标题不能为空’;Service层的检查是为了保证业务规则的正确性,防止有人绕过UI直接调用Service。两层都有检查,安全性更高。”
“明白了,就像考试既有老师巡视,又有摄像头,两道防线。” 豆豆笑道。
八、实现任务回顾对话框
“接下来写 ui/review_dialog.py——任务回顾界面。” 马老师说,”这个界面比较复杂,有搜索、排序和分页。你来说说分页的逻辑是什么?”
豆豆想了想:”每次显示10条,需要知道当前是第几页、总共有几页。点’下一页’就把页码加一,再重新查数据库。点’首页’就把页码设为1,’尾页’设为最大页码。”
“完全正确。来,我们写 ui/review_dialog.py:
from PyQt5.QtWidgets import ( QDialog, QVBoxLayout, QHBoxLayout, QLabel, QPushButton, QLineEdit, QComboBox, QTableWidget, QTableWidgetItem, QHeaderView, QAbstractItemView, QMessageBox, QWidget)from service.task_service import TaskServicefrom dao.task_dao import Taskclass ReviewDialog(QDialog): def __init__(self, parent=None, service: TaskService = None): super().__init__(parent) self.service = service self.cur_page = 1 self.total_page = 1 self.setWindowTitle("任务回顾") self.setGeometry(100, 100, 1024, 480) self.init_ui() def init_ui(self): layout = QVBoxLayout() # 搜索栏 query_layout = QHBoxLayout() query_layout.addWidget(QLabel("关键字:")) self.query_edit = QLineEdit() query_btn = QPushButton("查询") self.sort_combo = QComboBox() self.sort_combo.addItems(["按完成时间降序排序", "按完成时间升序排序"]) query_btn.clicked.connect(lambda: self._go_page(1)) self.sort_combo.currentIndexChanged.connect(lambda: self._go_page(1)) query_layout.addWidget(self.query_edit) query_layout.addWidget(query_btn) query_layout.addWidget(self.sort_combo) layout.addLayout(query_layout) # 任务列表 self.task_table = QTableWidget() self.task_table.setEditTriggers(QTableWidget.NoEditTriggers) self.task_table.setAlternatingRowColors(True) self.task_table.setSelectionBehavior(QAbstractItemView.SelectRows) self.task_table.setSelectionMode(QAbstractItemView.SingleSelection) self.task_table.setColumnCount(5) self.task_table.setHorizontalHeaderLabels( ["任务标题", "分类", "描述", "完成时间", "操作"]) self.task_table.horizontalHeader().setSectionResizeMode( 2, QHeaderView.Stretch) layout.addWidget(self.task_table) # 分页控制器 page_layout = QHBoxLayout() home_btn = QPushButton("首页") pre_btn = QPushButton("< 上一页") self.cur_page_label = QLabel("1") next_btn = QPushButton("下一页 >") final_btn = QPushButton("尾页") self.total_page_label = QLabel("共 1 页") skip_edit = QLineEdit() skip_edit.setFixedWidth(40) confirm_btn = QPushButton("跳转") home_btn.clicked.connect(lambda: self._go_page(1)) pre_btn.clicked.connect(lambda: self._go_page(self.cur_page - 1)) next_btn.clicked.connect(lambda: self._go_page(self.cur_page + 1)) final_btn.clicked.connect(lambda: self._go_page(self.total_page)) confirm_btn.clicked.connect( lambda: self._go_page(int(skip_edit.text())) if skip_edit.text().isdigit() else None) for w in [home_btn, pre_btn, self.cur_page_label, next_btn, final_btn, self.total_page_label, QLabel("跳到"), skip_edit, QLabel("页"), confirm_btn]: page_layout.addWidget(w) layout.addLayout(page_layout) self.setLayout(layout) self.refresh_tasks() def _go_page(self, page: int): page = max(1, min(page, self.total_page)) self.cur_page = page self.refresh_tasks() def refresh_tasks(self): result = self.service.get_completed_tasks( keyword=self.query_edit.text(), sort_option=self.sort_combo.currentText(), page=self.cur_page) if not result: return self.cur_page = result["page"] self.total_page = result["total_pages"] self.cur_page_label.setText(str(self.cur_page)) self.total_page_label.setText(f"共 {self.total_page} 页") tasks = result["tasks"] self.task_table.setRowCount(len(tasks)) for row, task in enumerate(tasks): self.task_table.setItem(row, 0, QTableWidgetItem(task.title)) self.task_table.setItem(row, 1, QTableWidgetItem(task.category or "")) self.task_table.setItem(row, 2, QTableWidgetItem(task.description or "")) self.task_table.setItem(row, 3, QTableWidgetItem(str(task.completed_at or ""))) del_btn = QPushButton("删除") del_btn.clicked.connect(lambda _, t=task: self.delete_task(t)) self.task_table.setCellWidget(row, 4, del_btn) def delete_task(self, task: Task): reply = QMessageBox.question( self, "确认删除", "确定要删除该任务记录吗?", QMessageBox.Yes | QMessageBox.No, QMessageBox.No) if reply == QMessageBox.Yes: self.service.delete_task(task.id) self.refresh_tasks()
“这个 _go_page 方法很简洁。” 豆豆说,”首页、上一页、下一页、尾页都调用它,只是传的页码不同。”
“对,这就是代码复用的思路——不要写四个几乎一样的方法,把共同逻辑提取出来,用参数区分。” 马老师说。
九、实现提醒功能
“现在来实现最后一个功能——任务提醒。” 马老师说,”提醒功能需要一个后台服务,程序启动后就一直在后台运行,定时检查有没有到期的提醒。”
“用什么来实现定时运行?” 豆豆问。
“我们用一个叫 APScheduler 的库。” 马老师说,”AP是 Advanced Python 的缩写,Scheduler是调度器,这个库可以让你很方便地设置定时任务。先安装它,还有用于发送桌面通知的 plyer 库:
pip install APScheduler plyer
首先写 dao/reminder_dao.py——专门负责查询到期提醒和标记已发送:
import sqlite3from datetime import datetimefrom dataclasses import dataclassfrom typing import List@dataclassclass TaskReminder: id: int = None task_id: int = None reminder_time: datetime = Noneclass TaskReminderDAO: def __init__(self, db_path="gtd.db"): self.db_path = db_path def get_due_reminders(self, current_time: datetime) -> List[TaskReminder]: """获取所有到期且未发送的提醒""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute(""" SELECT id, task_id, reminder_time FROM reminders WHERE reminder_time <= ? AND is_sent = 0 """, (current_time,)) rows = cursor.fetchall() return [TaskReminder(*row) for row in rows] def mark_reminder_sent(self, reminder_id: int): """标记提醒为已发送""" with sqlite3.connect(self.db_path) as conn: cursor = conn.cursor() cursor.execute( "UPDATE reminders SET is_sent = 1 WHERE id = ?", (reminder_id,)) conn.commit()
然后写 service/reminder_service.py:from plyer import notificationfrom apscheduler.schedulers.background import BackgroundSchedulerfrom datetime import datetimefrom dao.task_dao import TaskDAOfrom dao.reminder_dao import TaskReminderDAOclass ReminderService: def __init__(self): self.scheduler = BackgroundScheduler() self.task_dao = TaskDAO() self.reminder_dao = TaskReminderDAO() def start(self): """启动后台提醒服务""" self.scheduler.start() # 每10秒检查一次到期提醒 self.scheduler.add_job(self.check_reminders, 'interval', seconds=10) def check_reminders(self): """检查并发送到期提醒""" now = datetime.now() due_reminders = self.reminder_dao.get_due_reminders(now) for reminder in due_reminders: task = self.task_dao.get_task(reminder.task_id) if task and not task.is_completed: self.send_notification(task) self.reminder_dao.mark_reminder_sent(reminder.id) def send_notification(self, task): """发送桌面通知""" notification.notify( title=f"任务提醒:{task.title}", message=f"截止时间:{task.due_date}\n进度:{task.progress}%\n{task.description or''}", app_name='GTD任务管理系统', timeout=10 )
BackgroundScheduler 是后台调度器,启动后在后台线程里运行,不会阻塞主程序。” 马老师解释道,”每隔10秒,它就会自动调用一次 check_reminders,检查数据库里有没有到期的提醒。如果有,就用 plyer 发送一条系统桌面通知,然后把这条提醒标记为已发送,防止重复提醒。”
“原来桌面右下角弹出来的那种通知是这么实现的!” 豆豆说。
“对,plyer 库帮我们屏蔽了不同操作系统的差异,在Windows、Mac、Linux上都能发送通知,用法是一样的。”
十、最后一步:程序入口与界面美化
“所有模块都写好了,最后来写程序入口 main.py。” 马老师说,”这个文件负责把所有东西组装起来,启动程序。”
先写一个最简单的版本:
from plyer import notificationfrom apscheduler.schedulers.background import BackgroundSchedulerfrom datetime import datetimefrom dao.task_dao import TaskDAOfrom dao.reminder_dao import TaskReminderDAOclass ReminderService: def __init__(self): self.scheduler = BackgroundScheduler() self.task_dao = TaskDAO() self.reminder_dao = TaskReminderDAO() def start(self): """启动后台提醒服务""" self.scheduler.start() # 每10秒检查一次到期提醒 self.scheduler.add_job(self.check_reminders, 'interval', seconds=10) def check_reminders(self): """检查并发送到期提醒""" now = datetime.now() due_reminders = self.reminder_dao.get_due_reminders(now) for reminder in due_reminders: task = self.task_dao.get_task(reminder.task_id) if task and not task.is_completed: self.send_notification(task) self.reminder_dao.mark_reminder_sent(reminder.id) def send_notification(self, task): """发送桌面通知""" notification.notify( title=f"任务提醒:{task.title}", message=f"截止时间:{task.due_date}\n进度:{task.progress}%\n{task.description or''}", app_name='GTD任务管理系统', timeout=10 )
“好,现在可以运行了!” 马老师说,”在项目根目录下运行:
python main.py
豆豆迫不及待地运行了程序。窗口弹出来了,四个象限整整齐齐地排列着,虽然里面还没有数据,但整体结构一目了然。
“能跑起来了!” 豆豆兴奋地说,”不过……界面有点朴素。”
马老师笑了:”那我们来做最后一步——界面美化。我们用一个叫 qdarkstyle 的库,一行代码就能给程序换上深色主题。”
“一行代码就能换皮肤?”
“对,这就是使用第三方库的好处——别人已经把样式表写好了,我们直接用就行。安装它:
pip install qdarkstyle
“然后修改 main.py,加两行:
import sysfrom PyQt5.QtWidgets import QApplicationfrom ui.main_window import MainWindowfrom service.reminder_service import ReminderServiceimport qdarkstyleclass GTDApp(QApplication): def __init__(self, argv): super().__init__(argv) # 程序启动时就开始提醒服务 self.reminder_service = ReminderService() self.reminder_service.start()if __name__ == "__main__": app = GTDApp(sys.argv) app.setStyleSheet(qdarkstyle.load_stylesheet(qt_api='pyqt5')) window = MainWindow() window.show() sys.exit(app.exec_())
“这里我们把 QApplication 子类化成 GTDApp,在它的构造函数里启动提醒服务。” 马老师说,”这样提醒服务的生命周期和整个应用一致——程序启动,服务启动;程序关闭,服务也跟着结束。”
“比之前在外面直接创建 ReminderService 更规范。” 豆豆说。
“对,这是一种更好的组织方式。” 马老师说,”好,再运行一次!”
豆豆再次运行程序,深色主题的界面出现了——深灰色的背景,白色的文字,整个程序看起来相当专业。

“哇!” 豆豆眼睛一亮,”完全不一样的感觉!”
马老师点了点头:” requirements.txt 记得更新一下,把所有依赖库都写进去,方便以后在其它电脑上安装:
APScheduler==3.11.0plyer==2.1.0PyQt5==5.15.9QDarkStyle==3.2.3
“以后在新电脑上,只需要运行 pip install -r requirements.txt,就能一次性安装所有依赖。”
“这个我知道,就像食谱一样,把所有原料都列出来。” 豆豆说。
“对,这个比喻很贴切。”
十一、调试与完善
“程序能跑起来了,但还需要测试一下各个功能。” 马老师说,”豆豆,你来试试新建一个任务。”
豆豆点击”新建任务”,对话框弹出来了。他填上标题”数学作业”,截止日期选了明天,分类选”学习”,勾上”重要”和”紧急”,点保存。

“出现在第一象限了!” 豆豆说。
“再试试编辑和删除。” 马老师说。
豆豆点了编辑按钮,对话框弹出来,原来的数据都填好了。他把进度改成了50,保存,表格里的数字也跟着变了。
“统计信息也更新了。” 豆豆指着顶部说,”总任务1,待办1,第一象限1。”
“很好。” 马老师说,”再试试完成任务——勾选复选框。”
豆豆勾上了复选框,任务从第一象限消失了,统计信息变成了”已完成1”。
“再点’任务回顾’。”
豆豆点开任务回顾,刚才完成的任务出现在列表里,显示了标题、分类、完成时间。

“全部功能都正常!” 豆豆说,”马老师,这个程序我真的可以用了!”
随后豆豆又测试了提醒界面:

“不错。” 马老师笑道,”以后你有自己的提醒小助手了。不过这也不能算结束,后面在使用的过程中,你可能会发现一些小问题或者想要改进的地方,这是很正常的。软件开发就是这样一个不断迭代的过程。”
十二、回顾整个项目
“我们来回顾一下,从第一节课到今天,整个项目经历了哪些阶段?” 马老师说。
豆豆翻着笔记,认真地说:”第一节课,马老师给我讲了GTD方法,然后我们开始做需求分析——用例图、用例描述。第二节课,用原型图把界面画出来,确认了需求,然后做了系统架构设计、功能模块设计和数据库设计。今天这节课,我们按照设计,从底层开始一层层把代码写出来,还写了自动化测试,最后程序能跑起来了。”
“总结得很好。” 马老师说,”你觉得和以前我们做程序的方式有什么不同?”
豆豆想了一下:”以前做程序,都是马老师直接告诉我要做什么,我跟着写代码。这次不一样,我们先花了很多时间想清楚要做什么、怎么做,然后才开始写代码。感觉……写代码反而是最快的部分。”
“说得非常好。” 马老师点头,”这就是软件工程的价值。代码只是把设计变成现实的手段,真正决定软件质量的,是前期的分析和设计。”
“还有测试。” 豆豆补充道,”以前我从来不写测试,这次写了测试,感觉心里踏实多了。”
“对,测试是保证代码质量的重要手段。” 马老师说,”我们这次只对最核心的 TaskDAO 做了单元测试,在真正的项目里,每一层的代码都应该有对应的测试。”
“那 TaskService 也应该有测试?”
“是的,不过 TaskService 的测试方式稍微复杂一点,涉及到’模拟对象’(Mock)的概念,以后有机会再讲。” 马老师说,”你现在已经掌握了单元测试的基本思路,这就够了。”
豆豆把笔记本合上,看着屏幕上运行着的GTD程序,心里有一种说不出的满足感。
“马老师,我发现一件事。” 豆豆说,”这次我们做这个程序,感觉和做《万卷书库》那时候不一样了。以前我就是跟着做,现在……我好像能自己想了。”
“那是因为你开始有了系统性思维。” 马老师说,”不只是会写一段代码,而是能从需求到设计到实现,把整件事情想清楚。这是一个程序员最重要的能力之一。”
豆豆默默点了点头,把这句话记在了本子的最后一页。
十三、结语——新的起点
窗外,夕阳的余晖把天边染成了橘红色,凉风徐徐,带来了一丝秋末的清爽。
“马老师,” 豆豆站起来,背上书包,”咱们下次还学什么?”
马老师沉吟了片刻,说:”豆豆,你学Python编程也有一段时间了,从Scratch到Python,从命令行程序到图形界面,从单文件脚本到三层架构……你已经走过了很长的路。”
“是啊,感觉自己学了很多。” 豆豆说。
“但你有没有想过,现在有一种全新的学习方式——在AI的帮助下学习编程?” 马老师说,”这两年AI编程助手的能力越来越强,不仅能帮你写代码、解释错误、提供建议,甚至能做完整的项目……但这不是说你就不用学了,恰恰相反——你需要懂得如何跟AI协作,才能真正发挥它的价值。”
“就是要知道怎么问它,怎么用它?” 豆豆问。
“对,还要能判断它给的答案对不对,能看懂它写的代码,能发现它的错误。” 马老师说,”如果你连基础都不懂,AI给你一段代码,你都不知道能不能用,那才是真的被AI’替代’了。但如果你有了扎实的基础,再加上AI的辅助,你的能力会被放大好几倍。”
“那……我们下次就学这个?” 豆豆眼睛亮了。
“是的。” 马老师笑着说,”接下来,我们会开始新形式的学习——和AI一起学Python。我们会用真实的项目,一边学编程知识,一边学习如何正确地使用AI工具。你会学到如何用AI帮你分析需求、生成代码、排查Bug,也会学到AI的局限在哪里,什么时候要自己动脑筋。”
“听起来很有意思!” 豆豆说,”那比现在更难吗?”
“难度会提升,但方式会不一样。” 马老师说,”你会发现,有了AI的帮助,你能做的事情比以前大得多。就像有了计算器,你能解更复杂的数学题——但前提是你要会数学,不然你连答案对不对都不知道。”
豆豆若有所思地点点头,走向门口,忽然回头问:”马老师,那这个GTD程序……以后我们还会继续改进它吗?”
“当然可以。” 马老师说,”说不定在新的系列里,就会用它来做项目练习呢。”
豆豆笑了,推开门,走进了橘红色的傍晚里。傍晚的街道空荡荡的,但他感觉自己的内心从未像今天这样充实而坚定,他已经明白——
编程不只是写代码,它是一种把想法变成现实的能力。而这种能力,才刚刚开始。

《豆豆的Python生活》系列到这里就告一段落了。从当年ScratchJR中小猫的简单动画,到今天用Python做出一个有三层架构、有自动化测试、有桌面通知的完整应用——豆豆走过的,也是每一个初学者都会走过的路。感谢一路陪伴的读者们。新的系列《豆豆的AI时代》即将开始,再会!