def foo(x=[]):  # 危险！
    x.append(1)
    return x

foo()  # [1]
foo()  # [1, 1]  ← 不是 [1]

a = 10
print(id(a))  # 假设 140735281575776
a += 1
print(id(a))  # 假设 140735281575808  ← id 变了，说明是新对象

def append_item(item, lst=[]):
    lst.append(item)
    return lst

print(append_item(1))  # [1]
print(append_item(2))  # [1, 2]  ← 默认列表被复用了！

import copy

a = [[1, 2], [3, 4]]
b = copy.copy(a)    # 浅拷贝
c = copy.deepcopy(a) # 深拷贝

a[0].append(99)
# b[0] → [1, 2, 99]  受影响
# c[0] → [1, 2]      不受影响

original = [[1, 2, 3], [4, 5, 6]]

# 赋值 — 只是贴了个新标签
assigned = original

# 浅拷贝 — 新瓶子，旧内容
shallow = original[:]   # 等价于 copy.copy(original)

# 深拷贝 — 全部独立
deep = copy.deepcopy(original)

original[0][0] = 999
# assigned[0][0] → 999  （同一对象）
# shallow[0][0] → 999   （内层列表是共享的）
# deep[0][0] → 1        （完全独立）

a = [1, 2]
b = [1, 2]
a == b  # True   值相等
a is b  # False  不同对象

# 小整数缓存 [-5, 256]
x = 256; y = 256
x is y  # True（CPython 缓存）

x = 257; y = 257
x is y  # False（超出缓存范围）

import sys
sys.getswitchinterval()  # 默认 0.005 秒 = 5ms

def func(a, b, *args, key=None, **kwargs):
    pass

func(1, 2, 3, 4, key='v', x=10, y=20)
# a=1, b=2, args=(3,4), key='v', kwargs={'x':10, 'y':20}

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

args = [1, 2, 3]
add(*args)  # 6

kwargs = {'a': 1, 'b': 2, 'c': 3}
add(**kwargs)  # 6

def f(
    pos_only,          # 位置参数（无默认值）
    /,                 # 位置参数到此为止
    pos_or_kw,         # 可位置可关键字
    *args,             # 可变位置参数（位置参数到此为止）
    kw_only,           # 关键字参数开始
    default=None,      # 带默认值的关键字参数
    **kwargs           # 可变关键字参数
):
    pass

# 列表/元组解包
first, *middle, last = [1, 2, 3, 4, 5]
# first=1, middle=[2,3,4], last=5

# 合并多个列表
combined = [*list1, *list2]

# 合并多个字典
merged = {**dict1, **dict2}

# 基本装饰器
def timer(func):
    @functools.wraps(func)  # 保留原函数元信息
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        print(f'{func.__name__} took {time.time()-start:.4f}s')
        return result
    return wrapper

@timer  # 等价于 foo = timer(foo)
def foo():
    pass

def retry(max_retries=3):
    def decorator(func):
        @functools.wraps(func)
        def wrapper(*args, **kwargs):
            for i in range(max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    if i == max_retries - 1:
                        raise
        return wrapper
    return decorator

@retry(max_retries=5)
def call_api():
    pass

@timer
def foo():
    pass
# 此时 timer(foo) 已经执行了，wrapper 函数已经创建好了

@retry(max_retries=5)  # 1. 调用 retry(5) → 返回 decorator
def call_api():        # 2. decorator(call_api) 被调用 → 返回 wrapper
    pass
# call_api 现在 = wrapper

def bad_decorator(func):
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

@bad_decorator
def hello():
    """Say hello"""
    pass

hello.__name__  # 'wrapper'  ← 丢失了！
hello.__doc__   # None       ← 丢失了！

@decorator_a
@decorator_b
def func():
    pass
# 等价于 decorator_a(decorator_b(func))

# 列表推导式 —— 一次性创建完整列表，占用内存
[x * 2 for x in range(1000000)]

# 生成器表达式 —— 惰性求值，每次只产出一个元素
(x * 2 for x in range(1000000))

# 编译期就知道这是一个完整的列表创建
result = [x * 2 for x in range(1000000)]
# 内存中立即创建了一个含 100 万个元素的列表

result = (x * 2 for x in range(1000000))
# 几乎没有占用内存！只是创建了一个生成器对象
# 只有在迭代时才逐个计算

sum(x**2 for x in range(100))  
# 等价于 sum((x**2 for x in range(100)))
# 当生成器表达式是函数的唯一参数时，外层括号可以省略

reduce(lambda a, b: a + b, (x**2 for x in range(100)))  # 不能省略

matrix = [[1, 2], [3, 4], [5, 6]]
# 下面两段代码等价
flat1 = [num for row in matrix for num in row]   # 从左到右嵌套
# 等价于
flat2 = []
for row in matrix:
    for num in row:
        flat2.append(num)

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, name):
        # 注意：每次实例化都会执行
        self.name = name

MyClass() 调用
    ↓
1. __new__(cls)  → 分配内存，返回新实例
    ↓
2. __init__(self) → 初始化实例属性
    ↓
返回实例

class PositiveInteger(int):
    def __new__(cls, value):
        instance = super().__new__(cls, abs(value))
        return instance

p = PositiveInteger(-5)  # p == 5

class Singleton:
    _instance = None
    _initialized = False

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

    def __init__(self, name):
        if not getattr(self, '_initialized', False):
            self.name = name
            self._initialized = True

class DatabaseConnection:
    def __enter__(self):
        self.conn = create_connection()
        return self.conn

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.conn.close()
        return False  # 异常继续传播

from contextlib import contextmanager

@contextmanager
def db_connection():
    conn = create_connection()
    try:
        yield conn
    finally:
        conn.close()

with EXPR as VAR:
    BLOCK

# 等价于
manager = EXPR          # 1. 获取上下文管理器对象
value = manager.__enter__()  # 2. 调用 __enter__，返回值赋给 VAR
try:
    BLOCK               # 3. 执行代码块
except Exception as e:
    if not manager.__exit__(type(e), e, e.__traceback__):
        raise           # 4. 异常时：调用 __exit__，如果返回 False 则继续抛
else:
    manager.__exit__(None, None, None)  # 5. 无异常：正常调用 __exit__

class IgnoreValueError:
    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        if exc_type is ValueError:
            print(f'忽略 ValueError: {exc_val}')
            return True   # ← 返回 True 表示异常已被处理
        return False

with IgnoreValueError():
    raise ValueError('这个错误会被忽略')
print('程序继续执行！')  # 这句会执行

@contextmanager
def my_context():
    print('Enter')      # __enter__ 部分
    yield 'value'
    print('Exit')       # __exit__ 部分（正常退出时）

@contextmanager
def atomic_write(filepath):
    """原子写入：先写临时文件，成功后再重命名"""
    tmp = filepath + '.tmp'
    try:
        yield open(tmp, 'w')
        os.replace(tmp, filepath)
    except:
        os.remove(tmp)
        raise

# 可以
f = lambda x: x ** 2

# 条件表达式可以（这是表达式不是语句）
f = lambda x: x if x > 0 else -x

# 不可以
# lambda x: if x > 0: return x  ← 语法错误

# 除了语法，还有功能差异
f = lambda x: x**2
# f.__name__ → '<lambda>'  ← 调试困难！

def g(x):
    return x**2
# g.__name__ → 'g'

sorted(data, key=lambda x: x['age'])
filter(lambda x: x > 0, nums)
map(lambda x: x.strip(), lines)

# 错误
funcs = [lambda: i for i in range(5)]
[func() for func in funcs]  # [4, 4, 4, 4, 4] ← 全是 4！

# 正确
funcs = [lambda i=i: i for i in range(5)]
[func() for func in funcs]  # [0, 1, 2, 3, 4]

x = 10

def outer():
    y = 20
    def inner():
        nonlocal y  # 引用 outer 的 y
        y += 1
        global x    # 引用模块的 x
        x += 1
    inner()

x = 10

def func():
    x = 5   # 这里创建了局部变量 x，不会修改全局 x！
    print(x)  # 5

func()
print(x)   # 10  ← 全局 x 没变

def func():
    global x
    x = 5   # 现在修改的是全局 x

# 错误写法
def counter():
    count = 0
    def inc():
        count += 1  # UnboundLocalError!
        return count
    return inc

# 正确写法
def counter():
    count = 0
    def inc():
        nonlocal count
        count += 1
        return count
    return inc

c = counter()
c()  # 1
c()  # 2

class Point:
    __slots__ = ('x', 'y')

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

p = Point(1, 2)
p.z = 3  # AttributeError! 不能动态添加属性

普通实例的内存布局：
[ob_refcnt | ob_type | __dict__指针 → {"x": 1, "y": 2, ...}]

__slots__ 实例的内存布局：
[ob_refcnt | ob_type | x=1 | y=2 | ...]
          无 __dict__，属性直接内联在实例内存中

class Animal:
    __slots__ = ('name',)

class Dog(Animal):
    pass  # 子类没有定义 __slots__，所以 Dog 实例有 __dict__！

d = Dog()
d.anything = 'works'  # 因为 Dog 有 __dict__

class Dog(Animal):
    __slots__ = ('breed',)  # 子类也要定义

# yield from 简化前
def chain(*iterables):
    for it in iterables:
        for x in it:
            yield x

# yield from 简化后
def chain(*iterables):
    for it in iterables:
        yield from it

def gen():
    print('开始')
    yield 1
    print('继续')
    yield 2
    print('结束')

g = gen()           # 没有输出！
print(next(g))      # 开始 → 1
print(next(g))      # 继续 → 2
print(next(g))      # 结束 → StopIteration

def sub_gen():
    received = yield 'ready'
    print(f'Sub got: {received}')
    return 'done'

def main_gen():
    result = yield from sub_gen()
    print(f'Main got result: {result}')

g = main_gen()
print(next(g))            # 'ready'
print(g.send('hello'))    # Sub got: hello → Main got result: done → StopIteration

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("Radius cannot be negative")
        self._radius = value

    @property
    def area(self):
        return 3.14159 * self._radius ** 2

c = Circle(5)
c.radius = 10   # 调用 setter
print(c.area)   # 调用 getter，只读属性

# property 的简化实现
class Property:
    def __init__(self, fget=None, fset=None, fdel=None, doc=None):
        self.fget = fget
        self.fset = fset
        self.fdel = fdel
        self.__doc__ = doc

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        if self.fget is None:
            raise AttributeError("unreadable attribute")
        return self.fget(obj)

    def __set__(self, obj, value):
        if self.fset is None:
            raise AttributeError("can't set attribute")
        self.fset(obj, value)

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

    # 不推荐
    def get_upper_name(self):
        return self.name.upper()

    # 推荐
    @property
    def upper_name(self):
        return self.name.upper()

p = Person('Alice')
p.get_upper_name()    # 需要括号
p.upper_name          # 不需要，像属性一样

def read_data(source):
    # 不检查 source 的类型，只要有 read 方法即可
    return source.read()

