从基础到高级详解Python字符串I/O操作完全指南

import io

def basic_stringio_operations():
    """基础StringIO操作示例"""
    # 创建StringIO对象
    string_buffer = io.StringIO()
    
    # 写入数据
    string_buffer.write("Hello, World!\n")
    string_buffer.write("这是第二行文本\n")
    string_buffer.write("Python字符串I/O操作\n")
    
    # 获取当前位置
    print(f"当前位置: {string_buffer.tell()}")
    
    # 回到起始位置
    string_buffer.seek(0)
    
    # 读取数据
    content = string_buffer.read()
    print("全部内容:")
    print(content)
    
    # 再次定位并读取部分内容
    string_buffer.seek(7)  # 移动到"World"前面
    partial = string_buffer.read(5)  # 读取5个字符
    print(f"部分内容: '{partial}'")
    
    # 按行读取
    string_buffer.seek(0)
    lines = string_buffer.readlines()
    print("行列表:")
    for i, line in enumerate(lines, 1):
        print(f"行 {i}: {line.strip()}")
    
    # 检查缓冲区状态
    print(f"缓冲区大小: {string_buffer.tell()} 字符")
    print(f"是否可读: {string_buffer.readable()}")
    print(f"是否可写: {string_buffer.writable()}")
    
    # 清空缓冲区
    string_buffer.truncate(0)
    string_buffer.seek(0)
    string_buffer.write("新的开始")
    
    # 获取最终内容
    final_content = string_buffer.getvalue()
    print(f"最终内容: '{final_content}'")
    
    # 关闭缓冲区
    string_buffer.close()

# 执行示例
basic_stringio_operations()

def basic_bytesio_operations():
    """基础BytesIO操作示例"""
    # 创建BytesIO对象
    bytes_buffer = io.BytesIO()
    
    # 写入二进制数据
    bytes_buffer.write(b"Binary data\n")
    bytes_buffer.write("中文文本".encode('utf-8'))
    bytes_buffer.write(b"\x00\x01\x02\x03\x04\x05")  # 原始字节
    
    # 获取当前位置
    print(f"当前位置: {bytes_buffer.tell()} 字节")
    
    # 回到起始位置
    bytes_buffer.seek(0)
    
    # 读取数据
    binary_content = bytes_buffer.read()
    print("二进制内容:")
    print(f"长度: {len(binary_content)} 字节")
    print(f"十六进制: {binary_content.hex()}")
    
    # 尝试解码文本部分
    try:
        text_part = binary_content.split(b'\n')[0]
        decoded_text = text_part.decode('utf-8')
        print(f"解码文本: '{decoded_text}'")
    except UnicodeDecodeError:
        print("包含非文本数据")
    
    # 写入混合数据
    bytes_buffer.seek(0)
    bytes_buffer.truncate(0)  # 清空
    
    # 写入不同类型数据
    data_parts = [
        b"HEADER",
        struct.pack('>I', 12345),  # 打包整数
        struct.pack('>d', 3.14159),  # 打包浮点数
        "结束标记".encode('utf-8')
    ]
    
    for data in data_parts:
        bytes_buffer.write(data)
    
    # 解析结构化数据
    bytes_buffer.seek(0)
    header = bytes_buffer.read(6)
    int_data = struct.unpack('>I', bytes_buffer.read(4))[0]
    float_data = struct.unpack('>d', bytes_buffer.read(8))[0]
    footer = bytes_buffer.read().decode('utf-8')
    
    print(f"头部: {header.decode('utf-8')}")
    print(f"整数: {int_data}")
    print(f"浮点数: {float_data}")
    print(f"尾部: {footer}")
    
    # 关闭缓冲区
    bytes_buffer.close()

# 执行示例
basic_bytesio_operations()

def context_manager_usage():
    """上下文管理器使用示例"""
    # 使用with语句自动管理资源
    with io.StringIO() as buffer:
        buffer.write("使用上下文管理器\n")
        buffer.write("自动处理资源清理\n")
        
        content = buffer.getvalue()
        print("上下文管理器内容:")
        print(content)
    
    # 缓冲区已自动关闭
    print("缓冲区已自动关闭")
    
    # 异常处理示例
    try:
        with io.BytesIO() as byte_buffer:
            byte_buffer.write(b"测试数据")
            raise ValueError("模拟异常")
            # 不会执行到这里
    except ValueError as e:
        print(f"捕获异常: {e}")
        print("缓冲区仍然被正确关闭")
    
    # 自定义上下文管理器
    class SmartStringIO:
        """智能StringIO上下文管理器"""
        def __init__(self, initial_value=""):
            self.buffer = io.StringIO(initial_value)
            self.operation_count = 0
        
        def __enter__(self):
            return self.buffer
        
        def __exit__(self, exc_type, exc_val, exc_tb):
            self.operation_count += 1
            content = self.buffer.getvalue()
            print(f"退出上下文 (操作次数: {self.operation_count})")
            print(f"最终内容长度: {len(content)} 字符")
            self.buffer.close()
            
            if exc_type:
                print(f"发生异常: {exc_val}")
                return False  # 不抑制异常
    
    # 使用自定义上下文管理器
    with SmartStringIO("初始内容\n") as buffer:
        buffer.write("追加内容\n")
        buffer.write("更多内容\n")
        print("在上下文中操作缓冲区")
    
    print("自定义上下文管理器演示完成")

