「学习笔记」Python 进阶(一)：语言特质 — 写出地道 Python

# 变量标注
name: str = "HelloWorld"
count: int = 0
scores: list[float] = [90.5, 85.0, 92.3]

# 函数标注
def greet(name: str, times: int = 1) -> str:
    return f"Hello {name}! " * times

# 多返回值标注
def split_path(path: str) -> tuple[str, str]:
    import os
    return os.path.split(path)

from typing import Any, Optional, Callable, TypeVar, Generic

# Any — 任意类型（等于"不标注"，但语义更明确）
def parse(data: Any) -> Any: ...

# Optional — 可选类型，等价于 X | None
def find_user(user_id: int) -> Optional[str]: ...

# Callable — 可调用对象类型
# Callable[[参数类型...], 返回类型]
callback: Callable[[int, str], bool] = lambda i, s: i > len(s)

# TypeVar + Generic — 泛型（对标 Java <T>）
T = TypeVar("T")

def first(items: list[T]) -> T:
    return items[0]

class Stack(Generic[T]):
    def __init__(self) -> None:
        self._items: list[T] = []
    def push(self, item: T) -> None:
        self._items.append(item)
    def pop(self) -> T:
        return self._items.pop()

# 联合类型：X | Y 替代 Union[X, Y]
def process(data: str | bytes | dict) -> str: ...

# 内置泛型：list[str] 替代 List[str]
def count_words(text: str) -> dict[str, int]: ...

# 元组类型
def min_max(nums: list[int]) -> tuple[int, int]:
    return min(nums), max(nums)

# 模块级类型别名
Vector = list | np.ndarray     # 用 numpy 数组或普通列表表示向量

from typing import TypeAlias

# 语义化基本类型：让代码自解释
UserId: TypeAlias = int           # 明确"这是一个用户ID"，而不只是 int
UserName: TypeAlias = str         # 明确"这是一个用户名"，而不只是 str
JsonDict: TypeAlias = dict[str, "JsonDict"]  # 递归类型别名：嵌套 JSON

# 使用：类型清晰，一目了然
def get_user(user_id: UserId) -> UserName:
    # 虽然本质还是 int → str，但调用方一看就懂
    ...

# 简化复杂类型：避免重复写一长串
from typing import Callable, Coroutine

TimeoutException = Type[BaseException] | BaseException
OnTimeoutCallback = Callable[..., Any] | Coroutine[Any, Any, Any]
# 现在 OnTimeoutCallback 就是一个简洁的类型名，可以在多处复用

def fibonacci(n):
    """生成前 n 个斐波那契数"""
    a, b = 0, 1
    for _ in range(n):
        yield a          # 产出值，暂停执行，保存状态
        a, b = b, a + b  # 下次调用时从此处恢复

gen = fibonacci(5)
print(next(gen))  # 0
print(next(gen))  # 1
print(next(gen))  # 1
print(next(gen))  # 2
print(next(gen))  # 3（协议终止）

# 再次使用生成器
for num in fibonacci(10):
   print(num, end=' ') # 输出: 0 1 1 2 3 5 8 13 21 34

# 列表推导式：立即计算所有值，占用全部内存
squares_list = [x ** 2 for x in range(1_000_000)]  # ~8MB

# 生成器表达式：惰性计算，几乎不占内存
squares_gen = (x ** 2 for x in range(1_000_000))   # ~120 字节

# 可以传给 sum/max/min/any/all 等，不创建中间列表
total = sum(x ** 2 for x in range(1_000_000))

def flatten(nested):
    """展平嵌套列表"""
    for item in nested:
        if isinstance(item, (list, tuple)):
            yield from flatten(item)  # 递归委托：逐层展开
        else:
            yield item

list(flatten([1, [2, [3, 4], 5], 6]))
# [1, 2, 3, 4, 5, 6]

# 生成器不仅能产出值，还能从外部接收值
def accumulator():
    total = 0
    while True:
        value = yield total        # yield 右侧 = 产出的值
        total += value             # yield 左侧 = 接收 send() 的值

gen = accumulator()
next(gen)          # 0（启动生成器）
gen.send(10)       # 10（接收 10，产出累计值）
gen.send(20)       # 30
gen.send(5)        # 35

    """流式聊天：逐块产出 LLM 响应"""
    response = self.client.chat.completions.create(
        model=self.model_name,
        messages=history,
        stream=True,           # 关键：开启流式
        **gen_conf
    )
    ans = ""
    for resp in response:
        if not resp.choices:
            continue
        delta = resp.choices[0].delta.content or ""
        ans += delta
        yield ans, len(ans)    # 每次产出一段，外部逐字渲染

