# 基本文件讀寫操作
def basic_file_operations():
    """基礎(chǔ)文件操作示例"""
    # 寫入文本數(shù)據(jù)
    with open('example.txt', 'w', encoding='utf-8') as f:
        f.write("Hello, World!\n")
        f.write("這是第二行文本\n")
        f.write("Third line with numbers: 123\n")
    
    # 讀取整個(gè)文件
    with open('example.txt', 'r', encoding='utf-8') as f:
        content = f.read()
        print("文件全部?jī)?nèi)容:")
        print(content)
    
    # 逐行讀取
    with open('example.txt', 'r', encoding='utf-8') as f:
        print("\n逐行讀取:")
        for i, line in enumerate(f, 1):
            print(f"行 {i}: {line.strip()}")
    
    # 讀取所有行到列表
    with open('example.txt', 'r', encoding='utf-8') as f:
        lines = f.readlines()
        print(f"\n所有行列表: {lines}")

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

def advanced_file_modes():
    """高級(jí)文件模式使用示例"""
    # 讀寫模式（r+）
    with open('data.txt', 'w+', encoding='utf-8') as f:
        f.write("初始內(nèi)容\n")
        f.seek(0)  # 回到文件開頭
        content = f.read()
        print("讀寫模式內(nèi)容:", content)
    
    # 追加讀寫模式（a+）
    with open('data.txt', 'a+', encoding='utf-8') as f:
        f.write("追加的內(nèi)容\n")
        f.seek(0)
        content = f.read()
        print("追加后內(nèi)容:", content)
    
    # 二進(jìn)制讀寫
    with open('binary_data.bin', 'wb') as f:
        f.write(b'\x00\x01\x02\x03\x04\x05')
    
    with open('binary_data.bin', 'rb') as f:
        binary_content = f.read()
        print("二進(jìn)制內(nèi)容:", binary_content)

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

def encoding_handling():
    """文本編碼處理"""
    texts = [
        "Hello World",
        "你好世界",
        "こんにちは世界",
        "????? ??"
    ]
    
    # 不同編碼寫入
    encodings = ['utf-8', 'gbk', 'shift_jis', 'euc-kr']
    
    for i, (text, encoding) in enumerate(zip(texts, encodings)):
        try:
            with open(f'file_{i}.txt', 'w', encoding=encoding) as f:
                f.write(text)
            print(f"成功寫入 {encoding} 編碼文件")
        except UnicodeEncodeError as e:
            print(f"編碼錯(cuò)誤: {encoding} - {e}")
    
    # 自動(dòng)檢測(cè)編碼讀取
    import chardet
    
    for i in range(len(texts)):
        try:
            with open(f'file_{i}.txt', 'rb') as f:
                raw_data = f.read()
                detected = chardet.detect(raw_data)
                encoding = detected['encoding']
                confidence = detected['confidence']
                
                print(f"檢測(cè)到編碼: {encoding} (置信度: {confidence:.2f})")
                content = raw_data.decode(encoding)
                print(f"文件內(nèi)容: {content}")
                
        except FileNotFoundError:
            print(f"文件 file_{i}.txt 不存在")
        except UnicodeDecodeError as e:
            print(f"解碼錯(cuò)誤: {e}")

encoding_handling()

def encoding_conversion():
    """編碼轉(zhuǎn)換處理"""
    # 創(chuàng)建測(cè)試文件
    text = "中文測(cè)試 English Test 日本語テスト"
    
    # 以不同編碼保存
    with open('text_gbk.txt', 'w', encoding='gbk') as f:
        f.write(text)
    
    with open('text_utf8.txt', 'w', encoding='utf-8') as f:
        f.write(text)
    
    # 編碼轉(zhuǎn)換函數(shù)
    def convert_encoding(input_file, output_file, from_encoding, to_encoding):
        """轉(zhuǎn)換文件編碼"""
        try:
            with open(input_file, 'r', encoding=from_encoding) as f_in:
                content = f_in.read()
            
            with open(output_file, 'w', encoding=to_encoding) as f_out:
                f_out.write(content)
            
            print(f"成功轉(zhuǎn)換 {input_file} 從 {from_encoding} 到 {to_encoding}")
            
        except UnicodeDecodeError:
            print(f"解碼失敗: {input_file} 可能不是 {from_encoding} 編碼")
        except UnicodeEncodeError:
            print(f"編碼失敗: 無法用 {to_encoding} 編碼內(nèi)容")
    
    # 執(zhí)行轉(zhuǎn)換
    convert_encoding('text_gbk.txt', 'text_utf8_from_gbk.txt', 'gbk', 'utf-8')
    convert_encoding('text_utf8.txt', 'text_gbk_from_utf8.txt', 'utf-8', 'gbk')
    
    # Unicode規(guī)范化
    import unicodedata
    
    text_with_unicode = "café na?ve ni?a"
    normalized = unicodedata.normalize('NFC', text_with_unicode)
    print(f"原始文本: {text_with_unicode}")
    print(f"規(guī)范化后: {normalized}")

