import osimport reimport pdfplumberfrom PyPDF2 import PdfReaderimport pandas as pdfrom datetime import datetime# ==================== 配置区 ====================INPUT_DIR = "./business_pdfs"       # PDF文件夹OUTPUT_EXCEL = "summary_report.xlsx"  # 输出ExcelLOG_FILE = "process_log.txt"          # 处理日志# 正则表达式配置（根据实际PDF格式调整）PATTERNS = {    "date": r"(?:日期|时间|开票日期|签订日期)[：:]\s*(\d{4}[-/年]\d{1,2}[-/月]\d{1,2}日?)",    "amount": r"(?:金额|总价|合计|价税合计|总金额)[：:]\s*[￥¥]?\s*([\d,]+\.?\d*)",    "customer": r"(?:客户名称|购买方|甲方|客户|公司名称)[：:]\s*(.+?)(?:\n|$)",    "contract_no": r"(?:合同编号|发票号码|单号|编号)[：:]\s*([A-Za-z0-9\-]+)",    "type": r"(?:合同类型|业务类型|类别)[：:]\s*(.+?)(?:\n|$)",}# ==================== 模块1：文件遍历 ====================def scan_pdf_files(input_dir):    """遍历文件夹，收集所有PDF文件"""    pdf_files = []    for root, dirs, files in os.walk(input_dir):        for f in files:            if f.lower().endswith(".pdf"):                pdf_files.append(os.path.join(root, f))    print(f"扫描到 {len(pdf_files)} 个PDF文件")    return pdf_files# ==================== 模块2：PDF预处理 ====================def check_encrypted(pdf_path):    """检查PDF是否加密"""    try:        reader = PdfReader(pdf_path)        return reader.is_encrypted    except:        return Falsedef extract_text(pdf_path):    """提取PDF全文文本"""    try:        with pdfplumber.open(pdf_path) as pdf:            full_text = ""            for page in pdf.pages:                text = page.extract_text()                if text:                    full_text += text + "\n"            return full_text    except Exception as e:        return Nonedef is_scanned(text):    """判断是否为扫描件（提取到的文本极少）"""    if text is None:        return True    return len(text.strip()) < 50  # 少于50个字符认为是扫描件# ==================== 模块3：信息提取 ====================def extract_info(text, patterns):    """用正则表达式从文本中提取关键字段"""    result = {}    for field, pattern in patterns.items():        match = re.search(pattern, text)        if match:            value = match.group(1).strip()            # 清理金额中的逗号            if field == "amount":                value = value.replace(",", "")            result[field] = value        else:            result[field] = ""    return resultdef detect_pdf_type(text):    """根据文本内容自动识别PDF类型"""    text_lower = text.lower()    if "发票" in text or "invoice" in text_lower:        return "发票"    elif "合同" in text or "contract" in text_lower:        return "合同"    elif "报表" in text or "报告" in text:        return "报表"    elif "申请" in text:        return "申请单"    else:        return "其他"# ==================== 模块4：主处理流程 ====================def process_all_pdfs(pdf_files):    """处理所有PDF文件"""    results = []    log_entries = []    for i, pdf_path in enumerate(pdf_files, 1):        filename = os.path.basename(pdf_path)        print(f"[{i}/{len(pdf_files)}] 处理: {filename}")        record = {            "文件名": filename,            "文件路径": pdf_path,            "处理状态": "",            "PDF类型": "",            "日期": "",            "金额": "",            "客户名称": "",            "合同编号": "",            "备注": "",        }        # 检查加密        if check_encrypted(pdf_path):            record["处理状态"] = "加密-跳过"            record["备注"] = "PDF有密码保护，无法自动提取"            results.append(record)            log_entries.append(f"[跳过] {filename}: 加密文件")            continue        # 提取文本        text = extract_text(pdf_path)        # 检查扫描件        if is_scanned(text):            record["处理状态"] = "扫描件-需人工"            record["备注"] = "疑似扫描件，文本提取不足，需人工处理"            results.append(record)            log_entries.append(f"[人工] {filename}: 疑似扫描件")            continue        # 识别类型        pdf_type = detect_pdf_type(text)        record["PDF类型"] = pdf_type        # 提取信息        info = extract_info(text, PATTERNS)        record["日期"] = info.get("date", "")        record["金额"] = info.get("amount", "")        record["客户名称"] = info.get("customer", "")        record["合同编号"] = info.get("contract_no", "")        # 判断提取质量        if info.get("amount") or info.get("date"):            record["处理状态"] = "成功"            log_entries.append(f"[成功] {filename}: 类型={pdf_type}, 金额={info.get('amount', 'N/A')}")        else:            record["处理状态"] = "部分成功"            record["备注"] = "未提取到关键金额或日期，建议人工核对"            log_entries.append(f"[部分] {filename}: 关键字段缺失")        results.append(record)    return results, log_entries# ==================== 模块5：导出Excel和统计 ====================def export_to_excel(results, output_path):    """导出结果到Excel，包含数据和统计摘要"""    df = pd.DataFrame(results)    # 金额转为数值类型（用于统计）    df["金额数值"] = pd.to_numeric(df["金额"], errors="coerce")    # 生成统计摘要    total = len(df)    success = len(df[df["处理状态"] == "成功"])    partial = len(df[df["处理状态"] == "部分成功"])    encrypted = len(df[df["处理状态"] == "加密-跳过"])    scanned = len(df[df["处理状态"] == "扫描件-需人工"])    total_amount = df["金额数值"].sum()    summary = pd.DataFrame({        "统计项": ["总文件数", "成功", "部分成功", "加密跳过", "扫描件需人工", "提取总金额"],        "数量/金额": [total, success, partial, encrypted, scanned, f"{total_amount:,.2f}"],    })    # 按类型统计    type_stats = df[df["处理状态"] == "成功"]["PDF类型"].value_counts().reset_index()    type_stats.columns = ["PDF类型", "数量"]    # 写入Excel（多个Sheet）    with pd.ExcelWriter(output_path, engine="openpyxl") as writer:        df.drop(columns=["金额数值"]).to_excel(writer, sheet_name="详细数据", index=False)        summary.to_excel(writer, sheet_name="统计摘要", index=False)        type_stats.to_excel(writer, sheet_name="类型统计", index=False)    print(f"\n数据已导出到: {output_path}")    print(f"总计: {total} 份 | 成功: {success} | 部分: {partial} | 加密: {encrypted} | 扫描件: {scanned}")    print(f"提取总金额: {total_amount:,.2f}")def save_log(log_entries, log_file):    """保存处理日志"""    with open(log_file, "w", encoding="utf-8") as f:        f.write(f"PDF批量处理日志 - {datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n")        f.write("=" * 60 + "\n\n")        for entry in log_entries:            f.write(entry + "\n")    print(f"日志已保存到: {log_file}")# ==================== 主函数 ====================def main():    print("=" * 60)    print("PDF批量数据提取系统")    print("=" * 60)    # 1. 扫描文件    pdf_files = scan_pdf_files(INPUT_DIR)    if not pdf_files:        print("未找到PDF文件，请检查输入目录")        return    # 2. 处理所有PDF    print(f"\n开始处理...")    results, log_entries = process_all_pdfs(pdf_files)    # 3. 导出Excel    export_to_excel(results, OUTPUT_EXCEL)    # 4. 保存日志    save_log(log_entries, LOG_FILE)    print("\n全部处理完成！")if __name__ == "__main__":    main()

