Python高级数据结构实战:Collections模块深度解析

前面咱们一起闯过了错误与异常处理的关卡，是不是感觉自己的代码更加健壮了？但你想过没有，Python自带的数据结构（list、dict、set）虽然强大，但在某些特定场景下，是不是感觉有点“力不从心”？比如统计元素频率、处理缺失键、维护插入顺序、实现高效队列……别担心，今天咱们就来探索Python的 “神器宝库” ——Collections模块！

• • 需要统计一段文本中每个单词出现的次数，用普通字典写起来又啰嗦又慢？
• • 处理嵌套字典时，经常遇到KeyError，需要写大量if判断来避免？
• • 需要记住字典的插入顺序（比如配置项、操作历史），但Python 3.7之前的字典不保证顺序？
• • 要实现一个队列或栈，用list虽然可以，但性能不够优化？
• • 需要一个轻量级的对象来存储数据，又不想定义完整的类？

from collections import Counter# 传统方法：啰嗦且易错words = ['apple', 'banana', 'apple', 'orange', 'banana', 'apple']word_count = {}for word in words:if word in word_count:        word_count[word] += 1else:        word_count[word] = 1print(word_count)  # {'apple': 3, 'banana': 2, 'orange': 1}# Counter方法：一行搞定！word_counter = Counter(words)print(word_counter)  # Counter({'apple': 3, 'banana': 2, 'orange': 1})

# 1. 直接更新计数counter = Counter(['a', 'b', 'c', 'a', 'b'])print(counter)  # Counter({'a': 2, 'b': 2, 'c': 1})counter.update(['a', 'b', 'd'])  # 添加新元素print(counter)  # Counter({'a': 3, 'b': 3, 'c': 1, 'd': 1})# 2. 获取最常见元素print(counter.most_common(2))  # [('a', 3), ('b', 3)]# 3. 计数器运算c1 = Counter(a=3, b=2, c=1)c2 = Counter(a=1, b=2, c=3)print(c1 + c2)   # Counter({'a': 4, 'b': 4, 'c': 4})  # 加法print(c1 - c2)   # Counter({'a': 2})                  # 减法（只保留正数）print(c1 & c2)   # Counter({'a': 1, 'b': 2, 'c': 1})  # 交集（取最小值）print(c1 | c2)   # Counter({'a': 3, 'b': 2, 'c': 3})  # 并集（取最大值）# 4. 从任意可迭代对象创建text = "hello world hello python world"char_counter = Counter(text.replace(" ", ""))print(char_counter.most_common(3))  # [('l', 5), ('o', 4), ('h', 3)]

defanalyze_survey_responses(responses):"""分析调查问卷数据"""# 清洗数据：去除空值和无效回答    valid_responses = [r.strip().lower() for r in responses if r and r.strip()]# 使用Counter统计    counter = Counter(valid_responses)# 分析结果    total = len(valid_responses)print(f"有效回答数: {total}")print(f"选项分布:")for value, count in counter.most_common():        percentage = (count / total) * 100print(f"  {value}: {count}次 ({percentage:.1f}%)")# 返回最受欢迎的选择    most_common = counter.most_common(1)[0] if counter elseNonereturn most_common# 测试responses = ['A', 'B', 'A', 'C', 'A', '', 'B', 'B', 'D', None, 'A']result = analyze_survey_responses(responses)print(f"最受欢迎的选择: {result}")

# 传统方法：需要先检查键是否存在data = [('a', 1), ('b', 2), ('a', 3), ('c', 4)]grouped = {}for key, value in data:if key notin grouped:        grouped[key] = []  # 每次都要检查    grouped[key].append(value)# defaultdict方法：自动初始化from collections import defaultdictgrouped = defaultdict(list)  # 默认值为空列表for key, value in data:    grouped[key].append(value)  # 直接追加，不需要检查！

