Python itertools模块全解析:掌握迭代器工具,让你的代码效率飙升!

import itertools# 从10开始，步长为2的无限序列counter = itertools.count(start=10, step=2)print("前5个值:", [next(counter) for _ in range(5)])# 输出: [10, 12, 14, 16, 18]# 支持浮点数和负数float_counter = itertools.count(2.5, 0.5)print("浮点序列:", [next(float_counter) for _ in range(3)])# 输出: [2.5, 3.0, 3.5]

defgenerate_ids(prefix="ID"):"""生成带前缀的唯一ID序列"""    counter = itertools.count(1)whileTrue:yieldf"{prefix}_{next(counter):06d}"# 使用示例id_generator = generate_ids("USER")user_ids = [next(id_generator) for _ in range(5)]print("用户ID:", user_ids)# 输出: ['USER_000001', 'USER_000002', 'USER_000003', 'USER_000004', 'USER_000005']

import itertools# 无限循环颜色序列colors = itertools.cycle(['红', '黄', '绿'])traffic_lights = [next(colors) for _ in range(7)]print("交通灯序列:", traffic_lights)# 输出: ['红', '黄', '绿', '红', '黄', '绿', '红']# 循环有限次数的应用limited_cycle = zip(range(10), itertools.cycle(['A', 'B', 'C']))print("有限循环:", list(limited_cycle))# 输出: [(0, 'A'), (1, 'B'), (2, 'C'), (3, 'A'), ...]

classLoadBalancer:"""简单的负载均衡器 - 轮询分发请求"""def__init__(self, servers):        self.servers = list(servers)        self.cycle = itertools.cycle(self.servers)defget_next_server(self):"""获取下一个服务器"""return next(self.cycle)defadd_server(self, server):"""动态添加服务器"""        self.servers.append(server)# 重新创建cycle以包含新服务器        self.cycle = itertools.cycle(self.servers)# 使用示例lb = LoadBalancer(['server1', 'server2', 'server3'])for i in range(6):    print(f"请求{i+1} → {lb.get_next_server()}")# 输出: 请求1 → server1, 请求2 → server2, 请求3 → server3, 请求4 → server1, ...

import itertools# 无限重复infinite_zeros = itertools.repeat(0)print("无限零:", [next(infinite_zeros) for _ in range(5)])# 输出: [0, 0, 0, 0, 0]# 指定重复次数limited_repeat = itertools.repeat('Python', 3)print("有限重复:", list(limited_repeat))# 输出: ['Python', 'Python', 'Python']

definitialize_matrix(rows, cols, value=0):"""初始化二维矩阵"""return [list(itertools.repeat(value, cols)) for _ in range(rows)]# 使用示例matrix = initialize_matrix(3, 4, 0.0)print("初始化矩阵:")for row in matrix:    print(row)# 输出:# [0.0, 0.0, 0.0, 0.0]# [0.0, 0.0, 0.0, 0.0]# [0.0, 0.0, 0.0, 0.0]

import itertools# 连接多个列表list1 = [1, 2, 3]list2 = ['a', 'b', 'c']list3 = [True, False]chained = itertools.chain(list1, list2, list3)print("连接结果:", list(chained))# 输出: [1, 2, 3, 'a', 'b', 'c', True, False]# 扁平化嵌套列表（一层）nested = [[1, 2], [3, 4], [5, 6]]flattened = list(itertools.chain.from_iterable(nested))print("扁平化结果:", flattened)# 输出: [1, 2, 3, 4, 5, 6]

defmerge_log_files(file_paths):"""合并多个日志文件的内容"""defread_lines(file_path):with open(file_path, 'r', encoding='utf-8') as f:for line in f:yield line.strip()# 创建多个生成器    file_generators = [read_lines(path) for path in file_paths]# 合并所有生成器    merged = itertools.chain.from_iterable(file_generators)return merged# 模拟使用log_files = ['app.log', 'error.log', 'access.log']# 实际使用时替换为真实文件路径# merged_logs = merge_log_files(log_files)# for line in itertools.islice(merged_logs, 5):#     print(line)

