Python 标准库 Day 4：数据结构与算法工具

1. 问题引入

继续博客项目。这几个需求看起来零散，其实都涉及"内置 dict / list 不够用"的场景：

1. 统计文章标签出现次数——哪些标签最热门？
2. 最近 100 条评论——只保留最近的，老的自动丢弃
3. 找出阅读量前 10 的文章——不需要排所有的，只要 top 10
4. 缓存渲染结果——避免重复计算同一篇文章
5. 去重 + 保序——给文章加"相关推荐"，列表里不能重复，但要保留顺序
6. 生成所有"标签 × 月份"的统计组合

每个需求用普通 list/dict 都能写，但代码会很长、性能也差。今天讲的五个模块，每一个都是「我今天有了一个新工具，原来要 10 行的代码现在 1 行」的体验。

2. 主题讲解

今天五个模块按"使用频率"排序讲：collections → functools → itertools → heapq → bisect。前两个几乎天天用，后三个用得少但偶尔很管用。

2.1 collections —— 增强版的内置容器

Counter —— 计数神器

最常见的场景：「数一下每个东西出现了几次」。

from collections import Counter

tags = ["python", "web", "python", "tutorial", "web", "python"]
c = Counter(tags)
print(c)


c["python"]              # 3
c["不存在的"]             # 0   ← 不报错，返回 0！这是和普通 dict 的区别

c.most_common(2)         # [('python', 3), ('web', 2)]   ← 最多的前 2 个

Counter 是 dict 的子类——所有 dict 操作都能用，但额外多了 most_common、自动返回 0、加减乘除运算等。

c1 = Counter(['a', 'b', 'a'])
c2 = Counter(['a', 'c'])
c1 + c2     # Counter({'a': 3, 'b': 1, 'c': 1})  ← 合并
c1 - c2     # Counter({'a': 1, 'b': 1})           ← 减法（负数被丢弃）

典型场景：

• 词频统计
• 投票计数
• 日志里各 URL 的访问次数（Day 2 我们就用过！）
• 标签热度

defaultdict —— 自动初始化的 dict

经典痛点：

groups = {}
for tag, post in tag_post_pairs:
    if tag not in groups:
        groups[tag] = []
    groups[tag].append(post)


from collections import defaultdict
groups = defaultdict(list)        # 默认值是空 list
for tag, post in tag_post_pairs:
    groups[tag].append(post)       # 不用先判断 key 存不存在

defaultdict(list) 的意思是：「访问不存在的 key 时，自动创建一个空 list 作为默认值」。list 是个工厂函数，调用 list() 返回 []。

常见工厂：

defaultdict(list)        # 默认 []
defaultdict(int)         # 默认 0     ← int() == 0
defaultdict(set)         # 默认 set()
defaultdict(dict)        # 默认 {}
defaultdict(lambda: "默认值")   # 自定义默认值

Counter vs defaultdict(int)：

counter = Counter()
counter["x"] += 1

dd = defaultdict(int)
dd["x"] += 1

deque —— 双端队列

list 的问题：在头部插入/删除是 O(n)（要把所有元素往后挪）。deque 两端都是 O(1)。

from collections import deque

d = deque([1, 2, 3])
d.append(4)         # 右边加     [1, 2, 3, 4]
d.appendleft(0)     # 左边加     [0, 1, 2, 3, 4]
d.pop()             # 右边出 → 4
d.popleft()         # 左边出 → 0

核心场景：固定长度的"最近 N 条"队列

recent = deque(maxlen=100)

for comment in stream:
    recent.append(comment)
    # 当 deque 满了再 append，最旧的会自动从左边掉出去

maxlen=100 是关键——这就是开头需求 2 的标准答案，不需要任何手动判断。

**deque**** vs **list** 该选哪个**：

场景	选什么
只在末尾追加，随机访问多	list
经常在头部操作，或要"最近 N 条"队列	deque
实现广度优先搜索（BFS）的队列	deque

namedtuple —— 给 tuple 命名（用得少，知道就行）

from collections import namedtuple

Post = namedtuple("Post", ["title", "url", "pv"])
p = Post("Python 入门", "/python", 1500)

p.title         # 'Python 入门'   ← 像对象一样访问
p[0]            # 'Python 入门'   ← 也能像 tuple 索引

