graph LR
A[文件遍歷] --> B[格式識別]
B --> C1[PDF 提取器]
B --> C2[Word 提取器]
B --> C3[Excel 提取器]
B --> C4[PPT 提取器]
B --> C5[文本提取器]
C1 --> D[OCR?]
D --> E[文本后處理]
C2 --> E
C3 --> E
C4 --> E
C5 --> E
E --> F[DeepSeek API]
F --> G[信息匯總]
G --> H[輸出結(jié)果]
H --> I[日志/報告]

pip install requests pandas PyMuPDF python-docx openpyxl python-pptx pdfplumber pytesseract pillow opencv-python-headless tqdm

from pathlib import Path
import os

def find_files(root_dir, extensions=['.pdf', '.docx', '.xlsx', '.pptx', '.txt', '.jpg', '.png']):
    """
    查找指定擴(kuò)展名的文件
    :param root_dir: 根目錄
    :param extensions: 目標(biāo)擴(kuò)展名列表
    :return: 文件路徑列表
    """
    file_paths = []
    for ext in extensions:
        # 使用 glob 匹配 (pathlib 的 glob 也可)
        pattern = os.path.join(root_dir, '**', f'*{ext}')
        file_paths.extend([f for f in glob.glob(pattern, recursive=True) if os.path.isfile(f)])
    return file_paths

# 或者使用 pathlib
def find_files_pathlib(root_dir, extensions):
    root_path = Path(root_dir)
    file_paths = []
    for ext in extensions:
        file_paths.extend(list(root_path.rglob(f'*{ext}')))
    return [str(p) for p in file_paths]

def get_file_type(file_path):
    """
    根據(jù)后綴名判斷文件類型
    :param file_path: 文件路徑
    :return: 文件類型字符串 ('pdf', 'docx', 'xlsx', 'pptx', 'text', 'image')
    """
    ext = Path(file_path).suffix.lower()
    if ext == '.pdf':
        return 'pdf'
    elif ext == '.docx':
        return 'docx'
    elif ext in ['.xlsx', '.xls']:  # 可能需要處理舊版 .xls
        return 'xlsx'
    elif ext == '.pptx':
        return 'pptx'
    elif ext in ['.txt', '.csv', '.md']:
        return 'text'
    elif ext in ['.jpg', '.jpeg', '.png', '.bmp', '.tiff']:
        return 'image'
    else:
        return 'unknown'  # 或拋出異常

import fitz  # PyMuPDF

def extract_text_from_pdf(pdf_path):
    """
    提取PDF文本內(nèi)容。對于掃描件，此方法只能得到空字符串或亂碼。
    :param pdf_path: PDF文件路徑
    :return: 提取出的文本字符串
    """
    doc = fitz.open(pdf_path)
    full_text = ""
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        text = page.get_text("text")  # "text" 參數(shù)獲取純文本
        full_text += text + "\n"  # 添加換行分隔頁面
    doc.close()
    return full_text.strip()

from PIL import Image
import pytesseract
import cv2  # 可選，用于圖像預(yù)處理

def ocr_image(image_path, lang='chi_sim+eng'):
    """
    使用Tesseract OCR識別圖像中的文字
    :param image_path: 圖像文件路徑
    :param lang: 語言 (例如 'eng' 英文, 'chi_sim' 簡體中文)
    :return: 識別出的文本字符串
    """
    # 預(yù)處理 (可選, 根據(jù)圖像質(zhì)量調(diào)整)
    img = cv2.imread(image_path)
    gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)  # 灰度化
    # thresh = cv2.threshold(gray, 0, 255, cv2.THRESH_BINARY | cv2.THRESH_OTSU)[1]  # 二值化
    # ... 其他可能的預(yù)處理步驟 (降噪、糾偏等)
    # 使用PIL打開預(yù)處理后的圖像 (或原始圖像)
    # pil_img = Image.fromarray(thresh)
    pil_img = Image.open(image_path)
    text = pytesseract.image_to_string(pil_img, lang=lang)
    return text.strip()