import itertools# 基础筛选data = ['苹果', '香蕉', '橙子', '葡萄', '芒果']mask = [1, 0, 1, 0, 1]  # 1保留，0过滤selected = list(itertools.compress(data, mask))print("筛选结果:", selected)# 输出: ['苹果', '橙子', '芒果']# 动态生成掩码numbers = range(10)even_mask = [n % 2 == 0for n in numbers]evens = list(itertools.compress(numbers, even_mask))print("偶数:", evens)# 输出: [0, 2, 4, 6, 8]

deffilter_passed_students(students, scores, passing_score=60):"""筛选考试通过的学生"""# 生成通过掩码    passed_mask = [score >= passing_score for score in scores]# 筛选学生    passed_students = list(itertools.compress(students, passed_mask))    passed_scores = list(itertools.compress(scores, passed_mask))return list(zip(passed_students, passed_scores))# 使用示例students = ['张三', '李四', '王五', '赵六', '钱七']scores = [85, 52, 90, 48, 75]passed = filter_passed_students(students, scores, 60)print("通过学生及成绩:", passed)# 输出: [('张三', 85), ('王五', 90), ('钱七', 75)]

import itertools# 基础分组（注意需要先排序）data = [('A', 1), ('A', 2), ('B', 3), ('B', 4), ('C', 5), ('A', 6)]# 错误示范：直接分组会得到错误结果# 正确做法：先按键排序sorted_data = sorted(data, key=lambda x: x[0])# 按第一个元素分组for key, group in itertools.groupby(sorted_data, key=lambda x: x[0]):    group_list = list(group)    print(f"组 {key}: {group_list}, 数量: {len(group_list)}")# 输出:# 组 A: [('A', 1), ('A', 2)], 数量: 2# 组 B: [('B', 3), ('B', 4)], 数量: 2  # 组 C: [('C', 5)], 数量: 1# 注意：后面的('A', 6)在排序后不连续，所以不会与前面的A合并

from datetime import datetimeimport itertools# 模拟日志数据logs = [    {'time': '2024-03-17 08:30', 'level': 'INFO', 'message': '服务启动'},    {'time': '2024-03-17 09:15', 'level': 'ERROR', 'message': '数据库连接失败'},    {'time': '2024-03-17 09:20', 'level': 'WARN', 'message': '高内存使用'},    {'time': '2024-03-18 10:00', 'level': 'INFO', 'message': '定时任务执行'},    {'time': '2024-03-18 10:05', 'level': 'ERROR', 'message': '文件写入失败'},]# 提取日期作为分组键for log in logs:    log['date'] = log['time'].split()[0]# 按日期排序logs.sort(key=lambda x: x['date'])# 按日期分组统计for date, group in itertools.groupby(logs, key=lambda x: x['date']):    group_list = list(group)    error_count = sum(1for log in group_list if log['level'] == 'ERROR')    print(f"{date}: 共{len(group_list)}条日志, 其中{error_count}条错误")# 输出:# 2024-03-17: 共3条日志, 其中1条错误# 2024-03-18: 共2条日志, 其中1条错误

import itertools# 基础切片numbers = range(100)sliced = itertools.islice(numbers, 10, 20)  # 取索引10-19print("切片结果:", list(sliced))# 输出: [10, 11, 12, 13, 14, 15, 16, 17, 18, 19]# 无限序列切片infinite = itertools.count(0)first_5 = list(itertools.islice(infinite, 5))print("无限序列前5个:", first_5)# 输出: [0, 1, 2, 3, 4]# 指定步长with_step = itertools.islice(range(20), 0, 20, 3)print("步长为3:", list(with_step))# 输出: [0, 3, 6, 9, 12, 15, 18]

defread_file_in_batches(file_path, batch_size=1000):"""分批读取大文件，避免内存溢出"""with open(file_path, 'r', encoding='utf-8') as f:whileTrue:# 读取一批行            batch = list(itertools.islice(f, batch_size))ifnot batch:breakyield [line.strip() for line in batch]# 模拟使用# for batch_num, batch in enumerate(read_file_in_batches('large_data.csv', 10000)):#     print(f"处理第{batch_num+1}批，共{len(batch)}行")#     # 处理本批次数据...

