常用标准库精讲

from collections import defaultdict

# 普通字典访问不存在的 key 会报 KeyError
# defaultdict 自动为不存在的 key 创建默认值

# 统计字符出现次数
text = "hello world"
counter = defaultdict(int)
for char in text:
    counter[char] += 1

print(counter)
# defaultdict(<class 'int'>, {'h': 1, 'e': 1, 'l': 3, 'o': 2, ' ': 1, 'w': 1, 'r': 1, 'd': 1})

# 用列表作为默认值（分组）
students = [
    ("A班", "小明"),
    ("B班", "小红"),
    ("A班", "小刚"),
    ("B班", "小丽"),
]

classes = defaultdict(list)
for class_name, student in students:
    classes[class_name].append(student)

print(classes)
# {'A班': ['小明', '小刚'], 'B班': ['小红', '小丽']}

# 自定义工厂函数
import re
word_lengths = defaultdict(lambda: 0)
for word in text.split():
    word_lengths[word] += len(word)

from collections import Counter

words = [
    "python", "java", "python", "go", "python",
    "java", "rust", "go", "python"
]

# 统计频率
counter = Counter(words)
print(counter)
# Counter({'python': 4, 'java': 2, 'go': 2, 'rust': 1})

# 最常见的 2 个
print(counter.most_common(2))
# [('python', 4), ('java', 2)]

# 计数运算
counter1 = Counter(["a", "b", "c", "a"])
counter2 = Counter(["a", "b", "b", "d"])
print(counter1 + counter2)  # Counter({'a': 2, 'b': 3, 'c': 1, 'd': 1})
print(counter1 - counter2)  # Counter({'c': 1, 'a': 1})
print(counter1 & counter2)  # 交集: Counter({'a': 1, 'b': 1})
print(counter1 | counter2)  # 并集: Counter({'a': 2, 'b': 2, 'c': 1, 'd': 1})

from collections import deque

# 创建队列
queue = deque(["a", "b", "c"])
print(queue)  # deque(['a', 'b', 'c'])

# 两端操作
queue.append("d")           # 右端添加
queue.appendleft("z")       # 左端添加
print(queue)                # deque(['z', 'a', 'b', 'c', 'd'])

right = queue.pop()         # 右端弹出
left = queue.popleft()      # 左端弹出
print(f"弹出: 左={left}, 右={right}")

# 固定长度队列（自动丢弃旧元素）
recent = deque(maxlen=5)
for i in range(10):
    recent.append(f"item-{i}")
print(recent)
# deque(['item-5', 'item-6', 'item-7', 'item-8', 'item-9'], maxlen=5)

# 循环旋转
d = deque([1, 2, 3, 4, 5])
d.rotate(2)    # 右移2位
print(d)       # deque([4, 5, 1, 2, 3])
d.rotate(-2)   # 左移2位
print(d)       # deque([1, 2, 3, 4, 5])

from collections import OrderedDict

# Python 3.7+ 普通字典也保持插入顺序
# 但 OrderedDict 提供了额外方法

od = OrderedDict()
od["a"] = 1
od["b"] = 2
od["c"] = 3

# 移动到末尾/开头
od.move_to_end("a")           # 将 'a' 移到末尾
print(od)                     # OrderedDict([('b', 2), ('c', 3), ('a', 1)])

od.move_to_end("a", last=False)  # 将 'a' 移到开头
print(od)                     # OrderedDict([('a', 1), ('b', 2), ('c', 3)])

# 弹出最后/第一个元素
od.popitem(last=True)         # 弹出 ('c', 3)
od.popitem(last=False)        # 弹出 ('a', 1)

from collections import namedtuple

# 创建命名元组类型
Point = namedtuple("Point", ["x", "y"])
Person = namedtuple("Person", "name age city")

# 使用
p = Point(10, 20)
print(p.x, p.y)      # 10 20
print(p[0], p[1])    # 也支持索引访问

person = Person("小明", 25, "北京")
print(f"{person.name}, {person.age}岁, 来自{person.city}")

# _make：从可迭代对象创建
data = ["小红", 30, "上海"]
person2 = Person._make(data)
print(person2)