现代代码更推荐 dataclass（Day 5 之后会接触），但读老代码会遇到 namedtuple。

2.2 functools —— 函数工具

lru_cache / cache —— 自动缓存（最常用）

我们 Day 2 自己写过 disk_cache（缓存到磁盘）。functools.cache 是它的"内存版"——更简单、更快，进程退出就没了。

from functools import cache

@cache
def fibonacci(n):
    if n < 2:
        return n
    return fibonacci(n-1) + fibonacci(n-2)

fibonacci(100)    # 瞬间返回，不带 cache 会卡死

工作原理：函数被调用时，用参数当 key 查缓存——命中就直接返回，没命中才执行函数并存结果。

@cache 是 Python 3.9+ 的简化版，老代码用的是 @lru_cache(maxsize=128)：

from functools import lru_cache

@lru_cache(maxsize=128)        # 最多缓存 128 个结果，超出就丢最久没用的
def expensive_calc(x, y):
    ...

LRU = Least Recently Used（最近最少使用）。maxsize 控制内存占用，cache 等价于 lru_cache(maxsize=None)（无限大）。

使用条件：

1. 函数参数必须 hashable（list、dict 不行，tuple、str、int 可以）
2. 函数应该是纯函数（同样参数永远返回同样结果，不依赖外部状态）

@cache
def get_current_user():        # 这会一直返回第一次的结果！
    return db.query_current_user()

wraps —— 写装饰器必备

Day 2 题 C 用过：让装饰后的函数保留原函数的名字和文档。

from functools import wraps

def my_decorator(func):
    @wraps(func)            # ← 必加
    def wrapper(*args, **kwargs):
        return func(*args, **kwargs)
    return wrapper

不加 @wraps，被装饰的函数 __name__ 会变成 'wrapper'，调试和文档全乱套。写装饰器永远要加。

partial —— 偏函数（局部固定参数）

from functools import partial

def greet(greeting, name):
    return f"{greeting}, {name}!"


say_hello = partial(greet, "Hello")
say_hello("Alice")     # 'Hello, Alice!'
say_hello("Bob")       # 'Hello, Bob!'

典型场景：把多参数函数适配成"只接受一个参数"的形式，方便丢给 map、sorted、回调函数等。

posts = [...]
sorted(posts, key=lambda p: p["pv"])             # 用 lambda
sorted(posts, key=partial(get_field, "pv"))      # 用 partial（如果有现成函数）

实战中 lambda 用得更多，partial 在某些回调场景更清晰。

reduce —— 累积运算（用得少）

from functools import reduce


reduce(lambda a, b: a + b, [1, 2, 3, 4])    # 10


dicts = [{"a": 1}, {"b": 2}, {"c": 3}]
merged = reduce(lambda x, y: {**x, **y}, dicts)

90% 的场景能用 sum/min/max/任何循环替代，知道有这玩意就行。

2.3 itertools —— 迭代器工具

itertools 是个工具箱，里面是各种「生成迭代器」的函数。它们的特点：惰性求值，不一次性生成结果，省内存。

注意：下面所有 print(...) 输出，都得用 list(...) 包一下才能看见——因为它们返回的是迭代器，不是 list。

chain —— 把多个序列拼起来

from itertools import chain

a = [1, 2, 3]
b = [4, 5, 6]
list(chain(a, b))     # [1, 2, 3, 4, 5, 6]

zip_longest —— zip 的长度不一致版

from itertools import zip_longest

a = [1, 2, 3]
b = ["a", "b"]

list(zip(a, b))               # [(1, 'a'), (2, 'b')]   ← 短的截断
list(zip_longest(a, b, fillvalue="?"))

product —— 笛卡尔积（开头需求 6 的答案）

from itertools import product

tags = ["python", "web"]
months = ["2026-01", "2026-02"]

list(product(tags, months))

替代手写嵌套 for 循环：

for tag in tags:
    for month in months:
        process(tag, month)


for tag, month in product(tags, months):
    process(tag, month)

combinations 和 permutations —— 组合/排列

from itertools import combinations, permutations

list(combinations([1, 2, 3], 2))     


list(permutations([1, 2, 3], 2))     

数学题味比较浓，实际工作中偶尔用——比如"给我 5 个文章，所有两两配对"用 combinations(posts, 2)。