import itertools# 两个集合的笛卡尔积colors = ['红', '绿', '蓝']sizes = ['S', 'M', 'L']combinations = list(itertools.product(colors, sizes))print("颜色-尺寸组合:", combinations)# 输出: [('红', 'S'), ('红', 'M'), ('红', 'L'), ('绿', 'S'), ...]# 多个维度的组合dimensions = [    ['A', 'B'],    [1, 2],    ['X', 'Y', 'Z']]multi_comb = list(itertools.product(*dimensions))print(f"三维组合数量: {len(multi_comb)}")# 输出: 12 (2×2×3)

defgenerate_test_cases():"""生成全参数组合的测试用例"""    browsers = ['Chrome', 'Firefox', 'Safari']    operating_systems = ['Windows', 'macOS', 'Linux']    screen_resolutions = ['1920x1080', '1366x768', '1440x900']    languages = ['en-US', 'zh-CN', 'ja-JP']# 生成所有组合    test_cases = itertools.product(        browsers, operating_systems, screen_resolutions, languages    )for i, (browser, os, resolution, lang) in enumerate(test_cases, 1):        print(f"用例{i}: {browser} on {os}, {resolution}, {lang}")# 这里可以实际执行测试...# 使用示例（限制输出前3个）print("测试用例示例:")for case in itertools.islice(generate_test_cases(), 3):    print(case)

import itertools# 基础排列items = ['A', 'B', 'C']perms = list(itertools.permutations(items, 2))print("2个元素的所有排列:", perms)# 输出: [('A', 'B'), ('A', 'C'), ('B', 'A'), ('B', 'C'), ('C', 'A'), ('C', 'B')]# 全排列（默认r=序列长度）full_perms = list(itertools.permutations(items))print(f"全排列数量: {len(full_perms)}")# 输出: 6 (3!)

defgenerate_password_combinations(chars, length):"""生成指定字符集和长度的所有密码组合"""# 注意：排列数量可能非常大，需要谨慎使用return itertools.permutations(chars, length)# 安全使用示例（限制输出）chars = ['a', 'b', 'c', '1', '2']length = 3print(f"从{len(chars)}个字符中选取{length}个的所有排列:")for i, pwd in enumerate(itertools.islice(generate_password_combinations(chars, length), 10)):    print(f"  {i+1}: {''.join(pwd)}")

import itertools# 基础组合items = ['A', 'B', 'C', 'D']combs = list(itertools.combinations(items, 2))print("2个元素的所有组合:", combs)# 输出: [('A', 'B'), ('A', 'C'), ('A', 'D'), ('B', 'C'), ('B', 'D'), ('C', 'D')]# 计算组合数量（不实际生成）from math import combn = 10k = 3print(f"C({n},{k}) = {comb(n, k)}")# 输出: C(10,3) = 120

from itertools import combinationsimport numpy as npdefevaluate_feature_combinations(features, target, model, k=3):"""评估所有k个特征的组合"""    n_features = len(features)    feature_names = list(features.columns)    best_score = -np.inf    best_combo = None# 遍历所有k个特征的组合for combo in combinations(range(n_features), k):# 选择特征子集        X_subset = features.iloc[:, list(combo)]# 训练模型并评估（这里简化为随机分数）# 实际应用中需要使用交叉验证        score = np.random.random()  # 模拟评估分数if score > best_score:            best_score = score            best_combo = [feature_names[i] for i in combo]return best_combo, best_score# 模拟使用# import pandas as pd# features = pd.DataFrame(np.random.randn(100, 10))# target = pd.Series(np.random.randn(100))# best_features, score = evaluate_feature_combinations(features, target, None, k=3)

import itertoolsimport refrom collections import Counterdefprocess_logs(log_lines):"""日志处理流水线"""# 1. 过滤空行    non_empty = filter(lambda x: x.strip(), log_lines)# 2. 提取时间戳和消息    parsed = map(lambda line: parse_log_line(line), non_empty)# 3. 过滤错误日志    errors = filter(lambda x: x['level'] == 'ERROR', parsed)# 4. 按小时分组    errors_by_hour = {}for hour, group in itertools.groupby(errors, key=lambda x: x['time'].hour):        errors_by_hour[hour] = list(group)return errors_by_hourdefparse_log_line(line):"""解析单行日志（简化版）"""# 实际应用中这里会有更复杂的正则匹配return {'time': datetime.strptime(line[:19], '%Y-%m-%d %H:%M:%S'),'level': line[20:25].strip(),'message': line[26:]    }