# 1. 列表作为默认值（分组场景）list_dict = defaultdict(list)list_dict['fruits'].append('apple')list_dict['fruits'].append('banana')print(list_dict)  # defaultdict(<class 'list'>, {'fruits': ['apple', 'banana']})# 2. 整数作为默认值（计数场景）int_dict = defaultdict(int)int_dict['a'] += 1# 不需要初始化int_dict['b'] += 3print(int_dict)  # defaultdict(<class 'int'>, {'a': 1, 'b': 3})# 3. 集合作为默认值（去重分组）set_dict = defaultdict(set)set_dict['tags'].add('python')set_dict['tags'].add('programming')set_dict['tags'].add('python')  # 重复的不会添加print(set_dict)  # defaultdict(<class 'set'>, {'tags': {'python', 'programming'}})# 4. 自定义默认值工厂defdefault_person():return {'name': '', 'age': 0, 'score': 0.0}person_dict = defaultdict(default_person)person_dict['alice']['name'] = 'Alice'person_dict['alice']['age'] = 25print(person_dict['bob'])  # 访问不存在的键返回默认字典

classTreeBuilder:"""使用defaultdict构建树状结构"""def__init__(self):self.tree = defaultdict(lambda: defaultdict(list))defadd_file(self, path):"""添加文件路径到树中"""        parts = path.split('/')        current = self.treefor i, part inenumerate(parts):if i == len(parts) - 1:  # 文件                current['files'].append(part)else:  # 目录                current = current['dirs'][part]defprint_tree(self, node=None, indent=0):"""打印树状结构"""if node isNone:            node = self.tree# 打印目录for dir_name, sub_node in node['dirs'].items():print('  ' * indent + f"📁 {dir_name}/")self.print_tree(sub_node, indent + 1)# 打印文件for file_name in node['files']:print('  ' * indent + f"📄 {file_name}")# 使用builder = TreeBuilder()builder.add_file('src/utils/helper.py')builder.add_file('src/models/user.py')builder.add_file('docs/README.md')builder.print_tree()

from collections import OrderedDict# 1. 维护特定顺序（不是插入顺序）config = OrderedDict()config['version'] = '1.0'config['author'] = 'Python学习搭子'config['license'] = 'MIT'config['description'] = '高级数据结构教程'# 即使重新排列，也能保持指定顺序print(list(config.keys()))  # ['version', 'author', 'license', 'description']# 2. 顺序相关操作od = OrderedDict([('a', 1), ('b', 2), ('c', 3)])# 移动元素到末尾od.move_to_end('a')print(list(od.keys()))  # ['b', 'c', 'a']# 移动元素到开头od.move_to_end('a', last=False)print(list(od.keys()))  # ['a', 'b', 'c']# 弹出指定位置的元素first_item = od.popitem(last=False)  # 弹出第一个last_item = od.popitem(last=True)    # 弹出最后一个print(first_item, last_item)  # ('a', 1) ('c', 3)

classLRUCache:"""基于OrderedDict实现的LRU缓存"""def__init__(self, capacity: int):self.capacity = capacityself.cache = OrderedDict()defget(self, key):"""获取缓存值，并将键移动到末尾（表示最近使用）"""if key notinself.cache:return -1# 移动到最后self.cache.move_to_end(key)returnself.cache[key]defput(self, key, value):"""添加缓存值"""if key inself.cache:# 已存在，更新值并移动到最后self.cache.move_to_end(key)else:# 检查容量iflen(self.cache) >= self.capacity:# 弹出最久未使用的（第一个）self.cache.popitem(last=False)self.cache[key] = valuedef__str__(self):"""显示缓存内容（按使用时间从旧到新）"""        items = [f"{k}: {v}"for k, v inself.cache.items()]return"LRUCache: {" + ", ".join(items) + "}"# 测试cache = LRUCache(3)cache.put('a', 1)cache.put('b', 2)cache.put('c', 3)print(cache)  # LRUCache: {a: 1, b: 2, c: 3}print(cache.get('a'))  # 1，现在a变成最近使用的print(cache)  # LRUCache: {b: 2, c: 3, a: 1}cache.put('d', 4)  # 容量已满，淘汰最旧的bprint(cache)  # LRUCache: {c: 3, a: 1, d: 4}