# 执行示例
context_manager_usage()

def streaming_processing():
    """流式处理与迭代器示例"""
    # 生成大量数据
    def generate_large_data(num_lines=1000):
        """生成大量数据"""
        for i in range(num_lines):
            yield f"这是第 {i+1} 行数据，包含一些文本内容用于测试字符串I/O性能\n"
    
    # 使用StringIO进行流式处理
    with io.StringIO() as buffer:
        # 分批写入
        batch_size = 100
        data_generator = generate_large_data(1000)
        
        for i, line in enumerate(data_generator):
            buffer.write(line)
            
            # 每100行处理一次
            if (i + 1) % batch_size == 0:
                current_content = buffer.getvalue()
                processed = current_content.upper()  # 模拟处理
                buffer.seek(0)
                buffer.truncate(0)
                buffer.write(processed)
                print(f"已处理 {i+1} 行")
        
        # 处理剩余数据
        final_content = buffer.getvalue()
        print(f"最终内容长度: {len(final_content)} 字符")
        print(f"行数: {final_content.count('\n')}")
    
    # 使用迭代器接口
    with io.StringIO("第一行\n第二行\n第三行\n") as buffer:
        print("迭代器读取:")
        for line in buffer:
            print(f"读取: {line.strip()}")
        
        # 重置并使用readline
        buffer.seek(0)
        print("使用readline:")
        while True:
            line = buffer.readline()
            if not line:
                break
            print(f"行: {line.strip()}")
    
    # 性能对比：直接拼接 vs StringIO
    import time
    
    def direct_concatenation(data):
        """直接字符串拼接"""
        result = ""
        for item in data:
            result += item
        return result
    
    def stringio_concatenation(data):
        """使用StringIO拼接"""
        with io.StringIO() as buffer:
            for item in data:
                buffer.write(item)
            return buffer.getvalue()
    
    # 生成测试数据
    test_data = [f"数据块 {i} " * 10 + "\n" for i in range(10000)]
    
    # 测试性能
    start_time = time.time()
    result1 = direct_concatenation(test_data)
    direct_time = time.time() - start_time
    
    start_time = time.time()
    result2 = stringio_concatenation(test_data)
    stringio_time = time.time() - start_time
    
    print(f"直接拼接时间: {direct_time:.4f}秒")
    print(f"StringIO拼接时间: {stringio_time:.4f}秒")
    print(f"性能提升: {(direct_time/stringio_time):.2f}倍")
    print(f"结果相等: {result1 == result2}")

# 执行示例
streaming_processing()

def csv_in_memory_processing():
    """CSV数据内存处理示例"""
    import csv
    
    # 创建CSV数据
    csv_data = [
        ['姓名', '年龄', '城市', '职业'],
        ['张三', '25', '北京', '工程师'],
        ['李四', '30', '上海', '设计师'],
        ['王五', '28', '广州', '产品经理'],
        ['赵六', '35', '深圳', '架构师']
    ]
    
    # 使用StringIO处理CSV
    with io.StringIO() as csv_buffer:
        # 写入CSV
        writer = csv.writer(csv_buffer)
        writer.writerows(csv_data)
        
        # 获取CSV内容
        csv_content = csv_buffer.getvalue()
        print("生成的CSV内容:")
        print(csv_content)
        
        # 重置并读取
        csv_buffer.seek(0)
        reader = csv.reader(csv_buffer)
        print("\n读取CSV数据:")
        for row in reader:
            print(f"行: {row}")
    
    # 更复杂的CSV处理
    def process_csv_in_memory(data, processing_func):
        """在内存中处理CSV数据"""
        with io.StringIO() as buffer:
            # 写入原始数据
            writer = csv.writer(buffer)
            writer.writerows(data)
            
            # 处理数据
            buffer.seek(0)
            processed_lines = []
            reader = csv.reader(buffer)
            
            header = next(reader)  # 读取表头
            processed_lines.append(processing_func(header, is_header=True))
            
            for row in reader:
                processed_lines.append(processing_func(row, is_header=False))
            
            # 写入处理后的数据
            buffer.seek(0)
            buffer.truncate(0)
            writer = csv.writer(buffer)
            writer.writerows(processed_lines)
            
            return buffer.getvalue()
    
    # 示例处理函数：年龄加1，城市大写
    def age_increment(row, is_header=False):
        if is_header:
            return row
        else:
            modified = row.copy()
            if len(modified) >= 2:  # 确保有年龄字段
                try:
                    modified[1] = str(int(modified[1]) + 1)
                except ValueError:
                    pass
            if len(modified) >= 3:  # 确保有城市字段
                modified[2] = modified[2].upper()
            return modified
    
    # 处理数据
    processed_csv = process_csv_in_memory(csv_data, age_increment)
    print("\n处理后的CSV:")
    print(processed_csv)

# 执行示例
csv_in_memory_processing()

