一文详解Python如何处理无法解码文件名

import sys
import locale

def analyze_encoding_environment():
    """分析当前环境的编码设置"""
    print("=== 编码环境分析 ===")
    print(f"文件系统编码: {sys.getfilesystemencoding()}")
    print(f"默认编码: {sys.getdefaultencoding()}")
    print(f"区域设置编码: {locale.getpreferredencoding()}")
    
    # 检查常见的问题场景
    test_cases = [
        # (描述, 字节序列, 可能编码)
        ("中文文件名", "中文文件.txt".encode('gbk'), 'GBK'),
        ("日文文件名", "日本語.txt".encode('shift_jis'), 'Shift-JIS'),
        ("特殊字符", "café.txt".encode('iso-8859-1'), 'ISO-8859-1'),
        ("错误编码", b'\xff\xfe\x00\x00invalid', '错误字节序列'),
    ]
    
    print("\n=== 常见问题场景 ===")
    for desc, bytes_data, encoding in test_cases:
        try:
            decoded = bytes_data.decode(sys.getfilesystemencoding())
            status = "可解码"
        except UnicodeDecodeError:
            status = "无法解码"
        print(f"{desc}: {status} (可能编码: {encoding})")

# 运行分析
analyze_encoding_environment()

def demonstrate_error_handlers(problematic_bytes):
    """
    演示不同的错误处理方式
    """
    error_handlers = [
        ('strict', "严格模式（默认）"),
        ('ignore', "忽略错误字节"),
        ('replace', "替换错误字符"),
        ('backslashreplace', "使用反斜杠转义"),
        ('surrogateescape', "代理转义（特殊用途）"),
    ]
    
    print("=== 错误处理策略比较 ===")
    print(f"原始字节: {problematic_bytes}")
    print(f"字节长度: {len(problematic_bytes)}")
    
    for handler, description in error_handlers:
        try:
            result = problematic_bytes.decode('utf-8', errors=handler)
            print(f"{handler:15} {description:20} → '{result}'")
        except Exception as e:
            print(f"{handler:15} {description:20} → 错误: {e}")

# 测试用例
test_bytes = b'valid_part_\xff\xfe_invalid_part'  # 混合有效和无效字节
demonstrate_error_handlers(test_bytes)

def safe_filename_display(filename):
    """
    安全地显示可能无法解码的文件名
    """
    if isinstance(filename, bytes):
        # 字节文件名，需要谨慎处理
        try:
            # 首先尝试系统编码
            decoded = filename.decode(sys.getfilesystemencoding())
            return decoded
        except UnicodeDecodeError:
            # 系统编码失败，尝试常见编码
            for encoding in ['utf-8', 'gbk', 'shift_jis', 'iso-8859-1']:
                try:
                    decoded = filename.decode(encoding)
                    return f"{decoded} (检测编码: {encoding})"
                except UnicodeDecodeError:
                    continue
            
            # 所有编码都失败，使用安全表示
            hex_representation = filename.hex()
            if len(hex_representation) > 20:
                hex_representation = hex_representation[:20] + "..."
            return f"<无法解码: {hex_representation}>"
    
    else:
        # 已经是字符串，直接返回
        return str(filename)

# 使用示例
test_filenames = [
    "正常文件.txt",
    "中文文件.txt".encode('gbk'),
    "日本語.txt".encode('shift_jis'),
    b'invalid_\xff\xfe_bytes.txt'
]

print("=== 安全文件名显示 ===")
for filename in test_filenames:
    display = safe_filename_display(filename)
    print(f"原始: {filename!r} → 显示: {display}")

def demonstrate_surrogateescape():
    """
    演示surrogateescape错误处理程序的使用
    """
    # 创建包含无效UTF-8字节的文件名
    original_bytes = b'file_with_\xff\xfe_invalid_bytes.txt'
    
    print("=== surrogateescape 演示 ===")
    print(f"原始字节: {original_bytes}")
    
    # 使用surrogateescape解码
    decoded_with_escape = original_bytes.decode('utf-8', errors='surrogateescape')
    print(f"解码后: {decoded_with_escape!r}")
    print(f"解码长度: {len(decoded_with_escape)}")
    
    # 检查代理字符
    for i, char in enumerate(decoded_with_escape):
        if '\udc00' <= char <= '\udcff':
            print(f"位置 {i}: 代理字符 {ord(char):04x}")
    
    # 重新编码恢复原始字节
    try:
        reencoded = decoded_with_escape.encode('utf-8', errors='surrogateescape')
        print(f"重新编码: {reencoded}")
        print(f"匹配原始: {reencoded == original_bytes}")
    except Exception as e:
        print(f"重新编码错误: {e}")

