import io

def basic_stringio_operations():
    """基礎(chǔ)StringIO操作示例"""
    # 創(chuàng)建StringIO對(duì)象
    string_buffer = io.StringIO()
    
    # 寫(xiě)入數(shù)據(jù)
    string_buffer.write("Hello, World!\n")
    string_buffer.write("這是第二行文本\n")
    string_buffer.write("Python字符串I/O操作\n")
    
    # 獲取當(dāng)前位置
    print(f"當(dāng)前位置: {string_buffer.tell()}")
    
    # 回到起始位置
    string_buffer.seek(0)
    
    # 讀取數(shù)據(jù)
    content = string_buffer.read()
    print("全部?jī)?nèi)容:")
    print(content)
    
    # 再次定位并讀取部分內(nèi)容
    string_buffer.seek(7)  # 移動(dòng)到"World"前面
    partial = string_buffer.read(5)  # 讀取5個(gè)字符
    print(f"部分內(nèi)容: '{partial}'")
    
    # 按行讀取
    string_buffer.seek(0)
    lines = string_buffer.readlines()
    print("行列表:")
    for i, line in enumerate(lines, 1):
        print(f"行 {i}: {line.strip()}")
    
    # 檢查緩沖區(qū)狀態(tài)
    print(f"緩沖區(qū)大小: {string_buffer.tell()} 字符")
    print(f"是否可讀: {string_buffer.readable()}")
    print(f"是否可寫(xiě): {string_buffer.writable()}")
    
    # 清空緩沖區(qū)
    string_buffer.truncate(0)
    string_buffer.seek(0)
    string_buffer.write("新的開(kāi)始")
    
    # 獲取最終內(nèi)容
    final_content = string_buffer.getvalue()
    print(f"最終內(nèi)容: '{final_content}'")
    
    # 關(guān)閉緩沖區(qū)
    string_buffer.close()

# 執(zhí)行示例
basic_stringio_operations()

def basic_bytesio_operations():
    """基礎(chǔ)BytesIO操作示例"""
    # 創(chuàng)建BytesIO對(duì)象
    bytes_buffer = io.BytesIO()
    
    # 寫(xiě)入二進(jìn)制數(shù)據(jù)
    bytes_buffer.write(b"Binary data\n")
    bytes_buffer.write("中文文本".encode('utf-8'))
    bytes_buffer.write(b"\x00\x01\x02\x03\x04\x05")  # 原始字節(jié)
    
    # 獲取當(dāng)前位置
    print(f"當(dāng)前位置: {bytes_buffer.tell()} 字節(jié)")
    
    # 回到起始位置
    bytes_buffer.seek(0)
    
    # 讀取數(shù)據(jù)
    binary_content = bytes_buffer.read()
    print("二進(jìn)制內(nèi)容:")
    print(f"長(zhǎng)度: {len(binary_content)} 字節(jié)")
    print(f"十六進(jìn)制: {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("包含非文本數(shù)據(jù)")
    
    # 寫(xiě)入混合數(shù)據(jù)
    bytes_buffer.seek(0)
    bytes_buffer.truncate(0)  # 清空
    
    # 寫(xiě)入不同類型數(shù)據(jù)
    data_parts = [
        b"HEADER",
        struct.pack('>I', 12345),  # 打包整數(shù)
        struct.pack('>d', 3.14159),  # 打包浮點(diǎn)數(shù)
        "結(jié)束標(biāo)記".encode('utf-8')
    ]
    
    for data in data_parts:
        bytes_buffer.write(data)
    
    # 解析結(jié)構(gòu)化數(shù)據(jù)
    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"整數(shù): {int_data}")
    print(f"浮點(diǎn)數(shù): {float_data}")
    print(f"尾部: {footer}")
    
    # 關(guān)閉緩沖區(qū)
    bytes_buffer.close()

# 執(zhí)行示例
basic_bytesio_operations()

