如何检测归档文件（ZIP/RAR/7z）是否损坏-Python教程-PHP中文网

如何检测归档文件（ZIP/RAR/7z）是否损坏

本文介绍如何使用 python 批量、可靠地检测 zip、rar 和 7z 归档文件的完整性，涵盖各格式对应的异常类型、统一验证策略、异步高效处理方案，并提供可直接运行的健壮检测代码。

本文介绍如何使用 python 批量、可靠地检测 zip、rar 和 7z 归档文件的完整性，涵盖各格式对应的异常类型、统一验证策略、异步高效处理方案，并提供可直接运行的健壮检测代码。

在大规模下载场景（如从 Nexus Mods 批量获取模组）中，网络干扰（如 GFW 干扰）、服务降级（如 NexusMods 高负载导致连接中断）、低速限流（非会员 1 MiB/s）等因素极易导致归档文件部分写入或数据损坏。仅校验文件头魔数（如 PK\x03\x04）远远不够——它只能确认文件“看起来像” ZIP，无法保证内部结构完整、压缩流可解码、目录表无损。真正可靠的损坏检测必须触发实际解析与校验逻辑。

Python 生态中主流库对归档损坏的响应方式不同，需统一兜底处理：

zipfile.ZipFile：调用 .testzip() 时，若发现损坏条目会返回首个错误文件名（str），否则返回 None；但更常见的是直接抛出异常，如：
- zipfile.BadZipFile：魔数错误或全局结构损坏（如中央目录偏移无效）；
- OSError：底层 I/O 错误（如截断文件）；
- zlib.error 或 lzma.LZMAError：压缩数据流损坏（如 invalid distance too far back）；
rarfile.RarFile：.testrar() 在检测到损坏时通常抛出 rarfile.BadRarFile（子类为 rarfile.Error），消息含 Corrupt file 或 read failed 等关键词；
py7zr.SevenZipFile：初始化时即可能因头部损坏抛出 py7zr.Bad7zFile；调用 .test() 或 .testzip() 则可能触发 lzma.LZMAError、zstandard.ZstdError（若含 ZSTD 压缩）等底层解压异常。

因此，安全的检测逻辑 = 显式调用测试方法 + 捕获所有与归档解析相关的异常 + 判定返回值是否为真值。不应依赖单一异常类型，而应采用“宽捕获 + 日志 + 标记”策略。

Regie.ai

一款面向销售团队的 AI 销售内容与沟通平台，帮助企业生成个性化销售文案、客户触达内容并辅助开展规模化外呼和拓客工作。

下载

以下为生产就绪的批量检测实现，兼顾健壮性、性能与可维护性：

import asyncio
import concurrent.futures
from pathlib import Path
from zipfile import ZipFile
from rarfile import RarFile
from py7zr import SevenZipFile

# 定义归档类型与对应验证函数
def test_zip(path: str) -> bool:
    try:
        with ZipFile(path) as zf:
            return zf.testzip() is not None  # 返回文件名表示损坏，None 表示正常
    except Exception:
        return True  # 任何异常均视为损坏

def test_rar(path: str) -> bool:
    try:
        with RarFile(path) as rf:
            rf.testrar()  # 成功无返回，失败抛异常
        return False
    except Exception:
        return True

def test_7z(path: str) -> bool:
    try:
        with SevenZipFile(path) as szf:
            szf.test()  # 验证所有条目
        return False
    except Exception:
        return True

ARCHIVE_HANDLERS = {
    '.zip': test_zip,
    '.dazip': test_zip,  # Nexus 常见重命名 ZIP
    '.rar': test_rar,
    '.7z': test_7z,
}

# 全局损坏文件集合（线程安全需注意，此处用于演示）
BAD_FILES = set()

def verify_archive(file_path: Path) -> None:
    """单文件验证主逻辑"""
    ext = file_path.suffix.lower()
    if ext not in ARCHIVE_HANDLERS:
        return

    try:
        if ARCHIVE_HANDLERS[ext](str(file_path)):
            BAD_FILES.add(str(file_path))
            print(f"❌ CORRUPT: {file_path}")
        else:
            print(f"✅ OK: {file_path}")
    except Exception as e:
        BAD_FILES.add(str(file_path))
        print(f"❌ ERROR (treated as corrupt): {file_path} — {type(e).__name__}: {e}")

async def scan_directory_async(root: Path, executor: concurrent.futures.Executor) -> None:
    """异步遍历目录，提交验证任务"""
    loop = asyncio.get_running_loop()

    # 收集所有归档文件
    files = []
    for ext in ARCHIVE_HANDLERS:
        files.extend(root.rglob(f"*{ext}"))

    # 并行执行验证（利用 ThreadPoolExecutor 处理 CPU/IO 密集型阻塞操作）
    tasks = [
        loop.run_in_executor(executor, verify_archive, f)
        for f in files
    ]
    await asyncio.gather(*tasks)

# 使用示例
if __name__ == "__main__":
    download_root = Path("d:/downloads/mods")

    # 使用线程池避免阻塞事件循环（推荐 4–8 工作线程，依 CPU 核心数调整）
    with concurrent.futures.ThreadPoolExecutor(max_workers=6) as executor:
        asyncio.run(scan_directory_async(download_root, executor))

    print(f"\n? 扫描完成。共发现 {len(BAD_FILES)} 个损坏文件：")
    for bad in sorted(BAD_FILES):
        print(f"  • {bad}")

关键设计说明：

✅ 异常兜底而非穷举：不尝试枚举所有可能异常（BadZipFile, BadRarFile, Bad7zFile, LZMAError, zlib.error…），而是捕获 Exception 并统一标记为损坏——这符合“损坏即不可用”的工程原则，且避免遗漏新型异常。
✅ 轻量验证，不提取：仅执行 testzip()/.testrar()/.test()，不实际解压，内存占用低、速度快。
✅ 异步+线程混合架构：asyncio 负责目录遍历与任务调度，ThreadPoolExecutor 承载阻塞的归档解析，兼顾并发性与资源效率。
✅ 路径与扩展名健壮处理：支持 .dazip 等伪装 ZIP，忽略大小写，递归扫描子目录。
⚠️ 注意事项：
- 分卷归档（如 hdr.2.rar, hdr.3.rar）需单独处理——当前代码将其视为独立损坏文件（因缺少首卷无法验证），实际应用中应先聚合分卷再校验；
- 若需更高精度（如定位具体损坏文件），可捕获并解析异常消息，但多数场景“整体损坏”已足够触发重新下载；
- 对于超大归档（>1GB），.test() 可能较慢，可考虑添加超时机制（如 concurrent.futures.TimeoutError）。