Python动态处理文件编码的完整指南

import io
 
def demonstrate_io_layers():
    """
    演示Python的IO层次结构
    """
    # 创建一个示例文件
    with open('test_file.txt', 'w', encoding='utf-8') as f:
        f.write('Hello, 世界!')
    
    # 不同层次的打开方式
    print("=== Python IO层次结构演示 ===")
    
    # 1. 二进制层 - 最底层
    with open('test_file.txt', 'rb') as bin_file:
        print(f"二进制层: {type(bin_file)}")
        raw_data = bin_file.read()
        print(f"原始字节: {raw_data}")
    
    # 2. 文本层 - 带编码的文本处理
    with open('test_file.txt', 'r', encoding='utf-8') as text_file:
        print(f"文本层: {type(text_file)}")
        text_data = text_file.read()
        print(f"解码文本: {text_data}")
    
    # 3. 缓冲层 - 自动处理的缓冲IO
    with io.open('test_file.txt', 'r', encoding='utf-8') as buffered_file:
        print(f"缓冲IO层: {type(buffered_file)}")
    
    # 清理
    import os
    os.remove('test_file.txt')
 
# 运行演示
demonstrate_io_layers()

def common_encoding_scenarIOS():
    """
    常见的文件编码问题场景
    """
    scenarios = [
        {
            'name': 'UTF-8文件无BOM',
            'content': 'Hello, 世界!',
            'encoding': 'utf-8',
            'bom': False
        },
        {
            'name': 'UTF-8文件带BOM',
            'content': 'Hello, 世界!',
            'encoding': 'utf-8-sig',
            'bom': True
        },
        {
            'name': 'GBK中文文件',
            'content': '你好，世界!',
            'encoding': 'gbk',
            'bom': False
        },
        {
            'name': 'Shift-JIS日文文件',
            'content': 'こんにちは、世界!',
            'encoding': 'shift_jis',
            'bom': False
        },
        {
            'name': '混合编码问题',
            'content': 'Hello, 世界!',
            'encoding': 'iso-8859-1',  # 错误的编码
            'bom': False
        }
    ]
    
    print("=== 常见编码场景 ===")
    for scenario in scenarios:
        # 创建测试文件
        filename = f"test_{scenario['name']}.txt"
        with open(filename, 'w', encoding=scenario['encoding']) as f:
            if scenario['bom']:
                # 写入BOM（如果适用）
                f.write('\ufeff')
            f.write(scenario['content'])
        
        # 尝试用不同编码读取
        try:
            with open(filename, 'r', encoding='utf-8') as f:
                content = f.read()
            status = 'UTF-8读取成功'
        except UnicodeDecodeError:
            status = 'UTF-8读取失败'
        
        print(f"{scenario['name']:20} {scenario['encoding']:12} -> {status}")
        
        # 清理
        import os
        os.remove(filename)
 
common_encoding_scenarios()

import io
 
def demonstrate_text_iowrapper():
    """
    演示使用io.TextIOWrapper动态修改编码
    """
    # 创建测试文件
    with open('demo_file.txt', 'w', encoding='gbk') as f:
        f.write('中文内容测试')
    
    print("=== io.TextIOWrapper 演示 ===")
    
    # 1. 以二进制模式打开文件
    with open('demo_file.txt', 'rb') as binary_file:
        print(f"二进制文件对象: {type(binary_file)}")
        
        # 2. 使用TextIOWrapper添加编码
        text_wrapper = io.TextIOWrapper(
            binary_file,
            encoding='gbk',  # 正确编码
            line_buffering=True
        )
        
        print(f"包装后文本对象: {type(text_wrapper)}")
        content = text_wrapper.read()
        print(f"读取内容: {content}")
        
        # 重要：使用后分离包装器，避免重复关闭
        text_wrapper.detach()
    
    # 3. 动态重新编码示例
    with open('demo_file.txt', 'rb') as binary_file:
        # 第一次用GBK读取
        wrapper_gbk = io.TextIOWrapper(binary_file, encoding='gbk')
        content_gbk = wrapper_gbk.read()
        print(f"GBK读取: {content_gbk}")
        
        # 分离后重新包装
        wrapper_gbk.detach()
        binary_file.seek(0)  # 重置文件指针
        
        # 用UTF-8重新包装（虽然内容不对，但演示功能）
        wrapper_utf8 = io.TextIOWrapper(binary_file, encoding='utf-8')
        try:
            content_utf8 = wrapper_utf8.read()
            print(f"UTF-8读取: {content_utf8}")
        except UnicodeDecodeError as e:
            print(f"UTF-8读取失败: {e}")
        finally:
            wrapper_utf8.detach()
    
    # 清理
    import os
    os.remove('demo_file.txt')
 