# _asdict：转为字典
print(person._asdict())
# {'name': '小明', 'age': 25, 'city': '北京'}

from collections import ChainMap

# 将多个字典合并为一个视图
defaults = {"theme": "dark", "lang": "zh", "debug": False}
user_settings = {"theme": "light", "lang": "zh"}
runtime_overrides = {"debug": True}

# 优先级：runtime > user > defaults
settings = ChainMap(runtime_overrides, user_settings, defaults)

print(settings["theme"])  # light（来自 user_settings）
print(settings["debug"])  # True（来自 runtime_overrides）
print(settings["lang"])   # zh

# 添加新配置
settings["font_size"] = 14
print(font_size 在: {settings.maps[0]})

from itertools import count, cycle, repeat, chain, groupby, permutations, combinations, product
import operator

# count：无限计数
for i in count(10, 2):  # 从10开始，步长2
    if i > 20:
        break
    print(i, end=" ")   # 10 12 14 16 18 20

# cycle：无限循环
colors = cycle(["红", "绿", "蓝"])
for i, color in enumerate(colors):
    if i >= 6:
        break
    print(color, end=" ")  # 红 绿 蓝 红 绿 蓝

# repeat：重复
for item in repeat("A", 3):
    print(item, end=" ")   # A A A

# chain：连接多个迭代器
result = list(chain([1, 2, 3], [4, 5], [6]))
print(result)  # [1, 2, 3, 4, 5, 6]

# groupby：分组（需先排序）
data = [("A班", 85), ("A班", 92), ("B班", 78), ("B班", 88)]
data.sort(key=lambda x: x[0])

for class_name, group in groupby(data, key=lambda x: x[0]):
    scores = [score for _, score in group]
    print(f"{class_name}: {scores}")

# permutations：排列
items = [1, 2, 3]
print(list(permutations(items, 2)))
# [(1, 2), (1, 3), (2, 1), (2, 3), (3, 1), (3, 2)]

# combinations：组合
print(list(combinations(items, 2)))
# [(1, 2), (1, 3), (2, 3)]

# product：笛卡尔积
print(list(product([1, 2], ["a", "b"])))
# [(1, 'a'), (1, 'b'), (2, 'a'), (2, 'b')]

# 实用案例：累加
import operator
values = [1, 2, 3, 4, 5]
from itertools import accumulate
print(list(accumulate(values)))           # [1, 3, 6, 10, 15]
print(list(accumulate(values, operator.mul)))  # [1, 2, 6, 24, 120]

import functools
from typing import Callable

# partial：部分应用（固定部分参数）
def power(base, exponent):
    return base ** exponent

square = functools.partial(power, exponent=2)
cube = functools.partial(power, exponent=3)

print(square(5))   # 25
print(cube(5))     # 125

# 实用例子：调整回调函数参数
def send_email(to, subject, body):
    print(f"发送邮件到 {to}: {subject}")

send_to_admin = functools.partial(send_email, to="admin@example.com")
send_to_admin("系统通知", "服务器已重启")

# lru_cache：自动缓存
@functools.lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(100))
print(fibonacci.cache_info())
# CacheInfo(hits=98, misses=101, maxsize=128, currsize=101)

# singledispatch：单分派泛型
@functools.singledispatch
def process_data(data):
    """处理数据的基函数"""
    raise NotImplementedError(f"不支持的类型: {type(data)}")

@process_data.register(str)
def _(data: str):
    return f"处理字符串: {data.upper()}"

@process_data.register(int)
def _(data: int):
    return f"处理整数: {data * 2}"

@process_data.register(list)
def _(data: list):
    return f"处理列表: {sum(data)}"

print(process_data("hello"))  # 处理字符串: HELLO
print(process_data(42))       # 处理整数: 84
print(process_data([1, 2, 3]))  # 处理列表: 6

# wraps（用于装饰器，前面章节已详细讲解）
def my_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# reduce：归约
from functools import reduce
numbers = [1, 2, 3, 4, 5]
total = reduce(lambda x, y: x + y, numbers)
print(total)  # 15
# 等同于 sum(numbers)