# 以下都可以传入
read_data(open('file.txt'))     # 文件对象
read_data(io.StringIO('hello')) # StringIO
read_data(requests.Response())  # HTTP 响应

# Java（静态类型）
void process(Animal animal) {
    animal.makeSound();  // animal 必须是 Animal 类型
}

# Python（鸭子类型）
def process(animal):
    animal.make_sound()  # 任何有 make_sound 方法的对象都可以

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

add(1, 2)        # 3（数值加法）
add('Hello', ' World')  # 'Hello World'（字符串拼接）
add([1], [2])    # [1, 2]（列表合并）
add(1, 'hello')  # TypeError！运行时才暴露

from abc import ABC, abstractmethod

class Readable(ABC):
    @abstractmethod
    def read(self): ...

def read_data(source: Readable):  # 名义上是类型提示，实际还是鸭子类型
    return source.read()

from typing import Protocol

class Readable(Protocol):
    def read(self) -> str: ...

def read_data(source: Readable) -> str:
    return source.read()

# LBYL（Look Before You Leap）— 不推荐
if hasattr(obj, 'read'):
    data = obj.read()
else:
    handle_error()

# EAFP（Easier to Ask for Forgiveness than Permission）— Pythonic
try:
    data = obj.read()
except AttributeError:
    handle_error()

indices: [None, 0, None, 1, None, 2]
entries: [(hash1, key1, value1), (hash2, key2, value2), (hash3, key3, value3)]

旧方案（3.6 前）：
[hash1, key1, val1, 空, hash2, key2, val2, 空, ...]
多个空槽位，内存浪费严重

新方案（3.6+）：
indices（稀疏，占位少）:  [3, None, 0, None, 1]
entries（紧凑，无空位）:  [(hash1, k1, v1), (hash2, k2, v2), (hash3, k3, v3)]
                          ↑ 插入顺序就是存储顺序

# 伪代码
def find_entry(dict, key):
    idx = hash(key) % len(indices)
    while indices[idx] is not None:
        entry_idx = indices[idx]
        if entries[entry_idx].key == key:
            return entry_idx
        # 线性探测（实际是伪随机探测）
        idx = (idx * 5 + 1) % len(indices)
    raise KeyError(key)

内存布局：
[ptr0 | ptr1 | ptr2 | ptr3 | ... | ptrN]
   ↓      ↓      ↓      ↓           ↓
  obj0   obj1   obj2   obj3  ...   objN

# 简化的逻辑
new_allocated = new_size + (new_size >> 3) + (3 if new_size < 9 else 6)

lst = [1, 2, 3, 4, 5]
lst.insert(0, 0)
# 内部操作：
# [1, 2, 3, 4, 5, _]  ← 先扩容
# [1, 2, 3, 4, 4, 5]  ← 后移
# [1, 2, 3, 3, 4, 5]
# [1, 2, 2, 3, 4, 5]
# [1, 1, 2, 3, 4, 5]
# [0, 1, 2, 3, 4, 5]  ← 插入新值

# 不可哈希的对象不能放入 set
s = set()
s.add([1, 2, 3])  # TypeError: unhashable type: 'list'

class Bad:
    def __init__(self, x):
        self.x = x
    def __hash__(self):
        return hash(self.x)
    def __eq__(self, other):
        return self.x == other.x

b = Bad(1)
s = {b}
b.x = 2          # 修改了 b 的哈希值
print(b in s)    # False！因为 b 的哈希值变了，找不到了
s.add(b)         # 现在 s 中有两个"逻辑相同"的对象

hash((1, 2, 3))        # OK
hash((1, [2], 3))      # TypeError! 因为 tuple 中包含 list

from functools import lru_cache

@lru_cache(maxsize=128)
def expensive_func(n):
    return n ** n

from collections import OrderedDict

class LRUCache:
    def __init__(self, capacity):
        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)

# 它的底层也是 OrderedDict（Python 3.3+）
# 不同点：支持 typed 参数（True 时 1 和 1.0 视为不同 key）
# 支持 maxsize=None（无限制缓存）

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

# 迭代法
def reverse_list(head):
    prev = None
    curr = head
    while curr:
        next_node = curr.next
        curr.next = prev
        prev = curr
        curr = next_node
    return prev

# 递归法
def reverse_list_recursive(head):
    if not head or not head.next:
        return head
    new_head = reverse_list_recursive(head.next)
    head.next.next = head
    head.next = None
    return new_head

初始：
prev → None
curr → 1 → 2 → 3 → None

第一步：
next_node = curr.next  # 保存下一个节点
curr.next = prev       # 当前节点指向前一个
prev = curr            # prev 前移
curr = next_node       # curr 前移

结果：
None ← 1    2 → 3 → None
       ↑    ↑
     prev  curr

reverse_list_recursive(head) 的功能：
1. 如果 head 为空或 head.next 为空，返回 head
2. 递归反转 head.next 之后的链表
3. 反转后 head.next 是原链表的尾节点，让它的 next 指向 head
4. 设置 head.next = None（原头节点的 next 指向空）

原始： 1 → 2 → 3 → 4 → None
           ↓
递归反转 2→3→4： head.next 指向 new_head=4
           ↓
1 → 2 ← 3 ← 4
    ↓         ↓
  head.next  new_head
    ↓
1 ← 2 ← 3 ← 4 → None（1 和 2 之间的箭头还是旧的）
    ↓
head.next.next = head  # 2.next → 1
head.next = None       # 1.next → None

结果：None ← 1 ← 2 ← 3 ← 4
                                ↑
                              new_head

def has_cycle(head):
    slow = fast = head
    while fast and fast.next:
        slow = slow.next
        fast = fast.next.next
        if slow == fast:
            ptr = head
            while ptr != slow:
                ptr = ptr.next
                slow = slow.next
            return ptr
    return None

class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

# 前序遍历（根-左-右）- 迭代
def preorder_iterative(root):
    res, stack = [], [root]
    while stack:
        node = stack.pop()
        if node:
            res.append(node.val)
            stack.append(node.right)
            stack.append(node.left)
    return res

# 中序遍历（左-根-右）- 迭代
def inorder_iterative(root):
    res, stack = [], []
    curr = root
    while curr or stack:
        while curr:
            stack.append(curr)
            curr = curr.left
        curr = stack.pop()
        res.append(curr.val)
        curr = curr.right
    return res

# 后序遍历（左-右-根）= 前序(根-右-左)的反转
def postorder_iterative(root):
    res, stack = [], [root]
    while stack:
        node = stack.pop()
        if node:
            res.append(node.val)
            stack.append(node.left)
            stack.append(node.right)
    return res[::-1]

    1
   / \n  2   3
 / \n4   5

前序（根→左→右）：[1, 2, 4, 5, 3]  — 先访问当前节点，再递归子树
中序（左→根→右）：[4, 2, 5, 1, 3]  — 先左子树，再当前节点，再右子树
后序（左→右→根）：[4, 5, 2, 3, 1]  — 先子树，再当前节点

stack.append(node.right)  # 右子树先进栈（后出）
stack.append(node.left)   # 左子树后进栈（先出）

模拟过程：
stack = [1, 2, 4]  ← 一直向左
pop 4 → visit 4 → 转向 4 的右（None）
pop 2 → visit 2 → 转向 2 的右（5）
stack = [1, 5]
pop 5 → visit 5 → 转向 5 的右（None）
pop 1 → visit 1 → 转向 1 的右（3）
stack = [3]
pop 3 → visit 3 → 转向 3 的右（None）

def quicksort_inplace(arr, low=0, high=None):
    if high is None:
        high = len(arr) - 1
    if low < high:
        pi = partition(arr, low, high)
        quicksort_inplace(arr, low, pi - 1)
        quicksort_inplace(arr, pi + 1, high)

def partition(arr, low, high):
    pivot = arr[high]
    i = low - 1
    for j in range(low, high):
        if arr[j] <= pivot:
            i += 1
            arr[i], arr[j] = arr[j], arr[i]
    arr[i + 1], arr[high] = arr[high], arr[i + 1]
    return i + 1

pivot = arr[high] = 5
arr = [3, 7, 8, 5, 2, 1, 9, 5, 4]
       ↑ i  ↑ j
i 指向"最后的 ≤5 的区域"的边界
j 遍历数组
遇到 ≤5 的元素，i+1 并交换

def partition_three_way(arr, low, high):
    """将数组分为 <pivot, =pivot, >pivot 三部分"""
    lt, gt = low, high
    pivot = arr[low]
    i = low
    while i <= gt:
        if arr[i] < pivot:
            arr[lt], arr[i] = arr[i], arr[lt]
            lt += 1; i += 1
        elif arr[i] > pivot:
            arr[i], arr[gt] = arr[gt], arr[i]
            gt -= 1
        else:
            i += 1
    return lt, gt  # 等于 pivot 的范围是 [lt, gt]

import heapq

def top_k_largest(nums, k):
    heap = nums[:k]
    heapq.heapify(heap)  # O(k)
    for num in nums[k:]:
        if num > heap[0]:
            heapq.heapreplace(heap, num)
    return heap

# 时间 O(n log k)，空间 O(k)

max_heap = [-x for x in data]  # 入堆时取负数
heapq.heapify(max_heap)
largest = -heapq.heappop(max_heap)  # 出堆时再转回来

数据：[3, 7, 2, 9, 5, 1, 8, 6, 4]，K=3

初始化堆 [3, 7, 2] → heapify
堆顶 = 2
遇到 9：2 < 9 → 替换 → [3, 7, 9]
堆顶 = 3
遇到 5：3 < 5 → 替换 → [5, 7, 9]
遇到 1：5 > 1 → 跳过
遇到 8：5 < 8 → 替换 → [7, 8, 9]
遇到 6：7 > 6 → 跳过
遇到 4：7 > 4 → 跳过
最终：[7, 8, 9]