def context_manager_usage():
    """上下文管理器使用示例"""
    # 使用with語(yǔ)句自動(dòng)管理資源
    with io.StringIO() as buffer:
        buffer.write("使用上下文管理器\n")
        buffer.write("自動(dòng)處理資源清理\n")
        
        content = buffer.getvalue()
        print("上下文管理器內(nèi)容:")
        print(content)
    
    # 緩沖區(qū)已自動(dòng)關(guān)閉
    print("緩沖區(qū)已自動(dòng)關(guān)閉")
    
    # 異常處理示例
    try:
        with io.BytesIO() as byte_buffer:
            byte_buffer.write(b"測(cè)試數(shù)據(jù)")
            raise ValueError("模擬異常")
            # 不會(huì)執(zhí)行到這里
    except ValueError as e:
        print(f"捕獲異常: {e}")
        print("緩沖區(qū)仍然被正確關(guān)閉")
    
    # 自定義上下文管理器
    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"退出上下文 (操作次數(shù): {self.operation_count})")
            print(f"最終內(nèi)容長(zhǎng)度: {len(content)} 字符")
            self.buffer.close()
            
            if exc_type:
                print(f"發(fā)生異常: {exc_val}")
                return False  # 不抑制異常
    
    # 使用自定義上下文管理器
    with SmartStringIO("初始內(nèi)容\n") as buffer:
        buffer.write("追加內(nèi)容\n")
        buffer.write("更多內(nèi)容\n")
        print("在上下文中操作緩沖區(qū)")
    
    print("自定義上下文管理器演示完成")

# 執(zhí)行示例
context_manager_usage()

def streaming_processing():
    """流式處理與迭代器示例"""
    # 生成大量數(shù)據(jù)
    def generate_large_data(num_lines=1000):
        """生成大量數(shù)據(jù)"""
        for i in range(num_lines):
            yield f"這是第 {i+1} 行數(shù)據(jù)，包含一些文本內(nèi)容用于測(cè)試字符串I/O性能\n"
    
    # 使用StringIO進(jìn)行流式處理
    with io.StringIO() as buffer:
        # 分批寫(xiě)入
        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} 行")
        
        # 處理剩余數(shù)據(jù)
        final_content = buffer.getvalue()
        print(f"最終內(nèi)容長(zhǎng)度: {len(final_content)} 字符")
        print(f"行數(shù): {final_content.count('\n')}")
    
    # 使用迭代器接口
    with io.StringIO("第一行\(zhòng)n第二行\(zhòng)n第三行\(zhòng)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()}")
    
    # 性能對(duì)比：直接拼接 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()
    
    # 生成測(cè)試數(shù)據(jù)
    test_data = [f"數(shù)據(jù)塊 {i} " * 10 + "\n" for i in range(10000)]
    
    # 測(cè)試性能
    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"直接拼接時(shí)間: {direct_time:.4f}秒")
    print(f"StringIO拼接時(shí)間: {stringio_time:.4f}秒")
    print(f"性能提升: {(direct_time/stringio_time):.2f}倍")
    print(f"結(jié)果相等: {result1 == result2}")

# 執(zhí)行示例
streaming_processing()

def csv_in_memory_processing():
    """CSV數(shù)據(jù)內(nèi)存處理示例"""
    import csv
    
    # 創(chuàng)建CSV數(shù)據(jù)
    csv_data = [
        ['姓名', '年齡', '城市', '職業(yè)'],
        ['張三', '25', '北京', '工程師'],
        ['李四', '30', '上海', '設(shè)計(jì)師'],
        ['王五', '28', '廣州', '產(chǎn)品經(jīng)理'],
        ['趙六', '35', '深圳', '架構(gòu)師']
    ]
    
    # 使用StringIO處理CSV
    with io.StringIO() as csv_buffer:
        # 寫(xiě)入CSV
        writer = csv.writer(csv_buffer)
        writer.writerows(csv_data)
        
        # 獲取CSV內(nèi)容
        csv_content = csv_buffer.getvalue()
        print("生成的CSV內(nèi)容:")
        print(csv_content)
        
        # 重置并讀取
        csv_buffer.seek(0)
        reader = csv.reader(csv_buffer)
        print("\n讀取CSV數(shù)據(jù):")
        for row in reader:
            print(f"行: {row}")
    
    # 更復(fù)雜的CSV處理
    def process_csv_in_memory(data, processing_func):
        """在內(nèi)存中處理CSV數(shù)據(jù)"""
        with io.StringIO() as buffer:
            # 寫(xiě)入原始數(shù)據(jù)
            writer = csv.writer(buffer)
            writer.writerows(data)
            
            # 處理數(shù)據(jù)
            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))
            
            # 寫(xiě)入處理后的數(shù)據(jù)
            buffer.seek(0)
            buffer.truncate(0)
            writer = csv.writer(buffer)
            writer.writerows(processed_lines)
            
            return buffer.getvalue()
    
    # 示例處理函數(shù)：年齡加1，城市大寫(xiě)
    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
    
    # 處理數(shù)據(jù)
    processed_csv = process_csv_in_memory(csv_data, age_increment)
    print("\n處理后的CSV:")
    print(processed_csv)

