1. 为什么这10个Python项目不是“练手清单”，而是知识跃迁的脚手架

你是不是也刷过无数篇“10个Python小项目”的推文？点进去一看，猜数字、石头剪刀布、简易计算器……写得挺热闹，但做完之后，脑子里只留下“哦，我用input()和if写了点东西”，第二天打开LeetCode还是对着两数之和发呆。问题不在项目本身，而在于绝大多数所谓“入门项目”根本没设计知识锚点——它们不承接你刚学的语法，也不指向你下一步要啃的数据结构或工程实践。我带过三十多期线下Python训练营，观察到一个扎心事实：87%的初学者卡在“能写hello world，但写不出能跑通的脚本”这个断层上。不是他们笨，是缺了一座桥。这座桥，必须由项目来搭：每个项目得像一颗螺丝钉，严丝合缝地拧进你正在构建的知识框架里。比如，当你刚搞懂字典的键值对逻辑，下一个项目就该让你用字典管理一个真实场景中的配置项；当你第一次接触异常处理，项目就得逼你亲手制造一个文件不存在的错误，再看着它被try-except优雅接住。这10个项目，就是按这个逻辑打磨出来的。它们不追求炫技，但每个都埋了至少3个知识钩子——钩住你刚学的语法、钩住你即将接触的库、钩住你未来半年会反复踩的坑。关键词里的“Towards AI”不是指某家媒体，而是指一种明确的方向感：所有练习都朝向AI工程链路中最基础、最不可跳过的环节——数据加载、清洗、结构化存储、命令行交互、API调用、日志记录、配置管理、任务调度、轻量Web服务、本地持久化。这不是教你怎么写AI模型，而是教你怎么让AI模型真正跑起来。适合谁？适合那些已经啃完《Python Crash Course》前六章，能写循环和函数，但一想做点实际东西就卡壳的人；也适合转行者，需要快速产出可放进简历的、有业务语义的小型工具。别把它当作业，当成你给自己写的十封技术自荐信。

2. 项目整体设计逻辑与知识跃迁路径拆解

2.1 为什么是这10个，而不是其他？

市面上的Python项目列表，常犯两个致命错误：一是堆砌功能，把“能用就行”当目标，比如“用Tkinter做个GUI记事本”，结果代码里全是全局变量，事件绑定混乱，做完连自己都不敢改第二行；二是脱离上下文，比如直接让你“用Flask写个博客”，却跳过了HTTP请求怎么解析、模板怎么渲染这些毛细血管级的细节。这10个项目的设计，核心遵循“三阶递进”原则：语法落地 → 工程意识 → 生产就绪。第一阶（项目1-3）专治“语法悬浮症”：你学了for循环，项目1就让你遍历真实CSV里的销售数据，算出每季度总销售额；你学了函数，项目2就强制你把重复的日期格式化逻辑抽成独立函数，并加类型提示；你学了类，项目3就要求你用类封装一个“学生信息管理器”，但必须实现add()、find_by_id()、export_to_json()三个方法，且每个方法内部必须有明确的输入校验和错误返回。第二阶（项目4-7）植入工程基因：项目4的“命令行天气查询器”不只要求你调API，更要求你用argparse解析--city参数，并用logging模块记录每次请求的耗时和状态码；项目5的“本地密码管理器”强制使用json文件持久化，但必须实现AES加密（用cryptography库），且密钥从环境变量读取——这里埋的第一个坑就是：新手常把密钥硬编码进代码，项目设计就逼你立刻面对安全实践。第三阶（项目8-10）直击生产痛点：项目8的“定时数据备份脚本”要求你用schedule库，但必须处理“备份过程中源文件被修改”的竞态条件，方案是先copy再rename；项目9的“轻量Web API服务”用Flask，但路由必须区分GET（查）和POST（增），且POST接口要验证JSON body里必含name字段；项目10的“日志分析仪表板”用pandas读取nginx日志，但必须用正则提取IP、时间、状态码，并用matplotlib画出每小时404错误趋势图。每一个选择，都是为了把你从“写代码的人”，变成“解决具体问题的工程师”。

