Python 从入门到精通

一、环境搭建与基础

1.1 安装 Python

去官网 https://www.python.org 下载安装，勾选 “Add Python to PATH”。

# 验证安装
python --version
pip --version

1.2 第一个程序

print("Hello, Python!")

1.3 变量与数据类型

# 数字
age = 25
price = 19.99
count = 1_000_000  # 可读性写法

# 字符串
name = "张三"
greeting = f"你好，{name}，你今年{age}岁"  # f-string格式化
multiline = """
这是
多行字符串
"""

# 布尔
is_active = True
is_deleted = False

# None
result = None

# 类型转换
num_str = "123"
num = int(num_str)       # 字符串转整数
flt = float("3.14")     # 字符串转浮点
s = str(100)            # 数字转字符串

# 查看类型
print(type(age))        # <class 'int'>
print(isinstance(age, int))  # True

1.4 运算符

# 算术运算
a = 10
b = 3
print(a + b)    # 13
print(a - b)    # 7
print(a * b)    # 30
print(a / b)    # 3.3333 (浮点除法)
print(a // b)   # 3 (整除)
print(a % b)    # 1 (取余)
print(a ** b)   # 1000 (幂运算)

# 比较运算
print(a > b)    # True
print(a == b)   # False
print(a != b)   # True

# 逻辑运算
print(True and False)   # False
print(True or False)    # True
print(not True)         # False

# 三元表达式
x = 10
result = "大" if x > 5 else "小"  # "大"

# 海象运算符 (Python 3.8+)
data = [1, 2, 3, 4, 5, 6, 7, 8]
if (n := len(data)) > 5:
    print(f"数据量太大: {n}")

---

二、数据结构

2.1 列表 (List)

# 创建列表
fruits = ["苹果", "香蕉", "橘子"]
numbers = [1, 2, 3, 4, 5]
mixed = [1, "hello", True, 3.14]

# 访问元素
print(fruits[0])     # "苹果"
print(fruits[-1])    # "橘子" (倒数第一个)

# 切片
print(numbers[1:4])   # [2, 3, 4]
print(numbers[:3])    # [1, 2, 3]
print(numbers[::2])   # [1, 3, 5] (步长为2)
print(numbers[::-1])  # [5, 4, 3, 2, 1] (反转)

# 增删改
fruits.append("葡萄")            # 末尾添加
fruits.insert(1, "芒果")          # 指定位置插入
fruits.extend(["西瓜", "樱桃"])   # 合并列表
fruits.remove("香蕉")            # 删除指定元素
popped = fruits.pop()            # 弹出末尾元素
popped = fruits.pop(0)           # 弹出指定位置元素
fruits[0] = "榴莲"              # 修改元素

# 排序
nums = [3, 1, 4, 1, 5, 9]
nums.sort()                      # 原地排序 [1, 1, 3, 4, 5, 9]
nums.sort(reverse=True)          # 降序排序
sorted_nums = sorted(nums)      # 返回新列表，不改变原列表

# 列表推导式
squares = [x**2 for x in range(10)]              # [0, 1, 4, 9, 16, ...]
evens = [x for x in range(20) if x % 2 == 0]    # [0, 2, 4, 6, ...]
matrix = [[i*3+j for j in range(3)] for i in range(3)]  # 二维列表

# 常用操作
print(len(fruits))       # 长度
print("苹果" in fruits)  # 是否存在
print(numbers.count(1))  # 统计出现次数
print(numbers.index(3))  # 查找索引

# 解包
first, *rest = [1, 2, 3, 4, 5]  # first=1, rest=[2,3,4,5]
a, b, c = [10, 20, 30]          # a=10, b=20, c=30

2.2 元组 (Tuple)

# 不可变序列
point = (3, 4)
single = (42,)   # 单元素元组必须加逗号
x, y = point     # 解包

# 命名元组
from collections import namedtuple
Point = namedtuple("Point", ["x", "y"])
p = Point(3, 4)
print(p.x, p.y)  # 3 4

2.3 字典 (Dict)

# 创建字典
user = {
    "name": "张三",
    "age": 25,
    "email": "zhangsan@example.com"
}

# 访问
print(user["name"])              # "张三"
print(user.get("phone", "无"))  # "无" (带默认值)

# 增删改
user["phone"] = "13800138000"   # 添加/修改
del user["email"]               # 删除
removed = user.pop("age")      # 弹出

# 遍历
for key in user:
    print(key, user[key])

for key, value in user.items():
    print(f"{key}: {value}")

for value in user.values():
    print(value)

# 字典推导式
squares = {x: x**2 for x in range(6)}  # {0:0, 1:1, 2:4, 3:9, 4:16, 5:25}

# 合并字典 (Python 3.9+)
dict1 = {"a": 1, "b": 2}
dict2 = {"b": 3, "c": 4}
merged = dict1 | dict2   # {"a": 1, "b": 3, "c": 4}

# 嵌套字典
students = {
    "001": {"name": "张三", "score": 90},
    "002": {"name": "李四", "score": 85}
}
print(students["001"]["name"])  # "张三"

# defaultdict
from collections import defaultdict
word_count = defaultdict(int)
for word in ["apple", "banana", "apple", "cherry", "banana", "apple"]:
    word_count[word] += 1
# {'apple': 3, 'banana': 2, 'cherry': 1}

# Counter
from collections import Counter
text = "hello world"
counter = Counter(text)  # {'l': 3, 'o': 2, 'h': 1, ...}
print(counter.most_common(3))  # 前3个最常见

2.4 集合 (Set)

# 创建集合（无序、不重复）
colors = {"red", "green", "blue"}
nums = set([1, 2, 2, 3, 3, 3])  # {1, 2, 3}

# 操作
colors.add("yellow")
colors.remove("red")       # 不存在会报错
colors.discard("purple")   # 不存在不报错

# 集合运算
a = {1, 2, 3, 4}
b = {3, 4, 5, 6}
print(a & b)   # 交集 {3, 4}
print(a | b)   # 并集 {1, 2, 3, 4, 5, 6}
print(a - b)   # 差集 {1, 2}
print(a ^ b)   # 对称差集 {1, 2, 5, 6}

# 去重
lst = [1, 2, 2, 3, 3, 3]
unique = list(set(lst))  # [1, 2, 3]

---

三、流程控制

3.1 条件判断

score = 85

if score >= 90:
    grade = "A"
elif score >= 80:
    grade = "B"
elif score >= 70:
    grade = "C"
else:
    grade = "D"

print(f"成绩等级: {grade}")

# match-case (Python 3.10+)
command = "quit"
match command:
    case "start":
        print("启动")
    case "stop" | "quit":
        print("停止")
    case _:
        print("未知命令")

3.2 循环

# for循环
for i in range(5):        # 0,1,2,3,4
    print(i)

for i in range(2, 10, 2): # 2,4,6,8 (起始,结束,步长)
    print(i)

# 带索引遍历
fruits = ["苹果", "香蕉", "橘子"]
for index, fruit in enumerate(fruits):
    print(f"{index}: {fruit}")

# 同时遍历多个列表
names = ["张三", "李四", "王五"]
scores = [90, 85, 92]
for name, score in zip(names, scores):
    print(f"{name}: {score}分")

# while循环
count = 0
while count < 5:
    print(count)
    count += 1

# break 和 continue
for i in range(10):
    if i == 3:
        continue  # 跳过3
    if i == 7:
        break     # 到7停止
    print(i)

# for-else (循环正常结束时执行else)
for n in range(2, 10):
    for x in range(2, n):
        if n % x == 0:
            break
    else:
        print(f"{n} 是质数")

---

四、函数

4.1 基本函数

def greet(name, greeting="你好"):
    """问候函数（这是文档字符串）"""
    return f"{greeting}, {name}!"

print(greet("张三"))              # "你好, 张三!"
print(greet("李四", "早上好"))    # "早上好, 李四!"

4.2 参数类型

# 位置参数、默认参数、关键字参数
def create_user(name, age, city="北京", **kwargs):
    user = {"name": name, "age": age, "city": city}
    user.update(kwargs)
    return user

user = create_user("张三", 25, email="zs@test.com")
# {'name': '张三', 'age': 25, 'city': '北京', 'email': 'zs@test.com'}

# *args 可变位置参数
def sum_all(*args):
    return sum(args)

print(sum_all(1, 2, 3, 4))  # 10

# **kwargs 可变关键字参数
def print_info(**kwargs):
    for key, value in kwargs.items():
        print(f"{key}: {value}")

print_info(name="张三", age=25)

# 强制关键字参数
def connect(host, port, *, timeout=30, retry=3):
    """timeout和retry必须以关键字方式传入"""
    pass

connect("localhost", 8080, timeout=10)

# 仅位置参数 (Python 3.8+)
def divide(a, b, /):
    """a和b只能以位置方式传入"""
    return a / b

4.3 Lambda 表达式

# 匿名函数
square = lambda x: x ** 2
print(square(5))  # 25

# 常用于排序
students = [
    {"name": "张三", "score": 90},
    {"name": "李四", "score": 85},
    {"name": "王五", "score": 92},
]
students.sort(key=lambda s: s["score"], reverse=True)

# 常用于map/filter
numbers = [1, 2, 3, 4, 5, 6, 7, 8, 9, 10]
evens = list(filter(lambda x: x % 2 == 0, numbers))     # [2, 4, 6, 8, 10]
doubled = list(map(lambda x: x * 2, numbers))            # [2, 4, 6, ..., 20]

4.4 闭包与装饰器

# 闭包
def make_counter(start=0):
    count = start
    def counter():
        nonlocal count
        count += 1
        return count
    return counter

c = make_counter()
print(c())  # 1
print(c())  # 2
print(c())  # 3

# 装饰器 —— 不改变函数代码的前提下增强功能
import time
import functools

def timer(func):
    """计时装饰器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 耗时: {end - start:.4f}秒")
        return result
    return wrapper

@timer
def slow_function():
    time.sleep(1)
    return "完成"

slow_function()  # slow_function 耗时: 1.00xx秒

# 带参数的装饰器
def retry(max_attempts=3, delay=1):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for attempt in range(max_attempts):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt == max_attempts - 1:
                        raise
                    print(f"第{attempt+1}次失败: {e}, {delay}秒后重试...")
                    time.sleep(delay)
        return wrapper
    return decorator

@retry(max_attempts=3, delay=2)
def unstable_api_call():
    import random
    if random.random() < 0.7:
        raise ConnectionError("网络超时")
    return "成功"

4.5 生成器

# yield关键字
def fibonacci(n):
    """斐波那契数列生成器"""
    a, b = 0, 1
    for _ in range(n):
        yield a
        a, b = b, a + b

for num in fibonacci(10):
    print(num, end=" ")  # 0 1 1 2 3 5 8 13 21 34

# 生成器表达式（惰性求值，节省内存）
squares_gen = (x**2 for x in range(1000000))  # 不会立即计算
print(next(squares_gen))  # 0
print(next(squares_gen))  # 1

# 实际应用：逐行读取大文件
def read_large_file(file_path):
    with open(file_path, "r", encoding="utf-8") as f:
        for line in f:
            yield line.strip()

# for line in read_large_file("big_data.txt"):
#     process(line)

---

五、面向对象编程 (OOP)

5.1 类的基础

class Dog:
    # 类变量
    species = "犬科"

    def __init__(self, name, age):
        """构造函数"""
        self.name = name      # 实例变量
        self.age = age
        self._health = 100    # 约定私有（单下划线）

    def bark(self):
        return f"{self.name}: 汪汪!"

    def info(self):
        return f"{self.name}, {self.age}岁, 健康值: {self._health}"

    def __str__(self):
        """print时的显示"""
        return f"Dog({self.name})"

    def __repr__(self):
        """调试时的显示"""
        return f"Dog(name='{self.name}', age={self.age})"

dog = Dog("旺财", 3)
print(dog.bark())   # "旺财: 汪汪!"
print(dog)          # "Dog(旺财)"

5.2 继承与多态

class Animal:
    def __init__(self, name):
        self.name = name

    def speak(self):
        raise NotImplementedError("子类必须实现speak方法")

    def __str__(self):
        return f"{self.__class__.__name__}({self.name})"


class Cat(Animal):
    def speak(self):
        return f"{self.name}: 喵~"


class Dog(Animal):
    def speak(self):
        return f"{self.name}: 汪!"


class Duck(Animal):
    def speak(self):
        return f"{self.name}: 嘎嘎!"


# 多态
animals = [Cat("小花"), Dog("旺财"), Duck("唐老鸭")]
for animal in animals:
    print(animal.speak())

# 多重继承
class Flyable:
    def fly(self):
        return f"{self.name} 在飞"

class Swimmable:
    def swim(self):
        return f"{self.name} 在游泳"

class WildDuck(Duck, Flyable, Swimmable):
    pass

wd = WildDuck("野鸭")
print(wd.speak())  # 野鸭: 嘎嘎!
print(wd.fly())    # 野鸭 在飞
print(wd.swim())   # 野鸭 在游泳

5.3 Property 属性

class Circle:
    def __init__(self, radius):
        self._radius = radius

    @property
    def radius(self):
        return self._radius

    @radius.setter
    def radius(self, value):
        if value <= 0:
            raise ValueError("半径必须大于0")
        self._radius = value

    @property
    def area(self):
        """只读计算属性"""
        import math
        return math.pi * self._radius ** 2

c = Circle(5)
print(c.radius)     # 5
print(c.area)       # 78.54...
c.radius = 10       # 调用setter
# c.area = 100      # AttributeError: 只读属性

5.4 魔法方法（常用）

class Vector:
    def __init__(self, x, y):
        self.x = x
        self.y = y

    def __add__(self, other):
        return Vector(self.x + other.x, self.y + other.y)

    def __sub__(self, other):
        return Vector(self.x - other.x, self.y - other.y)

    def __mul__(self, scalar):
        return Vector(self.x * scalar, self.y * scalar)

    def __abs__(self):
        return (self.x**2 + self.y**2) ** 0.5

    def __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __len__(self):
        return 2

    def __getitem__(self, index):
        if index == 0: return self.x
        if index == 1: return self.y
        raise IndexError

    def __repr__(self):
        return f"Vector({self.x}, {self.y})"

v1 = Vector(3, 4)
v2 = Vector(1, 2)
print(v1 + v2)     # Vector(4, 6)
print(abs(v1))     # 5.0
print(v1 * 3)      # Vector(9, 12)
print(v1[0])       # 3

5.5 dataclass (Python 3.7+)

from dataclasses import dataclass, field
from typing import List

@dataclass
class Student:
    name: str
    age: int
    scores: List[int] = field(default_factory=list)

    @property
    def average(self):
        return sum(self.scores) / len(self.scores) if self.scores else 0

    def __post_init__(self):
        if self.age < 0:
            raise ValueError("年龄不能为负")

s = Student("张三", 20, [90, 85, 92])
print(s)           # Student(name='张三', age=20, scores=[90, 85, 92])
print(s.average)   # 89.0

# 自动生成 __init__, __repr__, __eq__
s2 = Student("张三", 20, [90, 85, 92])
print(s == s2)     # True

5.6 抽象类

from abc import ABC, abstractmethod

class Shape(ABC):
    @abstractmethod
    def area(self):
        pass

    @abstractmethod
    def perimeter(self):
        pass

    def description(self):
        return f"{self.__class__.__name__}: 面积={self.area():.2f}, 周长={self.perimeter():.2f}"

class Rectangle(Shape):
    def __init__(self, width, height):
        self.width = width
        self.height = height

    def area(self):
        return self.width * self.height

    def perimeter(self):
        return 2 * (self.width + self.height)

# shape = Shape()  # TypeError: 不能实例化抽象类
rect = Rectangle(5, 3)
print(rect.description())  # "Rectangle: 面积=15.00, 周长=16.00"

---

六、异常处理

# 基本异常处理
def divide(a, b):
    try:
        result = a / b
    except ZeroDivisionError:
        print("除数不能为零!")
        return None
    except TypeError as e:
        print(f"类型错误: {e}")
        return None
    else:
        # 没有异常时执行
        print(f"结果: {result}")
        return result
    finally:
        # 无论如何都执行（清理资源）
        print("计算完成")

divide(10, 2)   # 结果: 5.0 → 计算完成
divide(10, 0)   # 除数不能为零! → 计算完成

# 自定义异常
class ValidationError(Exception):
    def __init__(self, field, message):
        self.field = field
        self.message = message
        super().__init__(f"{field}: {message}")

class User:
    def __init__(self, name, age):
        if not name:
            raise ValidationError("name", "用户名不能为空")
        if age < 0 or age > 150:
            raise ValidationError("age", "年龄不合法")
        self.name = name
        self.age = age

try:
    user = User("", 25)
except ValidationError as e:
    print(f"验证失败 - {e.field}: {e.message}")

# 上下文管理器（自动资源管理）
class DatabaseConnection:
    def __init__(self, host):
        self.host = host

    def __enter__(self):
        print(f"连接到 {self.host}")
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        print(f"断开 {self.host} 连接")
        return False  # 不抑制异常

    def query(self, sql):
        return f"执行: {sql}"

with DatabaseConnection("localhost") as db:
    print(db.query("SELECT * FROM users"))
# 自动调用 __exit__

---

七、文件操作

7.1 文本文件

# 写入文件
with open("output.txt", "w", encoding="utf-8") as f:
    f.write("第一行\n")
    f.write("第二行\n")
    f.writelines(["第三行\n", "第四行\n"])

# 读取文件
with open("output.txt", "r", encoding="utf-8") as f:
    content = f.read()          # 读取全部内容
    # lines = f.readlines()     # 读取所有行(列表)
    # line = f.readline()       # 读取一行

# 逐行读取(内存友好)
with open("output.txt", "r", encoding="utf-8") as f:
    for line in f:
        print(line.strip())

# 追加模式
with open("output.txt", "a", encoding="utf-8") as f:
    f.write("追加的内容\n")

7.2 JSON 操作

import json

# Python对象 → JSON字符串
data = {
    "name": "张三",
    "age": 25,
    "hobbies": ["编程", "阅读"],
    "address": {"city": "北京", "zipcode": "100000"}
}

json_str = json.dumps(data, ensure_ascii=False, indent=2)
print(json_str)

# JSON字符串 → Python对象
parsed = json.loads(json_str)
print(parsed["name"])  # "张三"

# 写入JSON文件
with open("data.json", "w", encoding="utf-8") as f:
    json.dump(data, f, ensure_ascii=False, indent=2)

# 读取JSON文件
with open("data.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)

7.3 CSV 操作

import csv

# 写入CSV
headers = ["姓名", "年龄", "城市"]
rows = [
    ["张三", 25, "北京"],
    ["李四", 30, "上海"],
    ["王五", 28, "广州"],
]

with open("people.csv", "w", newline="", encoding="utf-8-sig") as f:
    writer = csv.writer(f)
    writer.writerow(headers)
    writer.writerows(rows)

# 读取CSV
with open("people.csv", "r", encoding="utf-8-sig") as f:
    reader = csv.DictReader(f)
    for row in reader:
        print(f"{row['姓名']}, {row['年龄']}岁, 来自{row['城市']}")

7.4 路径操作 (pathlib)

from pathlib import Path

# 路径操作
current = Path.cwd()           # 当前目录
home = Path.home()             # 用户主目录
file_path = Path("data") / "output" / "result.txt"

# 路径信息
p = Path("/home/user/documents/report.pdf")
print(p.name)       # "report.pdf"
print(p.stem)       # "report"
print(p.suffix)     # ".pdf"
print(p.parent)     # "/home/user/documents"
print(p.exists())   # 是否存在

# 遍历目录
for file in Path(".").glob("*.py"):       # 当前目录所有.py文件
    print(file)

for file in Path(".").rglob("*.txt"):     # 递归查找所有.txt
    print(file)

# 创建目录
Path("output/data").mkdir(parents=True, exist_ok=True)

# 读写文件
Path("hello.txt").write_text("你好世界", encoding="utf-8")
content = Path("hello.txt").read_text(encoding="utf-8")

---

八、模块与包

8.1 模块导入

# 标准导入
import os
import sys
from datetime import datetime, timedelta
from collections import defaultdict, Counter
from typing import List, Dict, Optional

# 别名
import numpy as np
import pandas as pd

# 导入自己的模块
# 假设项目结构:
# project/
#   main.py
#   utils/
#     __init__.py
#     helper.py
#     validator.py

# main.py中:
from utils.helper import format_date
from utils import validator

8.2 创建自己的包

# utils/__init__.py
from .helper import format_date, parse_date
from .validator import validate_email

__all__ = ["format_date", "parse_date", "validate_email"]

# utils/helper.py
from datetime import datetime

def format_date(dt: datetime, fmt: str = "%Y-%m-%d") -> str:
    return dt.strftime(fmt)

def parse_date(date_str: str, fmt: str = "%Y-%m-%d") -> datetime:
    return datetime.strptime(date_str, fmt)

# utils/validator.py
import re

def validate_email(email: str) -> bool:
    pattern = r'^[\w\.-]+@[\w\.-]+\.\w+$'
    return bool(re.match(pattern, email))

---

九、常用标准库

9.1 datetime 时间处理

from datetime import datetime, timedelta, date

# 当前时间
now = datetime.now()
today = date.today()
print(now.strftime("%Y-%m-%d %H:%M:%S"))  # "2026-07-31 10:30:00"

# 时间计算
tomorrow = now + timedelta(days=1)
next_week = now + timedelta(weeks=1)
diff = datetime(2026, 12, 31) - now
print(f"距离年底还有 {diff.days} 天")

# 字符串解析
dt = datetime.strptime("2026-07-31 10:30:00", "%Y-%m-%d %H:%M:%S")

# 时间戳
timestamp = now.timestamp()
dt_from_ts = datetime.fromtimestamp(timestamp)

9.2 os 与 subprocess

import os
import subprocess

# 环境变量
home = os.environ.get("HOME", os.environ.get("USERPROFILE"))
os.environ["MY_VAR"] = "hello"

# 执行系统命令
result = subprocess.run(
    ["python", "--version"],
    capture_output=True,
    text=True,
    timeout=10
)
print(result.stdout)  # "Python 3.x.x"
print(result.returncode)  # 0表示成功

9.3 re 正则表达式

import re

text = "我的邮箱是 test@example.com，电话是 13800138000"

# 查找
email = re.search(r'[\w\.-]+@[\w\.-]+\.\w+', text)
if email:
    print(email.group())  # "test@example.com"

# 查找所有
phones = re.findall(r'1[3-9]\d{9}', text)
print(phones)  # ["13800138000"]

# 替换
cleaned = re.sub(r'\d+', '***', text)
print(cleaned)  # "我的邮箱是 test@example.com，电话是 ***"

# 分割
parts = re.split(r'[,;，；\s]+', "张三,李四；王五 赵六")
print(parts)  # ['张三', '李四', '王五', '赵六']

# 编译正则（多次使用时提升性能）
pattern = re.compile(r'^(\d{4})-(\d{2})-(\d{2})$')
match = pattern.match("2026-07-31")
if match:
    year, month, day = match.groups()
    print(f"{year}年{month}月{day}日")

# 常用正则
patterns = {
    "手机号": r'^1[3-9]\d{9}$',
    "邮箱": r'^[\w\.-]+@[\w\.-]+\.\w+$',
    "身份证": r'^\d{17}[\dXx]$',
    "URL": r'https?://[\w\-\.]+(:\d+)?(/[\w\-\./?%&=]*)?',
    "中文": r'[\u4e00-\u9fa5]+',
    "IP地址": r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}',
}

9.4 itertools 迭代器工具

import itertools

# 排列组合
items = ['A', 'B', 'C']
print(list(itertools.permutations(items, 2)))    # 排列
# [('A','B'), ('A','C'), ('B','A'), ('B','C'), ('C','A'), ('C','B')]
print(list(itertools.combinations(items, 2)))    # 组合
# [('A','B'), ('A','C'), ('B','C')]

# 无限计数器
counter = itertools.count(start=1, step=2)  # 1, 3, 5, 7, ...

# 循环迭代
colors = itertools.cycle(["红", "绿", "蓝"])
for _, color in zip(range(6), colors):
    print(color, end=" ")  # 红 绿 蓝 红 绿 蓝

# 链接多个可迭代对象
combined = itertools.chain([1,2,3], [4,5,6], [7,8,9])
print(list(combined))  # [1,2,3,4,5,6,7,8,9]

# 分组
data = [
    {"city": "北京", "name": "张三"},
    {"city": "北京", "name": "李四"},
    {"city": "上海", "name": "王五"},
    {"city": "上海", "name": "赵六"},
]
data.sort(key=lambda x: x["city"])  # 先排序
for city, group in itertools.groupby(data, key=lambda x: x["city"]):
    print(f"{city}: {[p['name'] for p in group]}")

9.5 functools 函数工具

from functools import lru_cache, partial, reduce

# 缓存（记忆化）
@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

print(fibonacci(100))  # 瞬间计算出结果

# 偏函数
def power(base, exponent):
    return base ** exponent

square = partial(power, exponent=2)
cube = partial(power, exponent=3)
print(square(5))  # 25
print(cube(3))    # 27

# reduce
numbers = [1, 2, 3, 4, 5]
product = reduce(lambda x, y: x * y, numbers)  # 120

---

十、并发编程

10.1 多线程 (I/O密集型)

import threading
import time
from concurrent.futures import ThreadPoolExecutor

# 基本多线程
def download(url):
    print(f"开始下载: {url}")
    time.sleep(2)  # 模拟IO操作
    print(f"下载完成: {url}")
    return f"{url} 的内容"

# 线程池（推荐方式）
urls = [f"https://example.com/page/{i}" for i in range(5)]

with ThreadPoolExecutor(max_workers=3) as executor:
    futures = [executor.submit(download, url) for url in urls]
    results = [f.result() for f in futures]
    print(results)

# 线程锁
counter = 0
lock = threading.Lock()

def increment():
    global counter
    for _ in range(100000):
        with lock:
            counter += 1

threads = [threading.Thread(target=increment) for _ in range(5)]
for t in threads:
    t.start()
for t in threads:
    t.join()
print(counter)  # 500000

10.2 多进程 (CPU密集型)

from concurrent.futures import ProcessPoolExecutor
import multiprocessing

def cpu_heavy_task(n):
    """CPU密集计算"""
    return sum(i*i for i in range(n))

# 进程池
if __name__ == "__main__":
    numbers = [10_000_000] * 4

    with ProcessPoolExecutor(max_workers=4) as executor:
        results = list(executor.map(cpu_heavy_task, numbers))
    print(results)

10.3 异步编程 (asyncio)

import asyncio
import aiohttp  # pip install aiohttp

async def fetch_url(session, url):
    """异步HTTP请求"""
    async with session.get(url) as response:
        return await response.text()

async def main():
    urls = [
        "https://httpbin.org/get",
        "https://httpbin.org/ip",
        "https://httpbin.org/user-agent",
    ]

    async with aiohttp.ClientSession() as session:
        tasks = [fetch_url(session, url) for url in urls]
        results = await asyncio.gather(*tasks)
        for url, result in zip(urls, results):
            print(f"{url}: {len(result)} bytes")

# asyncio.run(main())

# 异步生成器
async def async_range(n):
    for i in range(n):
        await asyncio.sleep(0.1)
        yield i

async def consume():
    async for num in async_range(5):
        print(num)

# 信号量控制并发数
async def limited_fetch(sem, session, url):
    async with sem:
        return await fetch_url(session, url)

async def main_limited():
    sem = asyncio.Semaphore(10)  # 最多10个并发
    async with aiohttp.ClientSession() as session:
        urls = [f"https://example.com/{i}" for i in range(100)]
        tasks = [limited_fetch(sem, session, url) for url in urls]
        results = await asyncio.gather(*tasks, return_exceptions=True)

---

十一、网络请求

11.1 requests 库

import requests

# GET请求
response = requests.get("https://api.github.com/users/octocat")
print(response.status_code)  # 200
print(response.json())       # 解析JSON
print(response.headers)      # 响应头

# 带参数的GET
params = {"q": "python", "page": 1}
response = requests.get("https://api.github.com/search/repositories", params=params)

# POST请求
data = {"username": "test", "password": "123456"}
response = requests.post("https://httpbin.org/post", json=data)

# 表单提交
response = requests.post("https://httpbin.org/post", data={"key": "value"})

# 上传文件
files = {"file": open("report.pdf", "rb")}
response = requests.post("https://httpbin.org/post", files=files)

# 设置请求头
headers = {
    "Authorization": "Bearer your-token-here",
    "Content-Type": "application/json"
}
response = requests.get("https://api.example.com/data", headers=headers)

# 超时与重试
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

session = requests.Session()
retry = Retry(total=3, backoff_factor=1, status_forcelist=[500, 502, 503])
adapter = HTTPAdapter(max_retries=retry)
session.mount("https://", adapter)
session.mount("http://", adapter)

response = session.get("https://api.example.com", timeout=10)

# 下载大文件
def download_file(url, filename):
    with requests.get(url, stream=True) as r:
        r.raise_for_status()
        with open(filename, "wb") as f:
            for chunk in r.iter_content(chunk_size=8192):
                f.write(chunk)

---

十二、数据库操作

12.1 SQLite（内置）

import sqlite3

# 连接数据库（不存在则创建）
conn = sqlite3.connect("app.db")
conn.row_factory = sqlite3.Row  # 结果可以通过列名访问
cursor = conn.cursor()

# 创建表
cursor.execute("""
    CREATE TABLE IF NOT EXISTS users (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        name TEXT NOT NULL,
        email TEXT UNIQUE,
        age INTEGER,
        created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP
    )
""")

# 插入数据
cursor.execute(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    ("张三", "zhangsan@test.com", 25)
)

# 批量插入
users = [
    ("李四", "lisi@test.com", 30),
    ("王五", "wangwu@test.com", 28),
]
cursor.executemany(
    "INSERT INTO users (name, email, age) VALUES (?, ?, ?)",
    users
)
conn.commit()

# 查询
cursor.execute("SELECT * FROM users WHERE age > ?", (26,))
rows = cursor.fetchall()
for row in rows:
    print(f"{row['name']}, {row['age']}岁")

# 更新
cursor.execute("UPDATE users SET age = ? WHERE name = ?", (26, "张三"))
conn.commit()

# 删除
cursor.execute("DELETE FROM users WHERE name = ?", ("王五",))
conn.commit()

# 关闭连接
conn.close()

# 推荐用with自动管理
with sqlite3.connect("app.db") as conn:
    conn.execute("SELECT * FROM users")

12.2 SQLAlchemy ORM

from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey
from sqlalchemy.orm import declarative_base, sessionmaker, relationship
from datetime import datetime

# 创建引擎
engine = create_engine("sqlite:///app.db", echo=False)
Base = declarative_base()
Session = sessionmaker(bind=engine)

# 定义模型
class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String(50), nullable=False)
    email = Column(String(120), unique=True)
    created_at = Column(DateTime, default=datetime.now)
    posts = relationship("Post", back_populates="author")

    def __repr__(self):
        return f"<User(name='{self.name}')>"

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True)
    title = Column(String(200), nullable=False)
    content = Column(String)
    author_id = Column(Integer, ForeignKey("users.id"))
    author = relationship("User", back_populates="posts")

# 创建表
Base.metadata.create_all(engine)

# CRUD操作
session = Session()

# 创建
user = User(name="张三", email="zs@test.com")
session.add(user)
session.commit()

# 查询
users = session.query(User).filter(User.name.like("%张%")).all()
user = session.query(User).filter_by(email="zs@test.com").first()

# 更新
user.name = "张三丰"
session.commit()

# 删除
session.delete(user)
session.commit()

session.close()

---

十三、Web开发 (Flask)

13.1 基本Flask应用

from flask import Flask, request, jsonify, render_template

app = Flask(__name__)

# 路由
@app.route("/")
def index():
    return "Hello, Flask!"

# API接口
@app.route("/api/users", methods=["GET"])
def get_users():
    users = [
        {"id": 1, "name": "张三", "age": 25},
        {"id": 2, "name": "李四", "age": 30},
    ]
    return jsonify({"code": 0, "data": users})

@app.route("/api/users", methods=["POST"])
def create_user():
    data = request.get_json()
    name = data.get("name")
    age = data.get("age")
    if not name:
        return jsonify({"code": 1, "message": "name不能为空"}), 400
    # 保存到数据库...
    return jsonify({"code": 0, "message": "创建成功"}), 201

@app.route("/api/users/<int:user_id>", methods=["GET"])
def get_user(user_id):
    # 从数据库获取...
    return jsonify({"id": user_id, "name": "张三"})

# 错误处理
@app.errorhandler(404)
def not_found(e):
    return jsonify({"code": 404, "message": "资源不存在"}), 404

@app.errorhandler(500)
def server_error(e):
    return jsonify({"code": 500, "message": "服务器错误"}), 500

if __name__ == "__main__":
    app.run(debug=True, port=5000)

13.2 FastAPI（推荐，现代异步框架）

from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime

app = FastAPI(title="用户管理API", version="1.0")

# 数据模型
class UserCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=50, description="用户名")
    email: str = Field(..., description="邮箱")
    age: int = Field(..., ge=0, le=150, description="年龄")

class UserResponse(BaseModel):
    id: int
    name: str
    email: str
    age: int
    created_at: datetime

# 模拟数据库
users_db = []

@app.get("/api/users", response_model=List[UserResponse])
async def get_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    """获取用户列表"""
    return users_db[skip:skip+limit]

@app.post("/api/users", response_model=UserResponse, status_code=201)
async def create_user(user: UserCreate):
    """创建用户"""
    new_user = {
        "id": len(users_db) + 1,
        "created_at": datetime.now(),
        **user.dict()
    }
    users_db.append(new_user)
    return new_user

@app.get("/api/users/{user_id}", response_model=UserResponse)
async def get_user(user_id: int = Path(..., ge=1)):
    """获取指定用户"""
    for user in users_db:
        if user["id"] == user_id:
            return user
    raise HTTPException(status_code=404, detail="用户不存在")

@app.delete("/api/users/{user_id}")
async def delete_user(user_id: int):
    """删除用户"""
    for i, user in enumerate(users_db):
        if user["id"] == user_id:
            users_db.pop(i)
            return {"message": "删除成功"}
    raise HTTPException(status_code=404, detail="用户不存在")

# 运行: uvicorn main:app --reload
# 自动文档: http://localhost:8000/docs

---

十四、数据处理 (Pandas)

import pandas as pd
import numpy as np

# 创建DataFrame
df = pd.DataFrame({
    "姓名": ["张三", "李四", "王五", "赵六", "钱七"],
    "年龄": [25, 30, 28, 35, 22],
    "城市": ["北京", "上海", "北京", "广州", "上海"],
    "薪资": [15000, 20000, 18000, 25000, 12000],
    "入职日期": pd.to_datetime(["2020-01", "2019-06", "2021-03", "2018-09", "2022-01"])
})

# 基本信息
print(df.info())        # 类型、非空数量
print(df.describe())    # 数值统计
print(df.head(3))       # 前3行
print(df.shape)         # (5, 5)

# 列操作
print(df["姓名"])                # 选择列
print(df[["姓名", "薪资"]])      # 多列
df["年薪"] = df["薪资"] * 12    # 新增列
df.drop(columns=["年薪"], inplace=True)  # 删除列

# 行筛选
high_salary = df[df["薪资"] > 15000]
beijing = df[df["城市"] == "北京"]
combined = df[(df["年龄"] > 25) & (df["城市"] == "上海")]

# 排序
df_sorted = df.sort_values("薪资", ascending=False)

# 分组聚合
city_stats = df.groupby("城市").agg(
    平均薪资=("薪资", "mean"),
    人数=("姓名", "count"),
    最高薪资=("薪资", "max")
).reset_index()
print(city_stats)

# 缺失值处理
df2 = df.copy()
df2.loc[0, "薪资"] = np.nan
df2["薪资"].fillna(df2["薪资"].mean(), inplace=True)  # 均值填充
df2.dropna(subset=["薪资"], inplace=True)             # 删除缺失行

# 读写Excel/CSV
df.to_csv("output.csv", index=False, encoding="utf-8-sig")
df.to_excel("output.xlsx", index=False)
df_from_csv = pd.read_csv("output.csv")
df_from_excel = pd.read_excel("output.xlsx")

# 数据透视表
pivot = df.pivot_table(
    values="薪资",
    index="城市",
    aggfunc=["mean", "count", "max"]
)

# 合并数据
df_a = pd.DataFrame({"id": [1,2,3], "name": ["A","B","C"]})
df_b = pd.DataFrame({"id": [2,3,4], "score": [90,85,70]})
merged = pd.merge(df_a, df_b, on="id", how="left")  # 左连接

# apply自定义函数
df["薪资等级"] = df["薪资"].apply(
    lambda x: "高" if x > 20000 else ("中" if x > 15000 else "低")
)

---

十五、类型提示 (Type Hints)

from typing import (
    List, Dict, Tuple, Set, Optional,
    Union, Any, Callable, TypeVar, Generic
)
from dataclasses import dataclass

# 基本类型提示
def greet(name: str) -> str:
    return f"Hello, {name}!"

def add(a: int, b: int) -> int:
    return a + b

# 容器类型
def process_items(items: List[str]) -> Dict[str, int]:
    return {item: len(item) for item in items}

# Optional (可能为None)
def find_user(user_id: int) -> Optional[Dict[str, Any]]:
    """返回用户或None"""
    users = {1: {"name": "张三"}}
    return users.get(user_id)

# Union (多种类型)
def parse_input(value: Union[str, int]) -> int:
    if isinstance(value, str):
        return int(value)
    return value

# Callable (函数类型)
def apply_operation(
    numbers: List[int],
    operation: Callable[[int], int]
) -> List[int]:
    return [operation(n) for n in numbers]

result = apply_operation([1, 2, 3], lambda x: x * 2)  # [2, 4, 6]

# TypeVar 和 Generic
T = TypeVar("T")

class Stack(Generic[T]):
    def __init__(self):
        self._items: List[T] = []

    def push(self, item: T) -> None:
        self._items.append(item)

    def pop(self) -> T:
        return self._items.pop()

    def is_empty(self) -> bool:
        return len(self._items) == 0

# 使用
int_stack: Stack[int] = Stack()
int_stack.push(1)
int_stack.push(2)
print(int_stack.pop())  # 2

# Python 3.10+ 新语法
def process(value: str | int | None) -> str:
    if value is None:
        return "空"
    return str(value)

---

十六、测试

16.1 unittest

import unittest

# 被测试的代码
class Calculator:
    def add(self, a, b):
        return a + b

    def divide(self, a, b):
        if b == 0:
            raise ValueError("除数不能为零")
        return a / b

# 测试类
class TestCalculator(unittest.TestCase):
    def setUp(self):
        """每个测试方法前运行"""
        self.calc = Calculator()

    def test_add(self):
        self.assertEqual(self.calc.add(2, 3), 5)
        self.assertEqual(self.calc.add(-1, 1), 0)

    def test_divide(self):
        self.assertAlmostEqual(self.calc.divide(10, 3), 3.333, places=3)

    def test_divide_by_zero(self):
        with self.assertRaises(ValueError):
            self.calc.divide(10, 0)

if __name__ == "__main__":
    unittest.main()

16.2 pytest（推荐）

# pip install pytest
# 文件名: test_calculator.py

import pytest

def add(a, b):
    return a + b

def divide(a, b):
    if b == 0:
        raise ValueError("除数不能为零")
    return a / b

# 基本测试
def test_add():
    assert add(2, 3) == 5
    assert add(-1, 1) == 0

def test_divide():
    assert divide(10, 2) == 5.0

def test_divide_by_zero():
    with pytest.raises(ValueError, match="除数不能为零"):
        divide(10, 0)

# 参数化测试
@pytest.mark.parametrize("a,b,expected", [
    (1, 1, 2),
    (2, 3, 5),
    (-1, 1, 0),
    (0, 0, 0),
])
def test_add_params(a, b, expected):
    assert add(a, b) == expected

# fixture
@pytest.fixture
def sample_data():
    return {"users": [{"name": "张三", "age": 25}]}

def test_with_fixture(sample_data):
    assert len(sample_data["users"]) == 1
    assert sample_data["users"][0]["name"] == "张三"

# 运行: pytest test_calculator.py -v

---

十七、日志

import logging
from logging.handlers import RotatingFileHandler

# 基本配置
logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
    datefmt="%Y-%m-%d %H:%M:%S"
)

logger = logging.getLogger(__name__)

# 使用
logger.debug("调试信息")
logger.info("操作成功")
logger.warning("注意：磁盘空间不足")
logger.error("连接失败")
logger.exception("异常信息（自动附带堆栈）")

# 完整配置（文件+控制台）
def setup_logger(name, log_file, level=logging.INFO):
    logger = logging.getLogger(name)
    logger.setLevel(level)

    # 控制台处理器
    console_handler = logging.StreamHandler()
    console_handler.setLevel(logging.DEBUG)
    console_format = logging.Formatter("%(asctime)s [%(levelname)s] %(message)s")
    console_handler.setFormatter(console_format)

    # 文件处理器（自动轮转）
    file_handler = RotatingFileHandler(
        log_file, maxBytes=10*1024*1024, backupCount=5, encoding="utf-8"
    )
    file_handler.setLevel(logging.INFO)
    file_format = logging.Formatter(
        "%(asctime)s [%(levelname)s] %(name)s:%(lineno)d - %(message)s"
    )
    file_handler.setFormatter(file_format)

    logger.addHandler(console_handler)
    logger.addHandler(file_handler)
    return logger

app_logger = setup_logger("myapp", "app.log")
app_logger.info("应用启动")

---

十八、实用技巧与设计模式

18.1 单例模式

class Singleton:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            cls._instance = super().__new__(cls)
        return cls._instance

    def __init__(self, value=None):
        if not hasattr(self, '_initialized'):
            self.value = value
            self._initialized = True

s1 = Singleton("hello")
s2 = Singleton("world")
print(s1 is s2)      # True
print(s1.value)      # "hello" (不会被覆盖)

18.2 工厂模式

class Notification:
    def send(self, message):
        raise NotImplementedError

class EmailNotification(Notification):
    def send(self, message):
        print(f"[Email] {message}")

class SMSNotification(Notification):
    def send(self, message):
        print(f"[SMS] {message}")

class WeChatNotification(Notification):
    def send(self, message):
        print(f"[WeChat] {message}")

class NotificationFactory:
    _creators = {
        "email": EmailNotification,
        "sms": SMSNotification,
        "wechat": WeChatNotification,
    }

    @classmethod
    def create(cls, channel: str) -> Notification:
        creator = cls._creators.get(channel)
        if not creator:
            raise ValueError(f"不支持的渠道: {channel}")
        return creator()

# 使用
notifier = NotificationFactory.create("email")
notifier.send("你有新消息")

18.3 观察者模式

from typing import List, Callable

class EventEmitter:
    def __init__(self):
        self._listeners: dict[str, List[Callable]] = {}

    def on(self, event: str, callback: Callable):
        if event not in self._listeners:
            self._listeners[event] = []
        self._listeners[event].append(callback)

    def emit(self, event: str, *args, **kwargs):
        for callback in self._listeners.get(event, []):
            callback(*args, **kwargs)

    def off(self, event: str, callback: Callable):
        if event in self._listeners:
            self._listeners[event].remove(callback)

# 使用
emitter = EventEmitter()

def on_user_created(user):
    print(f"发送欢迎邮件给 {user['name']}")

def on_user_created_log(user):
    print(f"记录日志: 新用户 {user['name']} 注册")

emitter.on("user_created", on_user_created)
emitter.on("user_created", on_user_created_log)
emitter.emit("user_created", {"name": "张三", "email": "zs@test.com"})

18.4 上下文管理器

from contextlib import contextmanager
import time

@contextmanager
def timer(label="代码块"):
    """计时上下文管理器"""
    start = time.time()
    yield
    elapsed = time.time() - start
    print(f"{label} 耗时: {elapsed:.4f}秒")

with timer("数据处理"):
    total = sum(range(1000000))

@contextmanager
def temp_file(filename, mode="w"):
    """临时文件，退出时自动删除"""
    import os
    try:
        f = open(filename, mode)
        yield f
    finally:
        f.close()
        os.remove(filename)

with temp_file("tmp.txt") as f:
    f.write("临时数据")
# 文件已自动删除

18.5 配置管理

import os
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class DatabaseConfig:
    host: str = "localhost"
    port: int = 5432
    name: str = "mydb"
    user: str = "admin"
    password: str = ""

@dataclass
class AppConfig:
    debug: bool = False
    secret_key: str = "change-me"
    database: DatabaseConfig = field(default_factory=DatabaseConfig)

    @classmethod
    def from_env(cls):
        """从环境变量加载配置"""
        return cls(
            debug=os.getenv("DEBUG", "false").lower() == "true",
            secret_key=os.getenv("SECRET_KEY", "change-me"),
            database=DatabaseConfig(
                host=os.getenv("DB_HOST", "localhost"),
                port=int(os.getenv("DB_PORT", "5432")),
                name=os.getenv("DB_NAME", "mydb"),
                user=os.getenv("DB_USER", "admin"),
                password=os.getenv("DB_PASSWORD", ""),
            )
        )

config = AppConfig.from_env()
print(config.database.host)

---

十九、常用第三方库速查

19.1 数据验证 (Pydantic)

from pydantic import BaseModel, Field, validator, EmailStr
from typing import Optional, List
from datetime import datetime

class Address(BaseModel):
    city: str
    street: str
    zipcode: str = Field(..., regex=r'^\d{6}$')

class User(BaseModel):
    name: str = Field(..., min_length=2, max_length=50)
    email: EmailStr
    age: int = Field(..., ge=0, le=150)
    address: Optional[Address] = None
    tags: List[str] = []
    created_at: datetime = Field(default_factory=datetime.now)

    @validator("name")
    def name_must_not_be_empty(cls, v):
        if not v.strip():
            raise ValueError("姓名不能为空白")
        return v.strip()

# 使用
try:
    user = User(
        name="张三",
        email="zs@example.com",
        age=25,
        address={"city": "北京", "street": "朝阳路", "zipcode": "100000"}
    )
    print(user.dict())
    print(user.json(ensure_ascii=False))
except Exception as e:
    print(f"验证失败: {e}")

19.2 环境变量 (python-dotenv)

# .env 文件
# DATABASE_URL=postgresql://user:pass@localhost/db
# SECRET_KEY=my-secret-key
# DEBUG=true

from dotenv import load_dotenv
import os

load_dotenv()  # 加载.env文件

database_url = os.getenv("DATABASE_URL")
secret_key = os.getenv("SECRET_KEY")
debug = os.getenv("DEBUG", "false").lower() == "true"

19.3 定时任务 (schedule)

import schedule
import time

def job():
    print("执行定时任务...")

# 定时规则
schedule.every(10).seconds.do(job)
schedule.every().hour.do(job)
schedule.every().day.at("10:30").do(job)
schedule.every().monday.do(job)

# 运行调度器
while True:
    schedule.run_pending()
    time.sleep(1)

19.4 命令行工具 (Click)

import click

@click.group()
def cli():
    """我的命令行工具"""
    pass

@cli.command()
@click.option("--name", "-n", required=True, help="用户名")
@click.option("--age", "-a", type=int, default=18, help="年龄")
@click.option("--verbose", "-v", is_flag=True, help="详细输出")
def create(name, age, verbose):
    """创建用户"""
    click.echo(f"创建用户: {name}, {age}岁")
    if verbose:
        click.echo("详细模式已开启")

@cli.command()
@click.argument("user_id", type=int)
def delete(user_id):
    """删除用户"""
    if click.confirm(f"确认删除用户 {user_id}?"):
        click.echo(f"已删除用户 {user_id}")

if __name__ == "__main__":
    cli()

# 使用: python tool.py create -n 张三 -a 25 -v
#       python tool.py delete 1

---

二十、项目实战：完整的 TODO API

"""
完整的 TODO 应用 API
运行: pip install fastapi uvicorn sqlalchemy
      uvicorn todo_app:app --reload
"""
from fastapi import FastAPI, HTTPException, Depends, Query
from pydantic import BaseModel, Field
from typing import Optional, List
from datetime import datetime
from enum import Enum
import sqlite3
from contextlib import contextmanager

# ===== 数据库层 =====
DATABASE = "todos.db"

def init_db():
    with get_db() as db:
        db.execute("""
            CREATE TABLE IF NOT EXISTS todos (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                title TEXT NOT NULL,
                description TEXT DEFAULT '',
                status TEXT DEFAULT 'pending',
                priority INTEGER DEFAULT 0,
                created_at TEXT DEFAULT CURRENT_TIMESTAMP,
                updated_at TEXT DEFAULT CURRENT_TIMESTAMP
            )
        """)

@contextmanager
def get_db():
    conn = sqlite3.connect(DATABASE)
    conn.row_factory = sqlite3.Row
    try:
        yield conn
        conn.commit()
    finally:
        conn.close()

# ===== 数据模型 =====
class TodoStatus(str, Enum):
    pending = "pending"
    in_progress = "in_progress"
    done = "done"

class TodoCreate(BaseModel):
    title: str = Field(..., min_length=1, max_length=200)
    description: str = ""
    priority: int = Field(0, ge=0, le=5)

class TodoUpdate(BaseModel):
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    description: Optional[str] = None
    status: Optional[TodoStatus] = None
    priority: Optional[int] = Field(None, ge=0, le=5)

class TodoResponse(BaseModel):
    id: int
    title: str
    description: str
    status: str
    priority: int
    created_at: str
    updated_at: str

# ===== API =====
app = FastAPI(title="TODO API", version="1.0")

@app.on_event("startup")
def startup():
    init_db()

@app.get("/api/todos", response_model=List[TodoResponse])
def list_todos(
    status: Optional[TodoStatus] = None,
    skip: int = Query(0, ge=0),
    limit: int = Query(20, ge=1, le=100),
):
    """获取待办列表"""
    with get_db() as db:
        query = "SELECT * FROM todos"
        params = []
        if status:
            query += " WHERE status = ?"
            params.append(status.value)
        query += " ORDER BY priority DESC, created_at DESC LIMIT ? OFFSET ?"
        params.extend([limit, skip])
        rows = db.execute(query, params).fetchall()
        return [dict(row) for row in rows]

@app.post("/api/todos", response_model=TodoResponse, status_code=201)
def create_todo(todo: TodoCreate):
    """创建待办"""
    with get_db() as db:
        cursor = db.execute(
            "INSERT INTO todos (title, description, priority) VALUES (?, ?, ?)",
            (todo.title, todo.description, todo.priority)
        )
        row = db.execute("SELECT * FROM todos WHERE id = ?", (cursor.lastrowid,)).fetchone()
        return dict(row)

@app.get("/api/todos/{todo_id}", response_model=TodoResponse)
def get_todo(todo_id: int):
    """获取单个待办"""
    with get_db() as db:
        row = db.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
        if not row:
            raise HTTPException(status_code=404, detail="待办不存在")
        return dict(row)

@app.patch("/api/todos/{todo_id}", response_model=TodoResponse)
def update_todo(todo_id: int, todo: TodoUpdate):
    """更新待办"""
    with get_db() as db:
        existing = db.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
        if not existing:
            raise HTTPException(status_code=404, detail="待办不存在")

        updates = []
        params = []
        for field, value in todo.dict(exclude_unset=True).items():
            updates.append(f"{field} = ?")
            params.append(value)

        if updates:
            updates.append("updated_at = ?")
            params.append(datetime.now().isoformat())
            params.append(todo_id)
            db.execute(
                f"UPDATE todos SET {', '.join(updates)} WHERE id = ?",
                params
            )

        row = db.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
        return dict(row)

@app.delete("/api/todos/{todo_id}")
def delete_todo(todo_id: int):
    """删除待办"""
    with get_db() as db:
        existing = db.execute("SELECT * FROM todos WHERE id = ?", (todo_id,)).fetchone()
        if not existing:
            raise HTTPException(status_code=404, detail="待办不存在")
        db.execute("DELETE FROM todos WHERE id = ?", (todo_id,))
        return {"message": "删除成功"}

@app.get("/api/stats")
def get_stats():
    """统计信息"""
    with get_db() as db:
        total = db.execute("SELECT COUNT(*) as c FROM todos").fetchone()["c"]
        by_status = db.execute(
            "SELECT status, COUNT(*) as count FROM todos GROUP BY status"
        ).fetchall()
        return {
            "total": total,
            "by_status": {row["status"]: row["count"] for row in by_status}
        }

---

二十一、虚拟环境与依赖管理

# 创建虚拟环境
python -m venv venv

# 激活（Windows）
venv\Scripts\activate

# 激活（Mac/Linux）
source venv/bin/activate

# 安装依赖
pip install requests flask pandas

# 导出依赖
pip freeze > requirements.txt

# 从文件安装
pip install -r requirements.txt

# 退出虚拟环境
deactivate

requirements.txt 示例

requests==2.31.0
flask==3.0.0
pandas==2.1.0
sqlalchemy==2.0.20
pytest==7.4.0
python-dotenv==1.0.0
pydantic==2.4.0
fastapi==0.103.0
uvicorn==0.23.0
aiohttp==3.8.5
click==8.1.7

---

二十二、Python 编码最佳实践

# 1. 使用列表推导式替代简单循环
# ❌
result = []
for i in range(10):
    if i % 2 == 0:
        result.append(i ** 2)
# ✅
result = [i**2 for i in range(10) if i % 2 == 0]

# 2. 使用 with 语句管理资源
# ❌
f = open("file.txt")
content = f.read()
f.close()
# ✅
with open("file.txt") as f:
    content = f.read()

# 3. 使用 enumerate 而非手动索引
# ❌
for i in range(len(items)):
    print(i, items[i])
# ✅
for i, item in enumerate(items):
    print(i, item)

# 4. 使用 f-string 格式化
name, age = "张三", 25
# ❌
msg = "我是" + name + "，今年" + str(age) + "岁"
# ✅
msg = f"我是{name}，今年{age}岁"

# 5. 使用 dict.get() 避免 KeyError
user = {"name": "张三"}
# ❌
# phone = user["phone"]  # KeyError!
# ✅
phone = user.get("phone", "未知")

# 6. 用 defaultdict / Counter 替代手动计数
from collections import defaultdict, Counter
words = ["apple", "banana", "apple", "cherry"]
# ❌
count = {}
for w in words:
    count[w] = count.get(w, 0) + 1
# ✅
count = Counter(words)

# 7. 使用 any/all 替代循环判断
numbers = [2, 4, 6, 8, 10]
# ❌
all_even = True
for n in numbers:
    if n % 2 != 0:
        all_even = False
        break
# ✅
all_even = all(n % 2 == 0 for n in numbers)
has_big = any(n > 5 for n in numbers)

# 8. 字典合并
# Python 3.9+
merged = dict1 | dict2
# Python 3.5+
merged = {**dict1, **dict2}

# 9. 使用 dataclass 替代手写 __init__
from dataclasses import dataclass

@dataclass
class Point:
    x: float
    y: float

    def distance_to(self, other: "Point") -> float:
        return ((self.x - other.x)**2 + (self.y - other.y)**2) ** 0.5

# 10. 异常处理要精确
# ❌
try:
    do_something()
except:  # 捕获所有异常（太宽泛）
    pass
# ✅
try:
    do_something()
except (ValueError, TypeError) as e:
    logger.error(f"处理失败: {e}")
    raise

---

二十三、字符串高级操作

# ===== 字符串方法大全 =====
s = "  Hello, World! Python  "

# 清理
s.strip()            # "Hello, World! Python" (去两端空白)
s.lstrip()           # "Hello, World! Python  "
s.rstrip()           # "  Hello, World! Python"

# 大小写
"hello".upper()       # "HELLO"
"HELLO".lower()       # "hello"
"hello world".title() # "Hello World"
"hello world".capitalize()  # "Hello world"
"Hello".swapcase()    # "hELLO"

# 查找与替换
"hello world".find("world")      # 6 (找不到返回-1)
"hello world".index("world")     # 6 (找不到抛异常)
"hello world".count("l")         # 3
"hello world".replace("world", "python")  # "hello python"
"hello world".startswith("hello")  # True
"hello world".endswith("world")    # True

# 分割与连接
"a,b,c,d".split(",")             # ["a", "b", "c", "d"]
"a,b,c,d".split(",", maxsplit=2) # ["a", "b", "c,d"]
"line1\nline2\nline3".splitlines()  # ["line1", "line2", "line3"]
",".join(["a", "b", "c"])        # "a,b,c"
" ".join(["Hello", "World"])     # "Hello World"

# 判断
"123".isdigit()       # True
"abc".isalpha()       # True
"abc123".isalnum()    # True
"   ".isspace()       # True
"Hello".istitle()     # True

# 填充与对齐
"42".zfill(5)         # "00042"
"hi".center(10, "-")  # "----hi----"
"hi".ljust(10, ".")   # "hi........"
"hi".rjust(10, ".")   # "........hi"

# 格式化
name = "张三"
age = 25
score = 95.5678

# f-string 高级用法
f"{name:>10}"          # "        张三" (右对齐，宽度10)
f"{name:<10}"          # "张三        " (左对齐)
f"{name:^10}"          # "    张三    " (居中)
f"{score:.2f}"         # "95.57" (保留2位小数)
f"{age:05d}"           # "00025" (补零)
f"{1000000:,}"         # "1,000,000" (千位分隔)
f"{0.85:.1%}"          # "85.0%" (百分比)
f"{255:#x}"            # "0xff" (十六进制)
f"{255:#b}"            # "0b11111111" (二进制)
f"{255:#o}"            # "0o377" (八进制)

# 多行f-string
message = (
    f"用户: {name}\n"
    f"年龄: {age}\n"
    f"评分: {score:.1f}"
)

# 字符串模板（安全格式化，防注入）
from string import Template
t = Template("$name 的年龄是 $age")
result = t.substitute(name="张三", age=25)

# 编码与解码
text = "你好世界"
encoded = text.encode("utf-8")    # b'\xe4\xbd\xa0\xe5\xa5\xbd...'
decoded = encoded.decode("utf-8") # "你好世界"

# base64编码
import base64
data = "Hello Python".encode()
b64 = base64.b64encode(data).decode()     # "SGVsbG8gUHl0aG9u"
original = base64.b64decode(b64).decode()  # "Hello Python"

---

二十四、爬虫与数据采集

24.1 BeautifulSoup 网页解析

# pip install beautifulsoup4 lxml requests
import requests
from bs4 import BeautifulSoup

# 获取网页
url = "https://news.ycombinator.com/"
response = requests.get(url, headers={"User-Agent": "Mozilla/5.0"})
soup = BeautifulSoup(response.text, "lxml")

# 查找元素
title = soup.find("title").text
print(f"页面标题: {title}")

# 查找所有链接
links = soup.find_all("a", class_="titlelink")
for link in links[:10]:
    print(f"{link.text}: {link['href']}")

# CSS选择器
items = soup.select("tr.athing .titleline > a")
for item in items:
    print(item.get_text())

# 查找特定属性
divs = soup.find_all("div", attrs={"data-type": "article"})

# 获取嵌套内容
table = soup.find("table", class_="itemlist")
if table:
    rows = table.find_all("tr")
    for row in rows:
        cells = row.find_all("td")
        data = [cell.text.strip() for cell in cells]
        print(data)

24.2 实用爬虫模板

import requests
from bs4 import BeautifulSoup
import time
import csv
from urllib.parse import urljoin

class SimpleScraper:
    def __init__(self, base_url, delay=1):
        self.base_url = base_url
        self.delay = delay
        self.session = requests.Session()
        self.session.headers.update({
            "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
        })

    def fetch(self, url):
        """带延迟的请求"""
        time.sleep(self.delay)
        try:
            resp = self.session.get(url, timeout=10)
            resp.raise_for_status()
            resp.encoding = resp.apparent_encoding
            return resp.text
        except requests.RequestException as e:
            print(f"请求失败: {url} - {e}")
            return None

    def parse_list_page(self, html):
        """解析列表页，提取链接"""
        soup = BeautifulSoup(html, "lxml")
        items = []
        for article in soup.select("article.post"):
            title = article.select_one("h2 a")
            if title:
                items.append({
                    "title": title.text.strip(),
                    "url": urljoin(self.base_url, title["href"])
                })
        return items

    def parse_detail_page(self, html):
        """解析详情页"""
        soup = BeautifulSoup(html, "lxml")
        content = soup.select_one("div.content")
        return content.text.strip() if content else ""

    def run(self, output_file="result.csv"):
        """运行爬虫"""
        html = self.fetch(self.base_url)
        if not html:
            return

        items = self.parse_list_page(html)
        print(f"找到 {len(items)} 个条目")

        results = []
        for item in items:
            detail_html = self.fetch(item["url"])
            if detail_html:
                item["content"] = self.parse_detail_page(detail_html)
                results.append(item)
                print(f"已采集: {item['title']}")

        # 保存结果
        with open(output_file, "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=["title", "url", "content"])
            writer.writeheader()
            writer.writerows(results)
        print(f"结果已保存到 {output_file}")

# scraper = SimpleScraper("https://example.com/blog")
# scraper.run()

24.3 Selenium 浏览器自动化

# pip install selenium webdriver-manager
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.chrome.service import Service
from webdriver_manager.chrome import ChromeDriverManager

# 启动浏览器
options = webdriver.ChromeOptions()
options.add_argument("--headless")  # 无头模式
options.add_argument("--no-sandbox")
driver = webdriver.Chrome(
    service=Service(ChromeDriverManager().install()),
    options=options
)

try:
    # 打开页面
    driver.get("https://www.baidu.com")

    # 等待元素出现
    search_box = WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.ID, "kw"))
    )

    # 输入搜索
    search_box.send_keys("Python教程")
    search_box.send_keys(Keys.ENTER)

    # 等待结果加载
    WebDriverWait(driver, 10).until(
        EC.presence_of_element_located((By.CSS_SELECTOR, "#content_left"))
    )

    # 获取搜索结果
    results = driver.find_elements(By.CSS_SELECTOR, ".result h3 a")
    for r in results[:5]:
        print(r.text)

    # 截图
    driver.save_screenshot("screenshot.png")

    # 执行JS
    driver.execute_script("window.scrollTo(0, document.body.scrollHeight)")

finally:
    driver.quit()

---

二十五、图片处理 (Pillow)

# pip install Pillow
from PIL import Image, ImageDraw, ImageFont, ImageFilter, ImageEnhance

# ===== 基本操作 =====
# 打开图片
img = Image.open("photo.jpg")
print(f"尺寸: {img.size}, 格式: {img.format}, 模式: {img.mode}")

# 调整大小
resized = img.resize((800, 600))
# 等比缩放
img.thumbnail((800, 800))  # 原地修改，保持比例

# 裁剪 (left, upper, right, lower)
cropped = img.crop((100, 100, 500, 400))

# 旋转与翻转
rotated = img.rotate(45, expand=True)  # 旋转45度
flipped_h = img.transpose(Image.FLIP_LEFT_RIGHT)  # 水平翻转
flipped_v = img.transpose(Image.FLIP_TOP_BOTTOM)  # 垂直翻转

# 格式转换与保存
img.save("output.png")                      # 自动识别格式
img.save("output.jpg", quality=85)          # JPEG质量
img.convert("RGB").save("output.jpg")       # RGBA → RGB

# ===== 滤镜与增强 =====
blurred = img.filter(ImageFilter.GaussianBlur(radius=5))
sharpened = img.filter(ImageFilter.SHARPEN)
edges = img.filter(ImageFilter.FIND_EDGES)

# 亮度、对比度、颜色、锐度
enhancer = ImageEnhance.Brightness(img)
bright_img = enhancer.enhance(1.5)  # 1.0为原始，>1变亮
enhancer = ImageEnhance.Contrast(img)
contrast_img = enhancer.enhance(1.3)

# ===== 绘制图形和文字 =====
img = Image.new("RGB", (600, 400), color="white")
draw = ImageDraw.Draw(img)

# 画矩形
draw.rectangle([50, 50, 250, 200], outline="red", width=3)
# 画圆
draw.ellipse([300, 50, 500, 200], fill="blue", outline="navy")
# 画线
draw.line([(0, 0), (600, 400)], fill="green", width=2)

# 写文字
try:
    font = ImageFont.truetype("msyh.ttc", 36)  # 微软雅黑
except:
    font = ImageFont.load_default()
draw.text((50, 300), "Hello Python!", fill="black", font=font)

img.save("drawing.png")

# ===== 批量处理图片 =====
from pathlib import Path

def batch_resize(input_dir, output_dir, size=(800, 600)):
    """批量调整图片尺寸"""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    for img_path in Path(input_dir).glob("*.jpg"):
        with Image.open(img_path) as img:
            img.thumbnail(size)
            img.save(Path(output_dir) / img_path.name)
            print(f"处理完成: {img_path.name}")

# batch_resize("photos", "photos_resized")

# ===== 图片拼接 =====
def concat_images(images, direction="horizontal"):
    """拼接多张图片"""
    imgs = [Image.open(i) for i in images]
    if direction == "horizontal":
        total_width = sum(i.width for i in imgs)
        max_height = max(i.height for i in imgs)
        result = Image.new("RGB", (total_width, max_height))
        x_offset = 0
        for img in imgs:
            result.paste(img, (x_offset, 0))
            x_offset += img.width
    else:
        max_width = max(i.width for i in imgs)
        total_height = sum(i.height for i in imgs)
        result = Image.new("RGB", (max_width, total_height))
        y_offset = 0
        for img in imgs:
            result.paste(img, (0, y_offset))
            y_offset += img.height
    return result

---

二十六、Excel 操作 (openpyxl)

# pip install openpyxl
from openpyxl import Workbook, load_workbook
from openpyxl.styles import Font, Alignment, Border, Side, PatternFill
from openpyxl.chart import BarChart, Reference

# ===== 创建 Excel =====
wb = Workbook()
ws = wb.active
ws.title = "员工数据"

# 写入表头
headers = ["姓名", "部门", "薪资", "入职日期"]
ws.append(headers)

# 写入数据
data = [
    ["张三", "技术部", 15000, "2020-01-15"],
    ["李四", "市场部", 12000, "2019-06-20"],
    ["王五", "技术部", 18000, "2021-03-10"],
    ["赵六", "人事部", 13000, "2020-09-01"],
    ["钱七", "技术部", 20000, "2018-04-15"],
]
for row in data:
    ws.append(row)

# ===== 设置样式 =====
# 表头样式
header_font = Font(bold=True, size=12, color="FFFFFF")
header_fill = PatternFill(start_color="4472C4", end_color="4472C4", fill_type="solid")
header_alignment = Alignment(horizontal="center", vertical="center")

for cell in ws[1]:
    cell.font = header_font
    cell.fill = header_fill
    cell.alignment = header_alignment

# 列宽
ws.column_dimensions["A"].width = 15
ws.column_dimensions["B"].width = 15
ws.column_dimensions["C"].width = 12
ws.column_dimensions["D"].width = 15

# 数字格式
for row in ws.iter_rows(min_row=2, min_col=3, max_col=3):
    for cell in row:
        cell.number_format = "#,##0"

# ===== 添加图表 =====
chart = BarChart()
chart.title = "员工薪资"
chart.x_axis.title = "姓名"
chart.y_axis.title = "薪资(元)"

data_ref = Reference(ws, min_col=3, min_row=1, max_row=6)
cats_ref = Reference(ws, min_col=1, min_row=2, max_row=6)
chart.add_data(data_ref, titles_from_data=True)
chart.set_categories(cats_ref)
ws.add_chart(chart, "F2")

# ===== 添加公式 =====
ws["C8"] = "=SUM(C2:C6)"
ws["C9"] = "=AVERAGE(C2:C6)"
ws["A8"] = "总计"
ws["A9"] = "平均"

wb.save("employees.xlsx")
print("Excel 已生成")

# ===== 读取 Excel =====
wb = load_workbook("employees.xlsx")
ws = wb.active

for row in ws.iter_rows(min_row=2, max_row=ws.max_row, values_only=True):
    if row[0]:  # 非空行
        print(f"{row[0]} - {row[1]} - ¥{row[2]:,}")

# ===== 多个Sheet操作 =====
wb = Workbook()
ws1 = wb.active
ws1.title = "数据"
ws2 = wb.create_sheet("汇总")
ws3 = wb.create_sheet("图表")

# 复制数据到汇总表
ws2.append(["部门", "人数", "平均薪资"])
# ... 聚合逻辑

wb.save("report.xlsx")

---

二十七、邮件发送

import smtplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.mime.base import MIMEBase
from email import encoders
from pathlib import Path

def send_email(
    sender: str,
    password: str,
    recipients: list,
    subject: str,
    body: str,
    html: str = None,
    attachments: list = None,
    smtp_server: str = "smtp.qq.com",
    smtp_port: int = 465
):
    """发送邮件（支持附件和HTML）"""
    msg = MIMEMultipart("alternative")
    msg["From"] = sender
    msg["To"] = ", ".join(recipients)
    msg["Subject"] = subject

    # 纯文本内容
    msg.attach(MIMEText(body, "plain", "utf-8"))

    # HTML内容（可选）
    if html:
        msg.attach(MIMEText(html, "html", "utf-8"))

    # 附件
    if attachments:
        for filepath in attachments:
            path = Path(filepath)
            if path.exists():
                with open(path, "rb") as f:
                    part = MIMEBase("application", "octet-stream")
                    part.set_payload(f.read())
                encoders.encode_base64(part)
                part.add_header(
                    "Content-Disposition",
                    f"attachment; filename={path.name}"
                )
                msg.attach(part)

    # 发送
    try:
        with smtplib.SMTP_SSL(smtp_server, smtp_port) as server:
            server.login(sender, password)
            server.sendmail(sender, recipients, msg.as_string())
        print("邮件发送成功!")
    except Exception as e:
        print(f"发送失败: {e}")

# 使用示例
# send_email(
#     sender="your_email@qq.com",
#     password="授权码(非登录密码)",
#     recipients=["target@example.com"],
#     subject="测试邮件",
#     body="这是一封测试邮件",
#     html="<h1>Hello</h1><p>这是<b>HTML</b>邮件</p>",
#     attachments=["report.pdf", "data.xlsx"]
# )

---

二十八、Redis 缓存

# pip install redis
import redis
import json
from datetime import timedelta
from functools import wraps

# 连接Redis
r = redis.Redis(host="localhost", port=6379, db=0, decode_responses=True)

# ===== 基本操作 =====
# 字符串
r.set("name", "张三")
r.set("token", "abc123", ex=3600)  # 1小时过期
print(r.get("name"))  # "张三"
r.delete("name")

# 哈希
r.hset("user:1", mapping={"name": "张三", "age": "25", "city": "北京"})
print(r.hgetall("user:1"))  # {'name': '张三', 'age': '25', 'city': '北京'}
print(r.hget("user:1", "name"))  # "张三"

# 列表
r.rpush("queue", "task1", "task2", "task3")
print(r.lpop("queue"))  # "task1"
print(r.lrange("queue", 0, -1))  # ["task2", "task3"]

# 集合
r.sadd("online_users", "user1", "user2", "user3")
print(r.smembers("online_users"))  # {'user1', 'user2', 'user3'}
print(r.sismember("online_users", "user1"))  # True

# 有序集合（排行榜）
r.zadd("leaderboard", {"player1": 100, "player2": 85, "player3": 95})
print(r.zrevrange("leaderboard", 0, 2, withscores=True))
# [('player1', 100.0), ('player3', 95.0), ('player2', 85.0)]

# ===== 缓存装饰器 =====
def cache(expire: int = 300, prefix: str = "cache"):
    """Redis缓存装饰器"""
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # 生成缓存key
            key = f"{prefix}:{func.__name__}:{str(args)}:{str(kwargs)}"
            # 尝试获取缓存
            cached = r.get(key)
            if cached:
                return json.loads(cached)
            # 执行函数
            result = func(*args, **kwargs)
            # 存入缓存
            r.set(key, json.dumps(result, ensure_ascii=False), ex=expire)
            return result
        return wrapper
    return decorator

@cache(expire=600)
def get_user_info(user_id):
    """模拟数据库查询（耗时操作）"""
    import time
    time.sleep(2)
    return {"id": user_id, "name": "张三", "age": 25}

# 第一次调用：2秒（查数据库）
# 后续调用：毫秒级（读缓存）

---

二十九、WebSocket 实时通信

# pip install websockets fastapi uvicorn

# ===== 服务端 (FastAPI) =====
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
from typing import List

app = FastAPI()

class ConnectionManager:
    """WebSocket连接管理器"""
    def __init__(self):
        self.active_connections: List[WebSocket] = []

    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)

    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)

    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

@app.websocket("/ws/{username}")
async def websocket_endpoint(websocket: WebSocket, username: str):
    await manager.connect(websocket)
    await manager.broadcast(f"{username} 加入了聊天室")
    try:
        while True:
            data = await websocket.receive_text()
            await manager.broadcast(f"{username}: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"{username} 离开了聊天室")

# ===== 客户端 =====
import asyncio
import websockets

async def chat_client(username):
    uri = f"ws://localhost:8000/ws/{username}"
    async with websockets.connect(uri) as ws:
        # 接收消息的任务
        async def receive():
            async for message in ws:
                print(f"\r{message}\n> ", end="")

        # 发送消息的任务
        async def send():
            while True:
                message = await asyncio.get_event_loop().run_in_executor(
                    None, input, "> "
                )
                await ws.send(message)

        await asyncio.gather(receive(), send())

# asyncio.run(chat_client("张三"))

---

三十、调试技巧

# ===== 1. print 调试（快速但不专业） =====
def debug_print(*args, **kwargs):
    """带上下文的打印"""
    import inspect
    frame = inspect.currentframe().f_back
    filename = frame.f_code.co_filename.split("/")[-1]
    lineno = frame.f_lineno
    print(f"[DEBUG {filename}:{lineno}]", *args, **kwargs)

# ===== 2. breakpoint() 交互式调试 (Python 3.7+) =====
def calculate(data):
    result = []
    for item in data:
        processed = item * 2
        breakpoint()  # 程序会在这里暂停，进入pdb交互
        result.append(processed)
    return result

# pdb常用命令:
# n(next)    - 下一行
# s(step)    - 进入函数
# c(continue)- 继续执行
# p expr     - 打印表达式
# l(list)    - 显示当前代码
# q(quit)    - 退出调试
# w(where)   - 显示调用栈
# b 行号     - 设置断点

# ===== 3. 性能分析 =====
import cProfile
import pstats

def profile_function(func):
    """性能分析装饰器"""
    def wrapper(*args, **kwargs):
        profiler = cProfile.Profile()
        profiler.enable()
        result = func(*args, **kwargs)
        profiler.disable()
        stats = pstats.Stats(profiler)
        stats.sort_stats("cumulative")
        stats.print_stats(10)  # 打印前10个耗时函数
        return result
    return wrapper

@profile_function
def slow_code():
    total = 0
    for i in range(1000000):
        total += i ** 2
    return total

# ===== 4. 内存分析 =====
# pip install memory-profiler
# from memory_profiler import profile
#
# @profile
# def memory_hungry():
#     big_list = [i**2 for i in range(1000000)]
#     return sum(big_list)

# ===== 5. timeit 精确计时 =====
import timeit

# 比较两种实现的性能
time1 = timeit.timeit(
    'sum(range(1000))',
    number=10000
)
time2 = timeit.timeit(
    's = 0\nfor i in range(1000): s += i',
    number=10000
)
print(f"sum(): {time1:.4f}s")
print(f"for loop: {time2:.4f}s")

# ===== 6. traceback 详细错误信息 =====
import traceback

def risky_operation():
    try:
        result = 1 / 0
    except Exception:
        # 获取完整堆栈信息
        error_info = traceback.format_exc()
        print(f"错误详情:\n{error_info}")
        # 或写入日志文件
        with open("error.log", "a") as f:
            f.write(f"\n{'='*50}\n{error_info}")

# ===== 7. 断言调试 =====
def transfer(from_account, to_account, amount):
    assert amount > 0, f"转账金额必须大于0，当前: {amount}"
    assert from_account.balance >= amount, "余额不足"
    from_account.balance -= amount
    to_account.balance += amount
    # 后置断言：验证不变量
    assert from_account.balance >= 0, "转账后余额异常"

---

三十一、常用算法与数据结构

31.1 排序算法

# 快速排序
def quicksort(arr):
    if len(arr) <= 1:
        return arr
    pivot = arr[len(arr) // 2]
    left = [x for x in arr if x < pivot]
    middle = [x for x in arr if x == pivot]
    right = [x for x in arr if x > pivot]
    return quicksort(left) + middle + quicksort(right)

# 归并排序
def merge_sort(arr):
    if len(arr) <= 1:
        return arr
    mid = len(arr) // 2
    left = merge_sort(arr[:mid])
    right = merge_sort(arr[mid:])
    return merge(left, right)

def merge(left, right):
    result = []
    i = j = 0
    while i < len(left) and j < len(right):
        if left[i] <= right[j]:
            result.append(left[i])
            i += 1
        else:
            result.append(right[j])
            j += 1
    result.extend(left[i:])
    result.extend(right[j:])
    return result

31.2 搜索算法

# 二分查找
def binary_search(arr, target):
    left, right = 0, len(arr) - 1
    while left <= right:
        mid = (left + right) // 2
        if arr[mid] == target:
            return mid
        elif arr[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

# 二分查找（找左边界）
def bisect_left(arr, target):
    left, right = 0, len(arr)
    while left < right:
        mid = (left + right) // 2
        if arr[mid] < target:
            left = mid + 1
        else:
            right = mid
    return left

# BFS广度优先搜索
from collections import deque

def bfs(graph, start):
    visited = set()
    queue = deque([start])
    visited.add(start)
    result = []
    while queue:
        node = queue.popleft()
        result.append(node)
        for neighbor in graph[node]:
            if neighbor not in visited:
                visited.add(neighbor)
                queue.append(neighbor)
    return result

# DFS深度优先搜索
def dfs(graph, start, visited=None):
    if visited is None:
        visited = set()
    visited.add(start)
    result = [start]
    for neighbor in graph[start]:
        if neighbor not in visited:
            result.extend(dfs(graph, neighbor, visited))
    return result

# 示例图
graph = {
    'A': ['B', 'C'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F'],
    'D': ['B'],
    'E': ['B', 'F'],
    'F': ['C', 'E']
}
print(bfs(graph, 'A'))  # ['A', 'B', 'C', 'D', 'E', 'F']
print(dfs(graph, 'A'))  # ['A', 'B', 'D', 'E', 'F', 'C']

31.3 常用数据结构

from collections import deque
import heapq

# ===== 栈 (Stack) - 后进先出 =====
class Stack:
    def __init__(self):
        self._items = []

    def push(self, item):
        self._items.append(item)

    def pop(self):
        if self.is_empty():
            raise IndexError("栈为空")
        return self._items.pop()

    def peek(self):
        return self._items[-1] if self._items else None

    def is_empty(self):
        return len(self._items) == 0

    def size(self):
        return len(self._items)

# 应用：括号匹配
def is_valid_brackets(s):
    stack = []
    mapping = {')': '(', ']': '[', '}': '{'}
    for char in s:
        if char in '([{':
            stack.append(char)
        elif char in ')]}':
            if not stack or stack.pop() != mapping[char]:
                return False
    return len(stack) == 0

print(is_valid_brackets("({[]})"))  # True
print(is_valid_brackets("([)]"))    # False

# ===== 队列 (Queue) - 先进先出 =====
queue = deque()
queue.append("task1")     # 入队
queue.append("task2")
print(queue.popleft())    # 出队: "task1"

# ===== 优先队列 / 堆 =====
# 最小堆
heap = []
heapq.heappush(heap, 5)
heapq.heappush(heap, 1)
heapq.heappush(heap, 3)
print(heapq.heappop(heap))  # 1 (最小值)

# Top-K问题
numbers = [3, 1, 4, 1, 5, 9, 2, 6, 5, 3, 5]
top3 = heapq.nlargest(3, numbers)   # [9, 6, 5]
bottom3 = heapq.nsmallest(3, numbers)  # [1, 1, 2]

# ===== 链表 =====
class ListNode:
    def __init__(self, val=0, next=None):
        self.val = val
        self.next = next

class LinkedList:
    def __init__(self):
        self.head = None

    def append(self, val):
        if not self.head:
            self.head = ListNode(val)
            return
        curr = self.head
        while curr.next:
            curr = curr.next
        curr.next = ListNode(val)

    def to_list(self):
        result = []
        curr = self.head
        while curr:
            result.append(curr.val)
            curr = curr.next
        return result

    def reverse(self):
        prev, curr = None, self.head
        while curr:
            next_node = curr.next
            curr.next = prev
            prev = curr
            curr = next_node
        self.head = prev

# ===== LRU缓存 =====
from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity: int):
        self.cache = OrderedDict()
        self.capacity = capacity

    def get(self, key):
        if key not in self.cache:
            return -1
        self.cache.move_to_end(key)
        return self.cache[key]

    def put(self, key, value):
        if key in self.cache:
            self.cache.move_to_end(key)
        self.cache[key] = value
        if len(self.cache) > self.capacity:
            self.cache.popitem(last=False)

cache = LRUCache(3)
cache.put("a", 1)
cache.put("b", 2)
cache.put("c", 3)
cache.get("a")      # 访问a，a变为最近使用
cache.put("d", 4)   # 容量满，淘汰最久未使用的b

31.4 动态规划

# 最长公共子序列 (LCS)
def lcs(s1, s2):
    m, n = len(s1), len(s2)
    dp = [[0] * (n + 1) for _ in range(m + 1)]
    for i in range(1, m + 1):
        for j in range(1, n + 1):
            if s1[i-1] == s2[j-1]:
                dp[i][j] = dp[i-1][j-1] + 1
            else:
                dp[i][j] = max(dp[i-1][j], dp[i][j-1])
    return dp[m][n]

print(lcs("abcde", "ace"))  # 3 ("ace")

# 背包问题
def knapsack(weights, values, capacity):
    n = len(weights)
    dp = [[0] * (capacity + 1) for _ in range(n + 1)]
    for i in range(1, n + 1):
        for w in range(capacity + 1):
            if weights[i-1] <= w:
                dp[i][w] = max(
                    dp[i-1][w],
                    dp[i-1][w - weights[i-1]] + values[i-1]
                )
            else:
                dp[i][w] = dp[i-1][w]
    return dp[n][capacity]

weights = [2, 3, 4, 5]
values = [3, 4, 5, 6]
capacity = 8
print(knapsack(weights, values, capacity))  # 10

# 最长递增子序列 (LIS)
def lis(nums):
    if not nums:
        return 0
    dp = [1] * len(nums)
    for i in range(1, len(nums)):
        for j in range(i):
            if nums[j] < nums[i]:
                dp[i] = max(dp[i], dp[j] + 1)
    return max(dp)

print(lis([10, 9, 2, 5, 3, 7, 101, 18]))  # 4 ([2,3,7,101])

---

三十二、打包与发布

32.1 打包为可执行文件 (PyInstaller)

# 安装
pip install pyinstaller

# 打包单文件
pyinstaller --onefile --windowed app.py

# 带图标
pyinstaller --onefile --icon=app.ico app.py

# 带额外文件
pyinstaller --onefile --add-data "config.json;." app.py

32.2 发布到 PyPI

# 项目结构:
# my_package/
#   my_package/
#     __init__.py
#     core.py
#   tests/
#     test_core.py
#   setup.py (或 pyproject.toml)
#   README.md
#   LICENSE

# pyproject.toml (现代方式)
"""
[build-system]
requires = ["setuptools>=68.0", "wheel"]
build-backend = "setuptools.build_meta"

[project]
name = "my-awesome-package"
version = "0.1.0"
description = "一个很棒的Python包"
readme = "README.md"
license = {text = "MIT"}
requires-python = ">=3.8"
authors = [{name = "张三", email = "zs@example.com"}]
dependencies = [
    "requests>=2.28",
    "pydantic>=2.0",
]

[project.optional-dependencies]
dev = ["pytest", "black", "mypy"]

[project.scripts]
my-cli = "my_package.cli:main"
"""

# 构建与发布
# pip install build twine
# python -m build
# twine upload dist/*

---

三十三、网络编程 (Socket)

import socket
import threading

# ===== TCP 服务器 =====
def tcp_server(host="0.0.0.0", port=8888):
    server = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    server.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
    server.bind((host, port))
    server.listen(5)
    print(f"服务器启动: {host}:{port}")

    def handle_client(conn, addr):
        print(f"新连接: {addr}")
        try:
            while True:
                data = conn.recv(1024)
                if not data:
                    break
                message = data.decode("utf-8")
                print(f"收到 [{addr}]: {message}")
                conn.send(f"服务器收到: {message}".encode("utf-8"))
        finally:
            conn.close()
            print(f"断开连接: {addr}")

    try:
        while True:
            conn, addr = server.accept()
            thread = threading.Thread(target=handle_client, args=(conn, addr))
            thread.daemon = True
            thread.start()
    finally:
        server.close()

# ===== TCP 客户端 =====
def tcp_client(host="localhost", port=8888):
    client = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    client.connect((host, port))
    try:
        while True:
            message = input("发送: ")
            if message.lower() == "quit":
                break
            client.send(message.encode("utf-8"))
            response = client.recv(1024).decode("utf-8")
            print(f"响应: {response}")
    finally:
        client.close()

---

三十四、数据可视化 (Matplotlib)

# pip install matplotlib
import matplotlib.pyplot as plt
import numpy as np

# 设置中文字体
plt.rcParams["font.sans-serif"] = ["SimHei", "Microsoft YaHei"]
plt.rcParams["axes.unicode_minus"] = False

# ===== 折线图 =====
months = ["1月", "2月", "3月", "4月", "5月", "6月"]
sales = [120, 135, 148, 160, 155, 170]
costs = [80, 90, 95, 100, 98, 105]

plt.figure(figsize=(10, 6))
plt.plot(months, sales, "b-o", label="销售额", linewidth=2)
plt.plot(months, costs, "r--s", label="成本", linewidth=2)
plt.title("月度数据", fontsize=16)
plt.xlabel("月份")
plt.ylabel("金额(万元)")
plt.legend()
plt.grid(True, alpha=0.3)
plt.tight_layout()
plt.savefig("line_chart.png", dpi=150)
plt.close()

# ===== 柱状图 =====
categories = ["技术部", "市场部", "人事部", "财务部"]
values = [45, 30, 15, 10]

plt.figure(figsize=(8, 5))
bars = plt.bar(categories, values, color=["#4472C4", "#ED7D31", "#A5A5A5", "#FFC000"])
plt.title("各部门人数")
plt.ylabel("人数")
for bar, val in zip(bars, values):
    plt.text(bar.get_x() + bar.get_width()/2, bar.get_height() + 0.5,
             str(val), ha="center", fontsize=12)
plt.tight_layout()
plt.savefig("bar_chart.png", dpi=150)
plt.close()

# ===== 饼图 =====
labels = ["Python", "JavaScript", "Java", "Go", "其他"]
sizes = [35, 25, 20, 12, 8]
explode = (0.05, 0, 0, 0, 0)

plt.figure(figsize=(8, 8))
plt.pie(sizes, explode=explode, labels=labels, autopct="%1.1f%%",
        shadow=True, startangle=90, colors=plt.cm.Set3.colors)
plt.title("编程语言使用占比")
plt.tight_layout()
plt.savefig("pie_chart.png", dpi=150)
plt.close()

# ===== 子图组合 =====
fig, axes = plt.subplots(2, 2, figsize=(12, 10))

# 子图1 - 折线
axes[0,0].plot(range(10), [x**2 for x in range(10)])
axes[0,0].set_title("二次函数")

# 子图2 - 散点
x = np.random.randn(100)
y = np.random.randn(100)
axes[0,1].scatter(x, y, alpha=0.5)
axes[0,1].set_title("散点图")

# 子图3 - 直方图
data = np.random.normal(0, 1, 1000)
axes[1,0].hist(data, bins=30, edgecolor="black", alpha=0.7)
axes[1,0].set_title("正态分布")

# 子图4 - 热力图
matrix = np.random.rand(5, 5)
im = axes[1,1].imshow(matrix, cmap="YlOrRd")
plt.colorbar(im, ax=axes[1,1])
axes[1,1].set_title("热力图")

plt.tight_layout()
plt.savefig("subplots.png", dpi=150)
plt.close()

---

三十五、SSH 远程操作 (Paramiko)

# pip install paramiko
import paramiko

# ===== 执行远程命令 =====
def ssh_execute(host, username, password, command):
    """SSH执行远程命令"""
    client = paramiko.SSHClient()
    client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
    try:
        client.connect(host, port=22, username=username, password=password, timeout=10)
        stdin, stdout, stderr = client.exec_command(command)
        output = stdout.read().decode("utf-8")
        error = stderr.read().decode("utf-8")
        return output, error
    finally:
        client.close()

# output, error = ssh_execute("192.168.1.100", "root", "password", "ls -la /")

# ===== SFTP 文件传输 =====
def sftp_upload(host, username, password, local_path, remote_path):
    """上传文件到远程服务器"""
    transport = paramiko.Transport((host, 22))
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    try:
        sftp.put(local_path, remote_path)
        print(f"上传成功: {local_path} -> {remote_path}")
    finally:
        sftp.close()
        transport.close()

def sftp_download(host, username, password, remote_path, local_path):
    """从远程服务器下载文件"""
    transport = paramiko.Transport((host, 22))
    transport.connect(username=username, password=password)
    sftp = paramiko.SFTPClient.from_transport(transport)
    try:
        sftp.get(remote_path, local_path)
        print(f"下载成功: {remote_path} -> {local_path}")
    finally:
        sftp.close()
        transport.close()

---

三十六、多环境配置与项目结构

36.1 推荐项目结构

my_project/
├── README.md
├── pyproject.toml          # 项目配置
├── requirements.txt        # 生产依赖
├── requirements-dev.txt    # 开发依赖
├── .env                    # 环境变量（不提交git）
├── .env.example            # 环境变量模板
├── .gitignore
├── src/
│   └── my_project/
│       ├── __init__.py
│       ├── main.py         # 入口
│       ├── config.py       # 配置
│       ├── models/         # 数据模型
│       │   ├── __init__.py
│       │   └── user.py
│       ├── services/       # 业务逻辑
│       │   ├── __init__.py
│       │   └── user_service.py
│       ├── api/            # 接口层
│       │   ├── __init__.py
│       │   └── routes.py
│       └── utils/          # 工具函数
│           ├── __init__.py
│           └── helpers.py
├── tests/
│   ├── __init__.py
│   ├── conftest.py         # pytest配置
│   ├── test_models.py
│   └── test_services.py
├── scripts/                # 脚本
│   └── init_db.py
└── docs/                   # 文档
    └── api.md

36.2 配置管理最佳实践

# config.py
import os
from dataclasses import dataclass
from dotenv import load_dotenv

load_dotenv()

@dataclass
class Config:
    # 应用
    APP_NAME: str = os.getenv("APP_NAME", "MyApp")
    DEBUG: bool = os.getenv("DEBUG", "false").lower() == "true"
    SECRET_KEY: str = os.getenv("SECRET_KEY", "change-me-in-production")

    # 数据库
    DB_HOST: str = os.getenv("DB_HOST", "localhost")
    DB_PORT: int = int(os.getenv("DB_PORT", "5432"))
    DB_NAME: str = os.getenv("DB_NAME", "mydb")
    DB_USER: str = os.getenv("DB_USER", "postgres")
    DB_PASSWORD: str = os.getenv("DB_PASSWORD", "")

    @property
    def DATABASE_URL(self) -> str:
        return f"postgresql://{self.DB_USER}:{self.DB_PASSWORD}@{self.DB_HOST}:{self.DB_PORT}/{self.DB_NAME}"

    # Redis
    REDIS_URL: str = os.getenv("REDIS_URL", "redis://localhost:6379/0")

    # 日志
    LOG_LEVEL: str = os.getenv("LOG_LEVEL", "INFO")

config = Config()

---

三十七、协程高级模式

import asyncio
from typing import AsyncIterator

# ===== 异步队列 - 生产者消费者 =====
async def producer(queue: asyncio.Queue, n: int):
    for i in range(n):
        await asyncio.sleep(0.5)  # 模拟生产耗时
        item = f"item-{i}"
        await queue.put(item)
        print(f"生产: {item}")
    await queue.put(None)  # 结束信号

async def consumer(queue: asyncio.Queue, name: str):
    while True:
        item = await queue.get()
        if item is None:
            await queue.put(None)  # 传递结束信号给其他消费者
            break
        await asyncio.sleep(1)  # 模拟消费耗时
        print(f"[{name}] 消费: {item}")
        queue.task_done()

async def main():
    queue = asyncio.Queue(maxsize=5)
    await asyncio.gather(
        producer(queue, 10),
        consumer(queue, "worker-1"),
        consumer(queue, "worker-2"),
    )

# asyncio.run(main())

# ===== 异步超时与取消 =====
async def long_task():
    await asyncio.sleep(10)
    return "完成"

async def with_timeout():
    try:
        result = await asyncio.wait_for(long_task(), timeout=3.0)
    except asyncio.TimeoutError:
        print("任务超时!")

# ===== 异步锁 =====
class AsyncCounter:
    def __init__(self):
        self.count = 0
        self._lock = asyncio.Lock()

    async def increment(self):
        async with self._lock:
            current = self.count
            await asyncio.sleep(0.01)
            self.count = current + 1

async def test_lock():
    counter = AsyncCounter()
    tasks = [counter.increment() for _ in range(100)]
    await asyncio.gather(*tasks)
    print(f"最终计数: {counter.count}")  # 100（如果没有锁可能小于100）

# ===== 异步上下文管理器 =====
class AsyncDatabase:
    async def __aenter__(self):
        print("连接数据库...")
        await asyncio.sleep(0.1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        print("断开数据库...")
        await asyncio.sleep(0.1)

    async def query(self, sql):
        await asyncio.sleep(0.1)
        return f"结果: {sql}"

async def use_db():
    async with AsyncDatabase() as db:
        result = await db.query("SELECT * FROM users")
        print(result)

---

三十八、数据序列化与协议

# ===== Protocol Buffers 风格：msgpack（比JSON更快更小）=====
# pip install msgpack
import msgpack

data = {
    "name": "张三",
    "scores": [90, 85, 92],
    "metadata": {"level": 5, "active": True}
}

# 序列化
packed = msgpack.packb(data, use_bin_type=True)
print(f"msgpack大小: {len(packed)} bytes")

# 反序列化
unpacked = msgpack.unpackb(packed, raw=False)
print(unpacked)

# ===== pickle（Python对象序列化）=====
import pickle

class GameState:
    def __init__(self, level, score, items):
        self.level = level
        self.score = score
        self.items = items

state = GameState(5, 1000, ["sword", "shield"])

# 序列化到文件
with open("save.pkl", "wb") as f:
    pickle.dump(state, f)

# 从文件反序列化
with open("save.pkl", "rb") as f:
    loaded_state = pickle.load(f)
print(f"等级: {loaded_state.level}, 分数: {loaded_state.score}")

# ===== YAML 配置 =====
# pip install pyyaml
import yaml

config_yaml = """
server:
  host: 0.0.0.0
  port: 8080
  workers: 4

database:
  host: localhost
  port: 5432
  name: myapp
  pool_size: 10

logging:
  level: INFO
  file: app.log
"""

config = yaml.safe_load(config_yaml)
print(config["server"]["port"])  # 8080

# 写入YAML
with open("config.yaml", "w", encoding="utf-8") as f:
    yaml.dump(config, f, allow_unicode=True, default_flow_style=False)

# ===== TOML (Python 3.11+ 内置) =====
import tomllib  # Python 3.11+

with open("config.toml", "rb") as f:
    config = tomllib.load(f)

---

三十九、安全编码

import hashlib
import hmac
import secrets
from urllib.parse import quote, unquote

# ===== 密码哈希（永远不要明文存储密码）=====
# 推荐方式：使用 bcrypt 或 argon2
# pip install bcrypt
import bcrypt

def hash_password(password: str) -> str:
    """哈希密码"""
    salt = bcrypt.gensalt(rounds=12)
    hashed = bcrypt.hashpw(password.encode(), salt)
    return hashed.decode()

def verify_password(password: str, hashed: str) -> bool:
    """验证密码"""
    return bcrypt.checkpw(password.encode(), hashed.encode())

hashed = hash_password("my_secure_password")
print(verify_password("my_secure_password", hashed))  # True
print(verify_password("wrong_password", hashed))      # False

# ===== 生成安全随机数 =====
token = secrets.token_hex(32)          # 64字符十六进制
url_token = secrets.token_urlsafe(32)  # URL安全的随机字符串
otp = secrets.randbelow(1000000)       # 0-999999的随机数（验证码）

# ===== HMAC 签名 =====
secret_key = b"my-secret-key"
message = b"important data"
signature = hmac.new(secret_key, message, hashlib.sha256).hexdigest()

def verify_signature(message, signature, key):
    expected = hmac.new(key, message, hashlib.sha256).hexdigest()
    return hmac.compare_digest(signature, expected)  # 防时序攻击

# ===== SQL注入防护 =====
# ❌ 危险！SQL注入
# cursor.execute(f"SELECT * FROM users WHERE name = '{user_input}'")

# ✅ 安全：参数化查询
# cursor.execute("SELECT * FROM users WHERE name = ?", (user_input,))

# ===== XSS防护 =====
import html

user_input = '<script>alert("XSS")</script>'
safe_output = html.escape(user_input)
# '&lt;script&gt;alert(&quot;XSS&quot;)&lt;/script&gt;'

# ===== 文件路径安全 =====
from pathlib import Path

def safe_join(base_dir, user_path):
    """防止路径遍历攻击"""
    base = Path(base_dir).resolve()
    target = (base / user_path).resolve()
    if not str(target).startswith(str(base)):
        raise ValueError("非法路径!")
    return target

# safe_join("/app/uploads", "../../etc/passwd")  # ValueError!

---

四十、实用脚本合集

40.1 文件批量重命名

from pathlib import Path
import re

def batch_rename(directory, pattern, replacement):
    """批量重命名文件"""
    path = Path(directory)
    for file in path.iterdir():
        if file.is_file():
            new_name = re.sub(pattern, replacement, file.name)
            if new_name != file.name:
                file.rename(path / new_name)
                print(f"{file.name} → {new_name}")

# 示例：将 "IMG_20210101_001.jpg" 改为 "2021-01-01_001.jpg"
# batch_rename("photos", r"IMG_(\d{4})(\d{2})(\d{2})_", r"\1-\2-\3_")

40.2 监控文件变化

# pip install watchdog
from watchdog.observers import Observer
from watchdog.events import FileSystemEventHandler
import time

class FileChangeHandler(FileSystemEventHandler):
    def on_created(self, event):
        if not event.is_directory:
            print(f"新文件: {event.src_path}")

    def on_modified(self, event):
        if not event.is_directory:
            print(f"文件修改: {event.src_path}")

    def on_deleted(self, event):
        if not event.is_directory:
            print(f"文件删除: {event.src_path}")

def watch_directory(path="."):
    observer = Observer()
    observer.schedule(FileChangeHandler(), path, recursive=True)
    observer.start()
    print(f"监控目录: {path}")
    try:
        while True:
            time.sleep(1)
    except KeyboardInterrupt:
        observer.stop()
    observer.join()

# watch_directory("./my_project")

40.3 系统信息获取

import platform
import psutil  # pip install psutil

def system_info():
    """获取系统信息"""
    info = {
        "系统": platform.system(),
        "版本": platform.version(),
        "架构": platform.machine(),
        "处理器": platform.processor(),
        "Python版本": platform.python_version(),
        "CPU核心数": psutil.cpu_count(),
        "CPU使用率": f"{psutil.cpu_percent()}%",
        "内存总量": f"{psutil.virtual_memory().total / (1024**3):.1f} GB",
        "内存使用率": f"{psutil.virtual_memory().percent}%",
        "磁盘使用率": f"{psutil.disk_usage('/').percent}%",
    }
    for key, value in info.items():
        print(f"{key}: {value}")

# system_info()

40.4 进度条

# pip install tqdm
from tqdm import tqdm
import time

# 基本进度条
for i in tqdm(range(100), desc="处理中"):
    time.sleep(0.02)

# 手动更新
with tqdm(total=100, desc="下载") as pbar:
    for i in range(10):
        time.sleep(0.3)
        pbar.update(10)

# 嵌套进度条
for i in tqdm(range(3), desc="外层"):
    for j in tqdm(range(100), desc="内层", leave=False):
        time.sleep(0.01)

# 自定义简易进度条（不依赖第三方库）
import sys

def progress_bar(current, total, bar_length=40):
    percent = current / total
    filled = int(bar_length * percent)
    bar = "█" * filled + "░" * (bar_length - filled)
    sys.stdout.write(f"\r[{bar}] {percent:.1%} ({current}/{total})")
    sys.stdout.flush()
    if current == total:
        print()

for i in range(1, 101):
    progress_bar(i, 100)
    time.sleep(0.02)

40.5 并发下载器

import asyncio
import aiohttp
import aiofiles  # pip install aiofiles
from pathlib import Path
from tqdm.asyncio import tqdm

async def download_file(session, url, output_dir, semaphore):
    """下载单个文件"""
    filename = url.split("/")[-1]
    filepath = Path(output_dir) / filename

    async with semaphore:
        try:
            async with session.get(url) as resp:
                if resp.status == 200:
                    async with aiofiles.open(filepath, "wb") as f:
                        async for chunk in resp.content.iter_chunked(8192):
                            await f.write(chunk)
                    return f"✓ {filename}"
                else:
                    return f"✗ {filename} (HTTP {resp.status})"
        except Exception as e:
            return f"✗ {filename} ({e})"

async def batch_download(urls, output_dir="downloads", max_concurrent=5):
    """批量并发下载"""
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    semaphore = asyncio.Semaphore(max_concurrent)

    async with aiohttp.ClientSession() as session:
        tasks = [download_file(session, url, output_dir, semaphore) for url in urls]
        results = await asyncio.gather(*tasks)

    for result in results:
        print(result)

# urls = [f"https://example.com/file_{i}.zip" for i in range(20)]
# asyncio.run(batch_download(urls))

---

附录：常用内置函数速查

# 数学
abs(-5)              # 5
max(1, 2, 3)         # 3
min(1, 2, 3)         # 1
sum([1, 2, 3])       # 6
round(3.14159, 2)    # 3.14
pow(2, 10)           # 1024
divmod(17, 5)        # (3, 2) → 商和余数

# 序列操作
len([1, 2, 3])       # 3
sorted([3, 1, 2])    # [1, 2, 3]
reversed([1, 2, 3])  # 迭代器
enumerate(["a","b"]) # [(0,"a"), (1,"b")]
zip([1,2], [3,4])    # [(1,3), (2,4)]
map(str, [1,2,3])    # ["1","2","3"]
filter(None, [0,1,2])# [1, 2]

# 类型转换
int("42")            # 42
float("3.14")        # 3.14
str(100)             # "100"
bool(0)              # False
list((1,2,3))        # [1, 2, 3]
tuple([1,2,3])       # (1, 2, 3)
set([1,1,2,2])       # {1, 2}
dict([("a",1)])      # {"a": 1}

# 判断
isinstance(42, int)  # True
issubclass(bool, int)# True
callable(print)      # True
hasattr(obj, "name") # True/False

# 其他
id(obj)              # 对象内存地址
hash("hello")        # 哈希值
dir(obj)             # 对象所有属性方法
help(print)          # 查看帮助文档
vars(obj)            # 对象的 __dict__
globals()            # 全局变量字典
locals()             # 局部变量字典