def binary_search(nums, target):
    left, right = 0, len(nums) - 1
    while left <= right:
        mid = left + (right - left) // 2
        if nums[mid] == target:
            return mid
        elif nums[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

# 查找左边界
def lower_bound(nums, target):
    left, right = 0, len(nums)
    while left < right:
        mid = left + (right - left) // 2
        if nums[mid] < target:
            left = mid + 1
        else:
            right = mid
    return left

def lower_bound(nums, target):
    left, right = 0, len(nums)  # 注意 right = len(nums)，不是 len(nums)-1
    while left < right:
        mid = left + (right - left) // 2
        if nums[mid] < target:
            left = mid + 1  # mid 不够大，排除
        else:
            right = mid     # mid 可能是答案，保留
    return left  # left == right

# 合并两个有序链表
def merge_two_lists(l1, l2):
    dummy = ListNode()
    curr = dummy
    while l1 and l2:
        if l1.val <= l2.val:
            curr.next = l1
            l1 = l1.next
        else:
            curr.next = l2
            l2 = l2.next
        curr = curr.next
    curr.next = l1 or l2
    return dummy.next

# 合并两个有序数组（从后往前归并）
def merge(nums1, m, nums2, n):
    p1, p2, p = m - 1, n - 1, m + n - 1
    while p2 >= 0:
        if p1 >= 0 and nums1[p1] > nums2[p2]:
            nums1[p] = nums1[p1]
            p1 -= 1
        else:
            nums1[p] = nums2[p2]
            p2 -= 1
        p -= 1

dummy = ListNode()  # 哨兵节点，不用处理头节点为空的边界情况
curr = dummy        # curr 指向当前已合并链表的尾节点
return dummy.next   # 哨兵的下一个就是真正的头节点

nums1 = [1, 3, 5, 0, 0, 0]  m=3
nums2 = [2, 4, 6]            n=3

p1=2(指向5), p2=2(指向6), p=5
5 vs 6 → nums1[5]=6, p2=1, p=4
5 vs 4 → nums1[4]=5, p1=1, p=3
3 vs 4 → nums1[3]=4, p2=0, p=2
3 vs 2 → nums1[2]=3, p1=0, p=1
1 vs 2 → nums1[1]=2, p2=-1(结束)
最终：[1, 2, 3, 4, 5, 6]

class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word):
        node = self.root
        for ch in word:
            if ch not in node.children:
                node.children[ch] = TrieNode()
            node = node.children[ch]
        node.is_end = True

    def search(self, word):
        node = self._find(word)
        return node is not None and node.is_end

    def starts_with(self, prefix):
        return self._find(prefix) is not None

    def _find(self, prefix):
        node = self.root
        for ch in prefix:
            if ch not in node.children:
                return None
            node = node.children[ch]
        return node

root
 ├── a ── p ── p ── l ── e (is_end=True)
 │               └── s (is_end=True)      "apps"
 └── b ── a ── t (is_end=True)            "bat"
               └── h (is_end=True)         "bath"

class TrieNode:
    def __init__(self):
        self.children = [None] * 26  # 只支持 a-z
        self.is_end = False

from collections import Counter
c = Counter('abracadabra')
c.most_common(3)   # [('a', 5), ('b', 2), ('r', 2)]
c1 + c2            # 合并计数
c1 - c2            # 差集计数（只保留正数）

block1 ↔ block2 ↔ block3
[..., ...] [..., ...] [..., ...]
  ↑left              right↑

from collections import ChainMap

defaults = {'theme': 'light', 'lang': 'en'}
user_config = {'theme': 'dark'}

config = ChainMap(user_config, defaults)
config['theme']  # 'dark'  ← 先查 user_config
config['lang']   # 'en'    ← user_config 没有，查 defaults

import threading, queue

q = queue.Queue(maxsize=10)

def producer():
    for i in range(100):
        q.put(i)

def consumer():
    while True:
        item = q.get()
        q.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()
q.join()

async def producer(q):
    for i in range(100):
        await q.put(i)

async def consumer(q):
    while True:
        item = await q.get()
        q.task_done()

# Queue.put 的简化逻辑
def put(self, item, block=True, timeout=None):
    with self.mutex:
        if self.maxsize > 0 and self._qsize() >= self.maxsize:
            self.not_full.wait(timeout)
        self._put(item)
        self.not_empty.notify()

import heapq
import random

# 方法一：最小堆 O(n log k)
def find_kth_largest_heap(nums, k):
    return heapq.nlargest(k, nums)[-1]

# 方法二：快速选择 O(n) 平均（面试推荐）
def find_kth_largest(nums, k):
    def quickselect(left, right, k_smallest):
        if left == right:
            return nums[left]
        pivot_idx = random.randint(left, right)
        pivot_idx = partition(left, right, pivot_idx)
        if k_smallest == pivot_idx:
            return nums[k_smallest]
        elif k_smallest < pivot_idx:
            return quickselect(left, pivot_idx - 1, k_smallest)
        else:
            return quickselect(pivot_idx + 1, right, k_smallest)

    def partition(left, right, pivot_idx):
        pivot = nums[pivot_idx]
        nums[pivot_idx], nums[right] = nums[right], nums[pivot_idx]
        store = left
        for i in range(left, right):
            if nums[i] < pivot:
                nums[store], nums[i] = nums[i], nums[store]
                store += 1
        nums[store], nums[right] = nums[right], nums[store]
        return store

    return quickselect(0, len(nums) - 1, len(nums) - k)

T(n) = T(n/2) + O(n)  ← 每次只处理一半
递归展开：n + n/2 + n/4 + ... = 2n = O(n)

class Adder:
    def __init__(self, n):
        self.n = n

    def __call__(self, x):
        return self.n + x

add5 = Adder(5)
add5(10)  # 15，调用 __call__

callable(Adder(5))  # True
callable(42)        # False

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f'Called {self.count} times')
        return self.func(*args, **kwargs)

@CountCalls
def hello():
    print('Hello!')

hello()  # Called 1 times → Hello!
hello()  # Called 2 times → Hello!

class A:
    def method(self): return 'A'

class B(A):
    def method(self): return 'B'

class C(A):
    def method(self): return 'C'

class D(B, C):
    pass

D().method()  # 'B'
D.__mro__     # (D, B, C, A, object)

    A
   / \n  B   C
   \ /
    D

# 对于 D(B, C)
L(D) = D + merge(L(B), L(C), [B, C])

L(B) = B + merge(L(A), [A])
     = B + merge([A, O], [A])
     = B + A + merge([O])
     = B, A, O

L(C) = C, A, O

L(D) = D + merge([B, A, O], [C, A, O], [B, C])
     = D + B + merge([A, O], [C, A, O], [C])
     = D + B + C + merge([A, O], [A, O])
     = D + B + C + A + merge([O], [O])
     = D, B, C, A, O

D.__mro__  # (D, B, C, A, object)
D.mro()    # 同上

class A:
    def method(self):
        print('A')
        super().method()

class B(A):
    def method(self):
        print('B')
        super().method()

class C(A):
    def method(self):
        print('C')

class D(B, C):
    def method(self):
        print('D')
        super().method()

D().method()
# 输出: D → B → C → A
# MRO: D → B → C → A → object

class A:
    def method(self):
        print('A')

class B(A):
    def method(self):
        print('B')
        super().method()

class C(B):
    def method(self):
        print('C')
        super().method()

C.__mro__  # (C, B, A, object)
C().method()
# C → B → A

super(type, object_or_type)

class Saveable:
    def save(self):
        print('Saveable')

class Validatable:
    def save(self):
        print('Validate')
        super().save()

class Model(Validatable, Saveable):
    def save(self):
        print('Model')
        super().save()

Model().save()
# Model → Validate → Saveable

class MyClass:
    count = 0

    def instance_method(self):
        """实例方法：第一个参数是实例 self"""
        return self

    @classmethod
    def class_method(cls):
        """类方法：第一个参数是类 cls"""
        cls.count += 1
        return cls

    @staticmethod
    def static_method(x, y):
        """静态方法：无隐式参数"""
        return x + y

class Config:
    settings = {'debug': True}

    @classmethod
    def is_debug(cls):
        return cls.settings.get('debug', False)

class TestingConfig(Config):
    settings = {'debug': False}

Config.is_debug()      # True
TestingConfig.is_debug() # False  ← cls 是 TestingConfig！

class DateUtils:
    @staticmethod
    def is_valid_date(date_str):
        # 纯函数逻辑
        pass

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

# singleton.py
class _Singleton:
    pass
instance = _Singleton()

def singleton(cls):
    instances = {}
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

class SingletonMeta(type):
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Singleton(metaclass=SingletonMeta):
    pass

# config.py
class Config:
    def __init__(self):
        self.load_from_file()

config = Config()  # 模块导入时创建一次

# 其他文件
from config import config  # 拿到的是同一个实例

import threading

class ThreadSafeSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:  # 双重检查锁定
                    cls._instance = super().__new__(cls)
        return cls._instance

class ValidatedAttribute:
    def __set_name__(self, owner, name):
        self.private_name = f'_{name}'

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value):
        if not isinstance(value, int) or value < 0:
            raise ValueError(f"Invalid value: {value}")
        setattr(obj, self.private_name, value)

class Person:
    age = ValidatedAttribute()

p = Person()
p.age = 25    # OK
p.age = -1    # ValueError

1. 数据描述符（同时有 __get__ 和 __set__）→ 最高优先级
2. 实例的 __dict__（实例属性）
3. 非数据描述符（只有 __get__）
4. 类的 __dict__（类属性）

class Demo:
    @property
    def x(self):
        return 42

d = Demo()
d.__dict__['x'] = 100  # 存到实例字典
d.x  # 42！优先访问数据描述符 property

class MyClass:
    def method(self):
        return 'hello'

MyClass.method  # 普通函数
MyClass().method  # 绑定了实例的方法（bound method）

class UpperAttrMeta(type):
    def __new__(mcs, name, bases, namespace):
        new_namespace = {}
        for k, v in namespace.items():
            if not k.startswith('__'):
                new_namespace[k.upper()] = v
            else:
                new_namespace[k] = v
        return super().__new__(mcs, name, bases, new_namespace)

class Person(metaclass=UpperAttrMeta):
    name = 'Alice'

Person.NAME  # 'Alice'

# type 是最基本的元类
MyClass = type('MyClass', (Base,), {'attr': 'value'})
# 等价于
class MyClass(Base):
    attr = 'value'

class Foo(metaclass=Meta):
    bar = 1

class PluginBase:
    _registry = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        PluginBase._registry[cls.__name__] = cls

class MyPlugin(PluginBase):
    pass  # 自动注册

class Demo:
    def __init__(self):
        self.x = 1

    def __getattribute__(self, name):
        print(f'__getattribute__: {name}')
        return super().__getattribute__(name)

    def __getattr__(self, name):
        print(f'__getattr__: {name}')
        return f'{name} not found'

d = Demo()
d.x    # 先触发 __getattribute__，找到 x=1
d.y    # 先触发 __getattribute__，找不到，再触发 __getattr__

obj.attr
  ↓
__getattribute__('attr')   ← 每次访问都触发
  ↓
检查是否为数据描述符 → 是 → 调用描述符的 __get__
  ↓ 否
检查实例的 __dict__ → 有 → 返回值
  ↓ 无
检查非数据描述符 → 是 → 调用描述符的 __get__
  ↓ 否
检查类的 __dict__ → 有 → 返回值
  ↓ 无
触发 __getattr__('attr')   ← 只有找不到时才触发
  ↓
返回 __getattr__ 的结果 或 抛 AttributeError

class Bad:
    def __getattribute__(self, name):
        return self.__dict__[name]  # 无限递归！

# 动态属性示例
class DynamicConfig:
    def __init__(self, config_dict):
        self._config = config_dict

    def __getattr__(self, name):
        if name in self._config:
            return self._config[name]
        raise AttributeError(f'No config: {name}')

config = DynamicConfig({'host': 'localhost', 'port': 8080})
config.host  # 'localhost'