encoding_conversion()

class SafeFileHandler:
    """安全的文件處理器，帶異常處理"""
    def __init__(self, filename, mode='r', encoding='utf-8'):
        self.filename = filename
        self.mode = mode
        self.encoding = encoding
        self.file = None
    
    def __enter__(self):
        try:
            self.file = open(self.filename, self.mode, encoding=self.encoding)
            return self.file
        except FileNotFoundError:
            print(f"錯(cuò)誤: 文件 {self.filename} 不存在")
            raise
        except PermissionError:
            print(f"錯(cuò)誤: 沒有權(quán)限訪問 {self.filename}")
            raise
        except Exception as e:
            print(f"打開文件時(shí)發(fā)生未知錯(cuò)誤: {e}")
            raise
    
    def __exit__(self, exc_type, exc_val, exc_tb):
        if self.file:
            self.file.close()
        if exc_type:
            print(f"文件操作發(fā)生錯(cuò)誤: {exc_val}")
        return False  # 不抑制異常

# 使用示例
def safe_file_operations():
    """安全的文件操作示例"""
    try:
        with SafeFileHandler('example.txt', 'r') as f:
            content = f.read()
            print("安全讀取的內(nèi)容:", content)
    except Exception as e:
        print(f"操作失敗: {e}")
    
    # 寫入操作
    try:
        with SafeFileHandler('output.txt', 'w') as f:
            f.write("這是安全寫入的內(nèi)容\n")
    except Exception as e:
        print(f"寫入失敗: {e}")

safe_file_operations()

from pathlib import Path
import os

def path_operations():
    """現(xiàn)代文件路徑處理"""
    # 使用pathlib處理路徑
    current_dir = Path.cwd()
    print(f"當(dāng)前目錄: {current_dir}")
    
    # 創(chuàng)建文件路徑
    file_path = current_dir / 'data' / 'files' / 'example.txt'
    print(f"文件路徑: {file_path}")
    
    # 創(chuàng)建目錄
    file_path.parent.mkdir(parents=True, exist_ok=True)
    
    # 寫入文件
    file_path.write_text("這是使用pathlib寫入的內(nèi)容\n", encoding='utf-8')
    
    # 讀取文件
    content = file_path.read_text(encoding='utf-8')
    print(f"文件內(nèi)容: {content}")
    
    # 文件信息
    print(f"文件存在: {file_path.exists()}")
    print(f"是文件: {file_path.is_file()}")
    print(f"文件大小: {file_path.stat().st_size} 字節(jié)")
    
    # 遍歷目錄
    data_dir = current_dir / 'data'
    print("目錄內(nèi)容:")
    for item in data_dir.iterdir():
        print(f"  {item.name} - {'文件' if item.is_file() else '目錄'}")
    
    # 查找文件
    print("查找txt文件:")
    for txt_file in data_dir.rglob('*.txt'):
        print(f"  找到: {txt_file}")

path_operations()

def process_large_file(filename, chunk_size=1024 * 1024):  # 1MB chunks
    """處理大文件的迭代器方法"""
    with open(filename, 'r', encoding='utf-8') as f:
        while True:
            chunk = f.read(chunk_size)
            if not chunk:
                break
            # 處理塊數(shù)據(jù)
            yield from process_chunk(chunk)

def process_chunk(chunk):
    """處理數(shù)據(jù)塊的生成器"""
    lines = chunk.split('\n')
    for line in lines:
        if line.strip():  # 跳過空行
            yield line.strip()

# 使用示例
def large_file_example():
    """大文件處理示例"""
    # 創(chuàng)建一個(gè)大文件示例
    with open('large_file.txt', 'w', encoding='utf-8') as f:
        for i in range(100000):
            f.write(f"這是第 {i} 行數(shù)據(jù)，包含一些文本內(nèi)容用于測(cè)試\n")
    
    # 處理大文件
    line_count = 0
    for line in process_large_file('large_file.txt'):
        line_count += 1
        if line_count % 10000 == 0:
            print(f"已處理 {line_count} 行")
    
    print(f"總共處理了 {line_count} 行")

