从基础到高级详解Python迭代器手动访问完全指南

class SimpleIterator:
    """自定义迭代器示例"""
    def __init__(self, max_value):
        self.max = max_value
        self.current = 0
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.current < self.max:
            self.current += 1
            return self.current
        raise StopIteration

# 手动访问
it = SimpleIterator(5)
print(next(it))  # 1
print(next(it))  # 2
print(next(it))  # 3
print(next(it))  # 4
print(next(it))  # 5
try:
    print(next(it))  # 抛出StopIteration
except StopIteration:
    print("迭代结束")

def manual_iteration(iterable):
    """手动迭代通用模式"""
    it = iter(iterable)
    try:
        while True:
            item = next(it)
            # 处理元素
            print(f"处理: {item}")
    except StopIteration:
        print("迭代完成")

# 使用示例
manual_iteration([1, 2, 3, 4, 5])

class StatefulIterator:
    """带状态的手动迭代器"""
    def __init__(self, data):
        self.data = data
        self.index = 0
        self.state = 'active'
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.state == 'paused':
            raise StopIteration("迭代器已暂停")
        if self.index >= len(self.data):
            self.state = 'completed'
            raise StopIteration
        item = self.data[self.index]
        self.index += 1
        return item
    
    def pause(self):
        """暂停迭代"""
        self.state = 'paused'
    
    def resume(self):
        """恢复迭代"""
        if self.state == 'paused':
            self.state = 'active'
    
    def rewind(self, steps=1):
        """回退迭代"""
        self.index = max(0, self.index - steps)

# 使用示例
it = StatefulIterator([10, 20, 30, 40, 50])
print(next(it))  # 10
print(next(it))  # 20
it.rewind()      # 回退1步
print(next(it))  # 20
it.pause()
try:
    print(next(it))  # 抛出异常
except StopIteration as e:
    print(e)
it.resume()
print(next(it))  # 30

def multi_iterator_control(iterators):
    """多迭代器协同控制"""
    # 创建迭代器列表
    its = [iter(it) for it in iterators]
    active = [True] * len(its)
    
    while any(active):
        for i, it in enumerate(its):
            if not active[i]:
                continue
            try:
                item = next(it)
                yield (i, item)
            except StopIteration:
                active[i] = False

# 使用示例
list1 = [1, 2, 3]
list2 = ['a', 'b', 'c', 'd']
list3 = [10.5, 20.5]

for source, value in multi_iterator_control([list1, list2, list3]):
    print(f"来源 {source}: {value}")

def process_large_file(file_path, chunk_size=1024):
    """手动迭代处理大文件"""
    with open(file_path, 'r') as f:
        # 创建迭代器
        it = iter(lambda: f.read(chunk_size), '')
        
        try:
            while True:
                chunk = next(it)
                # 处理数据块
                process_chunk(chunk)
                
                # 条件中断
                if should_stop_processing():
                    print("处理中断")
                    break
        except StopIteration:
            print("文件处理完成")

def process_chunk(chunk):
    """处理数据块（示例）"""
    # 实际处理逻辑
    print(f"处理 {len(chunk)} 字节数据")

def should_stop_processing():
    """检查是否停止处理（示例）"""
    # 实际条件检查
    return False

# 使用示例
process_large_file('large_data.txt')

class StreamProcessor:
    """网络流手动迭代处理器"""
    def __init__(self, stream, buffer_size=4096):
        self.stream = stream
        self.buffer_size = buffer_size
        self.buffer = b''
        self.position = 0
        self.eof = False
    
    def __iter__(self):
        return self
    
    def __next__(self):
        """获取下一个完整数据包"""
        while not self.eof:
            # 检查缓冲区是否有完整数据包
            packet = self._extract_packet()
            if packet:
                return packet
            
            # 读取更多数据
            self._fill_buffer()
        
        # 处理剩余数据
        if self.buffer:
            packet = self.buffer
            self.buffer = b''
            return packet
        
        raise StopIteration
    
    def _fill_buffer(self):
        """填充缓冲区"""
        data = self.stream.read(self.buffer_size)
        if not data:
            self.eof = True
        else:
            self.buffer += data
    
    def _extract_packet(self):
        """从缓冲区提取数据包（示例）"""
        # 查找结束符
        end_pos = self.buffer.find(b'\n', self.position)
        if end_pos == -1:
            return None
        
        # 提取数据包
        packet = self.buffer[self.position:end_pos]
        self.position = end_pos + 1
        return packet