# 執(zhí)行示例
csv_in_memory_processing()

def json_in_memory_processing():
    """JSON數(shù)據(jù)內(nèi)存處理示例"""
    import json
    
    # 示例JSON數(shù)據(jù)
    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:
        # 寫(xiě)入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加載的數(shù)據(jù):")
        print(f"用戶數(shù)量: {len(loaded_data['users'])}")
        print(f"元數(shù)據(jù)版本: {loaded_data['metadata']['version']}")
    
    # JSON流式處理
    def stream_json_processing(data, chunk_size=1024):
        """流式處理大型JSON數(shù)據(jù)"""
        with io.StringIO() as buffer:
            # 使用生成器逐步寫(xiě)入
            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)
                
                # 模擬流式處理：定期處理數(shù)據(jù)
                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
    
    # 測(cè)試流式處理
    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"內(nèi)容預(yù)覽: {chunk[:50]}...")
    
    print(f"總大小: {total_size} 字符")

# 執(zhí)行示例
json_in_memory_processing()

def dynamic_html_generation():
    """動(dòng)態(tài)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>生成時(shí)間: $timestamp</p>
        <div id="users">
            $user_content
        </div>
    </body>
    </html>
    """)
    
    # 用戶數(shù)據(jù)
    users = [
        {"name": "張三", "email": "zhangsan@example.com", "active": True},
        {"name": "李四", "email": "lisi@example.com", "active": False},
        {"name": "王五", "email": "wangwu@example.com", "active": True}
    ]
    
    # 使用StringIO構(gòu)建動(dòng)態(tài)內(nèi)容
    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>狀態(tài): {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="系統(tǒng)用戶",
        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文件已保存")
    
    # 更復(fù)雜的模板系統(tǒng)
    class TemplateEngine:
        """簡(jiǎn)單的模板引擎"""
        def __init__(self):
            self.templates = {}
            self.partials = {}
        
        def register_template(self, name, content):
            """注冊(cè)模板"""
            self.templates[name] = content
        
        def register_partial(self, name, content):
            """注冊(cè)局部模板"""
            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>導(dǎo)航菜單</nav>
    </header>
    """)
    
    engine.register_partial('footer', """
    <footer>
        <p>版權(quán)所有 ? 2024</p>
    </footer>
    """)
    
    rendered = engine.render('page', 
                           title="模板引擎測(cè)試",
                           heading="歡迎使用",
                           content="這是主要內(nèi)容區(qū)域")
    
    print("\n模板引擎輸出:")
    print(rendered)

# 執(zhí)行示例
dynamic_html_generation()

def report_generation_system():
    """報(bào)告生成系統(tǒng)示例"""
    import csv
    from datetime import datetime, timedelta
    
    # 生成示例數(shù)據(jù)
    def generate_sales_data(days=30):
        """生成銷售數(shù)據(jù)"""
        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("銷售數(shù)據(jù)示例:")
    for item in sales_data:
        print(item)
    
    # 文本報(bào)告生成
    def generate_text_report(data):
        """生成文本格式報(bào)告"""
        with io.StringIO() as report:
            report.write("銷售日?qǐng)?bào)\n")
            report.write("=" * 40 + "\n")
            report.write(f"生成時(shí)間: {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n\n")
            
            report.write("日期        銷售額    客戶數(shù)  客單價(jià)\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"總計(jì)      {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文本報(bào)告:")
    print(text_report)
    
    # CSV報(bào)告生成
    def generate_csv_report(data):
        """生成CSV格式報(bào)告"""
        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報(bào)告:")
    print(csv_report)
    
    # HTML報(bào)告生成
    def generate_html_report(data):
        """生成HTML格式報(bào)告"""
        with io.StringIO() as html:
            html.write("""
            <!DOCTYPE html>
            <html>
            <head>
                <title>銷售報(bào)告</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>銷售報(bào)告</h1>
                <p>生成時(shí)間: """ + datetime.now().strftime('%Y-%m-%d %H:%M:%S') + """</p>
                <table>
                    <tr>
                        <th>日期</th>
                        <th>銷售額</th>
                        <th>客戶數(shù)</th>
                        <th>客單價(jià)</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>總計(jì)</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報(bào)告預(yù)覽:")
    print(html_report[:200] + "...")
    
    # 多格式報(bào)告生成器
    class ReportGenerator:
        """多格式報(bào)告生成器"""
        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'):
            """生成指定格式的報(bào)告"""
            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()
    
    # 使用報(bào)告生成器
    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} 報(bào)告: {filename}")
        if fmt == 'text':
            print("文本報(bào)告預(yù)覽:")
            print(report[:100] + "...")