import os

# 路径拼接（跨平台）
path = os.path.join("data", "images", "photo.jpg")
print(path)  # data\images\photo.jpg (Windows) 或 data/images/photo.jpg (Unix)

# 路径分解
full_path = r"C:\Users\admin\documents\report.txt"
print(os.path.dirname(full_path))   # C:\Users\admin\documents
print(os.path.basename(full_path))  # report.txt
print(os.path.splitext(full_path))  # ('C:\\Users\\admin\\documents\\report', '.txt')

# 路径检查
print(os.path.exists("config.json"))   # 文件/目录是否存在
print(os.path.isfile("config.json"))   # 是否是文件
print(os.path.isdir("config.json"))    # 是否是目录

# 文件信息
print(os.path.getsize("config.json"))  # 文件大小（字节）
print(os.path.getmtime("config.json")) # 最后修改时间（时间戳）

# 绝对路径
print(os.path.abspath("config.json"))

import os

# 环境变量
print(os.environ.get("PATH"))
print(os.environ.get("HOME", "/tmp"))

# 设置环境变量
os.environ["MY_CONFIG"] = "debug"

# 进程相关
print(os.getpid())        # 当前进程 ID
print(os.getcwd())        # 当前工作目录
# os.chdir("/tmp")        # 切换目录

# 目录操作
# os.mkdir("new_dir")     # 创建目录（父目录必须存在）
# os.makedirs("a/b/c")    # 创建目录（自动创建父目录）
# os.rmdir("empty_dir")   # 删除空目录

# 文件操作
# os.rename("old.txt", "new.txt")   # 重命名/移动
# os.remove("temp.txt")             # 删除文件

# 列出目录
entries = os.listdir(".")
print(entries[:5])  # 前5个条目

import sys

# 命令行参数
# 运行: python script.py arg1 arg2
print(f"脚本名: {sys.argv[0]}")
print(f"参数: {sys.argv[1:]}")

# Python 解释器信息
print(f"Python 版本: {sys.version}")
print(f"版本号: {sys.version_info}")
# sys.version_info(major=3, minor=13, micro=0, ...)

# 模块搜索路径
print(f"搜索路径: {sys.path[:3]}")  # 前3个

# 标准输入/输出/错误
sys.stdout.write("这是标准输出\n")
# sys.stderr.write("这是错误输出\n")

# 退出程序
# sys.exit(0)    # 正常退出
# sys.exit(1)    # 异常退出

# 获取当前编码
print(f"默认编码: {sys.getdefaultencoding()}")  # utf-8

# 内存大小
data = [1] * 1000000
print(f"列表大小: {sys.getsizeof(data)} 字节")

import json
from datetime import datetime

# Python 对象转 JSON
data = {
    "name": "小明",
    "age": 25,
    "scores": [85, 92, 78],
    "active": True,
    "address": None,
    "tags": ("student", "python")  # 元组会变成列表
}

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

# JSON 转 Python 对象
parsed = json.loads(json_str)
print(parsed["name"])   # 小明

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

with open("data.json", "r", encoding="utf-8") as f:
    loaded = json.load(f)

# 自定义编码器（处理 datetime 等类型）
class CustomEncoder(json.JSONEncoder):
    def default(self, obj):
        if isinstance(obj, datetime):
            return obj.isoformat()
        if isinstance(obj, bytes):
            return obj.decode("utf-8")
        return super().default(obj)

data_with_date = {
    "name": "事件",
    "time": datetime.now(),
    "data": b"binary data"
}

json_str = json.dumps(data_with_date, cls=CustomEncoder, ensure_ascii=False)
print(json_str)

import math
import statistics

# math 模块：数学函数
print(f"圆周率: {math.pi}")
print(f"自然常数: {math.e}")
print(f"无穷大: {math.inf}")

print(f"绝对值: {math.fabs(-5)}")           # 5.0
print(f"向上取整: {math.ceil(3.2)}")         # 4
print(f"向下取整: {math.floor(3.8)}")        # 3
print(f"截断: {math.trunc(3.8)}")             # 3
print(f"阶乘: {math.factorial(5)}")          # 120
print(f"最大公约数: {math.gcd(12, 18)}")      # 6