from collections import namedtuple# 1. 定义Point类型Point = namedtuple('Point', ['x', 'y'])# 2. 创建实例p1 = Point(10, 20)p2 = Point(x=5, y=15)print(p1)        # Point(x=10, y=20)print(p1.x, p1.y)  # 10 20（可以通过名称访问！）print(p1[0], p1[1])  # 10 20（也可以通过索引访问）# 3. 不可变性（安全！）# p1.x = 100  # 报错：AttributeError# 4. 内存效率对比import sysfrom dataclasses import dataclass# 普通类classRegularPoint:def__init__(self, x, y):self.x = xself.y = y# 数据类@dataclassclassDataPoint:    x: int    y: int# 对比内存占用nt_point = Point(10, 20)reg_point = RegularPoint(10, 20)data_point = DataPoint(10, 20)print(f"namedtuple: {sys.getsizeof(nt_point)} bytes")print(f"普通类: {sys.getsizeof(reg_point)} bytes")print(f"数据类: {sys.getsizeof(data_point)} bytes")

# 1. 默认值（Python 3.7+）Person = namedtuple('Person', ['name', 'age', 'city'], defaults=['Unknown', 0, 'Unknown'])p1 = Person('Alice')  # 只提供name，其他用默认值print(p1)  # Person(name='Alice', age=0, city='Unknown')# 2. 类型注解（Python 3.6+）from typing import NamedTupleclassEmployee(NamedTuple):id: int    name: str    department: str = '未分配'    salary: float = 0.0emp = Employee(1, '张三', '技术部', 15000.0)print(emp)  # Employee(id=1, name='张三', department='技术部', salary=15000.0)# 3. 方法扩展classPoint(namedtuple('Point', ['x', 'y'])):"""扩展namedtuple，添加自定义方法"""    __slots__ = ()  # 防止创建实例字典，保持内存效率defdistance_to(self, other):"""计算两点距离"""return ((self.x - other.x) ** 2 + (self.y - other.y) ** 2) ** 0.5def__add__(self, other):"""向量加法"""return Point(self.x + other.x, self.y + other.y)    @propertydefmagnitude(self):"""向量模长"""return (self.x ** 2 + self.y ** 2) ** 0.5# 使用扩展的Pointp1 = Point(3, 4)p2 = Point(1, 2)print(f"距离: {p1.distance_to(p2):.2f}")print(f"相加: {p1 + p2}")print(f"模长: {p1.magnitude:.2f}")

defprocess_student_records(records):"""处理学生记录"""# 定义命名元组    Student = namedtuple('Student', ['id', 'name', 'score', 'grade'])    processed = []for record in records:# 计算等级        score = record['score']if score >= 90:            grade = 'A'elif score >= 80:            grade = 'B'elif score >= 60:            grade = 'C'else:            grade = 'D'# 创建命名元组实例        student = Student(id=record['id'],            name=record['name'],            score=score,            grade=grade        )        processed.append(student)# 分析结果    grade_counter = Counter(s.grade for s in processed)    avg_score = sum(s.score for s in processed) / len(processed)print(f"学生总数: {len(processed)}")print(f"平均分: {avg_score:.1f}")print(f"等级分布: {dict(grade_counter)}")return processed# 测试records = [    {'id': 1, 'name': '张三', 'score': 85},    {'id': 2, 'name': '李四', 'score': 92},    {'id': 3, 'name': '王五', 'score': 78},    {'id': 4, 'name': '赵六', 'score': 95},    {'id': 5, 'name': '钱七', 'score': 62}]students = process_student_records(records)print(f"\n第一个学生: {students[0]}")print(f"第一个学生的姓名: {students[0].name}")

from collections import dequeimport time# 性能对比：从左侧添加/删除元素deftest_performance():"""测试deque和list在队列操作上的性能"""# 测试数据    test_size = 100000# 使用list    start = time.time()    lst = []for i inrange(test_size):        lst.insert(0, i)  # 从开头插入 - O(n)操作for i inrange(test_size):        lst.pop(0)  # 从开头弹出 - O(n)操作    list_time = time.time() - start# 使用deque    start = time.time()    dq = deque()for i inrange(test_size):        dq.appendleft(i)  # 从开头添加 - O(1)操作for i inrange(test_size):        dq.popleft()  # 从开头弹出 - O(1)操作    deque_time = time.time() - startprint(f"list耗时: {list_time:.4f}秒")print(f"deque耗时: {deque_time:.4f}秒")print(f"deque比list快 {list_time/deque_time:.1f}倍")test_performance()