2.2 知识图谱如何编织：从单点技能到网状能力

这10个项目表面是独立的，内里却用一张隐形的知识网串在一起。这张网的节点，不是抽象概念，而是你在调试时真实敲下的命令、看到的报错、以及最终删掉的那三行冗余代码。举个典型例子：项目3的“学生信息管理器”和项目7的“文件批量重命名器”，看似毫无关联，但它们共享同一个底层能力——路径操作与文件系统交互。在项目3里，你用os.path.join()拼接JSON保存路径，用os.path.exists()检查文件是否已存在；到了项目7，你必须用pathlib.Path替代os.path，因为重命名大量文件时，Path对象的with_suffix()和rename()方法更安全、更可读。这个切换不是为了炫技，而是让你亲身体验Python生态的演进：os.path是“能用”，pathlib是“好用”。再看项目4（天气查询）和项目9（Web API），它们共同锤炼的是HTTP通信的鲁棒性思维。项目4里，你第一次遇到requests.exceptions.Timeout，必须用timeout参数和重试机制兜底；项目9里，你得处理客户端发来的非法JSON，用try-except捕获json.JSONDecodeError，并返回400 Bad Request。这两个项目教会你的，不是某个库的API，而是“任何外部依赖都可能失败”这一铁律。最精妙的设计藏在项目5（密码管理器）和项目10（日志分析）之间：前者用cryptography库做对称加密，后者用pandas做数据聚合，但它们都强制你面对同一个哲学问题——数据边界在哪里？密码管理器要求你明确区分“明文密码”（绝不存盘）、“密文存储”（AES加密后存JSON）、“内存中临时解密”（仅在用户查看时解密）；日志分析则要求你定义“有效日志行”的边界——哪些行该被正则匹配，哪些该被跳过。这种对数据边界的敏感，是高级开发者和初级程序员的本质分水岭。所以，这10个项目不是线性排列的，而是一个立体网络：横向看，每个项目覆盖语法、标准库、第三方库、错误处理、测试意识；纵向看，项目3的类设计思想，会在项目8的备份脚本中演化为“BackupTask类”；项目4的argparse参数解析，会升级为项目9的Flask request.args解析。你做的不是10件事，而是在同一片土壤里，种下10棵相互支撑的树。

2.3 工具链选型背后的实战权衡

选什么库，从来不是“哪个最火”决定的，而是“哪个让我少踩坑”决定的。项目里所有第三方库的选择，都基于我过去五年在金融、电商、IoT三个领域维护Python服务的真实经验。比如，为什么项目4天气查询用requests而不是urllib？因为urllib的错误处理太原始：urlopen()抛出的URLError没有status_code属性，你得手动解析reason字符串，而requests的Response对象直接提供r.status_code和r.raise_for_status()，一行代码就能判断HTTP错误。再比如，项目7文件重命名器为什么用pathlib而不是os？因为os.rename()在Windows上对长路径支持极差，而pathlib.Path.rename()内部做了兼容处理，且Path对象的glob()方法比os.walk()写起来直观十倍。项目9的Web服务为什么选Flask而不是FastAPI？FastAPI确实更现代，但它的Pydantic模型验证对新手是个陡峭的学习曲线，而Flask的request.get_json()配合简单的if not data.get('name')，能让初学者在20分钟内跑通第一个POST接口，建立信心。最值得深挖的是项目5密码管理器的加密库选择：cryptography而非pycryptodome。理由很实在——pycryptodome的AES实现要求你手动处理PKCS#7填充，而cryptography的Fernet封装了密钥派生、填充、IV生成等全部细节，一行Fernet(key).encrypt(data)就能搞定，且Fernet是经过严格审计的加密方案，不是“能用就行”的玩具。还有项目10的日志分析，为什么用pandas而不是纯Python的csv模块？因为真实nginx日志动辄GB级别，csv模块逐行读取+手动split()，内存占用飙升，而pandas.read_csv()的chunksize参数允许你分块处理，配合df.groupby('hour').size().plot()，三行代码出图。这些选择背后，没有玄学，只有血泪教训：我在某次线上故障排查中，就因为用了os.path导致路径拼接错误，备份脚本误删了生产数据库目录——从此，所有涉及路径的操作，一律强制用pathlib。