# 调用方：
for answer, length in model.chat_streamly(history, config):
    print(f"\r{answer}", end="")  # 打字机效果

# 原理等价于：func = decorator(func)

def my_decorator(func):
    def wrapper(*args, **kwargs):
        print(f"调用前: {func.__name__}")
        result = func(*args, **kwargs)      # 调用原函数
        print(f"调用后: {func.__name__}")
        return result                        # 可以修改返回值
    return wrapper

@my_decorator
def say_hello(name):
    print(f"Hello, {name}!")

say_hello("World")
# 调用前: say_hello
# Hello, World!
# 调用后: say_hello

from functools import wraps

def my_decorator(func):
    @wraps(func)  # 必须加！把 func 的元信息复制到 wrapper
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@my_decorator
def say_hello(name):
    """打招呼"""...

print(say_hello.__name__)  # 有 @wraps → "say_hello"，没有 → "wrapper"
print(say_hello.__doc__)   # 有 @wraps → "打招呼"，没有 → None

from functools import wraps
import time

def retry(max_retries=3, delay=1.0):
    """带参数的装饰器 → 三层嵌套"""
    def decorator(func):                 # 第二层：接收函数
        @wraps(func)
        def wrapper(*args, **kwargs):     # 第三层：执行逻辑
            for attempt in range(max_retries + 1):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if attempt < max_retries:
                        time.sleep(delay * (2 ** attempt))  # 指数退避
                        continue
                    raise
        return wrapper
    return decorator                      # 第一层：返回装饰器

@retry(max_retries=5, delay=0.5)
def call_api(url): ...

# 等价于: call_api = retry(max_retries=5, delay=0.5)(call_api)

@decorator_a   # 第二步应用（最外层），执行时先进
@decorator_b   # 第一步应用（最内层），执行时后进
def func(): ...

# 等价于: func = decorator_a(decorator_b(func))

# Web 框架中的典型组合：
@manager.route("/conversation/set", methods=["POST"])  # 路由注册
@login_required                                        # 登录鉴权
@validate_request("name", "dialog_id")                 # 参数验证
def set_conversation():
    ...

from functools import wraps
from flask import request

def validate_request(*required_args):
    """验证请求中是否包含必需参数"""
    def wrapper(func):
        @wraps(func)
        def decorated_function(*_args, **_kwargs):
            input_arguments = request.json or request.form.to_dict()
            missing = [arg for arg in required_args if arg not in input_arguments]
            if missing:
                return {"error": f"Missing: {missing}"}, 400
            return func(*_args, **_kwargs)
        return decorated_function
    return wrapper

@manager.route("/site", methods=["POST"])
@validate_request("name", "url")
def create_site():
    req = request.json
    # 此时 name 和 url 一定存在
    ...

class DatabaseConnection:
    def __init__(self, db_url):
        self.db_url = db_url
        self.conn = None

    def __enter__(self):
        """进入 with 块时调用，返回值赋给 as 变量"""
        self.conn = connect(self.db_url)
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        """离开 with 块时调用（无论是否异常）"""
        if self.conn:
            self.conn.close()       # 确保资源释放
        return False                 # False/None → 异常继续传播
                                     # True → 吞掉异常（慎用）

with DatabaseConnection("mysql://localhost/db") as conn:
    conn.execute("SELECT 1")
# 离开 with 块后，__exit__ 自动调用

from contextlib import contextmanager

@contextmanager
def timer(label):
    import time
    start = time.time()
    try:
        yield                           # yield 之前 = __enter__
    finally:                            # yield 之后 = __exit__
        elapsed = time.time() - start
        print(f"{label}: {elapsed:.2f}s")

with timer("数据处理"):
    data = [x ** 2 for x in range(1_000_000)]
# 数据处理: 0.15s

from contextlib import suppress, ExitStack

# suppress — 忽略指定异常
with suppress(FileNotFoundError):
    os.remove("temp.txt")  # 文件不存在也不报错

# ExitStack — 动态管理多个上下文管理器
with ExitStack() as stack:
    files = [stack.enter_context(open(p)) for p in file_paths]
    # 离开 with 块时，所有文件自动关闭

from functools import lru_cache

@lru_cache(maxsize=128)
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n - 1) + fibonacci(n - 2)

print(fibonacci(100))  # 瞬间返回（有缓存时 O(1)）
print(fibonacci.cache_info())  # CacheInfo(hits=98, misses=101, ...)