# 运行演示
demonstrate_surrogateescape()

import chardet
from pathlib import Path

class SmartFilenameDecoder:
    """
    智能文件名解码器，结合多种策略
    """
    
    def __init__(self):
        self.common_encodings = [
            'utf-8', 'gbk', 'gb2312', 'shift_jis', 
            'euc-jp', 'iso-8859-1', 'windows-1252'
        ]
    
    def decode_with_fallback(self, byte_filename):
        """
        使用多种策略尝试解码文件名
        """
        # 策略1: 尝试系统默认编码
        try:
            return byte_filename.decode(sys.getfilesystemencoding()), 'system'
        except UnicodeDecodeError:
            pass
        
        # 策略2: 尝试常见编码
        for encoding in self.common_encodings:
            try:
                decoded = byte_filename.decode(encoding)
                # 简单验证：检查是否包含可打印字符
                if any(c.isprintable() for c in decoded):
                    return decoded, encoding
            except UnicodeDecodeError:
                continue
        
        # 策略3: 使用chardet自动检测
        try:
            detection = chardet.detect(byte_filename)
            if detection['confidence'] > 0.6:
                decoded = byte_filename.decode(detection['encoding'])
                return decoded, f"detected:{detection['encoding']}"
        except:
            pass
        
        # 策略4: 使用替代表示
        hex_repr = byte_filename.hex()
        if len(hex_repr) > 30:
            hex_repr = hex_repr[:30] + "..."
        return f"<undecodable:{hex_repr}>", 'hex'

    def safe_list_directory(self, directory_path):
        """
        安全地列出目录内容，处理编码问题
        """
        directory = Path(directory_path)
        results = []
        
        try:
            with directory.open('rb') as dir_fd:
                # 使用底层接口获取原始文件名
                for entry_bytes in os.listdir(dir_fd):
                    decoded, method = self.decode_with_fallback(entry_bytes)
                    results.append((decoded, method, entry_bytes))
        except Exception as e:
            print(f"列出目录错误: {e}")
            return []
        
        return results

# 使用示例
decoder = SmartFilenameDecoder()
test_dir = "/tmp"  # 替换为测试目录

print("=== 智能目录列表 ===")
entries = decoder.safe_list_directory(test_dir)
for name, method, raw_bytes in entries[:10]:  # 显示前10个
    print(f"{name:40} [方法: {method:15}] 原始: {raw_bytes!r}")

class SafeFileManager:
    """
    安全的文件管理器，处理编码问题
    """
    
    def __init__(self):
        self.decoder = SmartFilenameDecoder()
    
    def display_directory_tree(self, root_path, max_depth=3):
        """
        安全地显示目录树结构
        """
        root = Path(root_path)
        
        def _display_tree(current_path, current_depth=0, prefix=""):
            if current_depth > max_depth:
                return
                
            try:
                # 安全获取目录内容
                with current_path.open('rb') as dir_fd:
                    entries = []
                    for entry_bytes in os.listdir(dir_fd):
                        decoded, method, _ = self.decoder.decode_with_fallback(entry_bytes)
                        entries.append((decoded, current_path / entry_bytes))
                    
                    # 排序：目录在前，文件在后
                    entries.sort(key=lambda x: (not x[1].is_dir(), x[0].lower()))
                    
                    for i, (display_name, full_path) in enumerate(entries):
                        is_last = i == len(entries) - 1
                        
                        # 当前行的前缀
                        current_prefix = prefix + ("└── " if is_last else "├── ")
                        
                        if full_path.is_dir():
                            # 目录显示
                            print(f"{current_prefix}{display_name}/")
                            # 递归显示子目录
                            new_prefix = prefix + ("    " if is_last else "│   ")
                            _display_tree(full_path, current_depth + 1, new_prefix)
                        else:
                            # 文件显示
                            try:
                                size = full_path.stat().st_size
                                size_str = f" ({size} bytes)"
                            except:
                                size_str = " (无法获取大小)"
                            print(f"{current_prefix}{display_name}{size_str}")
                            
            except PermissionError:
                print(f"{prefix}└── [权限拒绝]")
            except Exception as e:
                print(f"{prefix}└── [错误: {e}]")
        
        print(f"{root_path}/")
        _display_tree(root)
    
    def safe_file_operation(self, operation_func, *args):
        """
        安全的文件操作包装器
        """
        try:
            return operation_func(*args)
        except UnicodeError as e:
            print(f"编码错误: {e}")
            # 尝试使用字节路径重试
            byte_args = []
            for arg in args:
                if isinstance(arg, (str, Path)):
                    try:
                        byte_args.append(str(arg).encode(sys.getfilesystemencoding()))
                    except UnicodeEncodeError:
                        byte_args.append(str(arg).encode('utf-8', errors='replace'))
                else:
                    byte_args.append(arg)
            
            try:
                return operation_func(*byte_args)
            except Exception as retry_error:
                print(f"重试失败: {retry_error}")
                raise