groupby —— 按 key 分组

使用前必须先排序！这是个大坑。

from itertools import groupby

data = [
    {"author": "Alice", "title": "P1"},
    {"author": "Bob", "title": "P2"},
    {"author": "Alice", "title": "P3"},
]


data.sort(key=lambda x: x["author"])

for author, group in groupby(data, key=lambda x: x["author"]):
    print(author, list(group))

90% 的场景，**defaultdict(list)**** 比 **groupby** 更直观**：

groups = defaultdict(list)
for item in data:
    groups[item["author"]].append(item)

groupby 主要在"已经有序的流式数据"场景才占优势（比如读已排序的 CSV）。

islice —— 切片的迭代器版

from itertools import islice


with open("huge.log") as f:
    for line in islice(f, 100):
        process(line)

还有一堆，记住"功能"就行

accumulate（累积和）、takewhile（满足条件就取）、dropwhile（满足条件就丢）、count（无限计数器）、cycle（循环）、repeat（重复）……

记忆策略：知道 itertools 里"基本啥都有"，写循环时如果觉得"应该有更优雅的写法"，先去翻一下 itertools 文档。

2.4 heapq —— 堆（用于 Top K 问题）

什么是堆

堆是一种特殊的列表，保证「最小元素始终在 list[0]」。Python 的 heapq 实现的是最小堆。

重点：heapq 就是普通 list，只是用一组特定函数维护它的"堆性质"。不是新数据结构。

基本操作

import heapq

h = []
heapq.heappush(h, 5)
heapq.heappush(h, 1)
heapq.heappush(h, 3)
print(h)                    # [1, 5, 3]   ← 不是排序，但 [0] 是最小

heapq.heappop(h)            # 1   ← 弹出最小的
print(h)                    # [3, 5]

核心场景：Top K 问题（开头需求 3）

「找出阅读量前 10 的文章」——朴素做法是排所有的再取前 10，复杂度 O(n log n)。用堆只需要 O(n log k)，n 大、k 小时性能差别巨大。

import heapq

posts = [
    {"title": "A", "pv": 100},
    {"title": "B", "pv": 5000},
    {"title": "C", "pv": 200},
    # ... 假设有 100 万篇
]


top_10 = heapq.nlargest(10, posts, key=lambda p: p["pv"])


bottom_10 = heapq.nsmallest(10, posts, key=lambda p: p["pv"])

nlargest 和 nsmallest 内部就是堆——你根本不用直接操作 heappush/heappop，知道有这两个函数就够了。

什么时候用 **sorted()[:k]** 反而更好：

• 数据量小（< 1000）：随便用哪个，sorted 更直观
• k 很接近 n（比如 n=100, k=80）：不如直接 sort
• 数据量大且 k << n（比如 n=百万, k=10）：用 nlargest 性能优势明显

2.5 bisect —— 二分查找（用得最少但很聪明）

什么是二分查找

在已排序的列表里查找元素，复杂度 O(log n)，比逐个找快得多。

import bisect

scores = [60, 70, 75, 80, 85, 90, 95]    # 已排序


bisect.bisect_left(scores, 80)     # 3   （第一个 >= 80 的位置）
bisect.bisect_right(scores, 80)    # 4   （第一个 > 80 的位置）


bisect.insort(scores, 78)
print(scores)                       # [60, 70, 75, 78, 80, 85, 90, 95]

经典应用：把数值映射到等级

breakpoints = [60, 70, 80, 90]      # 4 个分界点
grades = ['F', 'D', 'C', 'B', 'A']  # 5 个等级

def grade(score):
    return grades[bisect.bisect_right(breakpoints, score)]

grade(50)    # 'F'
grade(75)    # 'C'
grade(85)    # 'B'
grade(95)    # 'A'

这个一行替代了一坨 if/elif：

if score < 60:
    return 'F'
elif score < 70:
    return 'D'

bisect 的所有应用基本都是「根据连续区间映射到离散标签」这个模式：HTTP 状态码 → 类别、年龄 → 年龄段、价格 → 价格区间。

列表去重保序（开头需求 5）

bisect 不直接做这个，但 dict 从 Python 3.7 起保序了，可以这样：

related = ["a", "b", "a", "c", "b", "d"]
unique = list(dict.fromkeys(related))