# 使用示例（模拟网络流）
class MockStream:
    def __init__(self, data):
        self.data = data
        self.position = 0
    
    def read(self, size):
        if self.position >= len(self.data):
            return b''
        chunk = self.data[self.position:self.position+size]
        self.position += size
        return chunk

# 模拟数据流
data = b'packet1\npacket2\npartial'
stream = MockStream(data)
processor = StreamProcessor(stream)

for packet in processor:
    print(f"收到数据包: {packet.decode()}")

class TreeNode:
    """树节点"""
    def __init__(self, value):
        self.value = value
        self.children = []
    
    def add_child(self, node):
        self.children.append(node)

class TreeIterator:
    """树结构手动迭代器（深度优先）"""
    def __init__(self, root):
        self.stack = [root]
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if not self.stack:
            raise StopIteration
        
        # 弹出栈顶节点
        node = self.stack.pop()
        
        # 子节点逆序入栈（保证顺序）
        for child in reversed(node.children):
            self.stack.append(child)
        
        return node.value

# 使用示例
root = TreeNode('A')
b = TreeNode('B')
c = TreeNode('C')
d = TreeNode('D')
root.add_child(b)
root.add_child(c)
b.add_child(d)

it = TreeIterator(root)
print("深度优先遍历:")
for value in it:
    print(value)

class GraphIterator:
    """图结构手动迭代器（广度优先）"""
    def __init__(self, graph, start):
        self.graph = graph
        self.queue = collections.deque([start])
        self.visited = set([start])
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if not self.queue:
            raise StopIteration
        
        node = self.queue.popleft()
        
        # 添加未访问邻居
        for neighbor in self.graph[node]:
            if neighbor not in self.visited:
                self.visited.add(neighbor)
                self.queue.append(neighbor)
        
        return node

# 使用示例
graph = {
    'A': ['B', 'C'],
    'B': ['A', 'D', 'E'],
    'C': ['A', 'F'],
    'D': ['B'],
    'E': ['B', 'F'],
    'F': ['C', 'E']
}

print("广度优先遍历:")
it = GraphIterator(graph, 'A')
for node in it:
    print(node)

def coroutine_example():
    """协程手动控制示例"""
    print("协程启动")
    try:
        while True:
            value = yield
            print(f"接收值: {value}")
    except GeneratorExit:
        print("协程退出")

# 手动控制
coro = coroutine_example()
next(coro)  # 启动协程
coro.send(10)  # 发送值
coro.send(20)
coro.close()  # 关闭协程

import asyncio

class AsyncIterator:
    """异步迭代器"""
    def __init__(self, n):
        self.n = n
        self.current = 0
    
    def __aiter__(self):
        return self
    
    async def __anext__(self):
        if self.current >= self.n:
            raise StopAsyncIteration
        await asyncio.sleep(0.1)  # 模拟IO
        self.current += 1
        return self.current

async def manual_async_iteration():
    """手动控制异步迭代"""
    aiter = AsyncIterator(5)
    try:
        while True:
            value = await aiter.__anext__()
            print(f"异步值: {value}")
    except StopAsyncIteration:
        print("异步迭代结束")

# 运行
asyncio.run(manual_async_iteration())

def chain_iterators(*iterables):
    """手动链式迭代器"""
    for it in iterables:
        yield from it

# 使用示例
it1 = iter([1, 2, 3])
it2 = iter(['a', 'b'])
chained = chain_iterators(it1, it2)
print(list(chained))  # [1, 2, 3, 'a', 'b']

# 手动控制
chained = chain_iterators(it1, it2)
print(next(chained))  # 1
print(next(chained))  # 2
print(next(chained))  # 3
print(next(chained))  # 'a'

def large_data_iterator(data_size=1000000):
    """内存高效迭代器"""
    for i in range(data_size):
        # 生成数据（避免一次性加载）
        data = generate_data(i)
        yield data

def generate_data(index):
    """生成数据（示例）"""
    return f"数据项-{index}"

# 手动处理
it = large_data_iterator()
count = 0
try:
    while True:
        item = next(it)
        process_item(item)
        count += 1
        if count % 100000 == 0:
            print(f"已处理 {count} 项")
except StopIteration:
    print(f"总共处理 {count} 项")

def process_item(item):
    """处理数据项（示例）"""
    # 实际处理逻辑
    pass