# 使用示例
file_manager = SafeFileManager()
file_manager.display_directory_tree("/path/to/directory")  # 替换为实际路径

import logging
from datetime import datetime

class EncodingAwareLogger:
    """
    支持编码问题文件名的日志系统
    """
    
    def __init__(self, log_file=None):
        self.decoder = SmartFilenameDecoder()
        
        # 配置日志
        self.logger = logging.getLogger(__name__)
        self.logger.setLevel(logging.INFO)
        
        # 清除现有处理器
        self.logger.handlers.clear()
        
        # 控制台输出
        console_handler = logging.StreamHandler()
        console_formatter = logging.Formatter('%(levelname)s: %(message)s')
        console_handler.setFormatter(console_formatter)
        self.logger.addHandler(console_handler)
        
        # 文件输出（如果提供）
        if log_file:
            try:
                file_handler = logging.FileHandler(log_file, encoding='utf-8')
                file_formatter = logging.Formatter(
                    '%(asctime)s - %(levelname)s - %(message)s'
                )
                file_handler.setFormatter(file_formatter)
                self.logger.addHandler(file_handler)
            except Exception as e:
                self.logger.error(f"无法创建日志文件: {e}")
    
    def log_file_operation(self, operation, path, success=True, additional_info=None):
        """
        记录文件操作日志，安全处理文件名
        """
        # 安全处理路径显示
        if isinstance(path, bytes):
            display_path, method = self.decoder.decode_with_fallback(path)
            encoding_info = f" [编码: {method}]"
        else:
            display_path = str(path)
            encoding_info = ""
        
        # 构建日志消息
        status = "成功" if success else "失败"
        message = f"文件{operation} {status}: {display_path}{encoding_info}"
        
        if additional_info:
            message += f" | {additional_info}"
        
        if success:
            self.logger.info(message)
        else:
            self.logger.warning(message)
    
    def log_directory_scan(self, directory_path, file_count, error_count):
        """
        记录目录扫描结果
        """
        display_path = safe_filename_display(directory_path)
        self.logger.info(
            f"目录扫描完成: {display_path} | "
            f"文件: {file_count} | 错误: {error_count}"
        )

# 使用示例
def demo_logging():
    """日志系统演示"""
    logger = EncodingAwareLogger("file_operations.log")
    
    # 模拟各种文件操作
    test_operations = [
        ("读取", "正常文件.txt", True),
        ("写入", "中文文件.txt".encode('gbk'), True),
        ("删除", b'invalid_\xff\xfe_file.txt', False),
    ]
    
    for operation, path, success in test_operations:
        logger.log_file_operation(operation, path, success)
    
    logger.log_directory_scan("/tmp", 150, 3)

demo_logging()

import os
import errno