print(f"平方根: {math.sqrt(16)}")             # 4.0
print(f"幂运算: {math.pow(2, 10)}")           # 1024.0
print(f"对数: {math.log(100, 10)}")           # 2.0
print(f"自然对数: {math.log(math.e)}")         # 1.0

# 三角函数
print(f"sin(π/2): {math.sin(math.pi / 2)}")  # 1.0
print(f"角度转弧度: {math.radians(180)}")     # π

# statistics 模块：统计函数
data = [12, 15, 18, 20, 22, 25, 30]

print(f"均值: {statistics.mean(data)}")
print(f"中位数: {statistics.median(data)}")
print(f"众数: {statistics.mode([1, 1, 2, 3, 3, 3])}")  # 3
print(f"标准差: {statistics.stdev(data):.2f}")
print(f"方差: {statistics.variance(data):.2f}")

import random
import secrets

# random：伪随机数（适用于模拟、游戏等）
print(f"[0,1) 随机浮点数: {random.random()}")
print(f"[a,b) 随机整数: {random.randint(1, 10)}")
print(f"[a,b) 范围随机: {random.randrange(0, 100, 5)}")  # 5的倍数

fruits = ["苹果", "香蕉", "橙子", "葡萄"]
print(f"随机选一个: {random.choice(fruits)}")
print(f"随机选多个: {random.sample(fruits, 2)}")
random.shuffle(fruits)
print(f"打乱顺序: {fruits}")

# secrets：密码学安全的随机数（适用于安全场景）
print(f"安全随机 token: {secrets.token_hex(16)}")      # 32位十六进制
print(f"安全随机 URL: {secrets.token_urlsafe(16)}")     # URL 安全 base64
print(f"安全随机整数: {secrets.randbelow(100)}")         # [0, 100)

# 生成随机密码
def generate_password(length=12):
    alphabet = "abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789!@#$%^&*"
    return "".join(secrets.choice(alphabet) for _ in range(length))

print(f"随机密码: {generate_password()}")

import hashlib

# MD5（不推荐用于安全场景，速度太快）
text = "hello world"
md5_hash = hashlib.md5(text.encode()).hexdigest()
print(f"MD5: {md5_hash}")

# SHA-256（推荐）
sha256_hash = hashlib.sha256(text.encode()).hexdigest()
print(f"SHA-256: {sha256_hash}")

# 大文件的增量哈希
def hash_file(filepath: str, algorithm="sha256") -> str:
    """计算文件的哈希值"""
    h = hashlib.new(algorithm)
    with open(filepath, "rb") as f:
        while chunk := f.read(8192):  # 8KB 块
            h.update(chunk)
    return h.hexdigest()

# 密码加盐哈希
import secrets

def hash_password(password: str) -> tuple[str, str]:
    """返回 (salt, hashed_password)"""
    salt = secrets.token_hex(16)
    hashed = hashlib.sha256((salt + password).encode()).hexdigest()
    return salt, hashed

def verify_password(password: str, salt: str, hashed: str) -> bool:
    """验证密码"""
    return hashlib.sha256((salt + password).encode()).hexdigest() == hashed

salt, hashed = hash_password("mypassword123")
print(f"盐: {salt}")
print(f"哈希: {hashed}")
print(f"验证: {verify_password('mypassword123', salt, hashed)}")  # True
print(f"验证错误密码: {verify_password('wrong', salt, hashed)}")   # False

import argparse

def create_parser():
    parser = argparse.ArgumentParser(
        description="数据处理工具",
        formatter_class=argparse.RawDescriptionHelpFormatter,
        epilog="""
使用示例:
  python script.py input.csv -o output.json --verbose
  python script.py input.csv --encoding utf-8 --limit 100
        """
    )

    # 位置参数
    parser.add_argument("input", help="输入文件路径")

    # 可选参数
    parser.add_argument("-o", "--output", default="output.json",
                        help="输出文件路径 (默认: output.json)")
    parser.add_argument("--encoding", default="utf-8",
                        help="文件编码 (默认: utf-8)")
    parser.add_argument("--limit", type=int, default=0,
                        help="处理行数限制 (默认: 全部)")
    parser.add_argument("-v", "--verbose", action="store_true",
                        help="显示详细输出")
    parser.add_argument("--format", choices=["json", "csv", "yaml"],
                        default="json", help="输出格式")

    return parser