# 執(zhí)行示例
report_generation_system()

def testing_and_mocking():
    """測(cè)試與模擬框架應(yīng)用"""
    import unittest
    from unittest.mock import patch, MagicMock
    
    # 被測(cè)函數(shù)
    def process_data(data_source):
        """處理數(shù)據(jù)的函數(shù)"""
        content = data_source.read()
        return content.upper().strip()
    
    # 使用StringIO進(jìn)行單元測(cè)試
    class TestDataProcessing(unittest.TestCase):
        """數(shù)據(jù)處理測(cè)試用例"""
        
        def test_with_stringio(self):
            """使用StringIO測(cè)試"""
            test_data = "hello world\n測(cè)試數(shù)據(jù)"
            
            # 創(chuàng)建StringIO作為數(shù)據(jù)源
            with io.StringIO(test_data) as data_source:
                result = process_data(data_source)
                expected = "HELLO WORLD\n測(cè)試數(shù)據(jù)".upper().strip()
                self.assertEqual(result, expected)
        
        def test_empty_data(self):
            """測(cè)試空數(shù)據(jù)"""
            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()
    
    # 運(yùn)行測(cè)試
    print("運(yùn)行測(cè)試用例...")
    loader = unittest.TestLoader()
    suite = loader.loadTestsFromTestCase(TestDataProcessing)
    runner = unittest.TextTestRunner(verbosity=2)
    result = runner.run(suite)
    
    # 模擬標(biāo)準(zhǔn)輸出
    def function_that_prints():
        """一個(gè)會(huì)打印輸出的函數(shù)"""
        print("正常輸出")
        print("錯(cuò)誤輸出", file=sys.stderr)
        return "結(jié)果"
    
    def test_output_capture():
        """測(cè)試輸出捕獲"""
        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"函數(shù)結(jié)果: {result}")
                print(f"標(biāo)準(zhǔn)輸出: {stdout_content!r}")
                print(f"標(biāo)準(zhǔn)錯(cuò)誤: {stderr_content!r}")
                
                assert "正常輸出" in stdout_content
                assert "錯(cuò)誤輸出" in stderr_content
    
    test_output_capture()
    
    # 更復(fù)雜的模擬場(chǎng)景
    class DatabaseSimulator:
        """數(shù)據(jù)庫(kù)模擬器"""
        def __init__(self):
            self.data = io.StringIO()
            self._setup_sample_data()
        
        def _setup_sample_data(self):
            """設(shè)置示例數(shù)據(jù)"""
            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)
    
    # 使用數(shù)據(jù)庫(kù)模擬器
    db = DatabaseSimulator()
    print("\n數(shù)據(jù)庫(kù)查詢結(jié)果:")
    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)

# 執(zhí)行示例
testing_and_mocking()