# 危险：可能内存溢出# all_perms = list(itertools.permutations(range(15)))# 安全：使用islice限制或逐个处理for perm in itertools.islice(itertools.permutations(range(10)), 1000):    process(perm)

import itertools# tee会缓存未消费的数据data = range(1000000)a, b = itertools.tee(data, 2)# 如果两个迭代器消费速度差异大，可能内存激增# 解决方案：需要多次遍历时直接转换为listcached = list(data)

# 错误：未排序直接使用groupbydata = [('A', 1), ('B', 2), ('A', 3)]# 结果会得到3个单元素组# 正确：先排序sorted_data = sorted(data, key=lambda x: x[0])for key, group in itertools.groupby(sorted_data, key=lambda x: x[0]):    print(key, list(group))

import itertoolsimport operator# 累积乘积numbers = [1, 2, 3, 4, 5]cumulative_product = list(itertools.accumulate(numbers, operator.mul))print("累积乘积:", cumulative_product)# 输出: [1, 2, 6, 24, 120]# 按第二个元素排序data = [('apple', 5), ('banana', 2), ('cherry', 8)]sorted_by_second = sorted(data, key=operator.itemgetter(1))print("按数量排序:", sorted_by_second)# 输出: [('banana', 2), ('apple', 5), ('cherry', 8)]

import itertoolsfrom collections import Counterdefcount_ips_large_file(file_path, batch_size=10000):"""统计大文件中IP出现次数"""    ip_counter = Counter()with open(file_path, 'r') as f:whileTrue:# 读取一批行            batch = list(itertools.islice(f, batch_size))ifnot batch:break# 提取IP地址（假设IP在每行开头）            ips = [line.split()[0] for line in batch if line.strip()]            ip_counter.update(ips)return ip_counter# 使用示例（模拟）# top_ips = count_ips_large_file('access.log').most_common(10)

import itertoolsimport mathdeftsp_brute_force(distances):"""TSP暴力解法"""    n = len(distances)    cities = list(range(n))    best_path = None    min_distance = math.inf# 遍历所有排列for path in itertools.permutations(cities):# 计算路径长度        distance = 0for i in range(n):            from_city = path[i]            to_city = path[(i + 1) % n]            distance += distances[from_city][to_city]if distance < min_distance:            min_distance = distance            best_path = pathreturn best_path, min_distance# 使用示例（5个城市）# distances = [[0, 2, 9, 10], [1, 0, 6, 4], [15, 7, 0, 8], [6, 3, 12, 0]]# path, dist = tsp_brute_force(distances)

import itertoolsimport numpy as npdefdata_cleaning_pipeline(data_iter):"""数据清洗流水线"""# 1. 过滤空行    non_empty = filter(lambda x: x isnotNone, data_iter)# 2. 去除前后空格    trimmed = map(str.strip, non_empty)# 3. 解析为数值    parsed = map(parse_number, trimmed)# 4. 过滤异常值（假设范围0-100）    valid_range = filter(lambda x: 0 <= x <= 100, parsed)# 5. 分批处理（每批1000条）    batches = batched(valid_range, 1000)return batchesdefparse_number(s):"""安全解析数值"""try:return float(s)except (ValueError, TypeError):returnNonedefbatched(iterable, n):"""分批处理生成器（Python 3.12+有itertools.batched）"""    it = iter(iterable)while batch := list(itertools.islice(it, n)):yield batch

import itertoolsdata = [1, 2, 3]iter1, iter2 = itertools.tee(data, 2)print("迭代器1:", list(iter1))  # [1, 2, 3]print("迭代器2:", list(iter2))  # [1, 2, 3]

import itertools# 方法1：使用islice限制数量limited = itertools.islice(itertools.count(), 10)# 方法2：使用takewhile条件停止conditioned = itertools.takewhile(lambda x: x < 10, itertools.count())# 方法3：手动终止counter = itertools.count()for i in range(10):    print(next(counter))

# 小数据集：列表推导式更直观squares = [x**2for x in range(100)]# 大数据集或无限序列：itertools更高效from itertools import islice, countbig_squares = (x**2for x in count())first_100 = list(islice(big_squares, 100))