class LowLevelFileHandler:
    """
    底层文件系统接口，绕过Python的文件名解码
    """
    
    def __init__(self):
        self.decoder = SmartFilenameDecoder()
    
    def raw_list_directory(self, directory_path):
        """
        使用原始文件描述符列出目录
        """
        try:
            # 获取目录的文件描述符
            dir_fd = os.open(directory_path, os.O_RDONLY)
            
            entries = []
            try:
                # 读取目录内容（原始字节）
                with os.fdopen(dir_fd, 'rb') as f:
                    # 注意：这种方法在不同系统上可能表现不同
                    raw_entries = f.read()
                    
                    # 简单的目录条目解析（简化版）
                    # 实际实现需要处理系统特定的目录格式
                    pointer = 0
                    while pointer < len(raw_entries):
                        # 尝试解析目录条目（这是一个简化示例）
                        # 实际实现需要处理具体的文件系统格式
                        try:
                            # 假设条目以null字节结尾
                            null_pos = raw_entries.find(b'\x00', pointer)
                            if null_pos == -1:
                                break
                            
                            entry_bytes = raw_entries[pointer:null_pos]
                            if entry_bytes:  # 非空条目
                                entries.append(entry_bytes)
                            
                            pointer = null_pos + 1
                        except:
                            break
                
            except Exception as e:
                os.close(dir_fd)
                raise e
                
            return entries
            
        except OSError as e:
            if e.errno == errno.EACCES:
                print("权限拒绝")
            elif e.errno == errno.ENOENT:
                print("目录不存在")
            else:
                print(f"系统错误: {e}")
            return []
    
    def safe_file_operations(self, directory_path):
        """
        安全的文件操作演示
        """
        print(f"=== 底层目录列表: {directory_path} ===")
        
        # 获取原始目录条目
        raw_entries = self.raw_list_directory(directory_path)
        
        for entry_bytes in raw_entries:
            try:
                # 尝试解码显示
                display_name, method = self.decoder.decode_with_fallback(entry_bytes)
                print(f"条目: {display_name} [方法: {method}]")
                
                # 可以在这里进行文件操作，使用entry_bytes作为路径
                # 例如：stat操作
                try:
                    full_path = os.path.join(directory_path, entry_bytes)
                    stat_info = os.stat(full_path)
                    print(f"  大小: {stat_info.st_size} bytes")
                except OSError as e:
                    print(f"  无法获取信息: {e}")
                    
            except Exception as e:
                print(f"处理条目失败: {e}")
                print(f"原始字节: {entry_bytes.hex()}")

# 使用示例（注意：需要测试环境）
# handler = LowLevelFileHandler()
# handler.safe_file_operations("/tmp")

def cross_platform_low_level():
    """
    跨平台兼容的底层文件访问策略
    """
    strategies = []
    
    # 策略1: 使用os.scandir的字节能力（Python 3.5+）
    if hasattr(os, 'scandir'):
        def strategy_scandir(path):
            entries = []
            try:
                with os.scandir(path) as it:
                    for entry in it:
                        # 获取原始名称（如果可用）
                        if hasattr(entry, 'name_bytes'):
                            entries.append(entry.name_bytes)
                        else:
                            # 回退到字符串编码
                            entries.append(entry.name.encode(sys.getfilesystemencoding()))
            except Exception as e:
                print(f"scandir错误: {e}")
            return entries
        strategies.append(('os.scandir', strategy_scandir))
    
    # 策略2: 使用os.listdir的字节形式
    def strategy_listdir_bytes(path):
        try:
            dir_fd = os.open(path, os.O_RDONLY)
            try:
                return os.listdir(dir_fd)
            finally:
                os.close(dir_fd)
        except Exception as e:
            print(f"listdir字节错误: {e}")
            return []
    strategies.append(('os.listdir(bytes)', strategy_listdir_bytes))
    
    # 策略3: 使用pathlib的原始接口
    def strategy_pathlib_raw(path):
        try:
            path_obj = Path(path)
            entries = []
            for entry in path_obj.iterdir():
                # 尝试获取原始名称
                try:
                    entries.append(str(entry.name).encode(sys.getfilesystemencoding()))
                except:
                    entries.append(str(entry.name).encode('utf-8', errors='replace'))
            return entries
        except Exception as e:
            print(f"pathlib错误: {e}")
            return []
    strategies.append(('pathlib', strategy_pathlib_raw))
    
    return strategies

# 策略测试
def test_strategies(directory_path):
    """
    测试不同的底层访问策略
    """
    print("=== 底层访问策略比较 ===")
    
    strategies = cross_platform_low_level()
    results = {}
    
    for name, strategy in strategies:
        try:
            entries = strategy(directory_path)
            results[name] = {
                'count': len(entries),
                'success': True,
                'sample': entries[:3] if entries else []
            }
        except Exception as e:
            results[name] = {
                'count': 0,
                'success': False,
                'error': str(e)
            }
    
    # 显示结果
    for name, result in results.items():
        status = "成功" if result['success'] else f"失败: {result['error']}"
        print(f"{name:20}: {status} | 条目数: {result['count']}")
        if result['success'] and result['sample']:
            print(f"  样例: {result['sample'][:3]}")