def network_protocol_simulation():
    """網(wǎng)絡(luò)協(xié)議模擬示例"""
    import socket
    import threading
    import time
    
    # 簡(jiǎn)單的HTTP服務(wù)器模擬
    class HttpServerSimulator:
        """HTTP服務(wù)器模擬器"""
        def __init__(self):
            self.request_buffer = io.BytesIO()
            self.response_buffer = io.BytesIO()
            self.request_count = 0
        
        def handle_request(self, request_data):
            """處理HTTP請(qǐng)求"""
            self.request_count += 1
            self.request_buffer.write(request_data)
            
            # 解析請(qǐng)求
            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)
                
                # 生成響應(yīng)
                response_body = f"""
                <html>
                <head><title>模擬服務(wù)器</title></head>
                <body>
                    <h1>Hello from Simulator</h1>
                    <p>請(qǐng)求方法: {method}</p>
                    <p>請(qǐng)求路徑: {path}</p>
                    <p>協(xié)議版本: {protocol}</p>
                    <p>請(qǐng)求計(jì)數(shù): {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"
    
    # 測(cè)試HTTP模擬器
    simulator = HttpServerSimulator()
    
    # 模擬HTTP請(qǐng)求
    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"請(qǐng)求 {i+1} 響應(yīng):")
        print(response.decode('utf-8')[:200] + "...")
        print("-" * 50)
    
    # TCP協(xié)議模擬
    class TcpProtocolHandler:
        """TCP協(xié)議處理器"""
        def __init__(self):
            self.receive_buffer = io.BytesIO()
            self.send_buffer = io.BytesIO()
            self.sequence_number = 0
        
        def process_packet(self, packet_data):
            """處理數(shù)據(jù)包"""
            self.receive_buffer.write(packet_data)
            
            # 模擬協(xié)議處理
            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()
    
    # 測(cè)試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"數(shù)據(jù)包響應(yīng): {response.decode('utf-8')}")
    
    # 自定義協(xié)議格式處理
    class CustomProtocol:
        """自定義二進(jìn)制協(xié)議"""
        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))
                
                # 清理已處理的數(shù)據(jù)
                remaining = self.buffer.read()
                self.buffer.seek(0)
                self.buffer.truncate(0)
                self.buffer.write(remaining)
            
            return messages
    
    # 測(cè)試自定義協(xié)議
    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}, 長(zhǎng)度={len(data)}, 數(shù)據(jù)={data}")
    
    # 解碼消息（模擬網(wǎng)絡(luò)傳輸，可能分片）
    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)} 字節(jié)")
        messages = protocol.decode_messages(chunk)
        for msg_type, data in messages:
            print(f"  解碼消息: 類型={msg_type}, 數(shù)據(jù)={data.decode('utf-8')}")

# 執(zhí)行示例
network_protocol_simulation()

def memory_usage_optimization():
    """內(nèi)存使用優(yōu)化策略"""
    import tracemalloc
    import gc
    
    # 測(cè)試不同方法的內(nèi)存使用
    def test_memory_usage():
        """測(cè)試不同方法的內(nèi)存使用"""
        # 方法1: 直接字符串拼接
        def method_direct_concatenation():
            result = ""
            for i in range(10000):
                result += f"數(shù)據(jù) {i} "
            return result
        
        # 方法2: 列表拼接
        def method_list_join():
            parts = []
            for i in range(10000):
                parts.append(f"數(shù)據(jù) {i} ")
            return "".join(parts)
        
        # 方法3: StringIO
        def method_stringio():
            with io.StringIO() as buffer:
                for i in range(10000):
                    buffer.write(f"數(shù)據(jù) {i} ")
                return buffer.getvalue()
        
        # 測(cè)試內(nèi)存使用
        methods = [
            ("直接拼接", method_direct_concatenation),
            ("列表拼接", method_list_join),
            ("StringIO", method_stringio)
        ]
        
        results = {}
        
        for name, method in methods:
            # 清理內(nèi)存
            gc.collect()
            
            # 開(kāi)始內(nèi)存跟蹤
            tracemalloc.start()
            
            # 執(zhí)行方法
            result = method()
            results[name] = len(result)
            
            # 獲取內(nèi)存使用
            current, peak = tracemalloc.get_traced_memory()
            tracemalloc.stop()
            
            print(f"{name}:")
            print(f"  結(jié)果大小: {len(result)} 字符")
            print(f"  當(dāng)前內(nèi)存: {current / 1024:.2f} KB")
            print(f"  峰值內(nèi)存: {peak / 1024:.2f} KB")
            print(f"  效率: {(len(result) / (peak or 1)):.2f} 字符/字節(jié)")
        
        return results
    
    memory_results = test_memory_usage()
    
    # 大文件處理優(yōu)化
    def process_large_data_optimized():
        """大文件處理優(yōu)化"""
        # 生成模擬大文件
        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: 直接讀?。▋?nèi)存密集型）
        def read_directly():
            with open('large_file.txt', 'r', encoding='utf-8') as f:
                return f.read()
        
        # 方法2: 分塊讀?。▋?nèi)存友好）
        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
                        # 處理塊數(shù)據(jù)
                        processed_chunk = chunk.upper()  # 示例處理
                        buffer.write(processed_chunk)
                    return buffer.getvalue()
        
        # 方法3: 使用生成器（極低內(nèi)存）
        def process_with_generator():
            with open('large_file.txt', 'r', encoding='utf-8') as f:
                for line in f:
                    yield line.upper()  # 逐行處理
        
        # 測(cè)試性能
        import time
        
        print("\n大文件處理性能測(cè)試:")
        
        # 方法1測(cè)試
        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}秒, 峰值內(nèi)存: {peak/1024/1024:.2f}MB")
        
        # 方法2測(cè)試
        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}秒, 峰值內(nèi)存: {peak/1024/1024:.2f}MB")
        
        # 方法3測(cè)試
        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}秒, 峰值內(nèi)存: {peak/1024/1024:.2f}MB")
        
        # 驗(yàn)證結(jié)果一致性
        print(f"結(jié)果一致性: {result1 == result2 == result3}")
        
        # 清理文件
        os.remove('large_file.txt')
    
    process_large_data_optimized()
    
    # StringIO池化技術(shù)
    class StringIOPool:
        """StringIO對(duì)象池"""
        def __init__(self, max_pool_size=10):
            self.pool = []
            self.max_pool_size = max_pool_size
        
        def acquire(self):
            """獲取StringIO對(duì)象"""
            if self.pool:
                return self.pool.pop()
            return io.StringIO()
        
        def release(self, buffer):
            """釋放StringIO對(duì)象"""
            if len(self.pool) < self.max_pool_size:
                buffer.seek(0)
                buffer.truncate(0)
                self.pool.append(buffer)
        
        def clear(self):
            """清空對(duì)象池"""
            self.pool.clear()
    
    # 使用對(duì)象池
    pool = StringIOPool()
    
    # 模擬高頻率使用
    for i in range(100):
        buffer = pool.acquire()
        try:
            buffer.write(f"消息 {i}: 測(cè)試內(nèi)容")
            # 使用緩沖區(qū)...
            content = buffer.getvalue()
            # print(f"處理: {content}")
        finally:
            pool.release(buffer)
    
    print(f"對(duì)象池大小: {len(pool.pool)}")
    pool.clear()

# 執(zhí)行示例
memory_usage_optimization()

def performance_monitoring_analysis():
    """性能監(jiān)控與分析"""
    import time
    import cProfile
    import pstats
    from memory_profiler import profile
    
    # 性能測(cè)試函數(shù)
    def performance_test():
        """性能測(cè)試"""
        # 測(cè)試數(shù)據(jù)
        test_data = [f"行 {i}: 測(cè)試數(shù)據(jù) " * 10 + "\n" for i in range(10000)]
        
        # 測(cè)試1: 直接拼接
        start_time = time.time()
        result1 = ""
        for line in test_data:
            result1 += line
        time1 = time.time() - start_time
        
        # 測(cè)試2: 列表拼接
        start_time = time.time()
        result2 = "".join(test_data)
        time2 = time.time() - start_time
        
        # 測(cè)試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"結(jié)果相等: {result1 == result2 == result3}")
    
    # 運(yùn)行性能測(cè)試
    print("性能測(cè)試結(jié)果:")
    performance_test()
    
    # 使用cProfile進(jìn)行詳細(xì)性能分析
    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')
    
    # 內(nèi)存分析裝飾器
    @profile
    def memory_intensive_operation():
        """內(nèi)存密集型操作"""
        # 方法1: 直接操作
        big_string = ""
        for i in range(100000):
            big_string += f"數(shù)據(jù) {i} "
        
        # 方法2: StringIO
        with io.StringIO() as buffer:
            for i in range(100000):
                buffer.write(f"數(shù)據(jù) {i} ")
            result = buffer.getvalue()
        
        return len(big_string), len(result)
    
    # 運(yùn)行內(nèi)存分析（需要安裝memory_profiler）
    try:
        print("內(nèi)存分析:")
        result = memory_intensive_operation()
        print(f"結(jié)果大小: {result}")
    except ImportError:
        print("memory_profiler未安裝，跳過(guò)內(nèi)存分析")
    
    # 實(shí)時(shí)性能監(jiān)控
    class PerformanceMonitor:
        """性能監(jiān)控器"""
        def __init__(self):
            self.operations = []
            self.start_time = None
        
        def start(self):
            """開(kāi)始監(jiān)控"""
            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):
            """獲取性能報(bào)告"""
            report = io.StringIO()
            report.write("性能報(bào)告\n")
            report.write("=" * 50 + "\n")
            
            total_time = sum(op[1] for op in self.operations)
            report.write(f"總時(shí)間: {total_time:.4f}秒\n\n")
            
            report.write("操作耗時(shí):\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()
    
    # 使用性能監(jiān)控器
    monitor = PerformanceMonitor()
    monitor.start()
    
    # 模擬一些操作
    with io.StringIO() as buffer:
        monitor.record_operation("創(chuàng)建緩沖區(qū)")
        
        for i in range(1000):
            buffer.write(f"行 {i}\n")
        monitor.record_operation("寫(xiě)入數(shù)據(jù)")
        
        content = buffer.getvalue()
        monitor.record_operation("獲取內(nèi)容")
        
        lines = content.split('\n')
        monitor.record_operation("分割行")
    
    print("\n性能監(jiān)控報(bào)告:")
    print(monitor.get_report())

# 執(zhí)行示例
performance_monitoring_analysis()

def professional_stringio_template():
    """
    專業(yè)StringIO使用模板
    包含錯(cuò)誤處理、性能優(yōu)化、資源管理等最佳實(shí)踐
    """
    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):
            """安全寫(xiě)入數(shù)據(jù)"""
            try:
                if isinstance(data, bytes):
                    # 解碼字節(jié)數(shù)據(jù)
                    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"編碼錯(cuò)誤: {e}")
                # 嘗試錯(cuò)誤恢復(fù)
                try:
                    # 使用錯(cuò)誤處理策略
                    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"無(wú)法處理數(shù)據(jù): {inner_e}")
            
            except Exception as e:
                raise RuntimeError(f"寫(xiě)入失敗: {e}")
        
        def read(self, size=None):
            """安全讀取數(shù)據(jù)"""
            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):
            """獲取統(tǒng)計(jì)信息"""
            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"上下文退出時(shí)發(fā)生異常: {exc_val}")
            return False  # 不抑制異常
        
        def close(self):
            """關(guān)閉緩沖區(qū)"""
            if hasattr(self.buffer, 'close'):
                self.buffer.close()
    
    # 使用示例
    with ProfessionalStringIO("初始內(nèi)容\n", encoding='utf-8') as buffer:
        # 寫(xiě)入各種數(shù)據(jù)
        buffer.write("文本數(shù)據(jù)\n")
        buffer.write("中文內(nèi)容\n")
        buffer.write(b"Binary data with text\n")  # 自動(dòng)解碼
        
        # 讀取內(nèi)容
        content = buffer.read()
        print("緩沖區(qū)內(nèi)容:")
        print(content)
        
        # 查看統(tǒng)計(jì)
        stats = buffer.get_stats()
        print(f"操作統(tǒng)計(jì): {stats}")
    
    print("專業(yè)模板使用完成")

# 執(zhí)行示例
professional_stringio_template()