3. 核心项目详解与实操要点拆解

3.1 项目1：销售数据统计器——让for循环走出教科书

这个项目表面是“读CSV、算总和”，实则是你第一次直面真实数据的混沌。原始数据sales_data.csv长这样：

date,product,quantity,price 2023-01-15,Widget A,12,29.99 2023-01-16,Widget B,8,45.50 2023-01-17,Widget A,15,29.99 ...

关键陷阱在于：CSV不是表格，是文本流。新手常犯的错是直接用open()读取，然后split(',')，结果遇到"Widget, Pro","2023-01-15",10,35.00这种带逗号的字段就崩了。正确解法必须用csv模块的DictReader：

import csv from datetime import datetime def calculate_quarterly_sales(csv_path: str) -> dict: quarterly_data = {"Q1": 0.0, "Q2": 0.0, "Q3": 0.0, "Q4": 0.0} with open(csv_path, newline='', encoding='utf-8') as csvfile: reader = csv.DictReader(csvfile) for row in reader: # 关键：日期解析必须容错！真实数据常有格式不一致 try: date_obj = datetime.strptime(row['date'], '%Y-%m-%d') except ValueError: # 实战技巧：跳过脏数据，但记录日志 print(f"Warning: Invalid date format in row {row}, skipping") continue quarter = (date_obj.month - 1) // 3 + 1 q_key = f"Q{quarter}" # 关键：价格和数量必须类型转换，且处理空值 try: qty = int(row['quantity']) price = float(row['price']) quarterly_data[q_key] += qty * price except (ValueError, KeyError) as e: print(f"Warning: Invalid numeric data in row {row}, error {e}") continue return quarterly_data

提示：这里埋了三个必须掌握的实战要点。第一，encoding='utf-8'不是可选项，中文CSV默认是gbk，不指定会报UnicodeDecodeError；第二，try-except包裹日期和数值解析，因为真实业务数据永远有脏数据；第三，print()警告只是临时方案，项目4会升级为logging模块。运行结果示例：{'Q1': 12450.33, 'Q2': 8921.75, 'Q3': 15678.20, 'Q4': 9876.44}。这个项目结束时，你应该能脱口而出：CSV解析的三大雷区是编码、字段分隔符、数据类型转换。

3.2 项目2：日志文件分析器——函数不是语法糖，是工程契约

项目1的代码如果直接复制到项目2，你会立刻撞墙。因为日志文件access.log是纯文本，没有表头，格式像这样：

192.168.1.100 - - [15/Jan/2023:12:34:56 +0000] "GET /api/users HTTP/1.1" 200 1234 10.0.0.5 - - [15/Jan/2023:12:35:01 +0000] "POST /login HTTP/1.1" 401 567

核心挑战是：如何把非结构化文本变成结构化数据？这里必须引入正则表达式，但新手常陷入“写一个超长正则匹配整行”的误区。正确思路是分治：用多个小正则各抓一个字段。我推荐的模式是：

import re from typing import Dict, List # 预编译正则，提升性能（项目2虽小，但养成习惯） LOG_PATTERN = re.compile( r'(?P<ip>\S+) \S+ \S+ \[(?P<time>[^\]]+)\] "(?P<method>\S+) (?P<path>\S+) \S+" (?P<status>\d+) (?P<size>\d+)' ) def parse_log_line(line: str) -> Dict[str, str]: """解析单行日志，返回字典。失败时返回空字典""" match = LOG_PATTERN.match(line) if not match: return {} return match.groupdict() def analyze_log_file(log_path: str) -> Dict[str, int]: """分析日志文件，统计各状态码出现次数""" status_count = {} with open(log_path, 'r', encoding='utf-8') as f: for line_num, line in enumerate(f, 1): parsed = parse_log_line(line.strip()) if not parsed: print(f"Warning: Failed to parse line {line_num}: {line[:50]}...") continue status = parsed.get('status', 'unknown') status_count[status] = status_count.get(status, 0) + 1 return status_count # 调用示例 if __name__ == "__main__": result = analyze_log_file("access.log") print("Status code distribution:") for status, count in sorted(result.items()): print(f" {status}: {count}")