def json_in_memory_processing():
    """JSON数据内存处理示例"""
    import json
    
    # 示例JSON数据
    sample_data = {
        "users": [
            {"id": 1, "name": "张三", "email": "zhangsan@example.com", "active": True},
            {"id": 2, "name": "李四", "email": "lisi@example.com", "active": False},
            {"id": 3, "name": "王五", "email": "wangwu@example.com", "active": True}
        ],
        "metadata": {
            "version": "1.0",
            "timestamp": "2024-01-15T10:30:00Z",
            "count": 3
        }
    }
    
    # 使用StringIO处理JSON
    with io.StringIO() as json_buffer:
        # 写入JSON
        json.dump(sample_data, json_buffer, indent=2, ensure_ascii=False)
        
        # 获取JSON字符串
        json_string = json_buffer.getvalue()
        print("格式化的JSON:")
        print(json_string)
        
        # 从字符串加载
        json_buffer.seek(0)
        loaded_data = json.load(json_buffer)
        print("\n从StringIO加载的数据:")
        print(f"用户数量: {len(loaded_data['users'])}")
        print(f"元数据版本: {loaded_data['metadata']['version']}")
    
    # JSON流式处理
    def stream_json_processing(data, chunk_size=1024):
        """流式处理大型JSON数据"""
        with io.StringIO() as buffer:
            # 使用生成器逐步写入
            buffer.write('{"users": [')
            
            first = True
            for user in data['users']:
                if not first:
                    buffer.write(',')
                else:
                    first = False
                
                user_json = json.dumps(user, ensure_ascii=False)
                buffer.write(user_json)
                
                # 模拟流式处理：定期处理数据
                if buffer.tell() >= chunk_size:
                    chunk = buffer.getvalue()
                    yield chunk
                    buffer.seek(0)
                    buffer.truncate(0)
            
            buffer.write('], "metadata": ')
            buffer.write(json.dumps(data['metadata'], ensure_ascii=False))
            buffer.write('}')
            
            # 最后一部分
            final_chunk = buffer.getvalue()
            yield final_chunk
    
    # 测试流式处理
    print("\n流式JSON处理:")
    total_size = 0
    for chunk in stream_json_processing(sample_data, chunk_size=200):
        total_size += len(chunk)
        print(f"块大小: {len(chunk)} 字符")
        print(f"内容预览: {chunk[:50]}...")
    
    print(f"总大小: {total_size} 字符")

# 执行示例
json_in_memory_processing()

def dynamic_html_generation():
    """动态HTML生成示例"""
    from string import Template
    
    # HTML模板
    html_template = Template("""
    <!DOCTYPE html>
    <html>
    <head>
        <title>$title</title>
        <meta charset="utf-8">
        <style>
            body { font-family: Arial, sans-serif; margin: 40px; }
            .user { border: 1px solid #ddd; padding: 15px; margin: 10px 0; }
            .active { background-color: #e8f5e9; }
            .inactive { background-color: #ffebee; }
        </style>
    </head>
    <body>
        <h1>$heading</h1>
        <p>生成时间: $timestamp</p>
        <div id="users">
            $user_content
        </div>
    </body>
    </html>
    """)
    
    # 用户数据
    users = [
        {"name": "张三", "email": "zhangsan@example.com", "active": True},
        {"name": "李四", "email": "lisi@example.com", "active": False},
        {"name": "王五", "email": "wangwu@example.com", "active": True}
    ]
    
    # 使用StringIO构建动态内容
    with io.StringIO() as user_buffer:
        for user in users:
            css_class = "active" if user['active'] else "inactive"
            status_text = "活跃" if user['active'] else "非活跃"
            
            user_html = f"""
            <div class="user {css_class}">
                <h3>{user['name']}</h3>
                <p>邮箱: {user['email']}</p>
                <p>状态: {status_text}</p>
            </div>
            """
            user_buffer.write(user_html)
        
        user_content = user_buffer.getvalue()
    
    # 填充主模板
    from datetime import datetime
    html_content = html_template.substitute(
        title="用户列表",
        heading="系统用户",
        timestamp=datetime.now().strftime('%Y-%m-%d %H:%M:%S'),
        user_content=user_content
    )
    
    print("生成的HTML:")
    print(html_content[:200] + "..." if len(html_content) > 200 else html_content)
    
    # 保存到文件（可选）
    with open('users.html', 'w', encoding='utf-8') as f:
        f.write(html_content)
    print("HTML文件已保存")
    
    # 更复杂的模板系统
    class TemplateEngine:
        """简单的模板引擎"""
        def __init__(self):
            self.templates = {}
            self.partials = {}
        
        def register_template(self, name, content):
            """注册模板"""
            self.templates[name] = content
        
        def register_partial(self, name, content):
            """注册局部模板"""
            self.partials[name] = content
        
        def render(self, template_name, **context):
            """渲染模板"""
            if template_name not in self.templates:
                raise ValueError(f"模板未找到: {template_name}")
            
            content = self.templates[template_name]
            
            # 处理局部模板
            for partial_name, partial_content in self.partials.items():
                placeholder = f"{{{{ partial:{partial_name} }}}}"
                content = content.replace(placeholder, partial_content)
            
            # 处理变量
            template = Template(content)
            return template.substitute(**context)
    
    # 使用模板引擎
    engine = TemplateEngine()
    engine.register_template('page', """
    <html>
    <head><title>$title</title></head>
    <body>
        <h1>$heading</h1>
        {{ partial:header }}
        <main>$content</main>
        {{ partial:footer }}
    </body>
    </html>
    """)
    
    engine.register_partial('header', """
    <header>
        <nav>导航菜单</nav>
    </header>
    """)
    
    engine.register_partial('footer', """
    <footer>
        <p>版权所有 © 2024</p>
    </footer>
    """)
    
    rendered = engine.render('page', 
                           title="模板引擎测试",
                           heading="欢迎使用",
                           content="这是主要内容区域")
    
    print("\n模板引擎输出:")
    print(rendered)