large_file_example()

import mmap

def memory_mapped_operations():
    """內(nèi)存映射文件處理大型文本"""
    # 創(chuàng)建大型文本文件
    with open('large_text.txt', 'w', encoding='utf-8') as f:
        for i in range(100000):
            f.write(f"這是第 {i} 行，包含一些文本內(nèi)容用于測(cè)試內(nèi)存映射文件操作\n")
    
    # 使用內(nèi)存映射讀取
    with open('large_text.txt', 'r+', encoding='utf-8') as f:
        # 創(chuàng)建內(nèi)存映射
        with mmap.mmap(f.fileno(), 0, access=mmap.ACCESS_READ) as mm:
            # 搜索特定內(nèi)容
            position = mm.find(b"第 50000 行")
            if position != -1:
                mm.seek(position)
                line = mm.readline().decode('utf-8')
                print(f"找到的行: {line}")
            
            # 統(tǒng)計(jì)行數(shù)
            line_count = 0
            mm.seek(0)
            while True:
                line = mm.readline()
                if not line:
                    break
                line_count += 1
            
            print(f"文件總行數(shù): {line_count}")
            
            # 迭代處理每一行
            mm.seek(0)
            for i in range(5):  # 只顯示前5行
                line = mm.readline().decode('utf-8').strip()
                print(f"行 {i+1}: {line}")

memory_mapped_operations()

import csv
from collections import namedtuple

def csv_operations():
    """CSV文件讀寫操作"""
    # 寫入CSV文件
    with open('data.csv', 'w', newline='', encoding='utf-8') as f:
        writer = csv.writer(f)
        writer.writerow(['姓名', '年齡', '城市'])
        writer.writerow(['張三', 25, '北京'])
        writer.writerow(['李四', 30, '上海'])
        writer.writerow(['王五', 28, '廣州'])
    
    # 讀取CSV文件
    with open('data.csv', 'r', newline='', encoding='utf-8') as f:
        reader = csv.reader(f)
        header = next(reader)
        print("CSV頭部:", header)
        for row in reader:
            print(f"行數(shù)據(jù): {row}")
    
    # 使用字典方式讀寫CSV
    with open('data_dict.csv', 'w', newline='', encoding='utf-8') as f:
        fieldnames = ['name', 'age', 'city']
        writer = csv.DictWriter(f, fieldnames=fieldnames)
        writer.writeheader()
        writer.writerow({'name': '張三', 'age': 25, 'city': '北京'})
        writer.writerow({'name': '李四', 'age': 30, 'city': '上海'})
    
    # 讀取為字典
    with open('data_dict.csv', 'r', newline='', encoding='utf-8') as f:
        reader = csv.DictReader(f)
        for row in reader:
            print(f"字典行: {row}")

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

import json

def json_operations():
    """JSON文件讀寫操作"""
    data = {
        "users": [
            {"name": "張三", "age": 25, "hobbies": ["閱讀", "游泳"]},
            {"name": "李四", "age": 30, "hobbies": ["音樂", "旅行"]},
            {"name": "王五", "age": 28, "hobbies": ["攝影", "編程"]}
        ],
        "metadata": {
            "created": "2024-01-01",
            "version": "1.0"
        }
    }
    
    # 寫入JSON文件
    with open('data.json', 'w', encoding='utf-8') as f:
        json.dump(data, f, ensure_ascii=False, indent=2)
    
    # 讀取JSON文件
    with open('data.json', 'r', encoding='utf-8') as f:
        loaded_data = json.load(f)
        print("JSON數(shù)據(jù):", loaded_data)
    
    # 處理大型JSON流
    def generate_large_json():
        """生成大型JSON數(shù)據(jù)"""
        for i in range(1000):
            yield json.dumps({"id": i, "data": f"示例數(shù)據(jù) {i}"}) + '\n'
    
    # 寫入JSON流
    with open('large_data.jsonl', 'w', encoding='utf-8') as f:
        for item in generate_large_json():
            f.write(item)
    
    # 讀取JSON流
    with open('large_data.jsonl', 'r', encoding='utf-8') as f:
        for line in f:
            item = json.loads(line.strip())
            if item['id'] % 100 == 0:
                print(f"處理項(xiàng)目: {item}")

json_operations()

import re