# 處理掃描PDF: 需要先將PDF每一頁轉(zhuǎn)為圖像，再OCR
def ocr_pdf(pdf_path, output_dir='temp_ocr_images', dpi=300, lang='chi_sim+eng'):
    """
    OCR識別掃描PDF
    :param pdf_path: PDF路徑
    :param output_dir: 臨時存放圖像的目錄
    :param dpi: 渲染圖像的分辨率 (越高越清晰，但越慢)
    :param lang: OCR語言
    :return: 識別出的文本字符串
    """
    doc = fitz.open(pdf_path)
    Path(output_dir).mkdir(parents=True, exist_ok=True)
    full_text = ""
    for page_num in range(len(doc)):
        page = doc.load_page(page_num)
        pix = page.get_pixmap(matrix=fitz.Matrix(dpi/72, dpi/72))  # 提高DPI
        image_path = Path(output_dir) / f"page_{page_num}.png"
        pix.save(str(image_path))
        page_text = ocr_image(str(image_path), lang=lang)
        full_text += page_text + "\n\n"  # 添加空行分隔頁面
        os.remove(image_path)  # 刪除臨時圖像
    doc.close()
    return full_text.strip()

from docx import Document

def extract_text_from_docx(docx_path):
    """
    提取Word文檔文本內(nèi)容
    :param docx_path: .docx 文件路徑
    :return: 提取出的文本字符串
    """
    doc = Document(docx_path)
    full_text = []
    for para in doc.paragraphs:
        full_text.append(para.text)
    # 提取表格文本 (可選，按行按單元格)
    for table in doc.tables:
        for row in table.rows:
            for cell in row.cells:
                full_text.append(cell.text)
    return "\n".join(full_text).strip()

import pandas as pd

def extract_data_from_excel(excel_path, sheet_name=None):
    """
    提取Excel數(shù)據(jù)。這里簡單返回所有工作表的數(shù)據(jù)字典。
    :param excel_path: .xlsx 文件路徑
    :param sheet_name: 指定工作表名 (None表示所有)
    :return: 字典 {工作表名: pandas DataFrame}
    """
    xl = pd.ExcelFile(excel_path)
    if sheet_name:
        return {sheet_name: pd.read_excel(excel_path, sheet_name=sheet_name)}
    else:
        return {sheet: pd.read_excel(excel_path, sheet_name=sheet) for sheet in xl.sheet_names}
# 注意：提取的是結(jié)構(gòu)化數(shù)據(jù)，后續(xù)可能不需要DeepSeek處理整個表，而是處理特定列的描述或匯總統(tǒng)計。

from pptx import Presentation

def extract_text_from_pptx(pptx_path):
    """
    提取PPT幻燈片文本
    :param pptx_path: .pptx 文件路徑
    :return: 提取出的文本字符串
    """
    prs = Presentation(pptx_path)
    full_text = []
    for slide in prs.slides:
        # 提取幻燈片標(biāo)題 (如果有)
        if slide.shapes.title:
            full_text.append(slide.shapes.title.text)
        # 提取其他形狀中的文本
        for shape in slide.shapes:
            if hasattr(shape, "text") and shape != slide.shapes.title:  # 避免重復(fù)標(biāo)題
                if shape.text.strip():  # 跳過空文本框
                    full_text.append(shape.text)
    return "\n".join(full_text).strip()

def extract_text_from_text_file(txt_path):
    """
    提取純文本文件內(nèi)容
    :param txt_path: 文本文件路徑 (.txt, .md, .csv等)
    :return: 文件內(nèi)容字符串
    """
    with open(txt_path, 'r', encoding='utf-8') as f:
        return f.read().strip()

import re

def clean_text(raw_text):
    """
    清理文本：去除多余空白、常見頁眉頁腳模式、特定亂碼
    :param raw_text: 原始提取文本
    :return: 清理后的文本
    """
    # 合并多個空白符為一個空格
    cleaned = re.sub(r'\s+', ' ', raw_text)
    # 去除特定頁眉頁腳模式 (需要根據(jù)文檔特點調(diào)整正則表達(dá)式)
    cleaned = re.sub(r'機(jī)密\s*第\s*\d+\s*頁\s*共\s*\d+\s*頁', '', cleaned)  # 示例
    # 去除常見的OCR識別錯誤字符 (例如孤立的符號)
    cleaned = re.sub(r'^[^\w]*$', '', cleaned, flags=re.MULTILINE)  # 刪除僅包含非單詞字符的行
    # 去除首尾空白
    cleaned = cleaned.strip()
    return cleaned