# 实际使用：缓存 API 调用结果（如 Slack workspace URL）
@lru_cache()
def get_base_url(token: str) -> str:
    client = WebClient(token=token)
    return client.auth_test()["url"]

from functools import partial

# 预绑定部分参数，创建新函数
dumps_utf8 = partial(json.dumps, ensure_ascii=False)
dumps_pretty = partial(json.dumps, ensure_ascii=False, indent=2)

# 实际使用：全局替换 requests 的 JSON 编码器
requests.models.complexjson.dumps = functools.partial(
    json.dumps, cls=CustomJSONEncoder
)

from functools import reduce

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

# 拼接路径
path = reduce(lambda a, b: f"{a}/{b}", ["home", "user", "docs", "file.txt"])
# "home/user/docs/file.txt"

# 图像处理管道：每个步骤都是可调用对象
# 这是一种优雅的"责任链模式"实现

class ResizeImage:
    def __init__(self, target_size=(640, 640)):
        self.target_size = target_size

    def __call__(self, data):
        """使实例可像函数一样调用"""
        data['image'] = self._resize(data['image'])
        return data

class Pipeline:
    def __init__(self, *steps):
        self.steps = steps

    def __call__(self, data):
        for step in self.steps:
            data = step(data)       # 每个 step 都是可调用对象
        return data

pipe = Pipeline(
    ResizeImage((640, 640)),
    NormalizeImage(mean=[0.485], std=[0.229]),
)
result = pipe(raw_data)

class SparseVector:
    def __init__(self):
        self._data = {}

    def __getitem__(self, index: int) -> float:
        return self._data.get(index, 0.0)   # 不存在返回默认值

    def __setitem__(self, index: int, value: float):
        self._data[index] = value

    def __contains__(self, index: int) -> bool:
        return index in self._data

vec = SparseVector()
vec[0] = 0.5
vec[3] = 0.8
print(vec[0])        # 0.5
print(vec[1])        # 0.0（默认值）
print(3 in vec)      # True

class PageIterator:
    """分页迭代器：逐页加载数据，对外像普通迭代器一样使用"""
    def __init__(self, fetch_func, page_size=100):
        self.fetch = fetch_func
        self.page_size = page_size
        self.current_page = 0
        self.buffer = []
        self.exhausted = False

    def __iter__(self):
        return self

    def __next__(self):
        if not self.buffer and not self.exhausted:
            self.current_page += 1
            self.buffer = self.fetch(self.current_page, self.page_size)
            if not self.buffer:
                self.exhausted = True
        if not self.buffer:
            raise StopIteration
        return self.buffer.pop(0)

for item in PageIterator(fetch_from_db, page_size=50):
    process(item)

from pathlib import Path

# 创建路径对象
p = Path.home()       # 用户主目录
p = Path.cwd()        # 当前工作目录

# 路径拼接（用 / 运算符，跨平台安全）
config = Path("conf") / "service_conf.yaml"
# Windows → conf\service_conf.yaml
# Linux   → conf/service_conf.yaml

# 路径属性
p = Path("/home/user/docs/file.txt")
print(p.name)         # file.txt
print(p.stem)         # file（不含扩展名）
print(p.suffix)       # .txt
print(p.parent)       # /home/user/docs
print(p.parts)        # ('/', 'home', 'user', 'docs', 'file.txt')

# 判断
p.exists() / p.is_file() / p.is_dir()

# Path 对象自带读写方法
p = Path("output.txt")
p.write_text("Hello", encoding="utf-8")
content = p.read_text(encoding="utf-8")

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

# glob 模式匹配
for f in Path(".").glob("*.py"):     # 当前目录所有 .py 文件
    print(f)

for f in Path(".").rglob("*.py"):    # 递归搜索所有子目录
    print(f)

# 实际使用：自动发现所有 API 路由文件
pages_dir = Path(__file__).parent
for path in pages_dir.glob("*_app.py"):
    if not path.name.startswith("."):
        print(f"Found route: {path.stem}")

# 传统写法：先赋值再判断
line = input()
while line != "quit":
    process(line)
    line = input()

# 海象运算符：在条件中赋值
while (line := input()) != "quit":
    process(line)

# 在 if 语句中：避免重复 .get()
if (value := req.get("name")):
    use(value)

# 在列表推导式中：避免重复计算
results = [
    email for owner in file.get("owners", [])
    if (email := owner.get("emailAddress"))
    and "@" in email
]

# 组合使用：一次遍历，避免重复调用
upper_words = [
    wu for w in text.split()
    if len(w) > 3 and (wu := w.upper())
]