def regex_text_processing():
    """使用正則表達(dá)式處理文本"""
    # 示例文本
    text = """
    聯(lián)系人信息:
    張三: 電話 138-1234-5678, 郵箱 zhangsan@example.com
    李四: 電話 139-8765-4321, 郵箱 lisi@example.com
    王五: 電話 137-5555-6666, 郵箱 wangwu@example.com
    """
    
    # 提取電話號(hào)碼
    phone_pattern = r'\b\d{3}-\d{4}-\d{4}\b'
    phones = re.findall(phone_pattern, text)
    print("提取的電話號(hào)碼:", phones)
    
    # 提取郵箱地址
    email_pattern = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,}\b'
    emails = re.findall(email_pattern, text)
    print("提取的郵箱地址:", emails)
    
    # 提取姓名和聯(lián)系方式
    contact_pattern = r'([\u4e00-\u9fa5]+):\s*電話\s*(\d{3}-\d{4}-\d{4}),\s*郵箱\s*([^\s,]+)'
    contacts = re.findall(contact_pattern, text)
    print("完整聯(lián)系人信息:")
    for name, phone, email in contacts:
        print(f"姓名: {name}, 電話: {phone}, 郵箱: {email}")
    
    # 使用正則表達(dá)式替換
    replaced_text = re.sub(r'\d{3}-\d{4}-\d{4}', '***-****-****', text)
    print("脫敏后的文本:")
    print(replaced_text)

regex_text_processing()

from string import Template

def template_processing():
    """使用模板生成文本"""
    # 簡(jiǎn)單字符串模板
    template = Template("您好，$name！您的訂單#$order_id 已發(fā)貨，預(yù)計(jì)$delivery_date送達(dá)。")
    
    message = template.substitute(
        name="張三",
        order_id="12345",
        delivery_date="2024-01-15"
    )
    print("模板消息:", message)
    
    # 文件模板示例
    with open('template.txt', 'w', encoding='utf-8') as f:
        f.write("""
尊敬的$customer_name：

感謝您購(gòu)買我們的產(chǎn)品。

訂單詳情:
- 訂單號(hào): $order_id
- 產(chǎn)品: $product_name
- 數(shù)量: $quantity
- 總價(jià): ￥$total_price

預(yù)計(jì)發(fā)貨時(shí)間: $ship_date
如有問題，請(qǐng)聯(lián)系: $support_email

祝您購(gòu)物愉快！
$company_name 團(tuán)隊(duì)
        """)
    
    # 從文件讀取模板
    with open('template.txt', 'r', encoding='utf-8') as f:
        template_content = f.read()
    
    # 填充模板
    email_template = Template(template_content)
    email_content = email_template.substitute(
        customer_name="李四",
        order_id="67890",
        product_name="Python編程書籍",
        quantity=2,
        total_price="199.00",
        ship_date="2024-01-16",
        support_email="support@example.com",
        company_name="卓越圖書"
    )
    
    print("生成的郵件內(nèi)容:")
    print(email_content)
    
    # 批量生成內(nèi)容
    customers = [
        {"name": "王五", "order_id": "11111", "product": "筆記本電腦", "quantity": 1, "price": "5999.00"},
        {"name": "趙六", "order_id": "22222", "product": "智能手機(jī)", "quantity": 1, "price": "3999.00"},
    ]
    
    for customer in customers:
        message = template.substitute(
            name=customer["name"],
            order_id=customer["order_id"],
            delivery_date="2024-01-17"
        )
        print(f"給 {customer['name']} 的消息: {message}")

template_processing()

import time
import functools

def timeit(func):
    """計(jì)時(shí)裝飾器"""
    @functools.wraps(func)
    def wrapper(*args, **kwargs):
        start = time.time()
        result = func(*args, **kwargs)
        end = time.time()
        print(f"{func.__name__} 耗時(shí): {end - start:.4f}秒")
        return result
    return wrapper