import requests
import json
import time

DEEPSEEK_API_URL = "https://api.deepseek.com/v1/chat/completions"  # 假設(shè)的API地址，請?zhí)鎿Q為真實地址
DEEPSEEK_API_KEY = "your_deepseek_api_key_here"  # 替換為你的真實API密鑰

def call_deepseek_api(prompt, model="deepseek-chat", max_tokens=2048, temperature=0.7):
    """
    調(diào)用DeepSeek API進(jìn)行文本生成或理解
    :param prompt: 輸入的提示文本
    :param model: 使用的模型名稱
    :param max_tokens: 生成的最大token數(shù)
    :param temperature: 生成多樣性
    :return: API返回的響應(yīng)文本 (通常是字符串)
    """
    headers = {
        "Authorization": f"Bearer {DEEPSEEK_API_KEY}",
        "Content-Type": "application/json"
    }
    data = {
        "model": model,
        "messages": [{"role": "user", "content": prompt}],
        "max_tokens": max_tokens,
        "temperature": temperature,
    }
    try:
        response = requests.post(DEEPSEEK_API_URL, headers=headers, data=json.dumps(data))
        response.raise_for_status()  # 檢查HTTP錯誤
        response_data = response.json()
        return response_data['choices'][0]['message']['content'].strip()  # 提取模型回復(fù)內(nèi)容
    except requests.exceptions.RequestException as e:
        print(f"API請求失敗: {e}")
        return None
    except (KeyError, IndexError) as e:
        print(f"解析API響應(yīng)失敗: {e}")
        return None

cleaned_text = clean_text(extracted_text)
summary_prompt = f"請為以下文本生成一個簡潔的摘要：\n\n{cleaned_text[:3000]}"  # 注意token限制，可能需要分塊
summary = call_deepseek_api(summary_prompt)

info_prompt = f"""你是一個信息抽取助手。請從以下合同文本中提取以下信息，并以JSON格式返回：
- 合同編號 (contract_id)
- 甲方名稱 (party_a)
- 乙方名稱 (party_b)
- 合同金額 (amount) (帶單位)
- 簽訂日期 (sign_date) (YYYY-MM-DD格式)
- 合同有效期 (validity_period)
文本內(nèi)容：
{cleaned_text}
JSON格式示例：{{"contract_id": "HT2024001", "party_a": "XX公司", ...}}
請只返回JSON，不要有其他內(nèi)容。
"""
extracted_info_json = call_deepseek_api(info_prompt)
# 然后解析 JSON

question = "這份合同的主要交付物是什么？"
qa_prompt = f"""基于以下文本，回答用戶的問題。文本內(nèi)容：
{cleaned_text}
用戶問題：{question}
請直接給出答案。
"""
answer = call_deepseek_api(qa_prompt)

classify_prompt = f"""請判斷以下文本的主題類別（選項：技術(shù)文檔、財務(wù)報告、法律合同、新聞報道、其他）：
{cleaned_text}
請只返回類別名稱。
"""
category = call_deepseek_api(classify_prompt)

import pandas as pd

# 初始化一個空的DataFrame，定義好列名
summary_df = pd.DataFrame(columns=[
    'file_path', 'file_type', 'extracted_text_summary', 'contract_id', 'party_a', 'party_b',
    'amount', 'sign_date', 'validity_period', 'qa_answer', 'category', 'processing_time'
])