import pdfplumberwith pdfplumber.open("sample.pdf") as pdf:    text = pdf.pages[0].extract_text()    print(text)

合同编号：HT-2024-0089签订日期：2024年08月15日甲方：某某科技有限公司总金额：￥125,000.00

PATTERNS = {    "contract_no": r"合同编号[：:]\s*([A-Za-z0-9\-]+)",    "date": r"签订日期[：:]\s*(\d{4}年\d{1,2}月\d{1,2}日)",    "customer": r"甲方[：:]\s*(.+)",    "amount": r"总金额[：:]\s*[￥¥]?([\d,]+\.?\d*)",}

from multiprocessing import Pool, cpu_countdef process_single_pdf(pdf_path):    """处理单个PDF（独立函数，用于多进程）"""    # ... 把上面的单文件处理逻辑放这里 ...    return recorddef process_all_pdfs_parallel(pdf_files):    """多进程并行处理"""    num_workers = min(cpu_count(), 4)  # 最多4个进程    print(f"使用 {num_workers} 个进程并行处理")    with Pool(num_workers) as pool:        results = pool.map(process_single_pdf, pdf_files)    return results

def extract_text_first_pages(pdf_path, max_pages=2):    with pdfplumber.open(pdf_path) as pdf:        full_text = ""        for page in pdf.pages[:max_pages]:            text = page.extract_text()            if text:                full_text += text + "\n"        return full_text

import jsondef load_processed(log_file):    """加载已处理文件列表"""    if os.path.exists(log_file):        with open(log_file, "r", encoding="utf-8") as f:            return set(json.load(f))    return set()