from abc import ABC, abstractmethod

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

    @abstractmethod
    def perimeter(self):
        pass

    def describe(self):
        return f'Area: {self.area()}, Perimeter: {self.perimeter()}'

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

    def area(self):
        return 3.14159 * self.radius ** 2

    def perimeter(self):
        return 2 * 3.14159 * self.radius

# Shape()  → TypeError
Circle(5).describe()  # OK

from abc import ABC

class MyABC(ABC):
    @abstractmethod
    def do_something(self): ...

class Concrete:
    def do_something(self):
        print('doing')

MyABC.register(Concrete)
isinstance(Concrete(), MyABC)  # True

class Base(ABC):
    @abstractmethod
    def required(self): ...

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 在子类定义时检查是否实现了 required
        if not hasattr(cls, 'required') or cls.required == Base.required:
            raise TypeError(f'{cls.__name__} must implement required()')

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 __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __lt__(self, other):
        return (self.x**2 + self.y**2) < (other.x**2 + other.y**2)

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

    def __hash__(self):
        return hash((self.x, self.y))

    def __bool__(self):
        return self.x != 0 or self.y != 0

    def __len__(self):
        return 2

    def __getitem__(self, index):
        return (self.x, self.y)[index]

class Vector:
    def __mul__(self, other):
        """向量 * 数值"""
        if isinstance(other, (int, float)):
            return Vector(self.x * other, self.y * other)
        return NotImplemented  # ← 返回 NotImplemented 触发反射

    def __rmul__(self, other):
        """数值 * 向量"""
        return self.__mul__(other)  # 乘法交换律

v = Vector(2, 3)
v * 5   # 调用 __mul__
5 * v   # 调用 __rmul__

多进程（最高隔离）→ 多线程（共享内存）→ 协程（单线程高并发）
                                                      ↓
Python 3 的 asyncio + async/await

# 创建 10000 个并发任务
# 多进程：不可能（进程数有限）
# 多线程：可能但性能差（OS 线程数有限，切换开销大）
# 协程：轻松（每个协程只需几千字节内存）

import threading

local_data = threading.local()

def worker():
    local_data.value = threading.current_thread().name
    time.sleep(0.1)
    print(f'{threading.current_thread().name}: {local_data.value}')

t1 = threading.Thread(target=worker, name='Thread-1')
t2 = threading.Thread(target=worker, name='Thread-2')
t1.start(); t2.start()
# 各自输出自己的名字

# 全局变量被所有线程共享
user_context = {}

def handle_request(user):
    user_context['user'] = user  # 线程 A 设置
    do_work()                     # 线程 B 可能覆盖了 user!
    print(user_context['user'])   # 可能是 B 的用户！

from flask import Flask, request

app = Flask(__name__)

@app.route('/')
def index():
    # request 是全局变量，但每个线程看到自己的请求
    return f'Hello, {request.remote_addr}!'

import asyncio

async def fetch_data(url):
    print(f'Start fetching {url}')
    await asyncio.sleep(1)
    return f'data from {url}'

async def main():
    results = await asyncio.gather(
        fetch_data('url1'),
        fetch_data('url2'),
        fetch_data('url3'),
    )
    return results

asyncio.run(main())

while True:
    # 1. 检查是否有"可执行"的协程（没有在 await 等待的）
    for task in ready_tasks:
        task.step()

    # 2. 检查是否有"就绪"的 I/O 事件
    for event in io_events:
        wake_up(awaiting_task)

    # 3. 如果没有任务，等待新事件
    if not tasks:
        wait_for_events()

async def hello():
    print('Hello')
    await asyncio.sleep(1)  # 暂停 1 秒，让出控制权
    print('World')

async def main():
    # 创建 Task，协程立刻被调度执行
    task = asyncio.create_task(fetch_data('url1'))
    # 等待 Task 完成并获取结果
    result = await task

# 等同于
result = await fetch_data('url1')  # 直接等待协程

async def main():
    t1 = asyncio.create_task(fetch('url1'))
    t2 = asyncio.create_task(fetch('url2'))
    # 两个 fetch 同时运行！
    r1 = await t1
    r2 = await t2

import threading

# 1. Lock（互斥锁）
lock = threading.Lock()
with lock:
    pass  # 临界区

# 2. RLock（可重入锁）
rlock = threading.RLock()

# 3. Semaphore（信号量）—— 控制并发数
sem = threading.Semaphore(5)

# 4. Event（事件）—— 线程间通知
event = threading.Event()
event.wait()
event.set()

# 5. Condition（条件变量）—— 生产者消费者
cond = threading.Condition()
with cond:
    cond.wait()
    cond.notify()

# 6. Barrier（栅栏）—— 等待所有线程到达
barrier = threading.Barrier(3)

a += 1
# 可能对应：
# 1. LOAD a        读取 a 的值
# 2. LOAD_CONST 1  加载常量 1
# 3. BINARY_OP +   加法
# 4. STORE a       写回 a

lock = threading.Lock()

def recursive(n):
    with lock:  # 第一次获取成功
        if n > 0:
            recursive(n - 1)  # 第二次获取 → 死锁！

rlock = threading.RLock()
def recursive(n):
    with rlock:  # 第一次获取成功
        if n > 0:
            recursive(n - 1)  # 同一线程再次获取 → 成功（可重入）

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

# 线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    # 方式一：map
    results = list(executor.map(lambda x: x**2, range(10)))

    # 方式二：submit + as_completed
    futures = {executor.submit(fetch_url, url): url for url in urls}
    for future in as_completed(futures):
        try:
            result = future.result()
        except Exception as e:
            print(f'Failed: {e}')

# 进程池（CPU 密集型）
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(heavy_computation, data))

# CPU 密集型 → ProcessPoolExecutor
# I/O 密集型 → ThreadPoolExecutor

futures = [executor.submit(fetch, url) for url in urls]
# 按完成顺序处理，而非提交顺序
for future in as_completed(futures):
    process(future.result())

def transfer(a, b, amount):
    first, second = sorted([a, b], key=lambda x: id(x))
    with first.lock:
        with second.lock:
            pass

# 线程 1
lock_a.acquire()
lock_b.acquire()  # 等待 lock_b

# 线程 2
lock_b.acquire()
lock_a.acquire()  # 等待 lock_a

# 死锁！互相等待

if lock.acquire(timeout=5):
    try:
        # 临界区
        pass
    finally:
        lock.release()
else:
    # 超时处理，避免死锁
    handle_timeout()

from multiprocessing import Process, Queue, Pipe, Manager, Value, Array

# 1. Queue（最常用）
q = Queue()
q.put(data); data = q.get()

# 2. Pipe（双向管道）
parent_conn, child_conn = Pipe()

# 3. Manager（共享复杂对象）
manager = Manager()
shared_dict = manager.dict()
shared_list = manager.list()

# 4. Value / Array（共享内存，高效）
shared_val = Value('i', 0)
shared_arr = Array('d', [0.0]*10)

# 5. SharedMemory（Python 3.8+）
from multiprocessing.shared_memory import SharedMemory

# Queue 的数据流
Producer → [pickle 序列化] → Pipe → [pickle 反序列化] → Consumer

manager = Manager()
# 实际上启动了一个新的进程来管理共享对象
# 所有读写都通过 socket/pipe 与这个进程通信

import asyncio, concurrent.futures

# 在协程中调用阻塞函数
async def main():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_io, 'arg1')
    # 或指定线程池
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io, 'arg1')

# 在线程中运行协程
def thread_func():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    result = loop.run_until_complete(coro_func())

async def bad():
    time.sleep(1)  # 阻塞整个线程！所有协程都卡住
    requests.get('https://...')  # 也阻塞！

async def good():
    # blocking_io 在线程池中运行，不阻塞事件循环
    result = await loop.run_in_executor(None, blocking_io)

def thread_worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(async_func())
    finally:
        loop.close()

import asyncio

# 方式一：Semaphore
async def with_semaphore(urls, max_concurrent=10):
    sem = asyncio.Semaphore(max_concurrent)
    async def fetch(url):
        async with sem:
            return await aiohttp_get(url)
    await asyncio.gather(*[fetch(url) for url in urls])

# 方式二：分批处理
async def batch_process(urls, batch_size=10):
    for i in range(0, len(urls), batch_size):
        batch = urls[i:i+batch_size]
        await asyncio.gather(*[fetch(url) for url in batch])

# 方式三：TaskGroup（Python 3.11+）
async def with_task_group(urls):
    async with asyncio.TaskGroup() as tg:
        tasks = [tg.create_task(fetch(url)) for url in urls]

初始许可证：5
第一次获取：许可证 4
第二次获取：许可证 3
...
第六次获取：许可证 0，等待！
前五个完成一个 → 许可证 1 → 唤醒等待的协程

async with asyncio.TaskGroup() as tg:
    t1 = tg.create_task(fetch('url1'))
    t2 = tg.create_task(fetch('url2'))

# 所有任务完成后退出 with 块
# 如果有异常，所有任务被取消，异常被收集

import subprocess

result = subprocess.run(
    ['ls', '-la'],
    capture_output=True,
    text=True,
    timeout=30,
    check=True,
    cwd='/tmp',
)

print(result.stdout)

# 实时输出
proc = subprocess.Popen(['python', 'script.py'], stdout=subprocess.PIPE, text=True)
for line in proc.stdout:
    print(line, end='')

# 危险！
user_input = "rm -rf /; echo 'done'"
subprocess.run(f'echo {user_input}', shell=True)  # 危险命令注入！

# 安全！
subprocess.run(['echo', user_input])  # 参数化，不会执行 rm