demonstrate_text_iowrapper()

import chardet
from pathlib import Path
 
class DynamicEncodingAdapter:
    """
    动态编码检测与适配器
    """
    
    def __init__(self):
        self.common_encodings = [
            'utf-8', 'gbk', 'gb2312', 'shift_jis',
            'euc-jp', 'iso-8859-1', 'Windows-1252'
        ]
    
    def detect_encoding(self, file_path, sample_size=1024):
        """
        检测文件编码
        """
        with open(file_path, 'rb') as f:
            # 读取样本数据
            raw_data = f.read(sample_size)
            
            # 使用chardet检测
            detection = chardet.detect(raw_data)
            
            # 检查BOM（字节顺序标记）
            bom_encoding = self._check_bom(raw_data)
            if bom_encoding:
                return bom_encoding, True
            
            if detection['confidence'] > 0.7:
                return detection['encoding'], False
            
            # 尝试常见编码
            for encoding in self.common_encodings:
                try:
                    raw_data.decode(encoding)
                    return encoding, False
                except UnicodeDecodeError:
                    continue
            
            return 'utf-8', False  # 默认回退
    
    def _check_bom(self, data):
        """
        检查BOM标记
        """
        bom_signatures = {
            b'\xff\xfe': 'utf-16-le',
            b'\xfe\xff': 'utf-16-be',
            b'\xff\xfe\x00\x00': 'utf-32-le',
            b'\x00\x00\xfe编程xff': 'utf-32-be',
            b'\xef\xbb\xbf': 'utf-8-sig'
        }
        
        for signature, encoding in bom_signatures.items():
            if data.startswith(signature):
                return encoding
        
        return None
    
    def open_with_detected_encoding(self, file_path):
        """
        使用检测到的编码打开文件
        """
        encoding, has_bom = self.detect_encoding(file_path)
        print(f"检测到编码: {encoding} (BOM: {has_bom})")
        
        # 以二进制打开，然后动态包装
        binary_file = open(file_path, 'rb')
        
        # 跳过BOM（如果存在）
        if has_bom:
            bom_size = len(self._get_bom_bytes(encoding))
            binary_file.seek(bom_size)
        
        # 创建TextIOWrapper
        text_file = io.TextIOWrapper(
            binary_file,
            encoding=encoding,
            errors='replace'  # 替换无法解码的字符
        )
        
        return text_file
    
    def _get_bom_bytes(self, encoding):
        """
        获取编码对应的BOM字节
        """
        bom_map = {
            'utf-8-sig': b'\xef\xbb\xbf',
            'utf-16-le': b'\xff\xfe',
            'utf-16-be': b'\xfe\xff',
            'utf-32-le': b'\xff\xfe\x00\x00',
            'utf-32-be': b'\x00\x00\xfe\xff'
        }
        return bom_map.get(encoding, b'')
 
# 使用示例
def demo_dynamic_encoding():
    """动态编码演示"""
    adapter = DynamicEncodingAdapter()
    
    # 创建不同编码的测试文件
    test_files = [
        ('utf-8_file.txt', 'UTF-8内容', 'utf-8'),
        ('gbk_file.txt', 'GBK中文内容', 'gbk'),
    ]
    
    for filename, content, encoding in test_files:
        with open(filename, 'w', encoding=encoding) as f:
            f.write(content)
    
    # 动态检测和打开
    for filename, expected_content, expected_encoding in test_files:
        print(f"\n处理文件: {filename}")
        
        try:
            with adapter.open_with_detected_encoding(filename) as f:
                detected_content = f.read()
                print(f"预期: {expected_content}")
                print(f"读取: {detected_content}")
                print(f"匹配: {detected_content == expected_content}")
        except Exception as e:
            print(f"错误: {e}")
        
        # 清理
        import os
        os.remove(filename)
 
demo_dynamic_encoding()