class DataPipeline:
    """手动迭代数据管道"""
    def __init__(self):
        self.processors = []
    
    def add_processor(self, processor):
        """添加处理器"""
        self.processors.append(processor)
    
    def process(self, data_iter):
        """处理数据流"""
        it = iter(data_iter)
        for processor in self.processors:
            it = processor(it)
        return it

# 处理器示例
def filter_processor(predicate):
    """过滤处理器"""
    def process(input_iter):
        for item in input_iter:
            if predicate(item):
                yield item
    return process

def map_processor(mapper):
    """映射处理器"""
    def process(input_iter):
        for item in input_iter:
            yield mapper(item)
    return process

# 使用示例
pipeline = DataPipeline()
pipeline.add_processor(filter_processor(lambda x: x % 2 == 0))
pipeline.add_processor(map_processor(lambda x: x * 2))

data = [1, 2, 3, 4, 5, 6]
result_iter = pipeline.process(data)

# 手动控制
print(next(result_iter))  # 4 (2 * 2)
print(next(result_iter))  # 8 (4 * 2)
print(next(result_iter))  # 12 (6 * 2)

class RealTimeMonitor:
    """实时监控系统手动迭代"""
    def __init__(self, data_source):
        self.data_source = data_source
        self.iterator = None
        self.running = False
    
    def start(self):
        """启动监控"""
        self.iterator = iter(self.data_source)
        self.running = True
    
    def stop(self):
        """停止监控"""
        self.running = False
    
    def process_next(self):
        """处理下一个数据点"""
        if not self.running or self.iterator is None:
            return None
        
        try:
            data = next(self.iterator)
            self._analyze(data)
            return data
        except StopIteration:
            self.stop()
            return None
    
    def _analyze(self, data):
        """数据分析（示例）"""
        print(f"分析数据: {data}")
        # 实际分析逻辑

# 使用示例
class DataSource:
    """模拟数据源"""
    def __init__(self, max_count=5):
        self.count = 0
        self.max = max_count
    
    def __iter__(self):
        return self
    
    def __next__(self):
        if self.count >= self.max:
            raise StopIteration
        self.count += 1
        return f"数据-{self.count}"

monitor = RealTimeMonitor(DataSource())
monitor.start()

# 手动控制处理
while True:
    data = monitor.process_next()
    if data is None:
        break
    print(f"处理数据: {data}")
    # 可以在此添加控制逻辑
    if data == "数据-3":
        print("暂停处理")
        break

# 继续处理
print("继续处理")
monitor.start()  # 重新启动
while True:
    data = monitor.process_next()
    if data is None:
        break
    print(f"处理数据: {data}")

# 使用上下文管理器确保资源释放
class SafeIterator:
    def __init__(self, resource):
        self.resource = resource
        self.it = iter(resource)

    def __enter__(self):
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        self.resource.close()

    def __next__(self):
        return next(self.it)

with SafeIterator(open('file.txt')) as it:
    print(next(it))

def robust_next(iterator, default=None):
    """健壮的next函数"""
    try:
        return next(iterator)
    except StopIteration:
        return default
    except Exception as e:
        log_error(e)
        return default

# 使用
it = iter([1, 2])
print(robust_next(it))  # 1
print(robust_next(it))  # 2
print(robust_next(it))  # None

# 避免不必要的属性查找
def optimized_iteration(data):
    it = iter(data)
    next_item = it.__next__  # 缓存方法
    try:
        while True:
            item = next_item()
            process(item)
    except StopIteration:
        pass

# 使用生成器表达式
large_iter = (x * 2 for x in range(1000000))
# 手动处理
item = next(large_iter)

class CustomIterator:
    """
    自定义迭代器文档

    功能:
    - 支持手动next调用
    - 支持状态查询
    - 支持回退操作

    示例:
        it = CustomIterator(data)
        item = next(it)
    """
    # 实现代码

import unittest

class TestManualIteration(unittest.TestCase):
    def test_basic_next(self):
        it = iter([1, 2, 3])
        self.assertEqual(next(it), 1)
        self.assertEqual(next(it), 2)
        self.assertEqual(next(it), 3)
        with self.assertRaises(StopIteration):
            next(it)

    def test_custom_iterator(self):
        it = StatefulIterator([10, 20, 30])
        self.assertEqual(next(it), 10)
        it.rewind()
        self.assertEqual(next(it), 10)
        it.pause()
        with self.assertRaises(StopIteration):
            next(it)