# 创建dequedq = deque([1, 2, 3, 4, 5])print(f"初始deque: {dq}")  # deque([1, 2, 3, 4, 5])# 1. 两端操作dq.append(6)           # 右侧添加: deque([1, 2, 3, 4, 5, 6])dq.appendleft(0)       # 左侧添加: deque([0, 1, 2, 3, 4, 5, 6])right_item = dq.pop()  # 右侧弹出: 6left_item = dq.popleft()  # 左侧弹出: 0print(f"操作后: {dq}")  # deque([1, 2, 3, 4, 5])# 2. 批量添加dq.extend([6, 7, 8])     # 右侧批量添加dq.extendleft([-1, -2])  # 左侧批量添加（注意顺序！）print(f"批量添加后: {dq}")  # deque([-2, -1, 1, 2, 3, 4, 5, 6, 7, 8])# 3. 旋转操作dq.rotate(2)   # 向右旋转2位print(f"向右旋转2位: {dq}")  # deque([7, 8, -2, -1, 1, 2, 3, 4, 5, 6])dq.rotate(-2)  # 向左旋转2位print(f"向左旋转2位: {dq}")  # 恢复原状# 4. 限制最大长度limited_dq = deque(maxlen=3)for i inrange(5):    limited_dq.append(i)print(f"添加{i}: {limited_dq}")# 输出:# 添加0: deque([0], maxlen=3)# 添加1: deque([0, 1], maxlen=3)# 添加2: deque([0, 1, 2], maxlen=3)# 添加3: deque([1, 2, 3], maxlen=3)  # 自动淘汰最旧的0# 添加4: deque([2, 3, 4], maxlen=3)  # 自动淘汰最旧的1

classSlidingWindow:"""基于deque的滑动窗口统计器"""def__init__(self, window_size):self.window_size = window_sizeself.window = deque(maxlen=window_size)self.sum = 0defadd(self, value):"""添加新值到窗口"""iflen(self.window) == self.window_size:# 窗口已满，移除最旧的值            old_value = self.window[0]self.sum -= old_valueself.window.append(value)self.sum += valuedefget_average(self):"""计算窗口平均值"""ifnotself.window:return0returnself.sum / len(self.window)defget_median(self):"""计算窗口中位数"""ifnotself.window:return0        sorted_window = sorted(self.window)        n = len(sorted_window)if n % 2 == 1:return sorted_window[n // 2]else:            mid = n // 2return (sorted_window[mid - 1] + sorted_window[mid]) / 2defget_stats(self):"""获取窗口统计信息"""ifnotself.window:return {}return {'size': len(self.window),'sum': self.sum,'average': self.get_average(),'median': self.get_median(),'min': min(self.window),'max': max(self.window),'window': list(self.window)        }# 测试：实时数据流统计stream = [10, 20, 30, 40, 50, 60, 70, 80, 90, 100]window = SlidingWindow(5)print("滑动窗口统计:")for i, value inenumerate(stream, 1):    window.add(value)    stats = window.get_stats()print(f"第{i}个值{value}: 平均={stats['average']:.1f}, "f"中位数={stats['median']:.1f}, 窗口={stats['window']}")

from collections import ChainMap# 创建ChainMapdefault_config = {'theme': 'light', 'language': 'zh-CN'}user_config = {'theme': 'dark', 'font_size': 14}session_config = {'user_id': 123}# 合并多个配置（按优先级从高到低）config = ChainMap(session_config, user_config, default_config)print(f"主题: {config['theme']}")  # dark（从user_config获取）print(f"语言: {config['language']}")  # zh-CN（从default_config获取）# 更新只影响第一个字典config['theme'] = 'blue'print(user_config)  # {'theme': 'blue', 'font_size': 14}

from collections import UserDictclassCaseInsensitiveDict(UserDict):"""不区分大小写的字典"""def__init__(self, data=None):super().__init__()if data:# 转换所有键为小写self.data = {k.lower(): v for k, v in data.items()}def__setitem__(self, key, value):# 存储时键转为小写super().__setitem__(key.lower(), value)def__getitem__(self, key):# 获取时键转为小写returnsuper().__getitem__(key.lower())def__contains__(self, key):returnsuper().__contains__(key.lower())# 使用cid = CaseInsensitiveDict({'Name': 'Alice', 'AGE': 25})print(cid['name'])  # Aliceprint(cid['NAME'])  # Alicecid['Age'] = 26print(cid)  # {'name': 'Alice', 'age': 26}