class RealtimeTranscoder:
    """
    实时编码转换器
    """
    
    def __init__(self, source_encoding='auto', target_encoding='utf-8'):
        self.source_encoding = source_encoding
        self.target_encoding = target_encoding
        self.detector = DynamicEncodingAdapter()
    
    def transcode_file(self, source_path, target_path):
        """
        转换文件编码
        """
        # 确定源编码
        if self.source_encoding == 'auto':
            detected_encoding, has_bom = self.detector.detect_encoding(source_path)
            source_encoding = detected_encoding
        else:
            source_encoding = self.source_encoding
        
        print(f"转换: {source_encoding} -> {self.target_encoding}")
        
        # 使用二进制模式打开两个文件
        with open(source_path, 'rb') as src_binary, \
             open(target_path, 'wb') as tgt_binary:
            
            # 为源文件创建文本包装器
            src_text = io.TextIOWrapper(
                src_binary,
                encoding=source_encoding,
                errors='replace'
            )
            
            # 为目标文件创建文本包装器
            tgt_text = io.TextIOWrapper(
                tgt_binary,
                encoding=self.target_encoding,
                errors='replace',
                write_through=True  # 立即写入底层缓冲
            )
            
            # 逐块转换
            buffer_size = 4096
            while True:
                chunk = src_text.read(buffer_size)
                if not chunk:
                    break
                tgt_text.write(chunk)
            
            # 确保所有数据写入
            tgt_text.flush()
            
            # 分离包装器，避免关闭底层文件
            src_text.detach()
            tgt_text.detach()
        
        print(f"转换完成: {target_path}")
    
    def transcode_stream(self, input_stream, output_stream):
        """
        转换流编码
        """
        # 创建临时包装器
        input_wrapper = io.TextIOWrapper(
            input_stream,
            encoding=self.source_encoding,
            errors='replace'
        )
        
        output_wrapper = io.TextIOWrapper(
            output_stream,
            encoding=self.target_encoding,
            errors='replace',
            write_through=True
        )
        
        try:
            # 传输数据
            while True:
                chunk = input_wrapper.read(1024)
                if not chunk:
                    break
                output_wrapper.write(chunk)
            
            output_wrapper.flush()
            
        finally:
            # 分离包装器但不关闭底层流
            input_wrapper.detach()
            output_wrapper.detach()
 
# 使用示例
def demo_transcoding():
    """编码转换演示"""
    transcoder = RealtimeTranscoder('auto', 'utf-8')
    
    # 创建测试文件
    with open('source_gbk.txt', 'w', encoding='gbk') as f:
        f.write('这是GBK编码的中文内容')
    
    # 执行转换
    transcoder.transcode_file('source_gbk.txt', 'target_utf8.txt')
    
    # 验证结果
    with open('target_utf8.txt', 'r', encoding='utf-8') as f:
        content = f.read()
        print(f"转换结果: {content}")
    
    # 清理
    import os
    os.remove('source_gbk.txt')
    os.remove('target_utf8.txt')
 
demo_transcoding()