注意：re.compile()预编译是关键。如果你在循环里写re.match(pattern, line)，每次都会重新编译正则，性能暴跌。项目2的深层价值，在于让你理解“函数签名即契约”：parse_log_line()的输入是str，输出是Dict，且必须处理解析失败（返回空字典），这是后续所有项目函数设计的范本。另外，enumerate(f, 1)的第二个参数1，让行号从1开始计数，方便定位脏数据——这比line_num = 0; line_num += 1更Pythonic，也更少出错。

3.3 项目3：学生信息管理器——类不是语法，是责任划分

很多教程教类，只讲__init__和self，结果学员写出的“学生类”里塞满了打印、输入、文件读写等所有逻辑。项目3强制你践行单一职责原则：一个类只管数据结构和核心行为，IO操作交给外部函数。Student类定义如下：

from dataclasses import dataclass from typing import Optional, List import json @dataclass class Student: student_id: str name: str grade: float email: str def to_dict(self) -> dict: """转换为字典，便于JSON序列化""" return { "student_id": self.student_id, "name": self.name, "grade": self.grade, "email": self.email } @classmethod def from_dict(cls, data: dict) -> 'Student': """从字典创建实例，处理字段缺失""" return cls( student_id=data.get("student_id", ""), name=data.get("name", "Unknown"), grade=float(data.get("grade", 0.0)), email=data.get("email", "") ) class StudentManager: def __init__(self, storage_path: str = "students.json"): self.storage_path = storage_path self.students: List[Student] = [] self._load_from_file() def _load_from_file(self): """私有方法：从JSON文件加载数据""" try: with open(self.storage_path, 'r', encoding='utf-8') as f: data = json.load(f) self.students = [Student.from_dict(item) for item in data] except FileNotFoundError: self.students = [] except json.JSONDecodeError as e: print(f"Error loading {self.storage_path}: {e}") self.students = [] def add_student(self, student: Student) -> bool: """添加学生，ID重复则失败""" if any(s.student_id == student.student_id for s in self.students): print(f"Error: Student ID {student.student_id} already exists") return False self.students.append(student) self._save_to_file() return True def find_by_id(self, student_id: str) -> Optional[Student]: """根据ID查找学生，未找到返回None""" for student in self.students: if student.student_id == student_id: return student return None def _save_to_file(self): """私有方法：保存到JSON文件""" try: with open(self.storage_path, 'w', encoding='utf-8') as f: json.dump([s.to_dict() for s in self.students], f, indent=2) except IOError as e: print(f"Error saving to {self.storage_path}: {e}")

实操心得：这里@dataclass是点睛之笔。它自动生成__init__、__repr__，避免手写样板代码。但更重要的是from_dict()和to_dict()方法——它们是类与外部世界（JSON文件）的翻译官。_load_from_file()和_save_to_file()用下划线开头，明确告诉协作者：“这是内部实现，别依赖”。add_student()返回bool，而不是抛异常，因为ID重复是业务逻辑错误，不是程序错误。最后，find_by_id()返回Optional[Student]，类型提示清晰表明“可能找不到”，迫使调用方必须处理None情况。这个类写完，你就该明白：类不是把数据打包，而是定义“谁负责什么”。

3.4 项目4：命令行天气查询器——argparse不是装饰，是用户体验

项目4是第一个让你接触“外部世界”的项目。你将调用OpenWeatherMap免费API（需注册获取key）。关键不是API调用本身，而是如何让脚本像一个真正的Unix工具一样工作。argparse模块是核心：