这是 Python 中"去重保序"的标准写法。

2.6 五个模块的"什么时候用"速查

需求	用什么
计数	Counter
字典自动初始化	defaultdict
最近 N 条队列	deque(maxlen=N)
内存缓存	@cache / @lru_cache
写装饰器	@wraps
笛卡尔积 / 组合	itertools.product / combinations
大文件取前 N 行	itertools.islice
拼多个序列	itertools.chain
Top K 问题	heapq.nlargest / nsmallest
数值 → 区间标签	bisect.bisect_right
去重保序	list(dict.fromkeys(...))

3. Maybe Useful 旁注

旁注 1：Python dict 从 3.7 起保序：所以 "去重保序" 一行能写。这是个标准库内化的提升——以前要 OrderedDict，现在普通 dict 就行。OrderedDict 现在主要剩下 move_to_end 等少数特殊操作时才用。

旁注 2：**@cache**** 和"实例方法"的坑**：给类的方法加 @cache，会意外地让所有实例共享缓存——因为 self 也是参数的一部分，但实际是同一个对象类型。需要的话用第三方库 cachetools 或自己写缓存逻辑。

旁注 3：**itertools**** 的设计哲学**：函数都返回迭代器而不是 list，鼓励"流式处理"——大数据时省内存。这种设计源自 Haskell 等函数式语言的影响。

旁注 4：堆的"O(log n)" 是怎么来的：堆是个完全二叉树（虽然存在 list 里），插入/删除时元素只需要在树高度上下移动。一棵 n 个节点的二叉树高度是 log₂(n)。你不用懂证明，记住"堆操作是对数级"就行。

旁注 5：面试高频题：

• 「找前 K 大的元素」→ heapq
• 「LRU 缓存怎么实现」→ OrderedDict 或自己写
• 「单词频率统计」→ Counter
• 「为什么不要在循环里 list.insert(0, ...)」→ deque
• 「@lru_cache 怎么用？要注意什么？」→ 必答

4. 代码实践

我们围绕"博客后台分析工具"写一个综合 demo，把今天的工具都用上：

"""
blog_analytics.py - 用今天学的工具分析博客数据
"""
from collections import Counter, defaultdict, deque
from functools import cache
from itertools import islice
import heapq
import bisect


posts = [
    {"title": "Python 入门", "author": "Alice", "tags": ["python", "tutorial"], "pv": 1500},
    {"title": "Web 开发", "author": "Bob", "tags": ["python", "web"], "pv": 5000},
    {"title": "数据分析", "author": "Alice", "tags": ["python", "data"], "pv": 800},
    {"title": "前端基础", "author": "Charlie", "tags": ["web", "html"], "pv": 3000},
    {"title": "Django 教程", "author": "Bob", "tags": ["python", "web", "tutorial"], "pv": 2500},
    {"title": "Vue 实战", "author": "Charlie", "tags": ["web", "js"], "pv": 1200},
    # ... 假设有更多
]


def tag_popularity():
    """统计所有标签出现次数。"""
    all_tags = []
    for p in posts:
        all_tags.extend(p["tags"])
    return Counter(all_tags).most_common()


def group_by_author():
    """按作者分组文章。"""
    groups = defaultdict(list)
    for p in posts:
        groups[p["author"]].append(p["title"])
    return dict(groups)


def top_3_by_pv():
    """阅读量前 3 的文章。"""
    return heapq.nlargest(3, posts, key=lambda p: p["pv"])


recent_views = deque(maxlen=5)
def view_post(title):
    """记录用户浏览，自动只保留最近 5 篇。"""
    recent_views.append(title)


PV_BREAKPOINTS = [500, 2000, 5000]
PV_LEVELS = ["低", "中", "高", "爆款"]
def pv_level(pv: int) -> str:
    return PV_LEVELS[bisect.bisect_right(PV_BREAKPOINTS, pv)]


@cache
def render_post(post_title: str) -> str:
    """渲染文章（假装很慢）。"""
    print(f"  [真的在渲染 {post_title}]")
    # 假装做了复杂渲染
    return f"<h1>{post_title}</h1>..."