# 实际使用时取消注释
# parser = create_parser()
# args = parser.parse_args()
# print(f"输入: {args.input}, 输出: {args.output}")

from dataclasses import dataclass, field, asdict
from typing import Optional
from datetime import datetime

@dataclass
class User:
    """用户数据类"""
    id: int
    name: str
    email: str
    created_at: datetime = field(default_factory=datetime.now)
    is_active: bool = True
    tags: list[str] = field(default_factory=list)
    scores: Optional[list[int]] = None

    def __post_init__(self):
        """初始化后的验证"""
        if "@" not in self.email:
            raise ValueError(f"无效邮箱: {self.email}")
        if self.scores is None:
            self.scores = []

    @property
    def average_score(self) -> float:
        if not self.scores:
            return 0.0
        return sum(self.scores) / len(self.scores)

# 创建实例
user = User(
    id=1,
    name="小明",
    email="xiao@example.com",
    tags=["vip", "python"],
    scores=[85, 92, 78]
)

print(user)
print(f"平均分: {user.average_score:.1f}")
print(f"字典: {asdict(user)}")

# 冻结（不可变）数据类
@dataclass(frozen=True)
class Config:
    host: str
    port: int = 8080
    debug: bool = False

config = Config("localhost", 3000)
# config.port = 4000  # FrozenInstanceError
print(config)

from enum import Enum, auto, IntEnum

class Color(Enum):
    """颜色枚举"""
    RED = "红色"
    GREEN = "绿色"
    BLUE = "蓝色"

class Status(IntEnum):
    """状态码（整数值）"""
    PENDING = 1
    PROCESSING = 2
    COMPLETED = 3
    FAILED = 4

class Priority(Enum):
    """自动赋值"""
    LOW = auto()
    MEDIUM = auto()
    HIGH = auto()
    CRITICAL = auto()

# 使用枚举
print(Color.RED)               # Color.RED
print(Color.RED.value)         # 红色
print(Color.RED.name)          # RED

# 遍历
for color in Color:
    print(f"{color.name}: {color.value}")

# 比较
print(Status.PENDING < Status.COMPLETED)  # True
print(Priority.HIGH > Priority.LOW)       # True

# 枚举作为配置
def process_order(status: Status):
    if status == Status.PENDING:
        print("订单待处理")
    elif status == Status.PROCESSING:
        print("订单处理中")
    elif status == Status.COMPLETED:
        print("订单已完成")

process_order(Status.PENDING)

import json
import os
from collections import ChainMap
from dataclasses import dataclass, field
from typing import Optional

@dataclass
class AppConfig:
    """应用配置管理"""
    host: str = "0.0.0.0"
    port: int = 8080
    debug: bool = False
    database_url: str = "sqlite:///app.db"
    log_level: str = "INFO"
    max_workers: int = 4
    allowed_hosts: list[str] = field(default_factory=lambda: ["*"])

    @classmethod
    def load(cls, config_path: Optional[str] = None) -> "AppConfig":
        """加载配置：默认值 < 配置文件 < 环境变量"""
        # 1. 配置文件
        file_config = {}
        if config_path and os.path.exists(config_path):
            with open(config_path, encoding="utf-8") as f:
                file_config = json.load(f)

        # 2. 环境变量（前缀 APP_）
        env_config = {}
        for key, value in os.environ.items():
            if key.startswith("APP_"):
                env_key = key[4:].lower()
                env_config[env_key] = value

        # 3. 合并（环境变量优先级最高）
        merged = ChainMap(env_config, file_config)
        return cls(**{k: v for k, v in merged.items() if k in cls.__dataclass_fields__})

# 实际使用
# config = AppConfig.load("config.json")
# print(config)