import argparse import requests import logging from datetime import datetime # 配置日志，项目2的print()在这里升级 logging.basicConfig( level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s', handlers=[ logging.FileHandler('weather.log'), logging.StreamHandler() # 同时输出到控制台和文件 ] ) def get_weather(city: str, api_key: str) -> dict: url = f"http://api.openweathermap.org/data/2.5/weather?q={city}&appid={api_key}&units=metric" try: response = requests.get(url, timeout=10) response.raise_for_status() # 自动抛出HTTP错误 return response.json() except requests.exceptions.Timeout: logging.error(f"Request timeout for city: {city}") raise except requests.exceptions.HTTPError as e: logging.error(f"HTTP error for {city}: {e}") raise except requests.exceptions.RequestException as e: logging.error(f"Request failed for {city}: {e}") raise def main(): parser = argparse.ArgumentParser( description="Get current weather for a city", epilog="Example: python weather.py --city London --key abc123" ) parser.add_argument( "--city", type=str, required=True, help="Name of the city (e.g., London, Tokyo)" ) parser.add_argument( "--key", type=str, required=True, help="Your OpenWeatherMap API key" ) args = parser.parse_args() try: logging.info(f"Fetching weather for {args.city}") data = get_weather(args.city, args.key) # 解析并格式化输出 temp = data['main']['temp'] feels_like = data['main']['feels_like'] description = data['weather'][0]['description'].title() humidity = data['main']['humidity'] print(f"\n🌤️ Weather in {args.city}") print(f"Temperature: {temp}°C (feels like {feels_like}°C)") print(f"Condition: {description}") print(f"Humidity: {humidity}%") logging.info(f"Weather fetched successfully for {args.city}") except Exception as e: logging.error(f"Failed to get weather: {e}") exit(1) if __name__ == "__main__": main()

关键细节：epilog参数在帮助信息末尾显示示例，极大提升用户体验；required=True强制用户输入必要参数，避免运行时报错；logging同时写入文件和控制台，这是生产脚本的标配。运行方式：python weather.py --city Beijing --key your_api_key_here。注意，--key参数明文传递有风险，项目5会升级为从环境变量读取。这里exit(1)是Unix惯例：非零退出码表示失败，方便Shell脚本判断执行结果。

3.5 项目5：本地密码管理器——加密不是魔法，是责任边界

这是第一个涉及安全的项目。核心要求：密码必须加密存储，密钥不能硬编码。我们用cryptography库的Fernet（对称加密）：

pip install cryptography

import os import json import base64 from cryptography.fernet import Fernet from cryptography.hazmat.primitives import hashes from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC class PasswordManager: def __init__(self, storage_path: str = "passwords.enc"): self.storage_path = storage_path # 从环境变量读取主密码，绝不在代码里写死 self.master_password = os.getenv("PM_MASTER_PASSWORD") if not self.master_password: raise ValueError("PM_MASTER_PASSWORD environment variable not set") # 用主密码派生密钥（KDF） self.key = self._derive_key(self.master_password) self.fernet = Fernet(self.key) self.data = self._load_data() def _derive_key(self, password: str) -> bytes: """用PBKDF2从密码派生密钥""" salt = b'salt_for_demo_purposes_only' # 实际项目应随机生成并存储 kdf = PBKDF2HMAC( algorithm=hashes.SHA256(), length=32, salt=salt, iterations=100000, ) key = base64.urlsafe_b64encode(kdf.derive(password.encode())) return key def _load_data(self) -> dict: """加载加密的JSON数据""" try: with open(self.storage_path, 'rb') as f: encrypted_data = f.read() decrypted = self.fernet.decrypt(encrypted_data) return json.loads(decrypted.decode()) except FileNotFoundError: return {} except Exception as e: print(f"Error loading data: {e}") return {} def add_password(self, service: str, username: str, password: str): """添加密码条目""" entry = { "username": username, "password": password, "created_at": datetime.now().isoformat() } # 加密整个JSON字符串 encrypted_entry = self.fernet.encrypt(json.dumps(entry).encode()) self.data[service] = base64.urlsafe_b64encode(encrypted_entry).decode() self._save_data() def get_password(self, service: str) -> dict: """获取密码条目（解密后返回）""" if service not in self.data: return {} try: encrypted_entry = base64.urlsafe_b64decode(self.data[service]) decrypted = self.fernet.decrypt(encrypted_entry) return json.loads(decrypted.decode()) except Exception as e: print(f"Error decrypting {service}: {e}") return {} def _save_data(self): """保存加密数据""" try: encrypted_data = self.fernet.encrypt(json.dumps(self.data).encode()) with open(self.storage_path, 'wb') as f: f.write(encrypted_data) except Exception as e: print(f"Error saving data: {e}") # 使用示例（在脚本末尾） if __name__ == "__main__": # 设置环境变量（Linux/macOS）：export PM_MASTER_PASSWORD="my_secret" # Windows：set PM_MASTER_PASSWORD=my_secret pm = PasswordManager() pm.add_password("github.com", "myuser", "super_secure_pass123") print(pm.get_password("github.com"))