if __name__ == "__main__":
    print("标签热度:")
    for tag, count in tag_popularity():
        print(f"  {tag}: {count}")

    print("\n按作者分组:")
    for author, titles in group_by_author().items():
        print(f"  {author}: {titles}")

    print("\nTop 3 阅读量:")
    for p in top_3_by_pv():
        print(f"  {p['title']}: {p['pv']} ({pv_level(p['pv'])})")

    print("\n模拟浏览:")
    for title in ["Python 入门", "Web 开发", "数据分析", "前端基础", "Django 教程", "Vue 实战"]:
        view_post(title)
    print(f"  最近 5 篇: {list(recent_views)}")

    print("\n缓存测试:")
    render_post("Python 入门")    # 第一次：执行
    render_post("Python 入门")    # 第二次：从缓存返回，不打印 [真的在渲染]
    render_post("Web 开发")        # 不同参数：执行

预期输出：

标签热度:
  python: 4
  web: 4
  tutorial: 2
  data: 1
  html: 1
  js: 1

按作者分组:
  Alice: ['Python 入门', '数据分析']
  Bob: ['Web 开发', 'Django 教程']
  Charlie: ['前端基础', 'Vue 实战']

Top 3 阅读量:
  Web 开发: 5000 (高)
  前端基础: 3000 (中)
  Django 教程: 2500 (中)

模拟浏览:
  最近 5 篇: ['Web 开发', '数据分析', '前端基础', 'Django 教程', 'Vue 实战']

缓存测试:
  [真的在渲染 Python 入门]
  [真的在渲染 Web 开发]

注意 render_post("Python 入门") 第二次没打印 [真的在渲染]——@cache 起作用了。

5. 练习题

理论题

1. Counter 和 defaultdict(int) 用来计数有什么区别？什么时候选哪个？
2. deque(maxlen=10) 和普通 list 在"保留最近 10 条"这个需求下，性能差别在哪？
3. 给函数加 @cache 有什么前提条件？什么样的函数不能加？
4. heapq.nlargest(10, data) 比 sorted(data, reverse=True)[:10] 快在哪？什么时候快？
5. 写装饰器为什么要加 @wraps？不加会怎样？

代码实操题

题目 A：写一个函数 top_authors(posts, n)，返回总阅读量最高的 n 个作者。

提示：先用 defaultdict(int) 累加每个作者的 PV，再用 heapq.nlargest。

题目 B：写一个函数 running_average(stream, window)，输入一个数字流，返回每接收一个数字后最近 window 个数字的均值。

提示：用 deque(maxlen=window)。每次 append 后用 sum() / len() 算均值。

题目 C：给 Day 3 的 humanize 函数（"3 小时前"）改造——用 bisect 替代 if/elif 链。

提示：定义 BREAKPOINTS = [60, 3600, 86400, ...] 和对应的格式化函数列表。

思考题

你的博客文章列表越来越长（10 万篇）。每次访问首页都要：

1. 按发布时间倒序排列
2. 取最新 20 篇
3. 给每篇渲染 HTML

用今天学的工具，怎么优化这个流程？哪些步骤适合用 heapq？哪些适合用 @cache？如果文章会被编辑（缓存可能失效），怎么处理？

6. 当天总结

今天学了什么：

五个模块对应五类任务：

• collections —— 增强容器（Counter / defaultdict / deque）
• functools —— 函数工具（cache / wraps / partial）
• itertools —— 迭代器组合子（product / chain / islice）
• heapq —— Top K 问题
• bisect —— 已排序数据的查询

最重要的概念：

1. Counter / defaultdict / deque 是日常生产力工具——用熟它们能省一大半 if 判断
2. **@cache**** 是 Python 性能优化第一招**——递归、纯函数加上立刻飞起
3. **Top K 问题用 **heapq.nlargest**，不要 ****sorted()[:k]**
4. **想用 **groupby** 时，90% 应该用 ****defaultdict(list)**

需要反复练习：

• Counter.most_common
• @cache 的使用条件
• defaultdict 的工厂参数

和明天的衔接：

明天 Day 5 讲 argparse / logging / sys / pdb——把脚本变成命令行工具。具体衔接点：

• 今天的 blog_analytics.py 还是个固定脚本，明天会用 argparse 给它加 --top 10``--author Alice 这样的命令行参数
• print 调试以后会用 logging 替代
• 出 bug 时用 pdb 断点（之前提过 breakpoint()，明天系统讲）