@timeit
def optimized_text_processing():
    """優(yōu)化文本處理性能"""
    # 創(chuàng)建測(cè)試數(shù)據(jù)
    with open('perf_test.txt', 'w', encoding='utf-8') as f:
        for i in range(100000):
            f.write(f"這是測(cè)試行號(hào) {i}，包含一些文本內(nèi)容用于性能測(cè)試\n")
    
    # 方法1: 傳統(tǒng)逐行讀取
    def method1():
        with open('perf_test.txt', 'r', encoding='utf-8') as f:
            lines = []
            for line in f:
                lines.append(line.strip())
        return lines
    
    # 方法2: 使用列表推導(dǎo)式
    def method2():
        with open('perf_test.txt', 'r', encoding='utf-8') as f:
            return [line.strip() for line in f]
    
    # 方法3: 使用生成器表達(dá)式
    def method3():
        with open('perf_test.txt', 'r', encoding='utf-8') as f:
            return (line.strip() for line in f)
    
    # 方法4: 批量處理
    import itertools
    def method4():
        with open('perf_test.txt', 'r', encoding='utf-8') as f:
            while True:
                lines = [line.strip() for line in itertools.islice(f, 1000)]
                if not lines:
                    break
                yield lines
    
    print("性能測(cè)試開始:")
    result1 = method1()
    result2 = method2()
    result3 = method3()
    
    line_count = 0
    for batch in method4():
        line_count += len(batch)
    
    print(f"總行數(shù): {len(result1)}, 批量處理行數(shù): {line_count}")

optimized_text_processing()

def memory_optimization():
    """文本處理內(nèi)存優(yōu)化"""
    # 創(chuàng)建大型文件
    with open('large_memory_test.txt', 'w', encoding='utf-8') as f:
        for i in range(500000):
            f.write(f"行 {i}: 這是一個(gè)測(cè)試行，包含一些文本內(nèi)容用于內(nèi)存優(yōu)化測(cè)試\n")
    
    # 內(nèi)存密集型方法（不推薦）
    def memory_intensive():
        with open('large_memory_test.txt', 'r', encoding='utf-8') as f:
            lines = f.readlines()  # 一次性讀取所有行
            processed = [line.upper() for line in lines]
        return processed
    
    # 內(nèi)存友好方法（推薦）
    def memory_friendly():
        with open('large_memory_test.txt', 'r', encoding='utf-8') as f:
            for line in f:
                yield line.upper()  # 逐行生成結(jié)果
    
    # 測(cè)量?jī)?nèi)存使用
    import tracemalloc
    
    print("內(nèi)存使用測(cè)試:")
    
    tracemalloc.start()
    # 測(cè)試內(nèi)存密集型方法
    result1 = memory_intensive()
    current, peak = tracemalloc.get_traced_memory()
    print(f"內(nèi)存密集型 - 當(dāng)前: {current/1024/1024:.2f}MB, 峰值: {peak/1024/1024:.2f}MB")
    tracemalloc.stop()
    
    tracemalloc.start()
    # 測(cè)試內(nèi)存友好方法
    result2 = list(memory_friendly())  # 轉(zhuǎn)換為列表以便比較
    current, peak = tracemalloc.get_traced_memory()
    print(f"內(nèi)存友好型 - 當(dāng)前: {current/1024/1024:.2f}MB, 峰值: {peak/1024/1024:.2f}MB")
    tracemalloc.stop()
    
    # 驗(yàn)證結(jié)果一致性
    print(f"結(jié)果一致性: {result1 == result2}")

memory_optimization()

def professional_text_processor(input_file, output_file, processing_func):
    """
    專業(yè)文本處理器模板
    
    參數(shù):
        input_file: 輸入文件路徑
        output_file: 輸出文件路徑
        processing_func: 處理函數(shù)，接受一行文本返回處理結(jié)果
    """
    try:
        with open(input_file, 'r', encoding='utf-8') as infile, \
             open(output_file, 'w', encoding='utf-8') as outfile:
            
            # 使用生成器表達(dá)式進(jìn)行流式處理
            processed_lines = (processing_func(line) for line in infile)
            
            # 批量寫入提高性能
            batch_size = 1000
            batch = []
            
            for processed_line in processed_lines:
                batch.append(processed_line)
                if len(batch) >= batch_size:
                    outfile.writelines(batch)
                    batch = []
            
            # 寫入剩余行
            if batch:
                outfile.writelines(batch)
                
    except FileNotFoundError:
        print(f"錯(cuò)誤: 文件 {input_file} 不存在")
    except PermissionError:
        print(f"錯(cuò)誤: 沒有權(quán)限訪問文件")
    except Exception as e:
        print(f"處理過程中發(fā)生錯(cuò)誤: {e}")
    
    print("處理完成")

# 使用示例
def example_processor(line):
    """示例處理函數(shù): 轉(zhuǎn)換為大寫并添加行號(hào)"""
    return f"PROCESSED: {line.upper()}"

professional_text_processor('input.txt', 'output.txt', example_processor)