class MultiEncodingFileProcessor:
    """
    处理可能包含多种编码的文件
    """
    
    def __init__(self):
        self.detector = DynamicEncodingAdapter()
    
    def process_mixed_encoding_file(self, file_path):
        """
        处理可能包含多种编码的文件
        """
        results = {
            'sections': [],
            'encodings_found': set(),
            'errors': []
        }
        
        with open(file_path, 'rb') as binary_file:
            position = 0
            current_encoding = None
            current_buffer = bytearray()
            
            # 逐块分析文件
            while True:
                chunk = binary_file.read(1024)
                if not chunk:
                    break
                
                current_buffer.extend(chunk)
                
                # 尝试检测当前块的编码
                try:
                    detected_encoding, _ = self.detector.detect_encoding_from_bytes(
                        bytes(current_buffer)
                    )
                    
                    if current_encoding != detected_encoding:
                        # 编码变化，处理当前缓冲区
                        if current_encoding and current_buffer:
                            self._process_section(
                                bytes(current_buffer),
                                current_encoding,
                                position,
                                results
                            )
                            position += len(current_buffer)
                            current_buffer = bytearray()
                        
                        current_encoding = detected_encoding
                
                except Exception as e:
                    results['errors'].append(f"位置 {position}: {e}")
                    current_buffer = bytearray()
                    continue
            
            # 处理最后的部分
            if current_buffer and current_encoding:
                self._process_section(
                    bytes(current_buffer),
                    current_encoding,
                    position,
                    results
                )
        
        return results
    
    def _process_section(self, data, encoding, position, results):
        """
        处理文件的一个编码段落
        """
        try:
            decoded = data.decode(encoding, errors='replace')
            results['sections'].append({
                'position': position,
                'length': len(data),
                'encoding': encoding,
                'content': decoded,
                'success': True
            })
            results['encodings_found'].add(encoding)
        except Exception as e:
            results['sections'].append({
                'position': position,
                'length': len(data),
                'encoding': encoding,
                'error': str(e),
                'success': False
            })
            results['errors'].append(f"解码失败 {position}: {e}")
 
    def detect_encoding_from_bytes(self, data):
        """
        从字节数据检测编码
        """
        try:
            detection = chardet.detect(data)
            if detection['confidence'] > 0.5:
                return detection['encoding'], False
            
            # 尝试常见编码
            for encoding in self.common_encodings:
                try:
                    data.decode(encoding)
                    return encoding, False
                except UnicodeDecodeError:
                    continue
            
            return 'utf-8', False
        except:
            return 'utf-8', False
 
# 使用示例
def demo_mixed_processing():
    """混合编码处理演示"""
    processor = MultiEncodingFileProcessor()
    
    # 创建混合编码测试文件
    with open('mixed_encoding.txt', 'wb') as f:
        # UTF-8部分
        f.write('UTF-8部分: Hello, 世界!\n'.encode('utf-8'))
        # GBK部分
        f.write('GBK部分: 中文内容\n'.encode('gbk'))
        # 再回到UTF-8
        f.write('返回UTF-8: 继续内容\n'.encode(php'utf-8'))
    
    # 处理文件
    results = processor.process_mixed_encoding_file('mixed_encoding.txt')
    
    print("=== 混合编码处理结果 ===")
    print(f"找到编码: {results['encodings_found']}")
    print(f"段落数: {len(results['sections'])}")
    print(f"错误数: {len(results['errors'])}")
    
    for i, section in enumerate(results['sections']):
        print(f"\n段落 {i+1}:")
        print(f"  编码: {section['encoding']}")
        print(f"  位置: {section['position']}")
        print(f"  长度: {section['length']}")
        if section['success']:
            print(f"  内容: {section['content'][:50]}...")
        else:
            print(f"  错误: {section['error']}")
    
    # 清理
    import os
    os.remove('mixed_encoding.txt')
 
demo_mixed_processing()

import mmap
import io
 
class MappedFileEncoder:
    """
    使用内存映射高效处理大文件编码
    """
    
    def __init__(self):
        self.detector = DynamicEncodingAdapter()
    
    def process_large_file(self, file_path, target_encoding='utf-8'):
        """
        处理大文件的编码转换
        """
        results = {
            'processed_bytes': 0,
            'converted_chunks': 0,
            javascript'errors': []
        }
        
        with open(file_path, 'r+b') as f:
            # 创建内存映射
            with mmap.mmap(f.fileno(), 0, Access=mmap.ACCESS_READ) as mm:
                # 检测整体编码
                overall_encoding, _ = self.detector.detect_encoding_from_bytes(
                    mm[:min(len(mm), 4096)]
                )
                
                print(f"检测到整体编码: {overall_encoding}")
                
                # 分块处理
                chunk_size = 64 * 1024  # 64KB块
                position = 0
                
                while position < len(mm):
                    # 处理当前块
                    chunk_end = min(position + chunk_size, len(mm))
                    chunk = mm[position:chunk_end]
                    
                    try:
                        # 解码当前块
                        decoded = chunk.decode(overall_encoding, errors='replace')
                        
                        # 转换为目标编码
                        encoded = decoded.encode(target_encoding, errors='replace')
                        
                        results['processed_bytes'] += len(chunk)
                        results['converted_chunks'] += 1
                        
                        # 这里可以处理编码后的数据
                        # 例如写入新文件或进行其他处理
                        
                    except Exception as e:
                        results['errors'].append(f"位置 {position}: {e}")
                    
                    position = chunk_end
        
        return results
    
    def create_mapped_text_wrapper(self, file_path, encoding='utf-8'):
        """
        创建基于内存映射的文本包装器
        """
        # 打开文件并创建内存映射
        file_obj = open(file_path, 'r+b')
        mmapped = mmap.mmap(file_obj.fileno(), 0, access=mmap.ACCESS_READ)
        
        # 创建字节IO包装内存映射
        buffer = io.BytesIO(mmapped)
        
        # 创建文本包装器
        text_wrapper = io.TextIOWrapper(
            buffer,
            encoding=encoding,
            errors='replace'
        )
        
        return {
            'file_obj': file_obj,
            'mmapped': mmapped,
            'buffer': buffer,
            'text_wrapper': text_wrapper
        }
 