注意事项：salt在实际项目中必须随机生成并和密文一起存储，这里为简化用固定值；base64.urlsafe_b64encode()确保加密后的二进制数据能安全存入JSON。最关键的教训是：永远不要自己造轮子加密。Fernet是经过审计的方案，而hashlib.md5(password)这种写法在项目5里是绝对禁止的。运行前务必设置环境变量，否则会报错。这个项目完成后，你应该能解释清楚：为什么PBKDF2比直接哈希更安全？因为迭代次数增加了暴力破解成本。

4. 实操过程全记录与核心环节实现

4.1 从零搭建开发环境：虚拟环境不是可选项

所有项目必须在独立虚拟环境中运行，这是Python工程化的起点。我推荐venv（Python 3.3+内置），而非conda或pipenv，因为最轻量、最标准：

# 创建项目目录 mkdir python-projects && cd python-projects # 创建虚拟环境（命名为venv，这是约定俗成） python -m venv venv # 激活虚拟环境 # Linux/macOS: source venv/bin/activate # Windows: venv\Scripts\activate.bat # 升级pip（重要！旧版pip安装包常失败） pip install --upgrade pip # 安装项目所需库（以项目4为例） pip install requests cryptography pandas matplotlib

实操心得：激活后，命令行提示符前会显示(venv)，这是你的安全区。此时pip list看到的包，只属于这个项目。一旦你deactivate，就回到系统Python。我见过太多人跳过这步，结果全局pip安装了几十个包，版本冲突时哭都来不及。另一个关键是.gitignore文件：在项目根目录创建它，加入以下内容，防止虚拟环境和IDE配置被提交：

venv/ __pycache__/ *.pyc *.pyo *.pyd .Python env/ build/ develop-eggs/ dist/ downloads/ eggs/ .eggs/ lib/ lib64/ parts/ sdist/ var/ *.log .DS_Store

4.2 项目6：文件批量重命名器——pathlib如何拯救你的路径焦虑

项目6要求：将目录下所有.jpg文件重命名为photo_001.jpg,photo_002.jpg...。用os.path的噩梦是：os.path.join(dir, file)在Windows上生成\，在Linux上生成/，而某些老旧库不认反斜杠。pathlib彻底终结这个问题：

from pathlib import Path import re def batch_rename_jpgs(directory: str, prefix: str = "photo"): """批量重命名JPG文件""" dir_path = Path(directory) # 验证目录存在且可读 if not dir_path.exists() or not dir_path.is_dir(): raise ValueError(f"Directory {directory} does not exist or is not a directory") # 获取所有JPG文件，按修改时间排序（最新最先） jpg_files = sorted( list(dir_path.glob("*.jpg")) + list(dir_path.glob("*.JPG")), key=lambda x: x.stat().st_mtime, reverse=True ) # 生成新文件名，确保序号三位数（001, 002...） for i, file_path in enumerate(jpg_files, 1): new_name = f"{prefix}_{i:03d}{file_path.suffix}" new_path = dir_path / new_name # 关键：检查目标文件是否已存在，避免覆盖 if new_path.exists(): print(f"Warning: {new_path} already exists, skipping {file_path.name}") continue try: file_path.rename(new_path) print(f"Renamed: {file_path.name} -> {new_name}") except PermissionError: print(f"Error: Permission denied for {file_path.name}") except Exception as e: print(f"Error renaming {file_path.name}: {e}") # 使用 if __name__ == "__main__": batch_rename_jpgs("/path/to/your/photos", "vacation")