import refrom collections import Counter, defaultdictfrom typing importList, Dict, TupleclassTextAnalyzer:"""文本分析器：统计高频词、词频分布等"""def__init__(self, text: str):self.text = textself.words = self._extract_words(text)self.word_counter = Counter(self.words)def_extract_words(self, text: str) -> List[str]:"""提取文本中的单词（去除标点、转为小写）"""# 移除标点符号，只保留字母和数字        cleaned = re.sub(r'[^\w\s]', ' ', text.lower())# 分割单词        words = cleaned.split()return wordsdeftop_words(self, n: int = 10) -> List[Tuple[str, int]]:"""返回出现频率最高的n个单词"""returnself.word_counter.most_common(n)defword_frequency(self, word: str) -> int:"""返回特定单词的出现频率"""returnself.word_counter.get(word.lower(), 0)defvocabulary_size(self) -> int:"""返回词汇量（不重复单词数）"""returnlen(self.word_counter)defword_length_distribution(self) -> Dict[int, int]:"""统计单词长度分布"""        length_counter = Counter(len(word) for word inself.words)returndict(length_counter)defwords_by_length(self) -> defaultdict:"""按单词长度分组"""        grouped = defaultdict(list)for word inset(self.words):  # 去重            grouped[len(word)].append(word)return groupeddefgenerate_report(self) -> str:"""生成文本分析报告"""        report_lines = []# 基本信息        report_lines.append("=" * 50)        report_lines.append("文本分析报告")        report_lines.append("=" * 50)        report_lines.append(f"文本长度: {len(self.text)} 字符")        report_lines.append(f"单词总数: {len(self.words)}")        report_lines.append(f"词汇量: {self.vocabulary_size()} 个不重复单词")# 高频词        report_lines.append("\n🔝 高频词 Top 10:")for word, count inself.top_words(10):            percentage = (count / len(self.words)) * 100            report_lines.append(f"  {word}: {count}次 ({percentage:.1f}%)")# 单词长度分布        report_lines.append("\n📏 单词长度分布:")        length_dist = self.word_length_distribution()for length insorted(length_dist.keys()):            count = length_dist[length]            bar = "█" * (count // (max(length_dist.values()) // 20 + 1))            report_lines.append(f"  {length}字母: {count:3d}个 {bar}")# 按长度分组示例        report_lines.append("\n📂 按长度分组（示例）:")        grouped = self.words_by_length()for length insorted(grouped.keys())[:5]:  # 只显示前5组            words_sample = grouped[length][:5]  # 每组只显示前5个单词iflen(words_sample) == 5andlen(grouped[length]) > 5:                words_sample.append("...")            report_lines.append(f"  {length}字母: {', '.join(words_sample)}")return"\n".join(report_lines)# 使用示例sample_text = """Python is an interpreted, high-level, general-purpose programming language. Created by Guido van Rossum and first released in 1991, Python's design philosophy emphasizes code readability with its notable use of significant whitespace. Its language constructs and object-oriented approach aim to help programmers write clear, logical code for small and large-scale projects."""analyzer = TextAnalyzer(sample_text)print(analyzer.generate_report())# 单独使用功能print(f"\n'python'出现次数: {analyzer.word_frequency('python')}")print(f"单词长度分布: {analyzer.word_length_distribution()}")

