文档加载与处理(下):表格、图片、清洗去重、质量评估,手把手搭生产级管线

import pymupdf4llm# 直接转 Markdown，表格会自动变成 Markdown 表格格式md_text = pymupdf4llm.to_markdown("report.pdf")# 输出里表格会变成：# | 指标 | 2022 | 2023 |# |-----|------|------|# | 营收 | 100亿 | 120亿 |

import camelot# lattice 模式：适合有边框的表格tables = camelot.read_pdf("report.pdf", flavor="lattice", pages="1-5")# flavor 模式：适合无边框表格（基于文本对齐检测）tables = camelot.read_pdf("report.pdf", flavor="lattice", pages="1-5")for table in tables:    print(table.df)  # pandas DataFrame    print(table.parsing_report)  # 解析精度报告

|       营收情况       || 年份  | 营收 | 增长率 || 2022  | 100亿 | 15%  || 2023  | 120亿 | 20%  |

import pandas as pddef handle_merged_cells(table_df):    """处理合并单元格：向前填充空值"""    # 如果某行某列为空，继承上一行的值    return table_df.ffill()# 示例raw_df = pd.DataFrame([    ["营收情况", "", ""],    ["年份", "营收", "增长率"],    ["2022", "100亿", "15%"],    ["2023", "120亿", "20%"]])processed_df = handle_merged_cells(raw_df)print(processed_df)#       0     1    2# 0  营收情况  NaN  NaN# 1  年份    营收  增长率# 2  2022  100亿  15%# 3  2023  120亿  20%

def table_to_markdown(df, caption=""):    md = f"<!-- {caption} -->\n" if caption else ""    md += "| " + " | ".join(str(c) for c in df.columns) + " |\n"    md += "| " + " | ".join(["---"] * len(df.columns)) + " |\n"    for _, row in df.iterrows():        md += "| " + " | ".join(str(v) for v in row) + " |\n"    return md

def table_to_semantic(df):    """把表格转成自然语言描述"""    lines = []    headers = list(df.columns)    for _, row in df.iterrows():        desc = "，".join([f"{h}为{v}" for h, v in zip(headers, row)])        lines.append(desc)    return "；".join(lines)# 示例输出：# "年份为2022，营收为100亿，增长率为15%；年份为2023，营收为120亿，增长率为20%"

def table_to_json(df):    return {        "headers": list(df.columns),        "rows": df.values.tolist(),        "semantic_units": [            dict(zip(df.columns, row)) for _, row in df.iterrows()        ]    }

def detect_and_merge_cross_page_tables(pages_tables):    """检测并合并跨页表格"""    merged = []    prev_table = None    for table in pages_tables:        if prev_table is not None:            # 如果当前表格的表头是数字列（无文字表头）            first_row = table.df.iloc[0]            if all(is_cell_numeric(cell) for cell in first_row):                # 是上一页表格的延续，去掉第一行直接拼接                table.df = table.df.iloc[1:]                prev_table.df = pd.concat([prev_table.df, table.df])                continue        prev_table = table        merged.append(table)    return merged