# 执行示例
dynamic_html_generation()

def report_generation_system():
    """报告生成系统示例"""
    import csv
    from datetime import datetime, timedelta
    
    # 生成示例数据
    def generate_sales_data(days=30):
        """生成销售数据"""
        base_date = datetime.now() - timedelta(days=days)
        data = []
        
        for i in range(days):
            date = base_date + timedelta(days=i)
            sales = round(1000 + i * 50 * (0.8 + 0.4 * (i % 7) / 7), 2)
            customers = int(20 + i * 2 * (0.9 + 0.2 * (i % 5) / 5))
            
            data.append({
                'date': date.strftime('%Y-%m-%d'),
                'sales': sales,
                'customers': customers,
                'avg_sale': round(sales / customers, 2) if customers > 0 else 0
            })
        
        return data
    
    sales_data = generate_sales_data(7)
    print("销售数据示例:")
    for item in sales_data:
        print(item)
    
    # 文本报告生成
    def generate_text_report(data):
        """生成文本格式报告"""
        with io.StringIO() as report:
            report.write("销售日报\n")
            report.write("=" * 40 + "\n")
            report.write(f"生成时间: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
            
            report.write("日期        销售额    客户数  客单价\n")
            report.write("-" * 40 + "\n")
            
            total_sales = 0
            total_customers = 0
            
            for item in data:
                report.write(f"{item['date']}  {item['sales']:>8.2f}  {item['customers']:>6}  {item['avg_sale']:>7.2f}\n")
                total_sales += item['sales']
                total_customers += item['customers']
            
            report.write("-" * 40 + "\n")
            report.write(f"总计      {total_sales:>8.2f}  {total_customers:>6}  {total_sales/total_customers:>7.2f}\n")
            
            return report.getvalue()
    
    text_report = generate_text_report(sales_data)
    print("\n文本报告:")
    print(text_report)
    
    # CSV报告生成
    def generate_csv_report(data):
        """生成CSV格式报告"""
        with io.StringIO() as csv_buffer:
            writer = csv.DictWriter(csv_buffer, 
                                  fieldnames=['date', 'sales', 'customers', 'avg_sale'],
                                  extrasaction='ignore')
            
            writer.writeheader()
            writer.writerows(data)
            
            return csv_buffer.getvalue()
    
    csv_report = generate_csv_report(sales_data)
    print("CSV报告:")
    print(csv_report)
    
    # HTML报告生成
    def generate_html_report(data):
        """生成HTML格式报告"""
        with io.StringIO() as html:
            html.write("""
            <!DOCTYPE html>
            <html>
            <head>
                <title>销售报告</title>
                <style>
                    body { font-family: Arial, sans-serif; margin: 20px; }
                    table { border-collapse: collapse; width: 100%; }
                    th, td { border: 1px solid #ddd; padding: 8px; text-align: right; }
                    th { background-color: #f2f2f2; text-align: center; }
                    tr:nth-child(even) { background-color: #f9f9f9; }
                    .total { font-weight: bold; background-color: #e8f5e9; }
                </style>
            </head>
            <body>
                <h1>销售报告</h1>
                <p>生成时间: """ + datetime.now().strftime('%Y-%m-%d %H:%M:%S') + """</p>
                <table>
                    <tr>
                        <th>日期</th>
                        <th>销售额</th>
                        <th>客户数</th>
                        <th>客单价</th>
                    </tr>
            """)
            
            total_sales = 0
            total_customers = 0
            
            for item in data:
                html.write(f"""
                    <tr>
                        <td>{item['date']}</td>
                        <td>{item['sales']:.2f}</td>
                        <td>{item['customers']}</td>
                        <td>{item['avg_sale']:.2f}</td>
                    </tr>
                """)
                total_sales += item['sales']
                total_customers += item['customers']
            
            html.write(f"""
                    <tr class="total">
                        <td>总计</td>
                        <td>{total_sales:.2f}</td>
                        <td>{total_customers}</td>
                        <td>{(total_sales/total_customers):.2f}</td>
                    </tr>
                </table>
            </body>
            </html>
            """)
            
            return html.getvalue()
    
    html_report = generate_html_report(sales_data)
    print("HTML报告预览:")
    print(html_report[:200] + "...")
    
    # 多格式报告生成器
    class ReportGenerator:
        """多格式报告生成器"""
        def __init__(self):
            self.formatters = {
                'text': self._format_text,
                'csv': self._format_csv,
                'html': self._format_html,
                'json': self._format_json
            }
        
        def generate_report(self, data, format_type='text'):
            """生成指定格式的报告"""
            if format_type not in self.formatters:
                raise ValueError(f"不支持的格式: {format_type}")
            
            return self.formatters[format_type](data)
        
        def _format_text(self, data):
            """文本格式"""
            with io.StringIO() as buffer:
                # ... 文本格式化逻辑
                return buffer.getvalue()
        
        def _format_csv(self, data):
            """CSV格式"""
            with io.StringIO() as buffer:
                # ... CSV格式化逻辑
                return buffer.getvalue()
        
        def _format_html(self, data):
            """HTML格式"""
            with io.StringIO() as buffer:
                # ... HTML格式化逻辑
                return buffer.getvalue()
        
        def _format_json(self, data):
            """JSON格式"""
            with io.StringIO() as buffer:
                json.dump({
                    'metadata': {
                        'generated_at': datetime.now().isoformat(),
                        'record_count': len(data)
                    },
                    'data': data
                }, buffer, indent=2, ensure_ascii=False)
                return buffer.getvalue()
    
    # 使用报告生成器
    generator = ReportGenerator()
    
    formats = ['text', 'csv', 'html', 'json']
    for fmt in formats:
        report = generator.generate_report(sales_data, fmt)
        filename = f'sales_report.{fmt}'
        
        with open(filename, 'w', encoding='utf-8') as f:
            f.write(report)
        
        print(f"生成 {fmt} 报告: {filename}")
        if fmt == 'text':
            print("文本报告预览:")
            print(report[:100] + "...")

# 执行示例
report_generation_system()

def testing_and_mocking():
    """测试与模拟框架应用"""
    import unittest
    from unittest.mock import patch, MagicMock
    
    # 被测函数
    def process_data(data_source):
        """处理数据的函数"""
        content = data_source.read()
        return content.upper().strip()
    
    # 使用StringIO进行单元测试
    class TestDataProcessing(unittest.TestCase):
        """数据处理测试用例"""
        
        def test_with_stringio(self):
            """使用StringIO测试"""
            test_data = "hello world\n测试数据"
            
            # 创建StringIO作为数据源
            with io.StringIO(test_data) as data_source:
                result = process_data(data_source)
                expected = "HELLO WORLD\n测试数据".upper().strip()
                self.assertEqual(result, expected)
        
        def test_empty_data(self):
            """测试空数据"""
            with io.StringIO("") as data_source:
                result = process_data(data_source)
                self.assertEqual(result, "")
        
        def test_mock_file(self):
            """使用Mock模拟文件"""
            mock_file = MagicMock()
            mock_file.read.return_value = "mock data"
            
            result = process_data(mock_file)
            self.assertEqual(result, "MOCK DATA")
            mock_file.read.assert_called_once()
    
    # 运行测试
    print("运行测试用例...")
    loader = unittest.TestLoader()
    suite = loader.loadTestsFromTestCase(TestDataProcessing)
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(suite)
    
    # 模拟标准输出
    def function_that_prints():
        """一个会打印输出的函数"""
        print("正常输出")
        print("错误输出", file=sys.stderr)
        return "结果"
    
    def test_output_capture():
        """测试输出捕获"""
        with patch('sys.stdout', new_callable=io.StringIO) as mock_stdout:
            with patch('sys.stderr', new_callable=io.StringIO) as mock_stderr:
                result = function_that_prints()
                
                stdout_content = mock_stdout.getvalue()
                stderr_content = mock_stderr.getvalue()
                
                print(f"函数结果: {result}")
                print(f"标准输出: {stdout_content!r}")
                print(f"标准错误: {stderr_content!r}")
                
                assert "正常输出" in stdout_content
                assert "错误输出" in stderr_content
    
    test_output_capture()
    
    # 更复杂的模拟场景
    class DatabaseSimulator:
        """数据库模拟器"""
        def __init__(self):
            self.data = io.StringIO()
            self._setup_sample_data()
        
        def _setup_sample_data(self):
            """设置示例数据"""
            sample_data = [
                "1,Alice,alice@example.com,active",
                "2,Bob,bob@example.com,inactive",
                "3,Charlie,charlie@example.com,active"
            ]
            for line in sample_data:
                self.data.write(line + '\n')
            self.data.seek(0)
        
        def query(self, sql):
            """模拟查询"""
            results = []
            for line in self.data:
                if line.strip():  # 非空行
                    fields = line.strip().split(',')
                    results.append({
                        'id': fields[0],
                        'name': fields[1],
                        'email': fields[2],
                        'status': fields[3]
                    })
            return results
        
        def add_record(self, record):
            """添加记录"""
            line = f"{record['id']},{record['name']},{record['email']},{record['status']}\n"
            self.data.write(line)
    
    # 使用数据库模拟器
    db = DatabaseSimulator()
    print("\n数据库查询结果:")
    users = db.query("SELECT * FROM users")
    for user in users:
        print(user)
    
    # 添加新记录
    db.add_record({
        'id': '4',
        'name': 'Diana',
        'email': 'diana@example.com',
        'status': 'active'
    })
    
    print("\n添加记录后的查询:")
    users = db.query("SELECT * FROM users")
    for user in users:
        print(user)

# 执行示例
testing_and_mocking()

def network_protocol_simulation():
    """网络协议模拟示例"""
    import socket
    import threading
    import time
    
    # 简单的HTTP服务器模拟
    class HttpServerSimulator:
        """HTTP服务器模拟器"""
        def __init__(self):
            self.request_buffer = io.BytesIO()
            self.response_buffer = io.BytesIO()
            self.request_count = 0
        
        def handle_request(self, request_data):
            """处理HTTP请求"""
            self.request_count += 1
            self.request_buffer.write(request_data)
            
            # 解析请求
            request_text = request_data.decode('utf-8', errors='ignore')
            lines = request_text.split('\r\n')
            
            if lines and lines[0]:
                method, path, protocol = lines[0].split(' ', 2)
                
                # 生成响应
                response_body = f"""
                <html>
                <head><title>模拟服务器</title></head>
                <body>
                    <h1>Hello from Simulator</h1>
                    <p>请求方法: {method}</p>
                    <p>请求路径: {path}</p>
                    <p>协议版本: {protocol}</p>
                    <p>请求计数: {self.request_count}</p>
                </body>
                </html>
                """
                
                response = f"""HTTP/1.1 200 OK
Content-Type: text/html; charset=utf-8
Content-Length: {len(response_body.encode('utf-8'))}
Connection: close

{response_body}"""
                
                self.response_buffer.write(response.encode('utf-8'))
                return self.response_buffer.getvalue()
            
            return b"HTTP/1.1 400 Bad Request\r\n\r\n"
    
    # 测试HTTP模拟器
    simulator = HttpServerSimulator()
    
    # 模拟HTTP请求
    http_requests = [
        b"GET / HTTP/1.1\r\nHost: localhost\r\n\r\n",
        b"GET /api/users HTTP/1.1\r\nHost: localhost\r\n\r\n",
        b"POST /api/data HTTP/1.1\r\nHost: localhost\r\nContent-Length: 5\r\n\r\nhello"
    ]
    
    for i, request in enumerate(http_requests):
        response = simulator.handle_request(request)
        print(f"请求 {i+1} 响应:")
        print(response.decode('utf-8')[:200] + "...")
        print("-" * 50)
    
    # TCP协议模拟
    class TcpProtocolHandler:
        """TCP协议处理器"""
        def __init__(self):
            self.receive_buffer = io.BytesIO()
            self.send_buffer = io.BytesIO()
            self.sequence_number = 0
        
        def process_packet(self, packet_data):
            """处理数据包"""
            self.receive_buffer.write(packet_data)
            
            # 模拟协议处理
            response = f"ACK {self.sequence_number} Received {len(packet_data)} bytes"
            self.sequence_number += 1
            
            self.send_buffer.write(response.encode('utf-8'))
            return self.send_buffer.getvalue()
    
    # 测试TCP处理器
    tcp_handler = TcpProtocolHandler()
    
    test_packets = [b"DATA1", b"DATA2", b"DATA3" * 100]
    for packet in test_packets:
        response = tcp_handler.process_packet(packet)
        print(f"数据包响应: {response.decode('utf-8')}")
    
    # 自定义协议格式处理
    class CustomProtocol:
        """自定义二进制协议"""
        def __init__(self):
            self.buffer = io.BytesIO()
        
        def encode_message(self, message_type, data):
            """编码消息"""
            header = struct.pack('>HH', message_type, len(data))
            return header + data
        
        def decode_messages(self, packet_data):
            """解码消息"""
            self.buffer.write(packet_data)
            messages = []
            
            while True:
                # 检查是否有完整的消息头
                if self.buffer.tell() < 4:
                    break
                
                self.buffer.seek(0)
                header = self.buffer.read(4)
                if len(header) < 4:
                    break
                
                message_type, data_length = struct.unpack('>HH', header)
                
                # 检查是否有完整的消息体
                if self.buffer.tell() - 4 < data_length:
                    break
                
                # 读取消息体
                data = self.buffer.read(data_length)
                messages.append((message_type, data))
                
                # 清理已处理的数据
                remaining = self.buffer.read()
                self.buffer.seek(0)
                self.buffer.truncate(0)
                self.buffer.write(remaining)
            
            return messages
    
    # 测试自定义协议
    protocol = CustomProtocol()
    
    test_messages = [
        (1, b"Hello"),
        (2, b"World"),
        (3, b"Test message")
    ]
    
    # 编码消息
    encoded_packets = []
    for msg_type, data in test_messages:
        packet = protocol.encode_message(msg_type, data)
        encoded_packets.append(packet)
        print(f"编码消息: 类型={msg_type}, 长度={len(data)}, 数据={data}")
    
    # 解码消息（模拟网络传输，可能分片）
    received_data = b''.join(encoded_packets)
    
    # 模拟分片接收
    chunks = [received_data[:10], received_data[10:20], received_data[20:]]
    
    for i, chunk in enumerate(chunks):
        print(f"接收分片 {i+1}: {len(chunk)} 字节")
        messages = protocol.decode_messages(chunk)
        for msg_type, data in messages:
            print(f"  解码消息: 类型={msg_type}, 数据={data.decode('utf-8')}")

# 执行示例
network_protocol_simulation()

def memory_usage_optimization():
    """内存使用优化策略"""
    import tracemalloc
    import gc
    
    # 测试不同方法的内存使用
    def test_memory_usage():
        """测试不同方法的内存使用"""
        # 方法1: 直接字符串拼接
        def method_direct_concatenation():
            result = ""
            for i in range(10000):
                result += f"数据 {i} "
            return result
        
        # 方法2: 列表拼接
        def method_list_join():
            parts = []
            for i in range(10000):
                parts.append(f"数据 {i} ")
            return "".join(parts)
        
        # 方法3: StringIO
        def method_stringio():
            with io.StringIO() as buffer:
                for i in range(10000):
                    buffer.write(f"数据 {i} ")
                return buffer.getvalue()
        
        # 测试内存使用
        methods = [
            ("直接拼接", method_direct_concatenation),
            ("列表拼接", method_list_join),
            ("StringIO", method_stringio)
        ]
        
        results = {}
        
        for name, method in methods:
            # 清理内存
            gc.collect()
            
            # 开始内存跟踪
            tracemalloc.start()
            
            # 执行方法
            result = method()
            results[name] = len(result)
            
            # 获取内存使用
            current, peak = tracemalloc.get_traced_memory()
            tracemalloc.stop()
            
            print(f"{name}:")
            print(f"  结果大小: {len(result)} 字符")
            print(f"  当前内存: {current / 1024:.2f} KB")
            print(f"  峰值内存: {peak / 1024:.2f} KB")
            print(f"  效率: {(len(result) / (peak or 1)):.2f} 字符/字节")
        
        return results
    
    memory_results = test_memory_usage()
    
    # 大文件处理优化
    def process_large_data_optimized():
        """大文件处理优化"""
        # 生成模拟大文件
        def generate_large_file(filename, size_mb=10):
            """生成大文件"""
            chunk_size = 1024 * 1024  # 1MB
            with open(filename, 'w', encoding='utf-8') as f:
                for i in range(size_mb):
                    chunk = "x" * chunk_size
                    f.write(chunk)
                    print(f"生成 {i+1} MB")
        
        generate_large_file('large_file.txt', 5)  # 生成5MB文件
        
        # 方法1: 直接读取（内存密集型）
        def read_directly():
            with open('large_file.txt', 'r', encoding='utf-8') as f:
                return f.read()
        
        # 方法2: 分块读取（内存友好）
        def read_in_chunks(chunk_size=1024 * 1024):
            with open('large_file.txt', 'r', encoding='utf-8') as f:
                with io.StringIO() as buffer:
                    while True:
                        chunk = f.read(chunk_size)
                        if not chunk:
                            break
                        # 处理块数据
                        processed_chunk = chunk.upper()  # 示例处理
                        buffer.write(processed_chunk)
                    return buffer.getvalue()
        
        # 方法3: 使用生成器（极低内存）
        def process_with_generator():
            with open('large_file.txt', 'r', encoding='utf-8') as f:
                for line in f:
                    yield line.upper()  # 逐行处理
        
        # 测试性能
        import time
        
        print("\n大文件处理性能测试:")
        
        # 方法1测试
        start_time = time.time()
        tracemalloc.start()
        result1 = read_directly()
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        time1 = time.time() - start_time
        
        print(f"直接读取: {time1:.2f}秒, 峰值内存: {peak/1024/1024:.2f}MB")
        
        # 方法2测试
        start_time = time.time()
        tracemalloc.start()
        result2 = read_in_chunks()
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        time2 = time.time() - start_time
        
        print(f"分块读取: {time2:.2f}秒, 峰值内存: {peak/1024/1024:.2f}MB")
        
        # 方法3测试
        start_time = time.time()
        tracemalloc.start()
        result3 = ""
        for chunk in process_with_generator():
            result3 += chunk
        current, peak = tracemalloc.get_traced_memory()
        tracemalloc.stop()
        time3 = time.time() - start_time
        
        print(f"生成器处理: {time3:.2f}秒, 峰值内存: {peak/1024/1024:.2f}MB")
        
        # 验证结果一致性
        print(f"结果一致性: {result1 == result2 == result3}")
        
        # 清理文件
        os.remove('large_file.txt')
    
    process_large_data_optimized()
    
    # StringIO池化技术
    class StringIOPool:
        """StringIO对象池"""
        def __init__(self, max_pool_size=10):
            self.pool = []
            self.max_pool_size = max_pool_size
        
        def acquire(self):
            """获取StringIO对象"""
            if self.pool:
                return self.pool.pop()
            return io.StringIO()
        
        def release(self, buffer):
            """释放StringIO对象"""
            if len(self.pool) < self.max_pool_size:
                buffer.seek(0)
                buffer.truncate(0)
                self.pool.append(buffer)
        
        def clear(self):
            """清空对象池"""
            self.pool.clear()
    
    # 使用对象池
    pool = StringIOPool()
    
    # 模拟高频率使用
    for i in range(100):
        buffer = pool.acquire()
        try:
            buffer.write(f"消息 {i}: 测试内容")
            # 使用缓冲区...
            content = buffer.getvalue()
            # print(f"处理: {content}")
        finally:
            pool.release(buffer)
    
    print(f"对象池大小: {len(pool.pool)}")
    pool.clear()

# 执行示例
memory_usage_optimization()

def performance_monitoring_analysis():
    """性能监控与分析"""
    import time
    import cProfile
    import pstats
    from memory_profiler import profile
    
    # 性能测试函数
    def performance_test():
        """性能测试"""
        # 测试数据
        test_data = [f"行 {i}: 测试数据 " * 10 + "\n" for i in range(10000)]
        
        # 测试1: 直接拼接
        start_time = time.time()
        result1 = ""
        for line in test_data:
            result1 += line
        time1 = time.time() - start_time
        
        # 测试2: 列表拼接
        start_time = time.time()
        result2 = "".join(test_data)
        time2 = time.time() - start_time
        
        # 测试3: StringIO
        start_time = time.time()
        with io.StringIO() as buffer:
            for line in test_data:
                buffer.write(line)
            result3 = buffer.getvalue()
        time3 = time.time() - start_time
        
        print(f"直接拼接: {time1:.4f}秒")
        print(f"列表拼接: {time2:.4f}秒")
        print(f"StringIO: {time3:.4f}秒")
        print(f"速度比 (StringIO/直接): {time3/time1:.2f}")
        print(f"结果相等: {result1 == result2 == result3}")
    
    # 运行性能测试
    print("性能测试结果:")
    performance_test()
    
    # 使用cProfile进行详细性能分析
    def profile_stringio_operations():
        """StringIO操作性能分析"""
        with io.StringIO() as buffer:
            for i in range(100000):
                buffer.write(f"行 {i}\n")
            content = buffer.getvalue()
            lines = content.split('\n')
            return len(lines)
    
    print("\n性能分析:")
    cProfile.run('profile_stringio_operations()', sort='cumulative')
    
    # 内存分析装饰器
    @profile
    def memory_intensive_operation():
        """内存密集型操作"""
        # 方法1: 直接操作
        big_string = ""
        for i in range(100000):
            big_string += f"数据 {i} "
        
        # 方法2: StringIO
        with io.StringIO() as buffer:
            for i in range(100000):
                buffer.write(f"数据 {i} ")
            result = buffer.getvalue()
        
        return len(big_string), len(result)
    
    # 运行内存分析（需要安装memory_profiler）
    try:
        print("内存分析:")
        result = memory_intensive_operation()
        print(f"结果大小: {result}")
    except ImportError:
        print("memory_profiler未安装，跳过内存分析")
    
    # 实时性能监控
    class PerformanceMonitor:
        """性能监控器"""
        def __init__(self):
            self.operations = []
            self.start_time = None
        
        def start(self):
            """开始监控"""
            self.start_time = time.time()
        
        def record_operation(self, name):
            """记录操作"""
            if self.start_time is None:
                self.start_time = time.time()
            
            current_time = time.time()
            elapsed = current_time - self.start_time
            self.operations.append((name, elapsed))
            self.start_time = current_time
        
        def get_report(self):
            """获取性能报告"""
            report = io.StringIO()
            report.write("性能报告\n")
            report.write("=" * 50 + "\n")
            
            total_time = sum(op[1] for op in self.operations)
            report.write(f"总时间: {total_time:.4f}秒\n\n")
            
            report.write("操作耗时:\n")
            for name, duration in self.operations:
                percentage = (duration / total_time) * 100 if total_time > 0 else 0
                report.write(f"  {name}: {duration:.4f}秒 ({percentage:.1f}%)\n")
            
            return report.getvalue()
    
    # 使用性能监控器
    monitor = PerformanceMonitor()
    monitor.start()
    
    # 模拟一些操作
    with io.StringIO() as buffer:
        monitor.record_operation("创建缓冲区")
        
        for i in range(1000):
            buffer.write(f"行 {i}\n")
        monitor.record_operation("写入数据")
        
        content = buffer.getvalue()
        monitor.record_operation("获取内容")
        
        lines = content.split('\n')
        monitor.record_operation("分割行")
    
    print("\n性能监控报告:")
    print(monitor.get_report())

# 执行示例
performance_monitoring_analysis()

def professional_stringio_template():
    """
    专业StringIO使用模板
    包含错误处理、性能优化、资源管理等最佳实践
    """
    class ProfessionalStringIO:
        def __init__(self, initial_value="", encoding='utf-8'):
            self.buffer = io.StringIO(initial_value)
            self.encoding = encoding
            self.operation_count = 0
            self.total_bytes_written = 0
        
        def write(self, data):
            """安全写入数据"""
            try:
                if isinstance(data, bytes):
                    # 解码字节数据
                    data = data.decode(self.encoding)
                
                bytes_written = self.buffer.write(data)
                self.operation_count += 1
                self.total_bytes_written += bytes_written
                return bytes_written
                
            except UnicodeDecodeError as e:
                print(f"编码错误: {e}")
                # 尝试错误恢复
                try:
                    # 使用错误处理策略
                    decoded = data.decode(self.encoding, errors='replace')
                    bytes_written = self.buffer.write(decoded)
                    self.operation_count += 1
                    self.total_bytes_written += bytes_written
                    return bytes_written
                except Exception as inner_e:
                    raise ValueError(f"无法处理数据: {inner_e}")
            
            except Exception as e:
                raise RuntimeError(f"写入失败: {e}")
        
        def read(self, size=None):
            """安全读取数据"""
            try:
                if size is None:
                    return self.buffer.getvalue()
                else:
                    return self.buffer.read(size)
            except Exception as e:
                raise RuntimeError(f"读取失败: {e}")
        
        def get_stats(self):
            """获取统计信息"""
            return {
                'operations': self.operation_count,
                'bytes_written': self.total_bytes_written,
                'buffer_size': self.buffer.tell(),
                'encoding': self.encoding
            }
        
        def __enter__(self):
            return self
        
        def __exit__(self, exc_type, exc_val, exc_tb):
            self.close()
            if exc_type:
                print(f"上下文退出时发生异常: {exc_val}")
            return False  # 不抑制异常
        
        def close(self):
            """关闭缓冲区"""
            if hasattr(self.buffer, 'close'):
                self.buffer.close()
    
    # 使用示例
    with ProfessionalStringIO("初始内容\n", encoding='utf-8') as buffer:
        # 写入各种数据
        buffer.write("文本数据\n")
        buffer.write("中文内容\n")
        buffer.write(b"Binary data with text\n")  # 自动解码
        
        # 读取内容
        content = buffer.read()
        print("缓冲区内容:")
        print(content)
        
        # 查看统计
        stats = buffer.get_stats()
        print(f"操作统计: {stats}")
    
    print("专业模板使用完成")

# 执行示例
professional_stringio_template()