# 在處理每個文件后，將結(jié)果添加到DataFrame
def add_to_summary(summary_df, file_info, deepseek_results):
    """
    將單個文件處理結(jié)果添加到匯總DataFrame
    :param summary_df: 匯總DataFrame
    :param file_info: 文件基本信息字典 (如 path, type, size, extraction_time)
    :param deepseek_results: 字典，包含DeepSeek處理后的各種結(jié)果 (summary, extracted_info, qa_answer, category)
    :return: 更新后的DataFrame
    """
    new_row = {
        'file_path': file_info['path'],
        'file_type': file_info['type'],
        'extracted_text_summary': deepseek_results.get('summary', ''),
        # 假設(shè) extracted_info 是一個字典，包含合同字段
        'contract_id': deepseek_results.get('extracted_info', {}).get('contract_id', ''),
        'party_a': deepseek_results.get('extracted_info', {}).get('party_a', ''),
        'party_b': deepseek_results.get('extracted_info', {}).get('party_b', ''),
        'amount': deepseek_results.get('extracted_info', {}).get('amount', ''),
        'sign_date': deepseek_results.get('extracted_info', {}).get('sign_date', ''),
        'validity_period': deepseek_results.get('extracted_info', {}).get('validity_period', ''),
        'qa_answer': deepseek_results.get('qa_answer', ''),
        'category': deepseek_results.get('category', ''),
        'processing_time': file_info.get('processing_time', ''),
    }
    # 將新行轉(zhuǎn)換為DataFrame并拼接
    new_df = pd.DataFrame([new_row])
    return pd.concat([summary_df, new_df], ignore_index=True)

summary_df.to_csv('document_summary.csv', index=False, encoding='utf-8-sig')
summary_df.to_excel('document_summary.xlsx', index=False, engine='openpyxl')

from docx import Document
from docx.shared import Pt

def generate_word_report(summary_df, output_path):
    doc = Document()
    doc.add_heading('批量文檔處理匯總報告', level=0)
    for _, row in summary_df.iterrows():
        doc.add_heading(f"文件: {row['file_path']}", level=2)
        doc.add_paragraph(f"類型: {row['file_type']}")
        doc.add_paragraph(f"摘要:")
        doc.add_paragraph(row['extracted_text_summary'])
        if row['contract_id']:  # 如果有合同信息
            doc.add_paragraph("關(guān)鍵合同信息:")
            doc.add_paragraph(f"合同編號: {row['contract_id']}")
            doc.add_paragraph(f"甲方: {row['party_a']}")
            # ... 添加其他合同字段
        doc.add_paragraph("-" * 40)  # 分隔線
    doc.save(output_path)

with open('summary.md', 'w', encoding='utf-8') as md:
    md.write('# 批量文檔處理匯總\n\n')
    for _, row in summary_df.iterrows():
        md.write(f"## 文件: {row['file_path']}\n")
        md.write(f"- **類型**: {row['file_type']}\n")
        md.write(f"- **摘要**: {row['extracted_text_summary']}\n\n")
        # ... 添加其他信息

import logging
import sys
from tqdm import tqdm  # 進(jìn)度條