from openai import OpenAIimport base64client = OpenAI()def extract_chart_data(image_path):    """用视觉模型提取图表中的数据和趋势"""    with open(image_path, "rb") as f:        image_data = base64.b64encode(f.read()).decode("utf-8")    response = client.chat.completions.create(        model="gpt-4o",  # 或 claude-3.5-sonnet 等视觉模型        messages=[            {                "role": "user",                "content": [                    {"type": "text", "text": "请提取这张图表的完整数据。包含：1) X轴和Y轴标签 2) 所有数据点 3) 整体趋势描述。以JSON格式返回。"},                    {"type": "image_url", "image_url": {"url": f"data:image/png;base64,{image_data}"}}                ]            }        ]    )    return response.choices[0].message.content

{  "chart_type": "折线图",  "title": "2023年度营收趋势",  "x_axis": "月份",  "y_axis": "营收（万元）",  "data_points": [    {"month": "1月", "revenue": 80},    {"month": "2月", "revenue": 75}  ],  "trend": "上半年平稳，Q3开始下滑，Q4回暖"}

def dual_channel_index(document):    """文本通道 + 图片通道分别入库"""    # 通道一：文本通道（主通道）    text_chunks = chunk_text(document.text)    # 通道二：图片通道（辅通道）    image_chunks = []    for image in document.images:        # 生成图片描述        caption = generate_image_caption(image)        # 用描述文本来索引        image_chunks.append({            "text": f"[图片] {caption}",            "image_path": image.path,            "embedding": get_clip_embedding(image)  # 可选的：CLIP 向量        })    return text_chunks + image_chunks

import hashlibdef file_deduplicate(file_paths):    """MD5 哈希去重，完全一致的文件只保留一份"""    seen = {}    unique_files = []    for path in file_paths:        with open(path, "rb") as f:            md5 = hashlib.md5(f.read()).hexdigest()        if md5 not in seen:            seen[md5] = path            unique_files.append(path)        else:            print(f"重复文件，已跳过: {path} (与 {seen[md5]} 相同)")    return unique_files

from simhash import Simhashimport jiebadef simhash_deduplicate(documents, threshold=0.9):    """SimHash 去重，相似度 >= threshold 判定为重复"""    unique_docs = []    for doc in documents:        tokens = jieba.cut(doc["text"])        fingerprint = Simhash(" ".join(tokens))        is_duplicate = False        for existing in unique_docs:            sim = fingerprint.distance(existing["fingerprint"])            # SimHash 距离越小越相似，通常 <3 就高度相似            if sim < 3:                is_duplicate = True                break        if not is_duplicate:            doc["fingerprint"] = fingerprint            unique_docs.append(doc)    return [d for d in unique_docs if "fingerprint" in d]

import numpy as npfrom sklearn.metrics.pairwise import cosine_similaritydef chunk_deduplicate(embeddings, chunks, threshold=0.95):    """向量相似度去重，防止高度相似的 chunk 重复入库"""    unique_indices = []    unique_embs = []    for i, emb in enumerate(embeddings):        if len(unique_embs) == 0:            unique_embs.append(emb)            unique_indices.append(i)            continue        sims = cosine_similarity([emb], unique_embs)[0]        if np.max(sims) < threshold:            unique_embs.append(emb)            unique_indices.append(i)    return [chunks[i] for i in unique_indices]

import reimport unicodedataclass TextCleaner:    """RAG 文档清洗器"""    def __init__(self):        # 隐私脱敏规则        self.patterns = {            "phone": re.compile(r"1[3-9]\d{9}"),            "id_card": re.compile(r"\d{18}[\dXx]"),            "email": re.compile(r"\w+@\w+\.\w+"),        }        # 套话黑名单（可扩展）        self.cliches = [            "为进一步规范", "为了贯彻落实", "根据相关法律法规",            "在此背景下", "综上所述", "值得一提的是"        ]    def desensitize(self, text):        """脱敏：保留关键信息，隐藏隐私"""        text = self.patterns["phone"].sub(lambda m: m.group()[:3] + "****" + m.group()[-4:], text)        text = self.patterns["id_card"].sub(lambda m: m.group()[:6] + "********" + m.group()[-4:], text)        text = self.patterns["email"].sub(lambda m: m.group().split("@")[0][:2] + "***@" + m.group().split("@")[1], text)        return text    def normalize_unicode(self, text):        """统一 Unicode 编码：全角转半角、统一引号"""        text = unicodedata.normalize("NFKC", text)        # 全角字母数字转半角        result = []        for char in text:            code = ord(char)            if 0xFF01 <= code <= 0xFF5E:                result.append(chr(code - 0xFEE0))            else:                result.append(char)        return "".join(result)    def remove_cliches(self, text):        """去除公文套话"""        for cliche in self.cliches:            text = text.replace(cliche, "")        return text    def clean(self, text, level="standard"):        """        清洗入口        level: minimal(只脱敏) / standard(脱敏+归一) / aggressive(全量)        """        text = self.desensitize(text)        if level in ("standard", "aggressive"):            text = self.normalize_unicode(text)        if level == "aggressive":            text = self.remove_cliches(text)        return text.strip()# 使用示例cleaner = TextCleaner()raw_text = "为进一步规范公司管理，联系电话：13812345678"cleaned = cleaner.clean(raw_text, level="aggressive")print(cleaned)# 输出: 公司管理，联系电话：138****5678

def get_cleaning_strategy(doc_type):    strategies = {        "contract": {"level": "minimal", "note": "合同只做脱敏，不改内容"},        "meeting_minutes": {"level": "aggressive", "note": "会议纪要重度降噪"},        "tech_doc": {"level": "standard", "note": "技术文档保留代码和公式"},        "annual_report": {"level": "standard", "note": "年报保结构，去套话"},    }    return strategies.get(doc_type, {"level": "standard"})

class DocQualityEvaluator:    """文档处理质量评估器"""    def __init__(self, sample_size=100):        self.sample_size = sample_size    def check_structure_integrity(self, markdown_text):        """检测结构完整性"""        # 检查是否有标题层级断裂（h1 跳到 h3 没有 h2）        headings = re.findall(r"^(#{1,6})\s", markdown_text, re.MULTILINE)        issues = []        prev_level = 0        for h in headings:            level = len(h)            if level > prev_level + 1 and prev_level > 0:                issues.append(f"标题层级跳跃: {prev_level} -> {level}")            prev_level = level        return {            "total_headings": len(headings),            "structure_issues": issues,            "integrity_score": 1.0 - len(issues) / max(len(headings), 1)        }    def check_table_completeness(self, markdown_text):        """检测表格完整性"""        tables = re.findall(r"\|.+\|\n\|[-| ]+\|\n", markdown_text)        issues = []        for table in tables:            lines = table.strip().split("\n")            header_cells = len(lines[0].split("|")) - 2            for line in lines[2:]:                cells = len(line.split("|")) - 2                if cells != header_cells:                    issues.append(f"表格列数不一致: {header_cells} vs {cells}")        return {            "table_count": len(tables),            "table_issues": issues,            "completeness_score": 1.0 - len(issues) / max(len(tables), 1)        }    def evaluate(self, markdown_text):        """全量评估"""        return {            "structure": self.check_structure_integrity(markdown_text),            "tables": self.check_table_completeness(markdown_text)       

def evaluate_doc_quality(vector_store, test_queries, ground_truth):    """用测试集验证文档处理质量"""    results = []    for query, expected_doc in zip(test_queries, ground_truth):        retrieved = vector_store.similarity_search(query, k=5)        # 检查预期文档是否在检索结果中        retrieved_docs = [doc.metadata["source"] for doc in retrieved]        hit = expected_doc in retrieved_docs        results.append({"query": query, "hit": hit})    recall = sum(r["hit"] for r in results) / len(results)    return {"recall@5": recall, "details": results}

class DocProcessingPipeline:    """生产级文档处理管线"""    def __init__(self, cleaner=None, deduper=None, evaluator=None):        self.cleaner = cleaner or TextCleaner()        self.deduper = deduper or {}        self.evaluator = evaluator or DocQualityEvaluator()    def process(self, file_path, doc_type="general"):        """        处理单个文档：解析 → 清洗 → 结构化 → 评估        """        strategy = get_cleaning_strategy(doc_type)        results = {"source": file_path, "status": "success"}        # 1. 格式解析        ext = file_path.split(".")[-1].lower()        if ext == "pdf":            raw_text = self._parse_pdf(file_path)        elif ext in ("docx", "doc"):            raw_text = self._parse_docx(file_path)        elif ext in ("html", "htm"):            raw_text = self._parse_html(file_path)        else:            raw_text = self._parse_text(file_path)        results["raw_length"] = len(raw_text)        # 2. 清洗        cleaned_text = self.cleaner.clean(raw_text, level=strategy["level"])        results["cleaned_length"] = len(cleaned_text)        # 3. 提取表格        tables = extract_tables(file_path)        results["table_count"] = len(tables)        # 4. 提取图片描述        images = extract_image_captions(file_path)        results["image_count"] = len(images)        # 5. 合并输出        final_doc = {            "text": cleaned_text,            "tables": [table_to_markdown(t) for t in tables],            "images": images,            "metadata": {                "source": file_path,                "doc_type": doc_type,                "processed_at": datetime.now().isoformat()            }        }        # 6. 质量评估        quality = self.evaluator.evaluate(cleaned_text)        results["quality"] = quality        return final_doc, results    def _parse_pdf(self, path):        """生产级 PDF 解析"""        try:            # 优先用 PyMuPDF4LLM            import pymupdf4llm            return pymupdf4llm.to_markdown(path)        except ImportError:            # 回退到基础 PyMuPDF            import fitz            doc = fitz.open(path)            return "\n\n".join([page.get_text() for page in doc])    def _parse_docx(self, path):        from docx import Document        doc = Document(path)        return "\n".join([p.text for p in doc.paragraphs])    def _parse_html(self, path):        import trafilatura        with open(path, "r", encoding="utf-8") as f:            raw = f.read()        return trafilatura.extract(raw, output_format="markdown")    def _parse_text(self, path):        with open(path, "r", encoding="utf-8") as f:            return f.read()    def batch_process(self, file_paths, doc_types=None):        """批量处理文档"""        all_results = []        for i, path in enumerate(file_paths):            doc_type = doc_types[i] if doc_types else "general"            doc, report = self.process(path, doc_type)            all_results.append({"doc": doc, "report": report})        return all_results# 使用示例pipeline = DocProcessingPipeline()# 批量处理一批文档file_paths = ["report_2023.pdf", "meeting_notes.docx", "policy.html"]doc_types = ["annual_report", "meeting_minutes", "general"]results = pipeline.batch_process(file_paths, doc_types)# 查看处理报告for r in results:    report = r["report"]    print(f"{report['source']}: {report['status']}")    print(f"  原始长度: {report['raw_length']} → 清洗后: {report['cleaned_length']}")    print(f"  表格: {report['table_count']}个, 图片: {report['image_count']}个")