# 测试（需要实际目录路径）
# test_strategies("/tmp")

class ProductionReadyFilenameHandler:
    """
    生产环境使用的文件名处理器
    """
    
    def __init__(self, config=None):
        self.config = config or {
            'default_encoding': sys.getfilesystemencoding(),
            'fallback_encodings': ['utf-8', 'gbk', 'shift_jis', 'iso-8859-1'],
            'max_hex_display': 20,
            'enable_caching': True,
            'cache_size': 1000
        }
        
        self.decoder = SmartFilenameDecoder()
        self.cache = {} if self.config['enable_caching'] else None
    
    def safe_display(self, filename, context="display"):
        """
        安全地显示文件名，根据上下文调整策略
        """
        cache_key = None
        if self.cache is not None and isinstance(filename, bytes):
            cache_key = filename.hex()
            if cache_key in self.cache:
                return self.cache[cache_key]
        
        if isinstance(filename, bytes):
            # 字节文件名处理
            try:
                # 首先尝试默认编码
                decoded = filename.decode(self.config['default_encoding'])
                result = decoded
            except UnicodeDecodeError:
                # 尝试备选编码
                for encoding in self.config['fallback_encodings']:
                    try:
                        decoded = filename.decode(encoding)
                        # 基本验证
                        if any(c.isprintable() for c in decoded):
                            result = f"{decoded} ({encoding})"
                            break
                    except UnicodeDecodeError:
                        continue
                else:
                    # 所有编码都失败
                    hex_str = filename.hex()
                    if len(hex_str) > self.config['max_hex_display']:
                        hex_str = hex_str[:self.config['max_hex_display']] + "..."
                    
                    if context == "debug":
                        result = f"<undecodable: {hex_str}>"
                    else:
                        result = f"<invalid_filename>"
        else:
            # 字符串文件名
            result = str(filename)
        
        # 缓存结果
        if cache_key is not None:
            if len(self.cache) >= self.config['cache_size']:
                self.cache.clear()  # 简单的缓存清理策略
            self.cache[cache_key] = result
        
        return result
    
    def batch_process(self, file_list, processor_func):
        """
        批量处理文件列表，带有错误处理
        """
        results = []
        errors = []
        
        for file_path in file_list:
            try:
                # 安全显示用于日志
                display_name = self.safe_display(file_path, "log")
                
                # 处理文件
                result = processor_func(file_path)
                results.append((file_path, result, None))
                
            except UnicodeError as e:
                # 编码错误
                error_info = {
                    'type': 'encoding_error',
                    'message': str(e),
                    'bytes': file_path.hex() if isinstance(file_path, bytes) else None
                }
                errors.append((file_path, None, error_info))
                
            except OSError as e:
                # 文件系统错误
                error_info = {
                    'type': 'os_error',
                    'errno': e.errno,
                    'message': str(e)
                }
                errors.append((file_path, None, error_info))
                
            except Exception as e:
                # 其他错误
                error_info = {
                    'type': 'other_error',
                    'message': str(e)
                }
                errors.append((file_path, None, error_info))
        
        return {
            'successful': results,
            'failed': errors,
            'success_rate': len(results) / len(file_list) if file_list else 0
        }

# 使用示例
def demo_production_handler():
    """生产环境处理器演示"""
    handler = ProductionReadyFilenameHandler()
    
    # 测试文件列表
    test_files = [
        "normal_file.txt",
        "中文文件.txt".encode('gbk'),
        "日本語.txt".encode('shift_jis'),
        b'invalid_\xff\xfe_bytes.txt',
        "another_normal.py"
    ]
    
    # 简单的处理器函数
    def simple_processor(filepath):
        return f"processed_{safe_filename_display(filepath)}"
    
    # 批量处理
    results = handler.batch_process(test_files, simple_processor)
    
    print("=== 批量处理结果 ===")
    print(f"成功率: {results['success_rate']:.1%}")
    print(f"成功: {len(results['successful'])}")
    print(f"失败: {len(results['failed'])}")
    
    print("\n失败详情:")
    for file_path, result, error in results['failed']:
        display = handler.safe_display(file_path)
        print(f"  {display}: {error['type']} - {error['message']}")

demo_production_handler()