浅析Python中Generator的工作原理与应用

def read_log_file(filename):
    result = []
    with open(filename) as f:
        for line in f:
            if "ERROR" in line:
                result.append(line)
    return result

# 使用方式
errors = read_log_file("huge.log")
for error in errors:
    process_error(error)

def read_log_file(filename):
    with open(filename) as f:
        for line in f:
            if "ERROR" in line:
                yield line

# 使用方式
for error in read_log_file("huge.log"):
    process_error(error)

def counter():
    print("Starting")
    i = 0
    while True:
        print(f"Generating {i}")
        yield i
        i += 1

# 创建生成器对象
c = counter()  # 此时不会打印任何内容
print("Generator created")

# 获取前三个值
print(next(c))  # 打印 "Starting" 和 "Generating 0"，返回 0
print(next(c))  # 打印 "Generating 1"，返回 1
print(next(c))  # 打印 "Generating 2"，返回 2

def stream_completion(prompt):
    # 模拟 API 调用
    response = ["生成", "AI", "回复", "需要", "时间"]
    for token in response:
        yield token

# 使用方式
for chunk in stream_completion("你好"):
    print(chunk, end="", flush=True)

# 列表推导式
squares = [x*x for x in range(1000000)]  # 立即生成所有结果

# Generator 表达式
squares = (x*x for x in range(1000000))  # 按需生成

def averager():
    total = 0
    count = 0
    average = None
    while True:
        # yield 在这里扮演双重角色：
        # 1. 向外返回 average 值
        # 2. 接收外部发送的 value
        value = yield average
        if value is None:
            break
        total += value
        count += 1
        average = total / count

# 让我们一步步看发生了什么
avg = averager()          # 创建生成器对象，但函数体还未开始执行
print("第一步：创建生成器")

next(avg)                 # 启动生成器，运行到第一个 yield，返回 None
print("第二步：生成器已启动，等待第一个值")

print(avg.send(10))      # 1. send(10) 将 10 传给 value
                         # 2. 计算 average = 10/1 = 10.0
                         # 3. 到达 yield，返回 10.0
                         # 4. 生成器暂停，等待下一个值

print(avg.send(20))      # 1. value 获得值 20
                         # 2. 计算 average = 30/2 = 15.0
                         # 3. yield 返回 15.0

print(avg.send(30))      # 1. value 获得值 30
                         # 2. 计算 average = 60/3 = 20.0
                         # 3. yield 返回 20.0

from typing import Generator, Iterator
from typing import TypeVar, Generic

T_co = TypeVar('T_co', covariant=True)      # 生成器产出的类型
T_contra = TypeVar('T_contra', contravariant=True)  # send 方法接收的类型
V_co = TypeVar('V_co', covariant=True)      # return 语句返回的类型

def number_processor() -> Generator[int, str, float]:
    # 这个生成器：
    # - yield 产出 int
    # - 接收 str 类型的输入
    # - 最终 return float 类型的值
    count = 0
    total = 0.0
    while count < 3:
        text = yield count  # 产出 int，接收 str
        if text:
            total += len(text)
        count += 1
    return total  # 返回 float

# 使用示例
proc = number_processor()
next(proc)  # 启动生成器，返回 0
print(proc.send("hello"))  # 输出 1
print(proc.send("world"))  # 输出 2
try:
    proc.send("!")  # 生成器将结束，抛出 StopIteration
except StopIteration as e:
    print(f"最终结果：{e.value}")  # 打印 11.0 (len("hello") + len("world") + len("!"))

from typing import Generator, TypeVar

# 定义一些类来演示
class Animal:
    pass

class Dog(Animal):
    pass

class Chihuahua(Dog):
    pass

# 定义类型变量
T_co = TypeVar('T_co', covariant=True)
T_contra = TypeVar('T_contra', contravariant=True)

def dog_generator() -> Generator[Dog, Animal, None]:
    # 这个生成器:
    # - 产出 Dog (协变位置)
    # - 接收 Animal (逆变位置)
    dog = yield Dog()  # dog 的类型是 Animal
    
# 协变（T_co）示例：
# 如果一个函数返回 Dog，它也可以用在需要返回 Animal 的地方
generator1: Generator[Animal, Animal, None] = dog_generator()  # 没问题！
# 因为 Dog 是 Animal 的子类，所以可以用 Dog 替代 Animal

# 逆变（T_contra）示例：
# 如果一个函数接收 Animal，它也可以接收 Dog 或 Chihuahua
generator2 = dog_generator()
generator2.send(Chihuahua())  # 没问题！
# 因为函数期望接收 Animal，那接收 Animal 的子类当然也可以

from typing import Generator, Iterator
from itertools import chain

def read_chunks(file_path: str) -> Generator[str, None, None]:
    with open(file_path) as f:
        while chunk := f.read(1024):
            yield chunk

def process_chunk(chunk: str) -> Generator[str, None, None]:
    # 处理单个 chunk 中的行
    # 注意：chunk 可能在行中间截断，需要处理这种情况
    lines = chunk.split('\n')
    for line in lines:
        if line.strip():
            yield line.upper()

def filter_keywords(lines: Iterator[str]) -> Generator[str, None, None]:
    keywords = {'ERROR', 'WARNING', 'CRITICAL'}
    for line in lines:
        if any(k in line for k in keywords):
            yield line

# 正确的流式处理版本
def process_log_file(file_path: str):
    # 当前行的未完成部分
    partial_line = ''
    
    for chunk in read_chunks(file_path):
        # 处理可能被截断的行
        if partial_line:
            chunk = partial_line + chunk
            partial_line = ''
            
        # 分割成行，保留最后一个可能不完整的行
        lines = chunk.split('\n')
        if not chunk.endswith('\n'):
            partial_line = lines[-1]
            lines = lines[:-1]
            
        # 处理完整的行
        for line in lines:
            if line.strip():
                # 直接在这里处理，无需存储所有行
                upper_line = line.upper()
                if any(k in upper_line for k in {'ERROR', 'WARNING', 'CRITICAL'}):
                    print(upper_line)
    
    # 处理最后一个不完整的行（如果有的话）
    if partial_line and partial_line.strip():
        upper_line = partial_line.upper()
        if any(k in upper_line for k in {'ERROR', 'WARNING', 'CRITICAL'}):
            print(upper_line)

# 或者，使用更函数式的写法
def process_log_file_functional(file_path: str):
    def handle_chunks() -> Generator[str, None, None]:
        partial_line = ''
        for chunk in read_chunks(file_path):
            if partial_line:
                chunk = partial_line + chunk
                partial_line = ''
            
            lines = chunk.split('\n')
            if not chunk.endswith('\n'):
                partial_line = lines[-1]
                lines = lines[:-1]
            
            yield from (line for line in lines if line.strip())
        
        if partial_line and partial_line.strip():
            yield partial_line

    # 现在我们真正实现了流式处理
    lines = handle_chunks()
    upper_lines = (line.upper() for line in lines)
    filtered_lines = filter_keywords(upper_lines)
    
    for line in filtered_lines:
        print(line)