# 使用示例
def demo_mapped_processing():
    """内存映射处理演示"""
    encoder = MappedFileEncoder()
    
    # 创建测试大文件
    large_content = "测试内容\n" * 10000
    with open('large_file.txt', 'w', encoding='gbk') as f:
        f.write(large_content)
    
    # 处理文件
    results = encoder.process_large_file('large_file.txt', 'utf-8')
    
    print("=== 内存映射处理结果 ===")
    print(f"处理字节: {results['processed_bytes']}")
    print(f"处理块数: {results['converted_chunks']}")
    print(f"错误数: {len(results['errors'])}")
    
    # 清理
    import os
    os.remove('large_file.txt')
 
demo_mapped_processing()

class OptimizedEncodingProcessor:
    """
    优化的编码处理器
    """
    
    def __init__(self, buffer_size=8192, encoding_cache_size=1000):
        self.buffer_size = buffer_size
        self.encoding_cache = {}
        self.cache_size = encoding_cache_size
        self.detector = DynamicEncodingAdapter()
    
    def optimized_transcode(self, source_path, target_path, 
                          source_encoding=None, target_encoding='utf-8'):
        """
        优化的编码转换
        """
        # 检测源编码（如果未指定）
        if source_encoding is None:
            source_encoding, _ = self.detector.detect_encoding(source_path)
        
        # 使用缓冲策略
        with open(source_path, 'rb', buffering=self.buffer_size) as src, \
             open(target_path, 'wb', buffering=self.buffer_size) as tgt:
            
            # 创建缓冲的文本包装器
            src_text = io.TextIOWrapper(
                src,
                encoding=source_encoding,
                errors='replace',
                line_buffering=False
            )
            
            tgt_text = io.TextIOWrapper(
                tgt,
                encoding=target_encoding,
                errors='replace',
                write_through=True,
                line_buffering=False
            )
            
            # 使用大块传输
            while True:
                chunk = src_text.read(self.buffer_size)
                if not chunk:
                    break
                tgt_text.write(chunk)
            
            # 确保所有数据写入
            tgt_text.flush()
            
            # 分离包装器
            src_textpython.detach()
            tgt_text.detach()
    
    def BATch_process_files(self, file_list, target_encoding='utf-8'):
        """
        批量处理文件
        """
        results = []
        
        for file_path in file_list:
            try:
                # 检查编码缓存
                if file_path in self.encoding_cache:
                    source_encoding = self.encoding_cache[file_path]
                else:
                    source_encoding, _ = self.detector.detect_encoding(file_path)
                    # 更新缓存
                    if len(self.encoding_cache) >= self.cache_size:
                        self.encoding_cache.clear()
                    self.encoding_cache[file_path] = source_encoding
                
                # 处理文件
                temp_path = f"{file_path}.converted"
                self.optimized_transcode(
                    file_path, temp_path, source_encoding, target_encoding
                )
                
                results.append({
                    'file': file_path,
                    'success': True,
                    'source_encoding': source_encoding,
                    'target_encoding': target_encoding
                })
                
                # 这里可以替换原文件或进行其他操作
                
            except Exception as e:
                results.append({
                    'file': file_path,
                    'success': False,
                    'error': str(e)
                })
        
        return results
 
