你手头有几十上百个 PDF,需要统一加个密码,或者把一批带密码的合同批量“脱敏”成无密码版本.手动操作?一个文件点两次,一百个文件就

pip install pypdf[crypto]

from pypdf import PdfReader, PdfWriterreader = PdfReader("合同.pdf")writer = PdfWriter()for page in reader.pages:    writer.add_page(page)writer.encrypt(    user_password="OpenMe2026",    owner_password="AdminOnly2026",    algorithm="AES-256-R5",)with open("合同_加密.pdf", "wb") as f:    writer.write(f)

from pypdf.constants import UserAccessPermissionspermissions = (    UserAccessPermissions.PRINT    | UserAccessPermissions.EXTRACT_TEXT)writer.encrypt(    user_password="OpenMe2026",    owner_password="AdminOnly2026",    algorithm="AES-256-R5",    permissions_flag=permissions,)

from pypdf import PdfReader, PdfWriterreader = PdfReader("合同_加密.pdf")if reader.is_encrypted:    result = reader.decrypt("OpenMe2026")   # 传入用户密码或所有者密码    if not result:        raise ValueError("密码错误，无法解密")writer = PdfWriter()for page in reader.pages:    writer.add_page(page)with open("合同_解密.pdf", "wb") as f:    writer.write(f)

if reader.is_encrypted:    # 先试传入的密码，不行再试空密码（针对仅所有者密码的情况）    if not reader.decrypt(password) and not reader.decrypt(""):        raise ValueError("无法解密")

from pathlib import Pathfrom pypdf import PdfReader, PdfWriterfrom pypdf.constants import UserAccessPermissionsdef process_one(pdf_path: Path, mode: str, password: str, output_dir: Path):    """    mode: "encrypt" 或 "decrypt"    返回 (状态, 消息)，不抛异常，由调用方统一汇总    """    out_name = f"{pdf_path.stem}_{'encrypted'if mode == 'encrypt'else'decrypted'}.pdf"    out_path = output_dir / out_name    try:        reader = PdfReader(pdf_path)        if mode == "encrypt":            if reader.is_encrypted:                return ("skip", "已加密，跳过")            writer = PdfWriter()            for page in reader.pages:                writer.add_page(page)            writer.encrypt(                user_password=password,                owner_password=password,                algorithm="AES-256-R5",                permissions_flag=(                    UserAccessPermissions.PRINT                    | UserAccessPermissions.EXTRACT_TEXT                ),            )        else:  # decrypt            if not reader.is_encrypted:                return ("skip", "未加密，无需解密")            # 先试传入密码，再试空密码（处理仅权限加密）            if not reader.decrypt(password) and not reader.decrypt(""):                return ("fail", "密码错误或加密类型不支持")            writer = PdfWriter()            for page in reader.pages:                writer.add_page(page)        with open(out_path, "wb") as f:            writer.write(f)        return ("ok", str(out_path))    except Exception as e:        return ("fail", f"{type(e).__name__}: {e}")

def run_batch(input_dir: str, mode: str, password: str):    src = Path(input_dir)    out_dir = src / f"_{mode}_output"    out_dir.mkdir(exist_ok=True)    pdfs = sorted(src.rglob("*.pdf"))    if not pdfs:        print("没有找到 PDF 文件")        return    stats = {"ok": 0, "skip": 0, "fail": 0}    failed_files = []    for pdf in pdfs:        status, msg = process_one(pdf, mode, password, out_dir)        stats[status] += 1        icon = {"ok": "✓", "skip": "→", "fail": "✗"}[status]        print(f"  {icon}{pdf.name}{msg}")        if status == "fail":            failed_files.append((pdf.name, msg))    print(f"\n完成：成功 {stats['ok']}，跳过 {stats['skip']}，失败 {stats['fail']}")    if failed_files:        print("\n失败清单：")        for name, reason in failed_files:            print(f"  - {name}: {reason}")# 加密run_batch("./pdfs", "encrypt", "YourPassword2026")# 解密run_batch("./pdfs", "decrypt", "YourPassword2026")

from concurrent.futures import ThreadPoolExecutor, as_completeddef run_batch_parallel(input_dir: str, mode: str, password: str, workers: int = 8):    src = Path(input_dir)    out_dir = src / f"_{mode}_output"    out_dir.mkdir(exist_ok=True)    pdfs = sorted(src.rglob("*.pdf"))    results = []    with ThreadPoolExecutor(max_workers=workers) as executor:        futures = {            executor.submit(process_one, p, mode, password, out_dir): p            for p in pdfs        }        for future in as_completed(futures):            status, msg = future.result()            results.append((futures[future].name, status, msg))    ok = sum(1 for _, s, _ in results if s == "ok")    skip = sum(1 for _, s, _ in results if s == "skip")    fail = sum(1 for _, s, _ in results if s == "fail")    print(f"完成：成功 {ok}，跳过 {skip}，失败 {fail}")