关键技巧：Path.glob("*.jpg")比os.listdir()强大得多，它支持通配符且返回Path对象；file_path.stat().st_mtime获取修改时间戳，比os.path.getmtime()更直观；i:03d是f-string的格式化语法，确保序号为三位数。运行前务必用test目录先试，切勿直接在珍贵照片目录运行。这个项目教会你：pathlib不是语法糖，而是让你忘记路径分隔符的存在。

4.3 项目7：定时数据备份脚本——schedule库的生产级用法

项目7用schedule库（pip install schedule）实现每日凌晨2点备份。但schedule的默认用法在生产环境是危险的——它不处理异常，一旦备份失败，后续任务全停。必须包装：

import schedule import time import shutil import logging from datetime import datetime, timedelta # 配置日志 logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s') def backup_database(source: str, destination: str): """备份数据库文件，带原子性保证""" try: # 生成带时间戳的备份文件名 timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") backup_name = f"backup_{timestamp}.zip" backup_path = Path(destination) / backup_name # 关键：先复制到临时文件，再重命名为最终名，避免备份过程中文件被读取 temp_path = backup_path.with_suffix(".tmp") # 执行压缩（这里用shutil.copy2模拟，实际用zipfile或系统tar） shutil.copy2(source, temp_path) temp_path.rename(backup_path) logging.info(f"Backup successful: {backup_path.name}") # 清理7天前的备份 cleanup_old_backups(destination, days=7) except Exception as e: logging.error(f"Backup failed: {e}") def cleanup_old_backups(backup_dir: str, days: int): """清理指定天数前的备份文件""" backup_path = Path(backup_dir) cutoff_time = datetime.now() - timedelta(days=days) for backup_file in backup_path.glob("backup_*.zip"): try: # 从文件名提取时间戳 timestamp_str = backup_file.stem.split('_')[1] file_time = datetime.strptime(timestamp_str, "%Y%m%d_%H%M%S") if file_time < cutoff_time: backup_file.unlink() logging.info(f"Cleaned old backup: {backup_file.name}") except (ValueError, OSError) as e: logging.warning(f"Failed to process {backup_file.name}: {e}") # 调度任务 schedule.every().day.at("02:00").do(backup_database, source="/path/to/db.sqlite", destination="/path/to/backups") # 主循环（生产环境应使用systemd或cron，此处为演示） if __name__ == "__main__": logging.info("Backup scheduler started") while True: schedule.run_pending() time.sleep(60) # 每分钟检查一次

注意：shutil.copy2()保留文件元数据（如修改时间），比shutil.copy()更专业；temp_path.rename()是原子操作，确保备份文件要么完整，要么不存在；cleanup_old_backups()从文件名解析时间，而不是用stat().st_mtime，因为备份文件的修改时间是创建时间，更可靠。这个脚本上线前，务必在测试环境运行一周，观察日志。

4.4 项目8：轻量Web API服务——Flask路由设计的业务思维

项目8用Flask提供两个端点：GET /api/items返回所有商品，POST /api/items添加新商品。重点不是Flask语法，而是如何设计符合RESTful规范的API：

from flask import Flask, request, jsonify import json from pathlib import Path app = Flask(__name__) DATA_FILE = Path("items.json") # 初始化数据文件 if not DATA_FILE.exists(): DATA_FILE.write_text(json.dumps([], indent=2)) def load_items(): try: return json.loads(DATA_FILE.read_text()) except json.JSONDecodeError: return [] def save_items(items): DATA_FILE.write_text(json.dumps(items, indent=2)) @app.route('/api/items', methods=['GET']) def get_items(): items = load_items() return jsonify({ "success": True, "count": len(items), "data": items }) @app.route('/api/items', methods=['POST']) def add_item(): # 关键：验证请求体是JSON if not request.is_json: return jsonify({"success": False, "error": "Request must be JSON"}), 400 data = request.get_json() # 关键：业务字段校验 required_fields = ['name', 'price'] for field in required_fields: if field not in