# 使用示例
def demo_optimized_processing():
    """优化处理演示"""
    processor = OptimizedEncodingProcessor()
    
    # 创建测试文件
    test_files = []
    for i in range(3):
        filename = f'test_file_{i}.txt'
        encoding = 'gbk' if i % 2 == 0 else 'utf-8'
        with open(filename, 'w', encoding=encoding) as f:
            f.write(f'测试文件 {i} - 编码: {encoding}')
        test_files.append(filename)
    
    # 批量处理
    results = processor.batch_process_files(test_files)
    
    print("=== 批量处理结果 ===")
    for result in results:
        if result['success']:
            print(f"成功: {result['file']} "
                  f"({result['source_encoding']} -> {result['target_encoding']})")
        else:
            print(f"失败: {result['file']} - {result['error']}")
    
    # 清理
    import os
    for file in test_files:
        if os.path.exists(file):
            os.remove(file)
        temp_file = f"{file}.converted"
        if os.path.exists(temp_file):
            os.remove(temp_file)
 
demo_optimized_processing()

class RobustEncodingProcessor:
    """
    健壮的编码处理框架
    """
    
    def __init__(self):
        self.detector = DynamicEncodingAdapter()
        self.retry_strategies = [
            self._retry_with_different_encoding,
            self._retry_with_error_replacement,
            self._retry_with_byte_preservation
        ]
    
    def safe_read_file(self, file_path, preferred_encoding=None):
        """
        安全读取文件，使用多种恢复策略
        """
        attempts = []
        
        # 尝试1: 首选编码或自动检测
        try:
            if preferred_encoding:
                encoding = preferred_encoding
            else:
                encoding, _ = self.detector.detect_encoding(file_path)
            
            content = self._read_with_encoding(file_path, encoding)
            return {
                'success': True,
                'content': content,
                'encoding': encoding,
                'attempts': attempts
            }
            
        except Exception as first_error:
            attempts.append({
                'strategy': 'primary',
                'encoding': preferred_encoding,
                'error': str(first_error)
            })
        
        # 尝试恢复策略
        for strategy in self.retry_strategies:
            try:
                content, encoding = strategy(file_path)
                attempts.append({
                    'strategy': strategy.__name__,
                    'encoding': encoding,
                    'success': True
                })
                return {
                    'success': True,
                    'content': content,
                    'encoding': encoding,
                    'attempts': attempts
                }
            except Exception as e:
                attempts.append({
                    'strategy': strategy.__name__,
                    'error': str(e)
                })
        
        return {
            'success': False,
            'attempts': attempts,
            'error': '所有恢复策略都失败'
        }
    
    def _read_with_encoding(self, file_path, encoding):
        """使用指定编码读取文件"""
        with open(file_path, 'r', encoding=encoding, errors='strict') as f:
            return f.read()
    
    def _retry_with_different_encoding(self, file_path):
        """尝试不同编码"""
        for encoding in ['utf-8', 'gbk', 'iso-8859-1']:
            try:
                content = self._read_with_encoding(file_path, encoding)
                return content, encoding
            except:
                continue
        raise ValueError("所有备选编码都失败")
    
    def _retry_with_error_replacement(self, file_path):
        """使用错误替换策略"""
        with open(file_path, 'r', encoding='utf-8', errors='replace') as f:
            content = f.read()
        return content, 'utf-8-with-replace'
    
    def _retry_with_byte_preservation(self, file_path):
        """保留原始字节"""
        with open(file_path, 'rb') as f:
            content = f.read()
        return content.hex(), 'hex-encoded'
 
# 使用示例
def demo_robust_processing():
    """健壮处理演示"""
    processor = RobustEncodingProcessor()
    
    # 创建有问题的测试文件
    problematic_content = '正常内容'.encode('utf-8') + b'\xff\xfe' + '后续内容'.encode('utf-8')
    with open('problematic.txt', 'wb') as f:
        f.write(problematic_content)
    
    # 尝试安全读取
    result = processor.safe_read_file('problematic.txt')
    
    print("=== 健壮处理结果 ===")
    print(f"成功: {result['success']}")
    if result['success']:
        print(f"编码: {result['encoding']}")
        print(f"内容预览: {result['content'][:100]}...")
    else:
        print(f"错误: {result['error']}")
    
    print("\n尝试记录:")
    for attempt in result['attempts']:
        if 'success' in attempt:
            print(f"  ✓ {attempt['strategy']} ({attempt['encoding']})")
        else:
            print(f"  ✗ {attempt['strategy']}: {attempt['error']}")
    
    # 清理
    import oChina编程s
    os.remove('problematic.txt')
 
demo_robust_processing()