"""PDF 提取服务 — 从 PDF 计划书中提取结构化数据。""" import os import json import time import hashlib import logging import re import shutil import subprocess import tempfile from dataclasses import dataclass from typing import Callable, Optional logger = logging.getLogger(__name__) CACHE_VERSION = 6 def _format_exception(exc: Exception) -> str: message = str(exc) or repr(exc) return f"{exc.__class__.__name__}: {message}" @dataclass class ExtractionResult: pdf_path: str product_name: str plan_type: str # savings/ci/iul status: str # success/partial/error data: Optional[dict] = None usage: Optional[dict] = None error: Optional[str] = None duration_ms: float = 0 provenance: Optional[dict] = None # 字段级来源追踪 page_qualities: Optional[list] = None # 每页质量评分 def infer_plan_type(raw: dict) -> str: """从 LLM 输出推断产品类型。""" t = str(raw.get("product_type", "")).lower() if "ci" in t or "critical" in t: return "ci" if "iul" in t or "universal" in t: return "iul" rows = raw.get("benefit_illustration", []) if not isinstance(rows, list): rows = [] has_savings = any( r.get("total_surrender_value") is not None or r.get("guaranteed_cash_value") is not None or r.get("reversionary_bonus") is not None for r in rows if isinstance(r, dict) ) has_ci = any( r.get("surrender_value_total") is not None or r.get("death_benefit_total") is not None for r in rows if isinstance(r, dict) ) has_iul = any( r.get("cash_value") is not None or r.get("account_value") is not None or r.get("guaranteed_account_value") is not None or r.get("non_guaranteed_account_value") is not None or r.get("non_guaranteed_cash_value") is not None for r in rows if isinstance(r, dict) ) if has_iul: return "iul" if has_savings: return "savings" if has_ci: return "ci" policy = raw.get("policy", {}) if isinstance(policy, dict): if policy.get("index_account_rate") is not None or policy.get("capital_partition") is not None: return "iul" if policy.get("sum_insured") is not None: return "ci" return "savings" def _hash_file(file_path: str) -> str: """计算文件 SHA-256 哈希。""" h = hashlib.sha256() with open(file_path, "rb") as f: for chunk in iter(lambda: f.read(8192), b""): h.update(chunk) return h.hexdigest() def _get_cache_path(pdf_path: str, cache_dir: str) -> str: """获取缓存文件路径。""" file_hash = _hash_file(pdf_path) return os.path.join(cache_dir, f"{file_hash}.json") def _format_pdf_pages(text_parts: list[str]) -> str: """保留真实页码，供后续按封面、投保信息和利益表筛选关键页面。""" return "\n\n".join( f"[PAGE {page_num}]\n{content.strip()}" for page_num, content in enumerate(text_parts, start=1) if content and content.strip() ) def _extract_pdf_text(pdf_path: str, max_chars: int = 120000) -> tuple[str, list[dict]]: """提取 PDF 文本，支持多种 PDF 解析库。 返回 (text, page_qualities): - text: 合并后的全文 - page_qualities: 每页质量评分列表 [{"page": 1, "quality": "high", ...}, ...] 优先级：PyMuPDF > PyPDF2 > pdfplumber > pypdf """ text = "" # 尝试 PyMuPDF (fitz 或 pymupdf) try: try: import fitz except ImportError: import pymupdf as fitz doc = fitz.open(pdf_path) text_parts = [] for page in doc: text_parts.append(page.get_text()) doc.close() text = _format_pdf_pages(text_parts) if text.strip() and not _looks_corrupted(text): logger.info(f"使用 PyMuPDF 提取成功: {len(text)} 字符") truncated = text[:max_chars] if len(text) > max_chars else text page_qualities = _assess_page_qualities(text_parts) return truncated, page_qualities if text.strip(): logger.warning("PyMuPDF 提取结果疑似乱码，将尝试 OCR") except ImportError: logger.debug("PyMuPDF 未安装，尝试下一个库") except Exception as e: logger.warning(f"PyMuPDF 提取失败: {e}") # 尝试 PyPDF2 try: from PyPDF2 import PdfReader reader = PdfReader(pdf_path) text_parts = [] for page in reader.pages: page_text = page.extract_text() text_parts.append(page_text or "") text = _format_pdf_pages(text_parts) if text.strip() and not _looks_corrupted(text): logger.info(f"使用 PyPDF2 提取成功: {len(text)} 字符") truncated = text[:max_chars] if len(text) > max_chars else text page_qualities = _assess_page_qualities(text_parts) return truncated, page_qualities except ImportError: logger.debug("PyPDF2 未安装，尝试下一个库") except Exception as e: logger.warning(f"PyPDF2 提取失败: {e}") # 尝试 pypdf try: from pypdf import PdfReader reader = PdfReader(pdf_path) text_parts = [] for page in reader.pages: page_text = page.extract_text() text_parts.append(page_text or "") text = _format_pdf_pages(text_parts) if text.strip() and not _looks_corrupted(text): logger.info(f"使用 pypdf 提取成功: {len(text)} 字符") truncated = text[:max_chars] if len(text) > max_chars else text page_qualities = _assess_page_qualities(text_parts) return truncated, page_qualities except ImportError: logger.debug("pypdf 未安装，尝试下一个库") except Exception as e: logger.warning(f"pypdf 提取失败: {e}") # 尝试 pdfplumber try: import pdfplumber with pdfplumber.open(pdf_path) as pdf: text_parts = [] for page in pdf.pages: page_text = page.extract_text() text_parts.append(page_text or "") text = _format_pdf_pages(text_parts) if text.strip() and not _looks_corrupted(text): logger.info(f"使用 pdfplumber 提取成功: {len(text)} 字符") truncated = text[:max_chars] if len(text) > max_chars else text page_qualities = _assess_page_qualities(text_parts) return truncated, page_qualities except ImportError: logger.debug("pdfplumber 未安装") except Exception as e: logger.warning(f"pdfplumber 提取失败: {e}") ocr_text = _extract_pdf_text_ocr(pdf_path, max_chars=max_chars) if ocr_text: return ocr_text, [] logger.error( "无法提取 PDF 文本，请安装以下任一库:\n" " pip install PyMuPDF\n" " pip install PyPDF2\n" " pip install pypdf\n" " pip install pdfplumber" ) return "", [] def _assess_page_qualities(text_parts: list[str]) -> list[dict]: """评估每页文本质量。""" qualities = [] for i, page_text in enumerate(text_parts): score = _score_page_quality(page_text) score["page"] = i + 1 qualities.append(score) # 汇总日志 low_pages = [q for q in qualities if q["quality"] in ("corrupted", "low")] if low_pages: logger.info( f"[PageQuality] {len(qualities)} 页中 {len(low_pages)} 页质量低: " f"{[p['page'] for p in low_pages]}" ) return qualities def _looks_corrupted(text: str) -> bool: """检测 PDF 文本是否乱码。""" if not text or len(text) < 50: return True # 缺少 ToUnicode 映射时，部分解析器会返回大量 ``(cid:123)``。 # 这些占位符是 ASCII，不能只靠下方的可读字符率判断。 cid_placeholders = re.findall(r"\(cid:\d+\)", text, re.IGNORECASE) if len(cid_placeholders) >= 10: cid_chars = sum(len(value) for value in cid_placeholders) if cid_chars / max(len(text), 1) >= 0.02: return True visible = [c for c in text if not c.isspace()] if not visible: return True bad_chars = sum( 1 for c in visible if c in ("\ufffd", "\uffff") or ord(c) < 32 or 0x7F <= ord(c) <= 0x9F or 0xE000 <= ord(c) <= 0xF8FF ) readable_chars = sum( 1 for c in visible if ( (c.isascii() and (c.isalnum() or c in ".,:;!?%+-_/()[]{}$¥￥'")) or "\u3400" <= c <= "\u9fff" ) ) return bad_chars / len(visible) > 0.03 or readable_chars / len(visible) < 0.35 def _score_page_quality(page_text: str) -> dict: """评估单页文本质量。返回质量指标字典。 用于识别需要 OCR 重处理的低质量页。 """ if not page_text or not page_text.strip(): return {"chars": 0, "readable_ratio": 0, "label_hits": 0, "number_density": 0, "quality": "empty"} visible = [c for c in page_text if not c.isspace()] total = len(visible) if total == 0: return {"chars": 0, "readable_ratio": 0, "label_hits": 0, "number_density": 0, "quality": "empty"} readable = sum( 1 for c in visible if (c.isascii() and (c.isalnum() or c in ".,:;!?%+-_/()[]{}$")) or "㐀" <= c <= "鿿" or "　" <= c <= "〿" ) readable_ratio = readable / total # 保险相关标签命中 label_keywords = [ "保单", "保單", "保费", "保費", "受保人", "被保人", "投保", "现金价值", "現金價值", "红利", "紅利", "退保", "premium", "policy", "insured", "benefit", "cash value", "保证", "保證", "非保证", "非保證", "年度", "年齡", "年龄", ] text_lower = page_text.lower() label_hits = sum(1 for kw in label_keywords if kw in text_lower) # 数字密度 digits = sum(1 for c in visible if c.isdigit()) number_density = digits / total # 综合评分 if readable_ratio < 0.3: quality = "corrupted" elif readable_ratio < 0.5 and label_hits < 2: quality = "low" elif label_hits >= 3 and number_density >= 0.1: quality = "high" elif label_hits >= 1: quality = "medium" else: quality = "low" return { "chars": total, "readable_ratio": round(readable_ratio, 3), "label_hits": label_hits, "number_density": round(number_density, 3), "quality": quality, } def _extract_pdf_text_ocr( pdf_path: str, max_chars: int = 120000, max_pages: int = 40, ) -> str: """对扫描件或字体映射损坏的 PDF 使用 Tesseract OCR。 支持繁体中文（chi_tra）+ 简体中文（chi_sim）+ 英文。 如果 chi_tra 未安装，自动降级到 chi_sim+eng。 """ tesseract = shutil.which("tesseract") if not tesseract: logger.warning("PDF 文本疑似乱码，但未安装 Tesseract OCR") return "" # 检测可用的语言包：优先繁体+简体+英文，降级到简体+英文 ocr_lang = "chi_tra+chi_sim+eng" try: test_result = subprocess.run( [tesseract, "--list-langs"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10, check=False, ) available_langs = test_result.stdout.lower() if "chi_tra" not in available_langs: ocr_lang = "chi_sim+eng" logger.info("[OCR] 繁体语言包(chi_tra)未安装，降级到 chi_sim+eng") else: logger.info("[OCR] 使用繁体+简体+英文识别") except Exception: ocr_lang = "chi_sim+eng" try: try: import fitz except ImportError: import pymupdf as fitz doc = fitz.open(pdf_path) text_parts = [] page_count = min(len(doc), max_pages) with tempfile.TemporaryDirectory(prefix="insurance-pdf-ocr-") as temp_dir: for index in range(page_count): page = doc[index] # 300 DPI 基线，灰度模式 pixmap = page.get_pixmap( matrix=fitz.Matrix(3.0, 3.0), colorspace=fitz.csGRAY, ) image_path = os.path.join(temp_dir, f"page-{index + 1}.png") pixmap.save(image_path) # 使用 psm 6（统一文本块），适合表格和表单 completed = subprocess.run( [ tesseract, image_path, "stdout", "-l", ocr_lang, "--psm", "6", "--oem", "3", ], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=90, check=False, ) if completed.returncode == 0 and completed.stdout.strip(): text_parts.append(completed.stdout) else: text_parts.append("") logger.warning( "Tesseract OCR 第 %s 页失败: %s", index + 1, completed.stderr.strip()[:300], ) doc.close() text = _format_pdf_pages(text_parts) if text.strip() and not _looks_corrupted(text): logger.info("使用 Tesseract OCR 提取成功: %s 字符", len(text)) text = text[:max_chars] if len(text) > max_chars else text return f"[OCR]\n{text}" logger.warning("Tesseract OCR 未能产生可用文本") except Exception as exc: logger.warning("Tesseract OCR 提取失败: %s", _format_exception(exc)) return "" def _ocr_specific_pages( pdf_path: str, page_indices: list[int], ocr_lang: str = "chi_tra+chi_sim+eng", ) -> dict[int, str]: """对指定页码执行 OCR，返回 {page_index: ocr_text}。""" tesseract = shutil.which("tesseract") if not tesseract or not page_indices: return {} # 检测可用语言包 try: test_result = subprocess.run( [tesseract, "--list-langs"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=10, check=False, ) if "chi_tra" not in test_result.stdout.lower(): ocr_lang = "chi_sim+eng" except Exception: ocr_lang = "chi_sim+eng" results = {} try: try: import fitz except ImportError: import pymupdf as fitz doc = fitz.open(pdf_path) with tempfile.TemporaryDirectory(prefix="insurance-page-ocr-") as temp_dir: for page_idx in page_indices: if page_idx >= len(doc): continue page = doc[page_idx] pixmap = page.get_pixmap( matrix=fitz.Matrix(3.0, 3.0), colorspace=fitz.csGRAY, ) image_path = os.path.join(temp_dir, f"page-{page_idx + 1}.png") pixmap.save(image_path) completed = subprocess.run( [tesseract, image_path, "stdout", "-l", ocr_lang, "--psm", "6", "--oem", "3"], capture_output=True, text=True, encoding="utf-8", errors="replace", timeout=90, check=False, ) if completed.returncode == 0 and completed.stdout.strip(): results[page_idx] = completed.stdout.strip() doc.close() except Exception as exc: logger.warning(f"[OCR] 逐页 OCR 失败: {_format_exception(exc)}") return results def _merge_page_ocr(pdf_text: str, page_qualities: list[dict], pdf_path: str) -> str: """对低质量页执行 OCR 并合并回全文。 识别 quality 为 corrupted 或 low 的页面，OCR 替换其内容。 """ low_pages = [q for q in page_qualities if q.get("quality") in ("corrupted", "low")] if not low_pages: return pdf_text tesseract = shutil.which("tesseract") if not tesseract: logger.info("[OCR] 低质量页检测到但 Tesseract 未安装，跳过逐页 OCR") return pdf_text page_indices = [q["page"] - 1 for q in low_pages] # 转为 0-based logger.info(f"[OCR] 对 {len(page_indices)} 个低质量页执行 OCR: {[p+1 for p in page_indices]}") progress_callback = None # 这里的进度由上层管理 ocr_results = _ocr_specific_pages(pdf_path, page_indices) if not ocr_results: return pdf_text # 按 [PAGE N] 标记分割全文，替换低质量页 import re parts = re.split(r'(\[PAGE \d+\])', pdf_text) current_page = 0 merged_parts = [] for part in parts: page_match = re.match(r'\[PAGE (\d+)\]', part) if page_match: current_page = int(page_match.group(1)) - 1 # 转为 0-based merged_parts.append(part) elif current_page in ocr_results: # 用 OCR 结果替换该页内容 merged_parts.append(f"\n{ocr_results[current_page]}\n") logger.info(f"[OCR] 第 {current_page + 1} 页已用 OCR 替换") else: merged_parts.append(part) return "".join(merged_parts) def _normalized_product_name(data: dict) -> str: product_name = str(data.get("product_name") or "").strip() if not product_name: return "unknown" if product_name.lower() == "unknown": return "unknown" return product_name def _build_provenance( final_data: dict, regex_data: Optional[dict], method: str, used_ocr: bool, ) -> dict: """构建字段级来源追踪。 返回结构： { "product_name": {"source": "regex|llm|ocr", "confidence": 0.0-1.0}, "insured.age": {"source": "regex|llm", "confidence": 0.0-1.0}, "benefit_illustration": {"source": "regex|llm", "row_count": N, "confidence": 0.0-1.0}, ... } """ prov = {} base_source = "ocr" if used_ocr else ("regex" if method == "regex+analysis" else "llm") def _field_confidence(value, source) -> float: """根据值和来源计算置信度。""" if value is None or value == "" or value == "unknown": return 0.0 if source == "regex": return 0.9 # 正则匹配高置信 if source == "ocr": return 0.6 # OCR 中等置信 return 0.7 # LLM 中高置信 # 标量字段 product_name = _normalized_product_name(final_data) if regex_data and _normalized_product_name(regex_data) != "unknown": prov["product_name"] = {"source": "regex", "confidence": _field_confidence(product_name, "regex")} else: prov["product_name"] = {"source": base_source, "confidence": _field_confidence(product_name, base_source)} # insured 子字段 insured = final_data.get("insured") or {} regex_insured = (regex_data or {}).get("insured") or {} for field in ("age", "gender"): final_val = insured.get(field) regex_val = regex_insured.get(field) if regex_val is not None and regex_val != "": src = "regex" else: src = base_source prov[f"insured.{field}"] = {"source": src, "confidence": _field_confidence(final_val, src)} # policy 子字段 policy = final_data.get("policy") or {} regex_policy = (regex_data or {}).get("policy") or {} for field in ("currency", "sum_insured", "annual_premium", "premium_payment_period", "coverage_period"): final_val = policy.get(field) regex_val = regex_policy.get(field) if regex_val is not None and regex_val != "": src = "regex" else: src = base_source prov[f"policy.{field}"] = {"source": src, "confidence": _field_confidence(final_val, src)} # benefit_illustration benefit = final_data.get("benefit_illustration") or [] regex_benefit = (regex_data or {}).get("benefit_illustration") or [] if len(regex_benefit) >= len(benefit) and regex_benefit: ben_src = "regex" elif benefit: ben_src = base_source else: ben_src = "missing" prov["benefit_illustration"] = { "source": ben_src, "row_count": len(benefit), "confidence": min(1.0, len(benefit) / 20) if benefit else 0.0, } # withdrawal_illustration withdrawal = final_data.get("withdrawal_illustration") or [] if withdrawal: prov["withdrawal_illustration"] = { "source": base_source, "row_count": len(withdrawal), "confidence": min(1.0, len(withdrawal) / 10), } # 统计 sources = [v["source"] for v in prov.values() if isinstance(v, dict) and "source" in v] avg_conf = sum(v.get("confidence", 0) for v in prov.values() if isinstance(v, dict)) / max(len(prov), 1) prov["_summary"] = { "method": method, "used_ocr": used_ocr, "avg_confidence": round(avg_conf, 3), "source_counts": {s: sources.count(s) for s in set(sources)}, } return prov def _apply_filename_hints(data: dict, pdf_path: str, plan_type: str) -> dict: """用标准计划书文件名中的明确字段补齐 OCR 容易误读的数据。""" if not isinstance(data, dict): return data filename = os.path.basename(pdf_path) if plan_type == "iul" and re.search(r"(?:^|[_-])SIUL3(?:[_-]|$)", filename, re.IGNORECASE): def set_missing(target: dict, key: str, value) -> None: if target.get(key) in (None, "", "unknown"): target[key] = value data["product_name"] = "Manulife SIUL 3" insured = data.setdefault("insured", {}) policy = data.setdefault("policy", {}) if isinstance(policy, dict): policy["product_name"] = "Manulife SIUL 3" # 标准文件名示例：SIUL3_F-48-N-CN-USD-S3m-5x。 # 只补齐文件名明确编码且正文未识别的字段，绝不覆盖已识别值。 identity_match = re.search( r"(?:^|[_-])(?P[FM])-(?P\d{1,3})-(?P[NS])(?:[_-]|$)", filename, re.IGNORECASE, ) if identity_match and isinstance(insured, dict): set_missing(insured, "gender", "female" if identity_match.group("gender").upper() == "F" else "male") set_missing(insured, "age", int(identity_match.group("age"))) set_missing(insured, "smoker", "no" if identity_match.group("smoker").upper() == "N" else "yes") if isinstance(policy, dict): currency_match = re.search(r"(?:^|[_-])(USD|HKD|CNY|RMB)(?:[_-]|$)", filename, re.IGNORECASE) if currency_match: set_missing(policy, "currency", currency_match.group(1).upper()) sum_match = re.search(r"(?:^|[_-])S(?P\d+(?:\.\d+)?)(?P[mMkK])(?:[_-]|$)", filename) if sum_match: multiplier = 1_000_000 if sum_match.group("unit").lower() == "m" else 1_000 set_missing(policy, "sum_insured", float(sum_match.group("amount")) * multiplier) payment_match = re.search(r"(?:^|[_-])(?P\d{1,2})x(?:[_-]|$)", filename, re.IGNORECASE) if payment_match: set_missing(policy, "premium_payment_period", int(payment_match.group("years"))) return data def assess_extraction_payload(data: Optional[dict], plan_type: str) -> tuple[str, str]: """Return extraction status and a user-facing error when data is incomplete.""" if not isinstance(data, dict) or not data: return "partial", "结构化结果为空，请补充识别数据后再生成 PPT" insured = data.get("insured") or {} policy = data.get("policy") or {} benefit_rows = data.get("benefit_illustration") benefit_rows = benefit_rows if isinstance(benefit_rows, list) else [] product_name = _normalized_product_name(data) insured_age = insured.get("age") def has_positive_number(value) -> bool: try: return float(value) > 0 except (TypeError, ValueError): return False problems = [] if product_name == "unknown": problems.append("产品名称未识别") if not has_positive_number(insured_age): problems.append("被保人年龄缺失") if not benefit_rows: problems.append("利益演示为空") normalized_type = (plan_type or "").lower() if normalized_type == "iul": if not has_positive_number(policy.get("sum_insured")): problems.append("保额缺失") index_accounts = data.get("index_accounts") if not isinstance(index_accounts, list) or not index_accounts: problems.append("指数账户缺失") rows_with_iul_values = sum( 1 for row in benefit_rows if isinstance(row, dict) and any( row.get(field) is not None for field in ( "guaranteed_account_value", "non_guaranteed_account_value", "non_guaranteed_cash_value", "account_value", "cash_value", ) ) ) if benefit_rows and rows_with_iul_values / len(benefit_rows) < 0.5: problems.append("IUL利益字段缺失") elif normalized_type == "ci": if not has_positive_number(policy.get("sum_insured")): problems.append("保额缺失") coverage_items = data.get("coverage_items") if not isinstance(coverage_items, list) or not coverage_items: problems.append("保障项目缺失") else: if not has_positive_number(policy.get("annual_premium")): problems.append("年缴保费缺失") if problems: return "partial", "；".join(problems[:3]) return "success", "" def _payload_score(data: Optional[dict]) -> int: """用于比较两次提取结果，优先保留关键字段更完整的一次。""" if not isinstance(data, dict): return 0 insured = data.get("insured") if isinstance(data.get("insured"), dict) else {} policy = data.get("policy") if isinstance(data.get("policy"), dict) else {} rows = data.get("benefit_illustration") row_count = len(rows) if isinstance(rows, list) else 0 return ( row_count * 10 + (5 if _normalized_product_name(data) != "unknown" else 0) + (5 if insured.get("age") else 0) + sum(1 for value in policy.values() if value not in (None, "", [], {})) ) def _regex_quality_gate(regex_data: dict, plan_type: str) -> tuple[bool, list[str]]: """检查正则提取结果的关键字段完整性。 返回 (passed, missing_fields): - passed: True 表示关键字段完整，可以仅用 LLM 做分析 - missing_fields: 缺失的关键字段列表 """ missing = [] product_name = _normalized_product_name(regex_data) if product_name == "unknown": missing.append("product_name") insured = regex_data.get("insured") or {} if not insured.get("age"): missing.append("insured.age") policy = regex_data.get("policy") or {} normalized_type = (plan_type or "savings").lower() if normalized_type == "iul": try: if not (float(policy.get("sum_insured") or 0) > 0): missing.append("policy.sum_insured") except (TypeError, ValueError): missing.append("policy.sum_insured") index_accounts = regex_data.get("index_accounts") if not isinstance(index_accounts, list) or not index_accounts: missing.append("index_accounts") elif normalized_type == "ci": try: if not (float(policy.get("sum_insured") or 0) > 0): missing.append("policy.sum_insured") except (TypeError, ValueError): missing.append("policy.sum_insured") rows = [row for row in (regex_data.get("benefit_illustration") or []) if isinstance(row, dict)] years = [] for row in rows: try: year = int(row.get("policy_year")) except (TypeError, ValueError): continue if year > 0: years.append(year) if len(rows) >= 3 and ( len(years) != len(rows) or len(years) != len(set(years)) or years != sorted(years) ): missing.append("benefit_illustration.policy_year_sequence") age_offsets = [] for row in rows: try: if row.get("age") is not None and row.get("policy_year") is not None: age_offsets.append(int(row["age"]) - int(row["policy_year"])) except (TypeError, ValueError): continue if len(age_offsets) >= 2 and max(age_offsets) - min(age_offsets) > 1: missing.append("benefit_illustration.age_sequence") if normalized_type == "iul" and rows: iul_value_fields = ( "guaranteed_account_value", "non_guaranteed_account_value", "non_guaranteed_cash_value", "cost_of_insurance", "account_value", "cash_value", ) rows_with_iul_values = sum( 1 for row in rows if any(row.get(field) is not None for field in iul_value_fields) ) if rows_with_iul_values / len(rows) < 0.5: missing.append("benefit_illustration.iul_value_columns") try: annual_premium = float(policy.get("annual_premium") or 0) except (TypeError, ValueError): annual_premium = 0 positive_surrender = [] for row in rows: try: value = float((row or {}).get("total_surrender_value") or 0) except (TypeError, ValueError): value = 0 if value > 0: positive_surrender.append(value) if len(positive_surrender) >= 3: implausible_floor = max(150, annual_premium * 0.01) tiny = [value for value in positive_surrender if value < implausible_floor] if len(tiny) / len(positive_surrender) >= 0.6: missing.append("benefit_illustration.total_surrender_value_implausible") return (len(missing) == 0, missing) async def _llm_extract_split( pdf_text: str, plan_type: str, llm_client, progress_callback=None, ) -> tuple[dict, any]: """分块 LLM 提取：身份字段 + 利益表 + 提领表分开调用。 优势： - 每次调用输出更小，减少截断风险 - 利益表可以使用更多 token - 某块失败不影响其他块 返回 (merged_data, last_response)。 """ from insurance.ppt.prompts import select_key_pages extraction_text = select_key_pages(pdf_text, max_pages=10, max_chars=28000) merged_data = {} last_response = None normalized_type = (plan_type or "savings").lower() if normalized_type == "iul": identity_shape = ( '"policy": {"product_name": "产品名称", "currency": "USD/HKD/CNY", ' '"sum_insured": 数字, "annual_premium": 数字, "premium_payment_period": 数字, ' '"coverage_period": "终身", "target_premium": 数字或null, "minimum_premium": 数字或null}, ' '"coverage_items": null, ' '"index_accounts": [{"name": "账户名称", "allocation": 数字或null, ' '"current_rate": 数字或null, "guaranteed_floor": 数字或null}]' ) elif normalized_type == "ci": identity_shape = ( '"policy": {"product_name": "产品名称", "currency": "USD/HKD/CNY", ' '"sum_insured": 数字, "annual_premium": 数字, "premium_payment_period": 数字, ' '"coverage_period": "终身"}, ' '"coverage_items": [{"label": "保障项目", "amount": 数字, "description": null}], ' '"index_accounts": null' ) else: identity_shape = ( '"policy": {"product_name": "产品名称", "currency": "USD/HKD/CNY", ' '"sum_insured": 数字或null, "annual_premium": 数字, "premium_payment_period": 数字, ' '"coverage_period": "终身"}, "coverage_items": null, "index_accounts": null' ) # ── 第一次调用：身份 + 保单字段（小输出，高精度）── identity_prompt = ( "从以下 PDF 文本中提取保险计划书的身份和保单字段。\n" "只输出以下 JSON，不要输出利益演示表和提领表：\n" '{"product_name": "产品全称", "product_type": "savings/ci/iul", ' '"insured": {"name": null, "age": 数字, "gender": "male/female", "relation": null, "smoker": null}, ' f'{identity_shape}' + '}\n\n' "规则：\n" "1. gender 用 male/female\n" "2. premium_payment_period 只输出数字（年数）\n" "3. 无法确定的字段填 null\n" "4. 只输出 JSON，无 markdown\n\n" f"PDF 文本：\n{extraction_text}" ) if progress_callback: progress_callback(55, "正在识别产品和保单信息") try: identity_data, last_response = await llm_client.structured_output( prompt=identity_prompt, system_prompt="你是保险计划书数据提取专家。只输出 JSON。", schema={ "type": "object", "required": ["product_name", "insured", "policy"], "properties": { "product_name": {"type": "string"}, "insured": { "type": "object", "required": ["age", "gender"], }, "policy": { "type": "object", "required": [ "currency", "sum_insured", "annual_premium", "premium_payment_period", ], }, }, }, temperature=0, ) if isinstance(identity_data, dict): merged_data.update(identity_data) except Exception as e: logger.warning(f"[SplitExtract] 身份字段提取失败: {e}") # ── 第二次调用：利益演示表（大输出，用更多 token）── if normalized_type == "iul": benefit_shape = ( '{"benefit_illustration": [{"policy_year": 数字, "age": 数字或null, ' '"total_premium_paid": 数字或null, "guaranteed_account_value": 数字或null, ' '"guaranteed_cash_value": 数字或null, "non_guaranteed_account_value": 数字或null, ' '"non_guaranteed_cash_value": 数字或null, "total_surrender_value": 数字或null, ' '"non_guaranteed_death_benefit": 数字或null, "cost_of_insurance": 数字或null, ' '"source_page": 数字或null}]}' ) elif normalized_type == "ci": benefit_shape = ( '{"benefit_illustration": [{"policy_year": 数字, "age": 数字或null, ' '"total_premium_paid": 数字或null, "death_benefit": 数字或null, ' '"source_page": 数字或null}]}' ) else: benefit_shape = ( '{"benefit_illustration": [{"policy_year": 数字, "age": 数字或null, ' '"total_premium_paid": 数字或null, "guaranteed_cash_value": 数字或null, ' '"reversionary_bonus": 数字或null, "terminal_dividend": 数字或null, ' '"total_surrender_value": 数字或null, "death_benefit": 数字或null, ' '"source_page": 数字或null}]}' ) benefit_prompt = ( "从以下 PDF 文本中提取保险计划书的利益演示表。\n" "只输出 benefit_illustration 数组，不要输出其他字段：\n" f"{benefit_shape}\n\n" "规则：\n" "1. 扫描所有页面，提取全部保单年度\n" "2. 数值去逗号转数字，无法确定填 null，不要填 0\n" "3. 严禁编造数据\n" "4. 年龄和保单年度是不同列，年龄绝不能写入任何金额字段\n" "5. 只输出 JSON，无 markdown\n\n" f"PDF 文本：\n{extraction_text}" ) if progress_callback: progress_callback(70, "正在提取利益演示表") try: benefit_data, benefit_resp = await llm_client.structured_output( prompt=benefit_prompt, system_prompt="你是保险计划书数据提取专家。只输出 JSON。", schema={ "type": "object", "required": ["benefit_illustration"], "properties": {"benefit_illustration": {"type": "array"}}, }, temperature=0, ) if isinstance(benefit_data, dict) and "benefit_illustration" in benefit_data: merged_data["benefit_illustration"] = benefit_data["benefit_illustration"] last_response = benefit_resp except Exception as e: logger.warning(f"[SplitExtract] 利益表提取失败: {e}") # ── 第三次调用：提领表（可选，仅储蓄险/IUL）── if plan_type in ("savings", "iul"): withdrawal_prompt = ( "从以下 PDF 文本中提取保险计划书的提领/提款演示表。\n" "如果文本中没有提领表，输出空数组。\n" "只输出 withdrawal_illustration 数组：\n" '{"withdrawal_illustration": [{"policy_year": 数字, "annual_withdrawal": 数字或null, ' '"total_withdrawn": 数字或null, "surrender_value_before": 数字或null, ' '"surrender_value_after": 数字或null, "source_page": 数字或null}]}\n\n' "规则：\n" "1. 只取\"总额/Total\"列，不取子列\n" "2. 没有提领表时输出空数组 []\n" "3. 只输出 JSON，无 markdown\n\n" f"PDF 文本：\n{extraction_text}" ) if progress_callback: progress_callback(85, "正在提取提领表") try: withdrawal_data, withdrawal_resp = await llm_client.structured_output( prompt=withdrawal_prompt, system_prompt="你是保险计划书数据提取专家。只输出 JSON。", schema={ "type": "object", "required": ["withdrawal_illustration"], "properties": {"withdrawal_illustration": {"type": "array"}}, }, temperature=0, ) if isinstance(withdrawal_data, dict) and "withdrawal_illustration" in withdrawal_data: merged_data["withdrawal_illustration"] = withdrawal_data["withdrawal_illustration"] last_response = withdrawal_resp except Exception as e: logger.warning(f"[SplitExtract] 提领表提取失败: {e}") # 确保必要字段存在 merged_data.setdefault("benefit_illustration", []) merged_data.setdefault("withdrawal_illustration", []) return merged_data, last_response class ExtractionOrchestrator: """PDF 提取编排器。 提取策略（按优先级）： 1. 缓存命中 → 直接返回 2. 正则提取（零延迟）→ 关键字段完整则仅用 LLM 做轻量分析 3. 关键字段缺失或正则不足 → 回退到完整 LLM 提取 """ # 正则提取行数阈值：低于此值回退到 LLM REGEX_ROW_THRESHOLD = 3 def __init__(self, use_cache: bool = True, cache_dir: str = ".cache/insurance-ppt"): self.use_cache = use_cache self.cache_dir = cache_dir async def extract_plan( self, pdf_path: str, plan_type: str = "savings", force_reparse: bool = False, progress_callback: Optional[Callable[[int, str], None]] = None, company_id: str = "", product_id: str = "", product_name_hint: str = "", product_aliases: Optional[list[str]] = None, ) -> ExtractionResult: """从 PDF 提取结构化数据。 优先使用正则提取（~50ms），仅在行数不足或关键字段缺失时回退到 LLM。 Args: product_name_hint: 用户选择的产品标准名称（用于匹配和纠错） product_aliases: 产品别名列表（用于模糊匹配） """ from insurance.ppt.llm_client import llm_client from insurance.ppt.prompts import ( SAVINGS_PLAN_SYSTEM_PROMPT, CI_PLAN_SYSTEM_PROMPT, IUL_SYSTEM_PROMPT, ANALYSIS_SYSTEM_PROMPT, build_analysis_prompt, select_key_pages, ) from insurance.ppt.regex_extractor import extract_insurance_regex, count_benefit_rows start = time.time() abs_path = os.path.abspath(pdf_path) if not os.path.exists(abs_path): return ExtractionResult( pdf_path=abs_path, product_name="unknown", plan_type=plan_type, status="error", error="文件不存在", duration_ms=(time.time() - start) * 1000, ) # 检查缓存 if self.use_cache and not force_reparse: cached = self._load_from_cache(abs_path) if cached: if progress_callback: progress_callback(100, "已使用历史解析结果") cached.duration_ms = (time.time() - start) * 1000 return cached # 提取 PDF 文本 if progress_callback: progress_callback(10, "正在读取 PDF 文本") pdf_text, page_qualities = _extract_pdf_text(abs_path) if not pdf_text: return ExtractionResult( pdf_path=abs_path, product_name="unknown", plan_type=plan_type, status="error", error="无法提取 PDF 文本", duration_ms=(time.time() - start) * 1000, ) # 逐页 OCR：对低质量页单独 OCR 替换（仅在非全文 OCR 模式下） if not pdf_text.startswith("[OCR]") and page_qualities: low_count = sum(1 for q in page_qualities if q.get("quality") in ("corrupted", "low")) if low_count > 0: if progress_callback: progress_callback(20, f"检测到 {low_count} 个低质量页，正在 OCR 补充") pdf_text = _merge_page_ocr(pdf_text, page_qualities, abs_path) if progress_callback: progress_callback(30, "PDF 文本读取完成，正在识别数据表") # ─── 正则提取（第一阶段，零 LLM 调用）──────────────── regex_start = time.time() used_ocr = pdf_text.startswith("[OCR]") regex_data = extract_insurance_regex(pdf_text) regex_rows = count_benefit_rows(regex_data) regex_ms = (time.time() - regex_start) * 1000 if progress_callback: progress_callback( 50, f"规则识别完成，共识别 {regex_rows} 行利益数据", ) extraction_stats = { "regex_rows": regex_rows, "regex_ms": round(regex_ms, 1), "page_count": len(page_qualities), "low_quality_pages": [q["page"] for q in page_qualities if q["quality"] in ("corrupted", "low")], } response = None # LLM 响应（可能未调用） # 质量门：检查正则结果的关键字段完整性 quality_passed, quality_missing = _regex_quality_gate(regex_data, plan_type) extraction_stats["quality_passed"] = quality_passed extraction_stats["quality_missing"] = quality_missing if regex_rows >= self.REGEX_ROW_THRESHOLD and not used_ocr and quality_passed: # ─── 正则成功且关键字段完整：仅调用 LLM 做轻量分析 ──────────── extraction_stats["method"] = "regex+analysis" data = regex_data llm_start = time.time() try: # 选取关键页面（4-5页，~8000字符，远小于原来的 20000+） key_pages = select_key_pages(pdf_text, max_pages=5, max_chars=8000) analysis_prompt = build_analysis_prompt(key_pages, data) if progress_callback: progress_callback(60, "正在补充产品分析") analysis_result, response = await llm_client.structured_output( prompt=analysis_prompt, system_prompt=ANALYSIS_SYSTEM_PROMPT, ) # 将分析结果合并到 data if isinstance(analysis_result, dict): data["sales_insights"] = { "key_points": analysis_result.get("keyPoints", []), "gaps": analysis_result.get("gaps", []), "suggested_questions": analysis_result.get("suggestedQuestions", []), } extraction_stats["llm_tokens"] = { "input": response.tokens.get("input", 0) if response.tokens else 0, "output": response.tokens.get("output", 0) if response.tokens else 0, } extraction_stats["llm_ms"] = round((time.time() - llm_start) * 1000, 1) if progress_callback: progress_callback(90, "产品分析完成，正在校验数据") except Exception as e: # 分析失败不影响已提取的数据，只记日志 logger.warning(f"[ExtractionOrchestrator] LLM 分析失败（不影响数据提取）: {_format_exception(e)}") extraction_stats["llm_error"] = _format_exception(e) extraction_stats["llm_ms"] = round((time.time() - llm_start) * 1000, 1) else: # ─── 正则不足或关键字段缺失：回退到分块 LLM 提取 ──── if not quality_passed: logger.info( f"[ExtractionOrchestrator] 正则关键字段缺失({quality_missing}), " f"回退到分块 LLM 提取" ) else: logger.info( f"[ExtractionOrchestrator] 正则提取行数不足({regex_rows}), " f"回退到分块 LLM 提取" ) extraction_stats["method"] = "llm_split" llm_start = time.time() try: if progress_callback: progress_callback(55, "规则识别不足，正在等待 AI 分块提取") data, response = await _llm_extract_split( pdf_text, plan_type, llm_client, progress_callback, ) # 检查完整性，对缺失字段做针对性重试 initial_status, initial_error = assess_extraction_payload(data, plan_type) if initial_status == "partial" and initial_error: if progress_callback: progress_callback(88, "首次识别不完整，正在补充关键字段") # 用正则结果填补 LLM 缺失的字段 if regex_data: for key in ("product_name", "insured", "policy", "currency"): if key in regex_data and (key not in data or not data[key]): data[key] = regex_data[key] # 利益表：取行数更多的那个 regex_benefit = regex_data.get("benefit_illustration", []) llm_benefit = data.get("benefit_illustration", []) if quality_passed and len(regex_benefit) > len(llm_benefit): data["benefit_illustration"] = regex_benefit extraction_stats["llm_tokens"] = { "input": response.tokens.get("input", 0) if response and response.tokens else 0, "output": response.tokens.get("output", 0) if response and response.tokens else 0, } if progress_callback: progress_callback(90, "AI 分块提取完成，正在校验数据") except Exception as e: return ExtractionResult( pdf_path=abs_path, product_name="unknown", plan_type=plan_type, status="error", error=f"LLM 调用失败: {_format_exception(e)}", duration_ms=(time.time() - start) * 1000, ) extraction_stats["llm_ms"] = round((time.time() - llm_start) * 1000, 1) data = _apply_filename_hints(data, abs_path, plan_type) # 产品先验匹配：用用户选择的产品信息补正提取结果 if product_name_hint: extracted_name = _normalized_product_name(data) if extracted_name == "unknown": # 直接使用用户选择的产品名称 data.setdefault("policy", {})["product_name"] = product_name_hint if "product_name" in data: data["product_name"] = product_name_hint logger.info(f"[ExtractionOrchestrator] 产品名未知，采用用户选择: {product_name_hint}") elif product_aliases: # 检查提取的名称是否与别名匹配 extracted_lower = extracted_name.lower() all_names = [product_name_hint.lower()] + [a.lower() for a in product_aliases] if not any(n in extracted_lower or extracted_lower in n for n in all_names): logger.info( f"[ExtractionOrchestrator] 提取产品名 '{extracted_name}' " f"与用户选择 '{product_name_hint}' 不匹配，请用户确认" ) # 推断产品类型 detected_type = plan_type if plan_type in ("savings", "ci", "iul") else infer_plan_type(data) product_name = _normalized_product_name(data) status, extraction_error = assess_extraction_payload(data, detected_type) # 只缓存完整结果，避免后续复用错误或缺字段的解析。 if self.use_cache and status == "success": self._save_to_cache(abs_path, data) total_ms = (time.time() - start) * 1000 extraction_stats["total_ms"] = round(total_ms, 1) logger.info(f"[ExtractionOrchestrator] 提取完成: {extraction_stats}") if progress_callback: progress_callback(100, "数据校验完成") # 构建字段级来源追踪 method = extraction_stats.get("method", "unknown") provenance = _build_provenance(data, regex_data, method, used_ocr) return ExtractionResult( pdf_path=abs_path, product_name=product_name, plan_type=detected_type, status=status, data=data, usage={"input": response.tokens.get("input", 0), "output": response.tokens.get("output", 0)} if response and response.tokens else None, error=extraction_error or None, duration_ms=total_ms, provenance=provenance, page_qualities=page_qualities, ) async def extract_multiple(self, pdf_paths: list[str], plan_type: str = "savings") -> list[ExtractionResult]: """顺序提取多个 PDF。""" results = [] for pdf_path in pdf_paths: logger.info(f" 📄 {os.path.basename(pdf_path)}...") results.append(await self.extract_plan(pdf_path, plan_type)) return results def _load_from_cache(self, pdf_path: str) -> Optional[ExtractionResult]: """从缓存加载。""" try: cache_path = _get_cache_path(pdf_path, self.cache_dir) if not os.path.exists(cache_path): return None with open(cache_path, "r", encoding="utf-8") as f: raw = json.load(f) meta = raw.get("_meta", {}) if meta.get("cacheVersion") != CACHE_VERSION: return None data = raw.get("_data", raw) product_name = _normalized_product_name(data) plan_type = infer_plan_type(data) status, extraction_error = assess_extraction_payload(data, plan_type) return ExtractionResult( pdf_path=pdf_path, product_name=product_name, plan_type=plan_type, status=status, data=data, error=extraction_error or None, ) except Exception: return None def _save_to_cache(self, pdf_path: str, data: dict): """写入缓存。""" try: os.makedirs(self.cache_dir, exist_ok=True) file_hash = _hash_file(pdf_path) cache_data = { "_data": data, "_meta": { "cacheVersion": CACHE_VERSION, "originalFile": os.path.basename(pdf_path), "extractedAt": __import__("datetime").datetime.now().isoformat(), "fileHash": file_hash, }, } cache_path = os.path.join(self.cache_dir, f"{file_hash}.json") with open(cache_path, "w", encoding="utf-8") as f: json.dump(cache_data, f, ensure_ascii=False, indent=2) except Exception as e: logger.warning(f"缓存写入失败: {e}") async def extract_for_poster(self, filepath: str) -> dict: """提取海报所需的关键字段（精简 prompt，降低 LLM 成本）。 返回: {age, gender, currency, sum_assured, premium_term, annual_premium, coverage_period, key_benefits} """ from insurance.ppt.llm_client import llm_client abs_path = os.path.abspath(filepath) text, _page_qualities = _extract_pdf_text(abs_path) if not text: raise ValueError("无法提取 PDF 文本") text = text[:6000] system_prompt = ( "你是一位保险计划书解析专家。请从以下计划书内容中提取海报所需的关键字段。\n" "输出 JSON 格式（不要包含 markdown 代码块标记）：\n" '{"age": 35, "gender": "male", "currency": "USD", "sum_assured": 500000, ' '"premium_term": 5, "annual_premium": 100000, "coverage_period": "终身", ' '"key_benefits": ["身故赔偿", "全残保障"]}\n' "注意：数值用数字，不要带货币符号。gender 用 male/female。" ) result, _response = await llm_client.structured_output( text, system_prompt, schema={ "type": "object", "properties": { "age": {"type": "number"}, "gender": {"type": "string"}, "currency": {"type": "string"}, "sum_assured": {"type": "number"}, "premium_term": {"type": "number"}, "annual_premium": {"type": "number"}, "coverage_period": {"type": "string"}, "key_benefits": {"type": "array", "items": {"type": "string"}}, }, "required": ["age", "gender", "currency", "sum_assured", "annual_premium"], }, ) return result