proc = subprocess.Popen(
    ['ping', '8.8.8.8'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE,
    text=True,
    bufsize=1,  # 行缓冲
)

for line in proc.stdout:
    print(f'[PING] {line}', end='')

import sys
sys.getswitchinterval()  # 0.005（5ms）

# 伪代码
while True:
    if should_drop_gil():
        release_gil()      # 释放 GIL
        wait_for_gil()     # 等待重新获取
        acquire_gil()      # 获取 GIL
    execute_next_bytecode()  # 执行下一条字节码

# 简化逻辑
def acquire_gil(gil):
    while gil.locked:
        gil.cond.wait(gil.mutex)  # 等待其他线程释放 GIL
    gil.locked = True

def release_gil(gil):
    gil.locked = False
    gil.cond.notify_all()  # 唤醒等待的线程

// 在 time.sleep 的 C 实现中
Py_BEGIN_ALLOW_THREADS
sleep(seconds);  // 等待时释放 GIL，其他线程可执行
Py_END_ALLOW_THREADS

# 1. threading.Timer
from threading import Timer
def task():
    print('执行任务')
    Timer(60, task).start()

# 2. schedule 库
import schedule
schedule.every(10).minutes.do(task)
schedule.every().day.at("10:30").do(task)
while True:
    schedule.run_pending()
    time.sleep(1)

# 3. APScheduler（生产推荐）
from apscheduler.schedulers.asyncio import AsyncIOScheduler
scheduler = AsyncIOScheduler()
scheduler.add_job(task, 'interval', seconds=30)
scheduler.start()

Timer(60, task).start()
# 内部创建一个线程，time.sleep(60) 后执行 task
# 适合简单场景，但不精确（受 GIL 和系统调度影响）

from apscheduler.schedulers.blocking import BlockingScheduler

sched = BlockingScheduler()
@sched.scheduled_job('cron', day_of_week='mon-fri', hour='9')
def scheduled_job():
    print('工作日早上 9 点执行')

# Condition：等待条件满足
cond = threading.Condition()
with cond:
    while not items:  # 必须用 while 防止虚假唤醒
        cond.wait()
    item = items.pop(0)

# 等待服务就绪
ready = threading.Event()

def worker():
    ready.wait()  # 等待服务就绪
    print('开始工作')

threading.Thread(target=worker).start()
# 做一些初始化工作...
ready.set()  # 通知所有 worker 开始

# 正确
with cond:
    while not items:
        cond.wait()
    item = items.pop()

# 错误
with cond:
    if not items:
        cond.wait()
    item = items.pop()  # 如果虚假唤醒，items 可能还是空的！

import signal
import multiprocessing as mp

def worker(stop_event):
    while not stop_event.is_set():
        do_work()

def main():
    stop_event = mp.Event()

    def signal_handler(sig, frame):
        stop_event.set()

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    processes = [mp.Process(target=worker, args=(stop_event,)) for _ in range(4)]
    for p in processes:
        p.start()

    for p in processes:
        p.join(timeout=10)
        if p.is_alive():
            p.terminate()

用户按 Ctrl+C → 主进程收到 SIGINT
  ↓
信号处理函数设置 stop_event
  ↓
子进程的 while 循环检查 stop_event → 假 → 退出
  ↓
主进程 join(timeout=10) 等待子进程退出
  ↓
如果还有子进程存活 → terminate() 强制终止

process.terminate()  # 发送 SIGTERM，子进程可以注册处理函数
process.kill()       # 发送 SIGKILL，无法被捕获，强制杀死

with mp.Pool(4) as pool:
    try:
        results = pool.map_async(func, data).get(timeout=60)
    except KeyboardInterrupt:
        pool.terminate()  # 立即停止所有工作进程
        pool.join()

# 1. gather 的异常处理
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        print(f'Task failed: {r}')

# 2. TaskGroup（Python 3.11+）
try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(risky_task())
except* ValueError as eg:
    print(f'ValueErrors: {eg.exceptions}')

# 3. 单个 Task 异常处理
task = asyncio.create_task(risky_task())
try:
    result = await task
except ValueError as e:
    print(e)

# 默认行为：第一个异常就抛出
results = await asyncio.gather(task1, task2)  # task1 失败 → 抛异常

# 不立即抛异常，而是收集所有结果
results = await asyncio.gather(*tasks, return_exceptions=True)
for r in results:
    if isinstance(r, Exception):
        print(f'处理异常: {r}')

try:
    async with asyncio.TaskGroup() as tg:
        tg.create_task(maybe_value_error())
        tg.create_task(maybe_type_error())
except* ValueError as eg:
    print(f'处理 {len(eg.exceptions)} 个 ValueError')
except* TypeError as eg:
    print(f'处理 {len(eg.exceptions)} 个 TypeError')

task.cancel()  # 在协程中抛入 CancelledError

async def my_coro():
    try:
        await asyncio.sleep(100)
    except asyncio.CancelledError:
        # 清理资源
        await cleanup()
        raise  # 必须重新抛出 CancelledError！

import gc
gc.get_threshold()  # (700, 10, 10)
gc.collect()        # 手动触发全量 GC

a = []    # ob_refcnt = 1
b = a     # ob_refcnt = 2
del a     # ob_refcnt = 1
del b     # ob_refcnt = 0 → 回收

# 循环引用 —— 引用计数无法处理
a = []
b = []
a.append(b)
b.append(a)
# a 和 b 的引用计数都是 1，就算 del a, del b 也不会归零

# 手动触发 GC
import gc
gc.collect()       # 全量回收
gc.collect(0)      # 只回收第 0 代

# 1. objgraph 查看引用关系
import objgraph
objgraph.show_backrefs([obj], filename='backrefs.png')
objgraph.most_common_types()

# 2. tracemalloc 追踪内存分配
import tracemalloc
tracemalloc.start()
snapshot = tracemalloc.take_snapshot()
for stat in snapshot.statistics('lineno')[:10]:
    print(stat)

# 3. memory_profiler 逐行分析
from memory_profiler import profile

# 泄漏版本
_cache = {}

def get_data(key):
    if key not in _cache:
        _cache[key] = fetch_from_db(key)  # 缓存无限增长！
    return _cache[key]

# 修复版本
from functools import lru_cache

@lru_cache(maxsize=1000)
def get_data(key):
    return fetch_from_db(key)

def create_handler():
    large_data = load_large_file()

    def handler(event):
        print(event)
        # handler 闭包引用了 large_data，导致它没法回收
    return handler

class Leak:
    def __init__(self):
        self.other = None
    def __del__(self):
        pass  # __del__ 阻止 GC 回收循环引用

a = Leak()
b = Leak()
a.other = b
b.other = a  # 循环引用 + __del__ → 永不回收！

# 1. 上下文管理器（推荐）
class Resource:
    def __enter__(self):
        self.conn = acquire()
        return self.conn
    def __exit__(self, *args):
        self.conn.release()

# 2. weakref.finalize（Python 3.4+）
import weakref
weakref.finalize(conn, cleanup, conn)

# 3. atexit 注册清理函数
import atexit
atexit.register(cleanup)

class Bad:
    def __del__(self):
        print('被销毁了', self.file)
        self.file.close()

b = Bad()
# b 什么时候被销毁？不确定！
# - 引用计数归零时 → 可能立即
# - GC 回收时 → 不确定时间
# - 解释器退出时 → 可能某些模块已经没了！

import gc
gc.garbage  # [__del__ 循环引用的对象不会自动回收]

import weakref

class Resource:
    def __init__(self, name):
        self.name = name
        weakref.finalize(self, self._cleanup, name)

    @staticmethod
    def _cleanup(name):
        print(f'Cleaning up {name}')

r = Resource('db')
del r  # 触发 _cleanup

import weakref

obj = BigObject('huge')
ref = weakref.ref(obj)
ref()  # <BigObject>
del obj
ref()  # None

# WeakValueDictionary —— 缓存场景
cache = weakref.WeakValueDictionary()
cache['key'] = BigObject('temp')  # 不阻止回收

# WeakSet —— 观察者模式
observers = weakref.WeakSet()

# 强引用 — 阻止回收
obj = BigObject()    # 强引用，ob_refcnt += 1

# 弱引用 — 不阻止回收
ref = weakref.ref(obj)  # 弱引用，ob_refcnt 不变
del obj                # 对象被回收
ref()                  # None（对象已经没了）

class ImageCache:
    def __init__(self):
        self.cache = weakref.WeakValueDictionary()

    def get_image(self, path):
        if path not in self.cache:
            img = Image.open(path)
            self.cache[path] = img
        return self.cache[path]

# 当 Image 对象不再被其他地方引用时，自动从缓存中移除

class Subject:
    def __init__(self):
        self._observers = weakref.WeakSet()

    def attach(self, observer):
        self._observers.add(observer)

    def notify(self, data):
        for obs in self._observers:
            obs.update(data)

# 观察者被回收后自动从集合中移除，不需要手动 detach！

a = 'hello'
b = 'hello'
a is b  # True，短字符串和标识符自动驻留

import sys
a = sys.intern('hello world!')
b = sys.intern('hello world!')
a is b  # True

# 这些字符串自动驻留
a = 'hello'       # 编译期常量
b = 'hello'
a is b  # True

# 运行时创建的字符串不一定驻留
c = 'hello' * 3   # 运行时字符串连接
d = 'hello' * 3
c is d  # False

cities = [sys.intern(city) for city in raw_cities]
# 100 万行中的 100 个不同城市 → 100 个对象

应用程序 → pymalloc（≤512字节） → C malloc → 操作系统

Arena（256KB）：一次向系统申请
  ├── Pool (4KB) → Block(8B), Block(8B), ...
  ├── Pool (4KB) → Block(16B), Block(16B), ...
  └── Pool (4KB) → Block(32B), Block(32B), ...

x = 42
# 1. 42 < 512，走 pymalloc
# 2. 对齐到 8 的倍数：48 字节
# 3. 找大小 = 48 的 Pool
# 4. 从 Pool 中拿一个空闲 Block

import sys, array

# list：~36B/元素（指针 + 整数对象）
list_nums = list(range(1000000))

# array：4B/元素（连续 C 类型）
arr = array.array('i', range(1000000))
sys.getsizeof(arr)  # ~4MB

cities = [sys.intern(city) for city in raw_cities]

empty = []
full = list(range(1000000))

sys.getsizeof(empty)  # 56 字节
sys.getsizeof(full)   # ~8MB（仅指针数组，不含整数对象）

def deep_getsizeof(obj, seen=None):
    seen = seen or set()
    if id(obj) in seen:
        return 0
    seen.add(id(obj))
    size = sys.getsizeof(obj)
    if isinstance(obj, dict):
        size += sum(deep_getsizeof(k, seen) + deep_getsizeof(v, seen)
                    for k, v in obj.items())
    elif isinstance(obj, (list, tuple, set, frozenset)):
        size += sum(deep_getsizeof(item, seen) for item in obj)
    return size

import atexit

def cleanup_db():
    db_connection.close()

atexit.register(cleanup_db)

@atexit.register
def first():
    print('first')

@atexit.register
def second():
    print('second')

# 程序退出时：second → first

# 逐行读取
with open('big_file.txt') as f:
    for line in f:
        process(line)

# mmap（内存映射）
import mmap
with open('big_file', 'rb') as f:
    with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
        data = mm[1000:2000]

import ijson
with open('big.json') as f:
    for item in ijson.items(f, 'results.item'):
        process(item)

from itertools import *

count(10)           # 10, 11, 12, ...
cycle('ABC')        # A, B, C, A, B, C, ...
repeat(10, 3)       # 10, 10, 10

permutations('ABC', 2)   # AB, AC, BA, BC, CA, CB
combinations('ABC', 2)   # AB, AC, BC

accumulate([1, 2, 3, 4])  # 1, 3, 6, 10

chain(iter1, iter2)  # 链接多个迭代器

islice(range(10), 2, 8, 2)  # 2, 4, 6

product('AB', '12')  # A1, A2, B1, B2

# 36 选 7 的所有组合数
from math import comb
comb(36, 7)  # 8347680 种

# 但用 itertools 生成所有组合会消耗大量时间
# 通常不需要枚举全部

# 必须先排序！
data = [('a', 1), ('a', 2), ('b', 3), ('b', 4)]
for key, group in groupby(data, key=lambda x: x[0]):
    print(key, list(group))
# a [(a,1), (a,2)]
# b [(b,3), (b,4)]

chain([1, 2], [3, 4])           # 固定参数
chain.from_iterable([[1,2],[3,4]])  # 从迭代器展开

import functools

# lru_cache —— 缓存装饰器
@functools.lru_cache(maxsize=128)
def expensive(n):
    return n ** n

# partial —— 偏函数
square = functools.partial(pow, exp=2)

# wraps —— 保留原函数元信息
def my_decorator(func):
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

# singledispatch —— 单分派泛型函数
@functools.singledispatch
def process(data):
    raise NotImplementedError

@process.register(int)
def _(data):
    return f'Integer: {data}'

# reduce —— 累积计算
functools.reduce(lambda x, y: x + y, [1, 2, 3, 4])  # 10

# total_ordering —— 自动补全比较方法
@functools.total_ordering
class Student:
    def __eq__(self, other): ...
    def __lt__(self, other): ...
    # 自动生成 __le__, __gt__, __ge__

@singledispatch
def serialize(obj):
    raise NotImplementedError

@serialize.register(str)
def _(obj):
    return f'"{obj}"'

@serialize.register(int)
def _(obj):
    return str(obj)

# 提前绑定参数
def connect(host, port, user, password):
    pass

# 创建特定数据库的快捷连接
connect_db = partial(connect, host='localhost', port=3306)
connect_db(user='root', password='secret')

# tkinter 中绑定事件参数
button.config(command=partial(handle_click, item_id))

# Java：需要接口 + 实现类
# Python：
def quick_sort(data):
    return sorted(data)

def merge_sort(data):
    return sorted(data)

class Sorter:
    def __init__(self, strategy):
        self.strategy = strategy  # 直接传函数！
    
    def sort(self, data):
        return self.strategy(data)

Sorter(quick_sort).sort([3, 1, 2])  # 传函数引用即可

class PluginBase:
    _registry = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        PluginBase._registry[cls.__name__.lower()] = cls

class JsonPlugin(PluginBase):
    def process(self, data): ...

class YamlPlugin(PluginBase):
    def process(self, data): ...

# 自动注册，无需手动添加！
PluginBase._registry  # {'jsonplugin': JsonPlugin, 'yamlplugin': YamlPlugin}

class Service:
    def __init__(self, db: Database):
        self.db = db  # db 从外部注入

# 创建依赖
db = Database('postgresql://...')
# 注入
service = Service(db)

from fastapi import Depends, FastAPI

app = FastAPI()

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

@app.get('/users')
def get_users(db = Depends(get_db)):  # FastAPI 自动注入
    return db.query(User).all()

Request → [Middleware1] → [Middleware2] → [Handler] → [Middleware2] → [Middleware1] → Response
           before hook                    after hook

# 最简单的中间件——装饰器链
def logging_middleware(handler):
    def wrapper(request):
        print(f'Request: {request.method} {request.path}')
        response = handler(request)
        print(f'Response: {response.status}')
        return response
    return wrapper

def auth_middleware(handler):
    def wrapper(request):
        if not request.user:
            return Response(401)
        return handler(request)
    return wrapper

# 组合
handler = auth_middleware(logging_middleware(handle_request))

# requests — 同步，不能用于异步代码
resp = requests.get('https://api.example.com', timeout=10)

# httpx — 支持两种模式
import httpx

# 同步
resp = httpx.get('https://api.example.com')

# 异步
async with httpx.AsyncClient() as client:
    resp = await client.get('https://api.example.com')

# timeit — 比较两种写法谁更快
import timeit
timeit.timeit('"-".join(str(n) for n in range(100))', number=10000)

# cProfile — 找出最耗时的函数
python -m cProfile -s cumulative my_script.py

# line_profiler — 精确定位到行
@profile
def slow_function():
    total = 0
    for i in range(1000000):  # 这行最耗时
        total += i
    return total

# 纯 Python
def fib(n):
    a, b = 0, 1
    for i in range(n):
        a, b = b, a + b
    return a

# Cython — 加类型声明
def fib(int n):
    cdef int a = 0, b = 1, i
    for i in range(n):
        a, b = b, a + b
    return a

# 性能提升：10-100 倍

代码格式：black + ruff（自动格式化 + lint）
类型检查：mypy（静态类型检查）
测试：pytest + coverage（测试覆盖率）
预提交：pre-commit hooks（自动化检查）
CI/CD：GitHub Actions
文档：Sphinx + Google style docstring
依赖管理：poetry / uv

# 好的
except ValueError:
    ...
except (IOError, OSError):
    ...
    
# 不好的
except Exception:  # 太宽泛，可能隐藏 bug
    ...

# 最不好
except:  # 捕捉 KeyboardInterrupt、SystemExit！
    ...

# Python 风格
try:
    data = obj.read()
except AttributeError:
    handle_error()

# 非 Python 风格（LBYL - Look Before You Leap）
if hasattr(obj, 'read'):
    data = obj.read()
else:
    handle_error()

try:
    db_query()
except DatabaseError as e:
    raise ServiceError('服务暂时不可用') from e  # 保留原始异常

match command.split():
    case ['quit']:
        exit()
    case ['go', direction]:
        move(direction)
    case ['open', filename]:
        open_file(filename)
    case _:
        print('Unknown command')

数学原理：设 a 为头到入口距离，b 为入口到相遇点距离，c 为环剩余部分。`2(a+b) = a+b+nc → a = nc-b`，所以从相遇点和头部同时走必在入口相遇。

---

**🧠 深入解析**

**Floyd 判圈算法是 O(1) 空间判断链表环的最优解。**

**为什么快慢指针一定会在环中相遇？**

如果链表有环，快指针每次走 2 步，慢指针每次走 1 步。当慢指针进入环时，快指针已经在环中了。相对于慢指针，快指针以"每步 1 个节点"的速度靠近慢指针（`2-1=1`），所以必定会相遇。

**为什么相遇点和头节点同时走会在入口相遇？**

推导过程：
- 设头节点到环入口距离为 `a`
- 环入口到相遇点距离为 `b`
- 相遇点到环入口（绕一圈回来）距离为 `c`
- 慢指针走了 `a + b`
- 快指针走了 `a + b + n(b+c)`（n 圈）
- 快指针路程是慢指针的 2 倍：`2(a+b) = a+b+n(b+c)` → `a+b = n(b+c)` → `a = n(b+c)-b = (n-1)(b+c)+c`
- 所以从相遇点走 `c` 步到入口，从头走 `a` 步也到入口

**🎯 面试官考察点**
- 是否能推导出环入口位置的数学原理
- 是否知道为什么快慢指针一快一慢（快 2 慢 1）
- 是否能处理边界情况（空链表、单个节点、完整环）

**⚠️ 易错点**
- 快指针前进时要检查 `fast.next` 是否为 None，否则可能抛 AttributeError
- 无环时快指针会先到终点
- 完整环（头节点就在环上）的情况也能正确处理

**💡 生产实践**
- 面试高频算法题
- 实际应用中，链表环常出现在不正确的内存管理中（如 __del__ 中的循环引用）

---

### 22. Python 中如何实现二叉树的前中后序遍历（递归+迭代）？

**参考答案：**

```python
class TreeNode:
    def __init__(self, val=0, left=None, right=None):
        self.val = val
        self.left = left
        self.right = right

# 前序遍历（根-左-右）- 迭代
def preorder_iterative(root):
    res, stack = [], [root]
    while stack:
        node = stack.pop()
        if node:
            res.append(node.val)
            stack.append(node.right)
            stack.append(node.left)
    return res

# 中序遍历（左-根-右）- 迭代
def inorder_iterative(root):
    res, stack = [], []
    curr = root
    while curr or stack:
        while curr:
            stack.append(curr)
            curr = curr.left
        curr = stack.pop()
        res.append(curr.val)
        curr = curr.right
    return res

# 后序遍历（左-右-根）= 前序(根-右-左)的反转
def postorder_iterative(root):
    res, stack = [], [root]
    while stack:
        node = stack.pop()
        if node:
            res.append(node.val)
            stack.append(node.left)
            stack.append(node.right)
    return res[::-1]

    1
   / \n  2   3
 / \n4   5

前序（根→左→右）：[1, 2, 4, 5, 3]  — 先访问当前节点，再递归子树
中序（左→根→右）：[4, 2, 5, 1, 3]  — 先左子树，再当前节点，再右子树
后序（左→右→根）：[4, 5, 2, 3, 1]  — 先子树，再当前节点

stack.append(node.right)  # 右子树先进栈（后出）
stack.append(node.left)   # 左子树后进栈（先出）

模拟过程：
stack = [1, 2, 4]  ← 一直向左
pop 4 → visit 4 → 转向 4 的右（None）
pop 2 → visit 2 → 转向 2 的右（5）
stack = [1, 5]
pop 5 → visit 5 → 转向 5 的右（None）
pop 1 → visit 1 → 转向 1 的右（3）
stack = [3]
pop 3 → visit 3 → 转向 3 的右（None）

def quicksort_inplace(arr, low=0, high=None):
    if high is None:
        high = len(arr) - 1
    if low < high:
        pi = partition(arr, low, high)
        quicksort_inplace(arr, low, pi - 1)
        quicksort_inplace(arr, pi + 1, high)

def partition(arr, low, high):
    pivot = arr[high]
    i = low - 1
    for j in range(low, high):
        if arr[j] <= pivot:
            i += 1
            arr[i], arr[j] = arr[j], arr[i]
    arr[i + 1], arr[high] = arr[high], arr[i + 1]
    return i + 1

pivot = arr[high] = 5
arr = [3, 7, 8, 5, 2, 1, 9, 5, 4]
       ↑ i  ↑ j
i 指向"最后的 ≤5 的区域"的边界
j 遍历数组
遇到 ≤5 的元素，i+1 并交换

def partition_three_way(arr, low, high):
    """将数组分为 <pivot, =pivot, >pivot 三部分"""
    lt, gt = low, high
    pivot = arr[low]
    i = low
    while i <= gt:
        if arr[i] < pivot:
            arr[lt], arr[i] = arr[i], arr[lt]
            lt += 1; i += 1
        elif arr[i] > pivot:
            arr[i], arr[gt] = arr[gt], arr[i]
            gt -= 1
        else:
            i += 1
    return lt, gt  # 等于 pivot 的范围是 [lt, gt]

import heapq

def top_k_largest(nums, k):
    heap = nums[:k]
    heapq.heapify(heap)  # O(k)
    for num in nums[k:]:
        if num > heap[0]:
            heapq.heapreplace(heap, num)
    return heap

# 时间 O(n log k)，空间 O(k)

max_heap = [-x for x in data]  # 入堆时取负数
heapq.heapify(max_heap)
largest = -heapq.heappop(max_heap)  # 出堆时再转回来

数据：[3, 7, 2, 9, 5, 1, 8, 6, 4]，K=3

初始化堆 [3, 7, 2] → heapify
堆顶 = 2
遇到 9：2 < 9 → 替换 → [3, 7, 9]
堆顶 = 3
遇到 5：3 < 5 → 替换 → [5, 7, 9]
遇到 1：5 > 1 → 跳过
遇到 8：5 < 8 → 替换 → [7, 8, 9]
遇到 6：7 > 6 → 跳过
遇到 4：7 > 4 → 跳过
最终：[7, 8, 9]

def binary_search(nums, target):
    left, right = 0, len(nums) - 1
    while left <= right:
        mid = left + (right - left) // 2
        if nums[mid] == target:
            return mid
        elif nums[mid] < target:
            left = mid + 1
        else:
            right = mid - 1
    return -1

# 查找左边界
def lower_bound(nums, target):
    left, right = 0, len(nums)
    while left < right:
        mid = left + (right - left) // 2
        if nums[mid] < target:
            left = mid + 1
        else:
            right = mid
    return left

def lower_bound(nums, target):
    left, right = 0, len(nums)  # 注意 right = len(nums)，不是 len(nums)-1
    while left < right:
        mid = left + (right - left) // 2
        if nums[mid] < target:
            left = mid + 1  # mid 不够大，排除
        else:
            right = mid     # mid 可能是答案，保留
    return left  # left == right

# 合并两个有序链表
def merge_two_lists(l1, l2):
    dummy = ListNode()
    curr = dummy
    while l1 and l2:
        if l1.val <= l2.val:
            curr.next = l1
            l1 = l1.next
        else:
            curr.next = l2
            l2 = l2.next
        curr = curr.next
    curr.next = l1 or l2
    return dummy.next

# 合并两个有序数组（从后往前归并）
def merge(nums1, m, nums2, n):
    p1, p2, p = m - 1, n - 1, m + n - 1
    while p2 >= 0:
        if p1 >= 0 and nums1[p1] > nums2[p2]:
            nums1[p] = nums1[p1]
            p1 -= 1
        else:
            nums1[p] = nums2[p2]
            p2 -= 1
        p -= 1

dummy = ListNode()  # 哨兵节点，不用处理头节点为空的边界情况
curr = dummy        # curr 指向当前已合并链表的尾节点
return dummy.next   # 哨兵的下一个就是真正的头节点

nums1 = [1, 3, 5, 0, 0, 0]  m=3
nums2 = [2, 4, 6]            n=3

p1=2(指向5), p2=2(指向6), p=5
5 vs 6 → nums1[5]=6, p2=1, p=4
5 vs 4 → nums1[4]=5, p1=1, p=3
3 vs 4 → nums1[3]=4, p2=0, p=2
3 vs 2 → nums1[2]=3, p1=0, p=1
1 vs 2 → nums1[1]=2, p2=-1(结束)
最终：[1, 2, 3, 4, 5, 6]

class TrieNode:
    def __init__(self):
        self.children = {}
        self.is_end = False

class Trie:
    def __init__(self):
        self.root = TrieNode()

    def insert(self, word):
        node = self.root
        for ch in word:
            if ch not in node.children:
                node.children[ch] = TrieNode()
            node = node.children[ch]
        node.is_end = True

    def search(self, word):
        node = self._find(word)
        return node is not None and node.is_end

    def starts_with(self, prefix):
        return self._find(prefix) is not None

    def _find(self, prefix):
        node = self.root
        for ch in prefix:
            if ch not in node.children:
                return None
            node = node.children[ch]
        return node

root
 ├── a ── p ── p ── l ── e (is_end=True)
 │               └── s (is_end=True)      "apps"
 └── b ── a ── t (is_end=True)            "bat"
               └── h (is_end=True)         "bath"

class TrieNode:
    def __init__(self):
        self.children = [None] * 26  # 只支持 a-z
        self.is_end = False

from collections import Counter
c = Counter('abracadabra')
c.most_common(3)   # [('a', 5), ('b', 2), ('r', 2)]
c1 + c2            # 合并计数
c1 - c2            # 差集计数（只保留正数）

block1 ↔ block2 ↔ block3
[..., ...] [..., ...] [..., ...]
  ↑left              right↑

from collections import ChainMap

defaults = {'theme': 'light', 'lang': 'en'}
user_config = {'theme': 'dark'}

config = ChainMap(user_config, defaults)
config['theme']  # 'dark'  ← 先查 user_config
config['lang']   # 'en'    ← user_config 没有，查 defaults

import threading, queue

q = queue.Queue(maxsize=10)

def producer():
    for i in range(100):
        q.put(i)

def consumer():
    while True:
        item = q.get()
        q.task_done()

threading.Thread(target=producer).start()
threading.Thread(target=consumer, daemon=True).start()
q.join()

async def producer(q):
    for i in range(100):
        await q.put(i)

async def consumer(q):
    while True:
        item = await q.get()
        q.task_done()

# Queue.put 的简化逻辑
def put(self, item, block=True, timeout=None):
    with self.mutex:
        if self.maxsize > 0 and self._qsize() >= self.maxsize:
            self.not_full.wait(timeout)
        self._put(item)
        self.not_empty.notify()

import heapq
import random

# 方法一：最小堆 O(n log k)
def find_kth_largest_heap(nums, k):
    return heapq.nlargest(k, nums)[-1]

# 方法二：快速选择 O(n) 平均（面试推荐）
def find_kth_largest(nums, k):
    def quickselect(left, right, k_smallest):
        if left == right:
            return nums[left]
        pivot_idx = random.randint(left, right)
        pivot_idx = partition(left, right, pivot_idx)
        if k_smallest == pivot_idx:
            return nums[k_smallest]
        elif k_smallest < pivot_idx:
            return quickselect(left, pivot_idx - 1, k_smallest)
        else:
            return quickselect(pivot_idx + 1, right, k_smallest)

    def partition(left, right, pivot_idx):
        pivot = nums[pivot_idx]
        nums[pivot_idx], nums[right] = nums[right], nums[pivot_idx]
        store = left
        for i in range(left, right):
            if nums[i] < pivot:
                nums[store], nums[i] = nums[i], nums[store]
                store += 1
        nums[store], nums[right] = nums[right], nums[store]
        return store

    return quickselect(0, len(nums) - 1, len(nums) - k)

T(n) = T(n/2) + O(n)  ← 每次只处理一半
递归展开：n + n/2 + n/4 + ... = 2n = O(n)

class Adder:
    def __init__(self, n):
        self.n = n

    def __call__(self, x):
        return self.n + x

add5 = Adder(5)
add5(10)  # 15，调用 __call__

callable(Adder(5))  # True
callable(42)        # False

class CountCalls:
    def __init__(self, func):
        self.func = func
        self.count = 0

    def __call__(self, *args, **kwargs):
        self.count += 1
        print(f'Called {self.count} times')
        return self.func(*args, **kwargs)

@CountCalls
def hello():
    print('Hello!')

hello()  # Called 1 times → Hello!
hello()  # Called 2 times → Hello!

class A:
    def method(self): return 'A'

class B(A):
    def method(self): return 'B'

class C(A):
    def method(self): return 'C'

class D(B, C):
    pass

D().method()  # 'B'
D.__mro__     # (D, B, C, A, object)

    A
   / \n  B   C
   \ /
    D

# 对于 D(B, C)
L(D) = D + merge(L(B), L(C), [B, C])

L(B) = B + merge(L(A), [A])
     = B + merge([A, O], [A])
     = B + A + merge([O])
     = B, A, O

L(C) = C, A, O

L(D) = D + merge([B, A, O], [C, A, O], [B, C])
     = D + B + merge([A, O], [C, A, O], [C])
     = D + B + C + merge([A, O], [A, O])
     = D + B + C + A + merge([O], [O])
     = D, B, C, A, O

D.__mro__  # (D, B, C, A, object)
D.mro()    # 同上

class A:
    def method(self):
        print('A')
        super().method()

class B(A):
    def method(self):
        print('B')
        super().method()

class C(A):
    def method(self):
        print('C')

class D(B, C):
    def method(self):
        print('D')
        super().method()

D().method()
# 输出: D → B → C → A
# MRO: D → B → C → A → object

class A:
    def method(self):
        print('A')

class B(A):
    def method(self):
        print('B')
        super().method()

class C(B):
    def method(self):
        print('C')
        super().method()

C.__mro__  # (C, B, A, object)
C().method()
# C → B → A

super(type, object_or_type)

class Saveable:
    def save(self):
        print('Saveable')

class Validatable:
    def save(self):
        print('Validate')
        super().save()

class Model(Validatable, Saveable):
    def save(self):
        print('Model')
        super().save()

Model().save()
# Model → Validate → Saveable

class MyClass:
    count = 0

    def instance_method(self):
        """实例方法：第一个参数是实例 self"""
        return self

    @classmethod
    def class_method(cls):
        """类方法：第一个参数是类 cls"""
        cls.count += 1
        return cls

    @staticmethod
    def static_method(x, y):
        """静态方法：无隐式参数"""
        return x + y

class Config:
    settings = {'debug': True}

    @classmethod
    def is_debug(cls):
        return cls.settings.get('debug', False)

class TestingConfig(Config):
    settings = {'debug': False}

Config.is_debug()      # True
TestingConfig.is_debug() # False  ← cls 是 TestingConfig！

class DateUtils:
    @staticmethod
    def is_valid_date(date_str):
        # 纯函数逻辑
        pass

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

# singleton.py
class _Singleton:
    pass
instance = _Singleton()

def singleton(cls):
    instances = {}
    @functools.wraps(cls)
    def get_instance(*args, **kwargs):
        if cls not in instances:
            instances[cls] = cls(*args, **kwargs)
        return instances[cls]
    return get_instance

class SingletonMeta(type):
    _instances = {}
    def __call__(cls, *args, **kwargs):
        if cls not in cls._instances:
            cls._instances[cls] = super().__call__(*args, **kwargs)
        return cls._instances[cls]

class Singleton(metaclass=SingletonMeta):
    pass

# config.py
class Config:
    def __init__(self):
        self.load_from_file()

config = Config()  # 模块导入时创建一次

# 其他文件
from config import config  # 拿到的是同一个实例

import threading

class ThreadSafeSingleton:
    _instance = None
    _lock = threading.Lock()

    def __new__(cls, *args, **kwargs):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:  # 双重检查锁定
                    cls._instance = super().__new__(cls)
        return cls._instance

class ValidatedAttribute:
    def __set_name__(self, owner, name):
        self.private_name = f'_{name}'

    def __get__(self, obj, objtype=None):
        if obj is None:
            return self
        return getattr(obj, self.private_name, None)

    def __set__(self, obj, value):
        if not isinstance(value, int) or value < 0:
            raise ValueError(f"Invalid value: {value}")
        setattr(obj, self.private_name, value)

class Person:
    age = ValidatedAttribute()

p = Person()
p.age = 25    # OK
p.age = -1    # ValueError

1. 数据描述符（同时有 __get__ 和 __set__）→ 最高优先级
2. 实例的 __dict__（实例属性）
3. 非数据描述符（只有 __get__）
4. 类的 __dict__（类属性）

class Demo:
    @property
    def x(self):
        return 42

d = Demo()
d.__dict__['x'] = 100  # 存到实例字典
d.x  # 42！优先访问数据描述符 property

class MyClass:
    def method(self):
        return 'hello'

MyClass.method  # 普通函数
MyClass().method  # 绑定了实例的方法（bound method）

class UpperAttrMeta(type):
    def __new__(mcs, name, bases, namespace):
        new_namespace = {}
        for k, v in namespace.items():
            if not k.startswith('__'):
                new_namespace[k.upper()] = v
            else:
                new_namespace[k] = v
        return super().__new__(mcs, name, bases, new_namespace)

class Person(metaclass=UpperAttrMeta):
    name = 'Alice'

Person.NAME  # 'Alice'

# type 是最基本的元类
MyClass = type('MyClass', (Base,), {'attr': 'value'})
# 等价于
class MyClass(Base):
    attr = 'value'

class Foo(metaclass=Meta):
    bar = 1

class PluginBase:
    _registry = {}

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        PluginBase._registry[cls.__name__] = cls

class MyPlugin(PluginBase):
    pass  # 自动注册

class Demo:
    def __init__(self):
        self.x = 1

    def __getattribute__(self, name):
        print(f'__getattribute__: {name}')
        return super().__getattribute__(name)

    def __getattr__(self, name):
        print(f'__getattr__: {name}')
        return f'{name} not found'

d = Demo()
d.x    # 先触发 __getattribute__，找到 x=1
d.y    # 先触发 __getattribute__，找不到，再触发 __getattr__

obj.attr
  ↓
__getattribute__('attr')   ← 每次访问都触发
  ↓
检查是否为数据描述符 → 是 → 调用描述符的 __get__
  ↓ 否
检查实例的 __dict__ → 有 → 返回值
  ↓ 无
检查非数据描述符 → 是 → 调用描述符的 __get__
  ↓ 否
检查类的 __dict__ → 有 → 返回值
  ↓ 无
触发 __getattr__('attr')   ← 只有找不到时才触发
  ↓
返回 __getattr__ 的结果 或 抛 AttributeError

class Bad:
    def __getattribute__(self, name):
        return self.__dict__[name]  # 无限递归！

# 动态属性示例
class DynamicConfig:
    def __init__(self, config_dict):
        self._config = config_dict

    def __getattr__(self, name):
        if name in self._config:
            return self._config[name]
        raise AttributeError(f'No config: {name}')

config = DynamicConfig({'host': 'localhost', 'port': 8080})
config.host  # 'localhost'

from abc import ABC, abstractmethod

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

    @abstractmethod
    def perimeter(self):
        pass

    def describe(self):
        return f'Area: {self.area()}, Perimeter: {self.perimeter()}'

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

    def area(self):
        return 3.14159 * self.radius ** 2

    def perimeter(self):
        return 2 * 3.14159 * self.radius

# Shape()  → TypeError
Circle(5).describe()  # OK

from abc import ABC

class MyABC(ABC):
    @abstractmethod
    def do_something(self): ...

class Concrete:
    def do_something(self):
        print('doing')

MyABC.register(Concrete)
isinstance(Concrete(), MyABC)  # True

class Base(ABC):
    @abstractmethod
    def required(self): ...

    def __init_subclass__(cls, **kwargs):
        super().__init_subclass__(**kwargs)
        # 在子类定义时检查是否实现了 required
        if not hasattr(cls, 'required') or cls.required == Base.required:
            raise TypeError(f'{cls.__name__} must implement required()')

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 __eq__(self, other):
        return self.x == other.x and self.y == other.y

    def __lt__(self, other):
        return (self.x**2 + self.y**2) < (other.x**2 + other.y**2)

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

    def __hash__(self):
        return hash((self.x, self.y))

    def __bool__(self):
        return self.x != 0 or self.y != 0

    def __len__(self):
        return 2

    def __getitem__(self, index):
        return (self.x, self.y)[index]

class Vector:
    def __mul__(self, other):
        """向量 * 数值"""
        if isinstance(other, (int, float)):
            return Vector(self.x * other, self.y * other)
        return NotImplemented  # ← 返回 NotImplemented 触发反射

    def __rmul__(self, other):
        """数值 * 向量"""
        return self.__mul__(other)  # 乘法交换律

v = Vector(2, 3)
v * 5   # 调用 __mul__
5 * v   # 调用 __rmul__

多进程（最高隔离）→ 多线程（共享内存）→ 协程（单线程高并发）
                                                      ↓
Python 3 的 asyncio + async/await

# 创建 10000 个并发任务
# 多进程：不可能（进程数有限）
# 多线程：可能但性能差（OS 线程数有限，切换开销大）
# 协程：轻松（每个协程只需几千字节内存）

import threading

local_data = threading.local()

def worker():
    local_data.value = threading.current_thread().name
    time.sleep(0.1)
    print(f'{threading.current_thread().name}: {local_data.value}')

t1 = threading.Thread(target=worker, name='Thread-1')
t2 = threading.Thread(target=worker, name='Thread-2')
t1.start(); t2.start()
# 各自输出自己的名字

# 全局变量被所有线程共享
user_context = {}

def handle_request(user):
    user_context['user'] = user  # 线程 A 设置
    do_work()                     # 线程 B 可能覆盖了 user!
    print(user_context['user'])   # 可能是 B 的用户！

from flask import Flask, request

app = Flask(__name__)

@app.route('/')
def index():
    # request 是全局变量，但每个线程看到自己的请求
    return f'Hello, {request.remote_addr}!'

import asyncio

async def fetch_data(url):
    print(f'Start fetching {url}')
    await asyncio.sleep(1)
    return f'data from {url}'

async def main():
    results = await asyncio.gather(
        fetch_data('url1'),
        fetch_data('url2'),
        fetch_data('url3'),
    )
    return results

asyncio.run(main())

while True:
    # 1. 检查是否有"可执行"的协程（没有在 await 等待的）
    for task in ready_tasks:
        task.step()

    # 2. 检查是否有"就绪"的 I/O 事件
    for event in io_events:
        wake_up(awaiting_task)

    # 3. 如果没有任务，等待新事件
    if not tasks:
        wait_for_events()

async def hello():
    print('Hello')
    await asyncio.sleep(1)  # 暂停 1 秒，让出控制权
    print('World')

async def main():
    # 创建 Task，协程立刻被调度执行
    task = asyncio.create_task(fetch_data('url1'))
    # 等待 Task 完成并获取结果
    result = await task

# 等同于
result = await fetch_data('url1')  # 直接等待协程

async def main():
    t1 = asyncio.create_task(fetch('url1'))
    t2 = asyncio.create_task(fetch('url2'))
    # 两个 fetch 同时运行！
    r1 = await t1
    r2 = await t2

import threading

# 1. Lock（互斥锁）
lock = threading.Lock()
with lock:
    pass  # 临界区

# 2. RLock（可重入锁）
rlock = threading.RLock()

# 3. Semaphore（信号量）—— 控制并发数
sem = threading.Semaphore(5)

# 4. Event（事件）—— 线程间通知
event = threading.Event()
event.wait()
event.set()

# 5. Condition（条件变量）—— 生产者消费者
cond = threading.Condition()
with cond:
    cond.wait()
    cond.notify()

# 6. Barrier（栅栏）—— 等待所有线程到达
barrier = threading.Barrier(3)

a += 1
# 可能对应：
# 1. LOAD a        读取 a 的值
# 2. LOAD_CONST 1  加载常量 1
# 3. BINARY_OP +   加法
# 4. STORE a       写回 a

lock = threading.Lock()

def recursive(n):
    with lock:  # 第一次获取成功
        if n > 0:
            recursive(n - 1)  # 第二次获取 → 死锁！

rlock = threading.RLock()
def recursive(n):
    with rlock:  # 第一次获取成功
        if n > 0:
            recursive(n - 1)  # 同一线程再次获取 → 成功（可重入）

from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

# 线程池
with ThreadPoolExecutor(max_workers=5) as executor:
    # 方式一：map
    results = list(executor.map(lambda x: x**2, range(10)))

    # 方式二：submit + as_completed
    futures = {executor.submit(fetch_url, url): url for url in urls}
    for future in as_completed(futures):
        try:
            result = future.result()
        except Exception as e:
            print(f'Failed: {e}')

# 进程池（CPU 密集型）
with ProcessPoolExecutor(max_workers=4) as executor:
    results = list(executor.map(heavy_computation, data))

# CPU 密集型 → ProcessPoolExecutor
# I/O 密集型 → ThreadPoolExecutor

futures = [executor.submit(fetch, url) for url in urls]
# 按完成顺序处理，而非提交顺序
for future in as_completed(futures):
    process(future.result())

def transfer(a, b, amount):
    first, second = sorted([a, b], key=lambda x: id(x))
    with first.lock:
        with second.lock:
            pass

# 线程 1
lock_a.acquire()
lock_b.acquire()  # 等待 lock_b

# 线程 2
lock_b.acquire()
lock_a.acquire()  # 等待 lock_a

# 死锁！互相等待

if lock.acquire(timeout=5):
    try:
        # 临界区
        pass
    finally:
        lock.release()
else:
    # 超时处理，避免死锁
    handle_timeout()

from multiprocessing import Process, Queue, Pipe, Manager, Value, Array

# 1. Queue（最常用）
q = Queue()
q.put(data); data = q.get()

# 2. Pipe（双向管道）
parent_conn, child_conn = Pipe()

# 3. Manager（共享复杂对象）
manager = Manager()
shared_dict = manager.dict()
shared_list = manager.list()

# 4. Value / Array（共享内存，高效）
shared_val = Value('i', 0)
shared_arr = Array('d', [0.0]*10)

# 5. SharedMemory（Python 3.8+）
from multiprocessing.shared_memory import SharedMemory

# Queue 的数据流
Producer → [pickle 序列化] → Pipe → [pickle 反序列化] → Consumer

manager = Manager()
# 实际上启动了一个新的进程来管理共享对象
# 所有读写都通过 socket/pipe 与这个进程通信

import asyncio, concurrent.futures

# 在协程中调用阻塞函数
async def main():
    loop = asyncio.get_event_loop()
    result = await loop.run_in_executor(None, blocking_io, 'arg1')
    # 或指定线程池
    with concurrent.futures.ThreadPoolExecutor() as pool:
        result = await loop.run_in_executor(pool, blocking_io, 'arg1')

# 在线程中运行协程
def thread_func():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    result = loop.run_until_complete(coro_func())

async def bad():
    time.sleep(1)  # 阻塞整个线程！所有协程都卡住
    requests.get('https://...')  # 也阻塞！

async def good():
    # blocking_io 在线程池中运行，不阻塞事件循环
    result = await loop.run_in_executor(None, blocking_io)

def thread_worker():
    loop = asyncio.new_event_loop()
    asyncio.set_event_loop(loop)
    try:
        result = loop.run_until_complete(async_func())
    finally:
        loop.close()