from collections import OrderedDictfrom datetime import datetime, timedeltafrom typing importAny, Optional, Callableimport timeclassEnhancedLRUCache:"""    增强版LRU缓存，支持：    1. 容量限制    2. TTL（过期时间）    3. 命中率统计    4. 回调函数（淘汰时触发）    """def__init__(self, capacity: int = 100, default_ttl: int = 3600):self.capacity = capacityself.default_ttl = default_ttl  # 默认过期时间（秒）self.cache = OrderedDict()self.hits = 0self.misses = 0self.evictions = 0self.callbacks = {}def_evict_if_needed(self):"""如果需要，淘汰最久未使用的条目"""whilelen(self.cache) > self.capacity:            key, (value, expiry, callback_key) = self.cache.popitem(last=False)self.evictions += 1# 触发回调if callback_key inself.callbacks:try:self.callbacks[callback_key](key, value)except Exception:pass# 回调出错不中断主流程def_is_expired(self, expiry: float) -> bool:"""检查是否过期"""return time.time() > expirydefset(self, key: str, value: Any,             ttl: Optional[int] = None,            on_evict: Optional[Callable] = None) -> None:"""        设置缓存值        参数:            key: 缓存键            value: 缓存值            ttl: 过期时间（秒），None表示使用默认值            on_evict: 淘汰时的回调函数        """# 计算过期时间        actual_ttl = ttl if ttl isnotNoneelseself.default_ttl        expiry = time.time() + actual_ttl# 存储回调        callback_key = Noneif on_evict:            callback_key = f"{key}_{id(on_evict)}"self.callbacks[callback_key] = on_evict# 存储到缓存self.cache[key] = (value, expiry, callback_key)self.cache.move_to_end(key)  # 标记为最近使用# 检查容量self._evict_if_needed()defget(self, key: str) -> Optional[Any]:"""获取缓存值"""if key notinself.cache:self.misses += 1returnNone        value, expiry, callback_key = self.cache[key]# 检查是否过期ifself._is_expired(expiry):delself.cache[key]if callback_key inself.callbacks:delself.callbacks[callback_key]self.misses += 1returnNone# 标记为最近使用self.cache.move_to_end(key)self.hits += 1return valuedefdelete(self, key: str) -> bool:"""删除缓存值"""if key inself.cache:            _, _, callback_key = self.cache[key]delself.cache[key]if callback_key inself.callbacks:delself.callbacks[callback_key]returnTruereturnFalsedefclear(self):"""清空缓存"""self.cache.clear()self.callbacks.clear()self.hits = 0self.misses = 0self.evictions = 0defcleanup(self):"""清理过期条目"""        keys_to_delete = []for key, (value, expiry, callback_key) inself.cache.items():ifself._is_expired(expiry):                keys_to_delete.append((key, callback_key))for key, callback_key in keys_to_delete:delself.cache[key]if callback_key inself.callbacks:delself.callbacks[callback_key]self.evictions += 1defstats(self) -> dict:"""获取缓存统计信息"""        total = self.hits + self.misses        hit_rate = (self.hits / total * 100) if total > 0else0# 计算平均TTL        now = time.time()        total_ttl = 0        valid_entries = 0for _, (_, expiry, _) inself.cache.items():ifnotself._is_expired(expiry):                total_ttl += (expiry - now)                valid_entries += 1        avg_ttl = total_ttl / valid_entries if valid_entries > 0else0return {'capacity': self.capacity,'size': len(self.cache),'hits': self.hits,'misses': self.misses,'evictions': self.evictions,'hit_rate': f"{hit_rate:.1f}%",'avg_ttl': f"{avg_ttl:.1f}秒",'callbacks': len(self.callbacks)        }def__str__(self) -> str:"""显示缓存内容"""        lines = [f"EnhancedLRUCache(容量={self.capacity}, 当前大小={len(self.cache)})"]        lines.append("-" * 60)for i, (key, (value, expiry, _)) inenumerate(self.cache.items(), 1):            time_left = expiry - time.time()if time_left > 0:                lines.append(f"{i:2d}. {key}: {value} (剩余{time_left:.0f}秒)")        lines.append("-" * 60)        lines.append(f"命中率: {self.stats()['hit_rate']}")return"\n".join(lines)# 使用示例：数据库查询缓存defexpensive_db_query(user_id: int):"""模拟昂贵的数据库查询"""print(f"执行数据库查询: user_id={user_id}")    time.sleep(0.5)  # 模拟查询耗时returnf"用户{user_id}的数据"# 淘汰回调defon_cache_evict(key: str, value: Any):print(f"缓存条目被淘汰: {key} -> {value}")# 创建缓存cache = EnhancedLRUCache(capacity=3, default_ttl=10)# 测试print("=== 测试LRU缓存 ===")# 添加条目for i inrange(1, 6):    cache.set(f"user_{i}", expensive_db_query(i), on_evict=on_cache_evict)print(f"添加 user_{i} 后的缓存状态:")print(cache)print()# 访问条目改变顺序print("访问 user_3...")cache.get("user_3")print("访问后的缓存状态:")print(cache)print()# 查看统计print("缓存统计:")for key, value in cache.stats().items():print(f"  {key}: {value}")# 清理过期条目print("\n等待11秒让缓存过期...")time.sleep(11)cache.cleanup()print(f"清理后缓存大小: {len(cache.cache)}")