# 配置日志
logging.basicConfig(
    filename='document_processor.log',
    level=logging.INFO,
    format='%(asctime)s - %(name)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger('DocProcessor')

def process_all_files(root_dir):
    file_paths = find_files(root_dir)
    summary_df = pd.DataFrame(...)  # 初始化空的匯總DF
    error_files = []  # 記錄處理失敗的文件

    for file_path in tqdm(file_paths, desc="Processing Documents"):
        try:
            start_time = time.time()
            file_type = get_file_type(file_path)
            logger.info(f"開始處理文件: {file_path} ({file_type})")

            # Step 1: 提取內(nèi)容
            if file_type == 'pdf':
                raw_text = extract_text_from_pdf(file_path)
                if len(raw_text) < 100:  # 簡單判斷是否是掃描件
                    logger.warning(f"{file_path} 可能是掃描PDF，嘗試OCR...")
                    raw_text = ocr_pdf(file_path)
            elif file_type == 'docx':
                raw_text = extract_text_from_docx(file_path)
            # ... 其他類型處理
            cleaned_text = clean_text(raw_text)
            logger.info(f"文本提取完成，長度: {len(cleaned_text)} 字符")

            # Step 2: DeepSeek 處理 (示例：生成摘要)
            summary_prompt = f"請為以下文本生成一個簡潔的摘要：\n\n{cleaned_text[:3000]}"
            summary = call_deepseek_api(summary_prompt)
            if summary is None:
                raise Exception("DeepSeek API調(diào)用失敗")
            logger.info(f"摘要生成完成")

            # Step 3: 信息匯總 (這里假設(shè)只存摘要)
            file_info = {'path': file_path, 'type': file_type}
            deepseek_results = {'summary': summary}
            summary_df = add_to_summary(summary_df, file_info, deepseek_results)

            proc_time = time.time() - start_time
            logger.info(f"文件處理成功，耗時: {proc_time:.2f}秒")

        except Exception as e:
            logger.error(f"處理文件 {file_path} 時出錯: {str(e)}", exc_info=True)
            error_files.append(file_path)

    # Step 4: 輸出匯總結(jié)果
    summary_df.to_excel('document_summary.xlsx', index=False)
    logger.info(f"匯總結(jié)果已輸出到 document_summary.xlsx")

    # 輸出錯誤文件列表
    if error_files:
        logger.warning(f"以下文件處理失敗:")
        for ef in error_files:
            logger.warning(ef)
        with open('error_files.txt', 'w') as ferr:
            ferr.write("\n".join(error_files))

    return summary_df, error_files

import os
import time
import logging
import pandas as pd
import fitz  # PyMuPDF
from docx import Document
from pathlib import Path
import pytesseract
from PIL import Image
import requests
import json
from tqdm import tqdm

# ... (省略所有函數(shù)定義，如 find_files, get_file_type, extract_text_from_pdf, ocr_pdf, extract_text_from_docx, call_deepseek_api 等) ...

def main():
    # 配置
    input_folder = "/path/to/your/documents"
    deepseek_api_key = "your_api_key"  # 實際使用時從安全的地方獲取
    output_summary_excel = "document_summary.xlsx"
    log_file = "processing.log"

    # 設(shè)置日志
    logging.basicConfig(filename=log_file, level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
    logger = logging.getLogger('DocAutoProcessor')

    # 查找文件
    logger.info("開始查找目標(biāo)文件...")
    target_files = find_files(input_folder)
    logger.info(f"找到 {len(target_files)} 個待處理文件.")

    # 初始化匯總 DataFrame
    summary_cols = ['file_path', 'file_type', 'content_summary', 'processing_time_sec']
    summary_df = pd.DataFrame(columns=summary_cols)
    error_list = []

    # 處理每個文件
    for file_path in tqdm(target_files, desc="Processing"):
        try:
            start_time = time.time()
            file_type = get_file_type(file_path)
            logger.info(f"Processing: {file_path} ({file_type})")

            # 內(nèi)容提取
            if file_type == 'pdf':
                text = extract_text_from_pdf(file_path)
                if len(text) < 100:  # 簡單判斷是否需要OCR
                    logger.warning("Low text count, attempting OCR...")
                    text = ocr_pdf(file_path)
            elif file_type == 'docx':
                text = extract_text_from_docx(file_path)
            elif file_type == 'text':
                text = extract_text_from_text_file(file_path)
            else:
                logger.warning(f"Unsupported file type: {file_type}. Skipping.")
                continue
            clean_text = clean_text(text)
            logger.info(f"Extracted text length: {len(clean_text)} chars")

            # DeepSeek 處理 - 生成摘要
            prompt = f"請為以下文本生成一個簡潔的摘要：\n\n{clean_text[:3000]}"  # 注意截斷
            summary = call_deepseek_api(prompt, api_key=deepseek_api_key)
            if summary is None:
                raise Exception("DeepSeek API call failed or returned no summary")

            # 記錄結(jié)果
            proc_time = time.time() - start_time
            new_row = {
                'file_path': file_path,
                'file_type': file_type,
                'content_summary': summary,
                'processing_time_sec': proc_time
            }
            summary_df = pd.concat([summary_df, pd.DataFrame([new_row])], ignore_index=True)
            logger.info(f"Processed successfully in {proc_time:.2f} sec. Summary: {summary[:50]}...")

        except Exception as e:
            logger.error(f"Error processing {file_path}: {str(e)}", exc_info=True)
            error_list.append(file_path)

    # 輸出匯總結(jié)果
    try:
        summary_df.to_excel(output_summary_excel, index=False)
        logger.info(f"Summary report saved to {output_summary_excel}")
    except Exception as e:
        logger.error(f"Failed to save summary: {str(e)}")

    # 記錄錯誤文件
    if error_list:
        logger.warning(f"{len(error_list)} files failed processing:")
        with open("error_files.txt", 'w') as ferr:
            ferr.write("\n".join(error_list))
        logger.info("Error file list saved to error_files.txt")

    logger.info("Batch document processing completed!")

if __name__ == "__main__":
    main()