利用Python+DeepSeek实现多格式文件内容提取与汇总

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[输出结果]
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']):
    """
    查找指定扩展名的文件
    :param root_dir: 根目录
    :param extensions: 目标扩展名列表
    :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):
    """
    根据后缀名判断文件类型
    :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文本内容。对于扫描件，此方法只能得到空字符串或乱码。
    :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" 参数获取纯文本
        full_text += text + "\n"  # 添加换行分隔页面
    doc.close()
    return full_text.strip()

from pil import image
import pytesseract
import cv2  # 可选，用于图像预处理

def ocr_image(image_path, lang='chi_sim+eng'):
    """
    使用tesseract ocr识别图像中的文字
    :param image_path: 图像文件路径
    :param lang: 语言 (例如 'eng' 英文, 'chi_sim' 简体中文)
    :return: 识别出的文本字符串
    """
    # 预处理 (可选, 根据图像质量调整)
    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]  # 二值化
    # ... 其他可能的预处理步骤 (降噪、纠偏等)
    # 使用pil打开预处理后的图像 (或原始图像)
    # 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每一页转为图像，再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文档文本内容
    :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数据。这里简单返回所有工作表的数据字典。
    :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}
# 注意：提取的是结构化数据，后续可能不需要deepseek处理整个表，而是处理特定列的描述或汇总统计。

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:
        # 提取幻灯片标题 (如果有)
        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:  # 避免重复标题
                if shape.text.strip():  # 跳过空文本框
                    full_text.append(shape.text)
    return "\n".join(full_text).strip()

def extract_text_from_text_file(txt_path):
    """
    提取纯文本文件内容
    :param txt_path: 文本文件路径 (.txt, .md, .csv等)
    :return: 文件内容字符串
    """
    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)
    # 去除特定页眉页脚模式 (需要根据文档特点调整正则表达式)
    cleaned = re.sub(r'机密\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"  # 假设的api地址，请替换为真实地址
deepseek_api_key = "your_deepseek_api_key_here"  # 替换为你的真实api密钥

def call_deepseek_api(prompt, model="deepseek-chat", max_tokens=2048, temperature=0.7):
    """
    调用deepseek api进行文本生成或理解
    :param prompt: 输入的提示文本
    :param model: 使用的模型名称
    :param max_tokens: 生成的最大token数
    :param temperature: 生成多样性
    :return: api返回的响应文本 (通常是字符串)
    """
    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()  # 提取模型回复内容
    except requests.exceptions.requestexception as e:
        print(f"api请求失败: {e}")
        return none
    except (keyerror, indexerror) as e:
        print(f"解析api响应失败: {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)
文本内容：
{cleaned_text}
json格式示例：{{"contract_id": "ht2024001", "party_a": "xx公司", ...}}
请只返回json，不要有其他内容。
"""
extracted_info_json = call_deepseek_api(info_prompt)
# 然后解析 json

question = "这份合同的主要交付物是什么？"
qa_prompt = f"""基于以下文本，回答用户的问题。文本内容：
{cleaned_text}
用户问题：{question}
请直接给出答案。
"""
answer = call_deepseek_api(qa_prompt)

classify_prompt = f"""请判断以下文本的主题类别（选项：技术文档、财务报告、法律合同、新闻报道、其他）：
{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'
])

# 在处理每个文件后，将结果添加到dataframe
def add_to_summary(summary_df, file_info, deepseek_results):
    """
    将单个文件处理结果添加到汇总dataframe
    :param summary_df: 汇总dataframe
    :param file_info: 文件基本信息字典 (如 path, type, size, extraction_time)
    :param deepseek_results: 字典，包含deepseek处理后的各种结果 (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', ''),
        # 假设 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', ''),
    }
    # 将新行转换为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("关键合同信息:")
            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  # 进度条

# 配置日志
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: 提取内容
            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调用失败")
            logger.info(f"摘要生成完成")

            # step 3: 信息汇总 (这里假设只存摘要)
            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: 输出汇总结果
    summary_df.to_excel('document_summary.xlsx', index=false)
    logger.info(f"汇总结果已输出到 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

# ... (省略所有函数定义，如 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"

    # 设置日志
    logging.basicconfig(filename=log_file, level=logging.info, format='%(asctime)s - %(levelname)s - %(message)s')
    logger = logging.getlogger('docautoprocessor')

    # 查找文件
    logger.info("开始查找目标文件...")
    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})")

            # 内容提取
            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")

            # 记录结果
            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)

    # 输出汇总结果
    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()