371 lines
13 KiB
Python
371 lines
13 KiB
Python
"""PDF 提取服务 — 从 PDF 计划书中提取结构化数据。"""
|
||
import os
|
||
import re
|
||
import json
|
||
import time
|
||
import hashlib
|
||
import logging
|
||
from dataclasses import dataclass, field
|
||
from typing import Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
CACHE_VERSION = 3
|
||
|
||
|
||
@dataclass
|
||
class ExtractionResult:
|
||
pdf_path: str
|
||
product_name: str
|
||
plan_type: str # savings/ci/iul
|
||
status: str # success/cached/error
|
||
data: Optional[dict] = None
|
||
usage: Optional[dict] = None
|
||
error: Optional[str] = None
|
||
duration_ms: float = 0
|
||
|
||
|
||
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
|
||
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 _extract_pdf_text(pdf_path: str, max_chars: int = 30000) -> str:
|
||
"""提取 PDF 文本，支持多种 PDF 解析库。
|
||
|
||
优先级：PyMuPDF > PyPDF2 > pdfplumber > pypdf
|
||
"""
|
||
text = ""
|
||
|
||
# 尝试 PyMuPDF (fitz 或 pymupdf)
|
||
try:
|
||
# PyMuPDF 1.24.0+ 使用 pymupdf，旧版本使用 fitz
|
||
try:
|
||
import fitz # PyMuPDF
|
||
except ImportError:
|
||
import pymupdf as fitz # 新版本 PyMuPDF
|
||
doc = fitz.open(pdf_path)
|
||
text_parts = []
|
||
for page in doc:
|
||
text_parts.append(page.get_text())
|
||
doc.close()
|
||
text = "\n".join(text_parts)
|
||
if text.strip():
|
||
logger.info(f"使用 PyMuPDF 提取成功: {len(text)} 字符")
|
||
return text[:max_chars] if len(text) > max_chars else text
|
||
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()
|
||
if page_text:
|
||
text_parts.append(page_text)
|
||
text = "\n".join(text_parts)
|
||
if text.strip():
|
||
logger.info(f"使用 PyPDF2 提取成功: {len(text)} 字符")
|
||
return text[:max_chars] if len(text) > max_chars else text
|
||
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()
|
||
if page_text:
|
||
text_parts.append(page_text)
|
||
text = "\n".join(text_parts)
|
||
if text.strip():
|
||
logger.info(f"使用 pypdf 提取成功: {len(text)} 字符")
|
||
return text[:max_chars] if len(text) > max_chars else text
|
||
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()
|
||
if page_text:
|
||
text_parts.append(page_text)
|
||
text = "\n".join(text_parts)
|
||
if text.strip():
|
||
logger.info(f"使用 pdfplumber 提取成功: {len(text)} 字符")
|
||
return text[:max_chars] if len(text) > max_chars else text
|
||
except ImportError:
|
||
logger.debug("pdfplumber 未安装")
|
||
except Exception as e:
|
||
logger.warning(f"pdfplumber 提取失败: {e}")
|
||
|
||
# 所有库都失败
|
||
logger.error(
|
||
"无法提取 PDF 文本，请安装以下任一库:\n"
|
||
" pip install PyMuPDF\n"
|
||
" pip install PyPDF2\n"
|
||
" pip install pypdf\n"
|
||
" pip install pdfplumber"
|
||
)
|
||
return ""
|
||
|
||
|
||
def _looks_corrupted(text: str) -> bool:
|
||
"""检测 PDF 文本是否乱码。"""
|
||
if not text or len(text) < 50:
|
||
return True
|
||
# 计算坏字符比例
|
||
bad_chars = sum(1 for c in text if ord(c) < 32 and c not in "\n\r\t")
|
||
return bad_chars / len(text) > 0.1
|
||
|
||
|
||
class ExtractionOrchestrator:
|
||
"""PDF 提取编排器。"""
|
||
|
||
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) -> ExtractionResult:
|
||
"""从 PDF 提取结构化数据。"""
|
||
from insurance.ppt.llm_client import llm_client
|
||
from insurance.ppt.prompts import (
|
||
SAVINGS_PLAN_SYSTEM_PROMPT, CI_PLAN_SYSTEM_PROMPT, IUL_SYSTEM_PROMPT,
|
||
build_savings_prompt,
|
||
)
|
||
|
||
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:
|
||
cached.duration_ms = (time.time() - start) * 1000
|
||
return cached
|
||
|
||
# 提取 PDF 文本
|
||
pdf_text = _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,
|
||
)
|
||
|
||
# 选择 prompt
|
||
prompts = {
|
||
"savings": SAVINGS_PLAN_SYSTEM_PROMPT,
|
||
"ci": CI_PLAN_SYSTEM_PROMPT,
|
||
"iul": IUL_SYSTEM_PROMPT,
|
||
}
|
||
system_prompt = prompts.get(plan_type, SAVINGS_PLAN_SYSTEM_PROMPT)
|
||
|
||
# 调用 LLM
|
||
try:
|
||
data, response = await llm_client.structured_output(
|
||
prompt=f"请从以下PDF文本中提取保险计划书数据：\n\n{pdf_text[:20000]}",
|
||
system_prompt=system_prompt,
|
||
)
|
||
except Exception as e:
|
||
return ExtractionResult(
|
||
pdf_path=abs_path, product_name="unknown", plan_type=plan_type,
|
||
status="error", error=f"LLM 调用失败: {e}",
|
||
duration_ms=(time.time() - start) * 1000,
|
||
)
|
||
|
||
# 储蓄险数据修复：确保 total >= gcv
|
||
if plan_type == "savings" and isinstance(data.get("benefit_illustration"), list):
|
||
for row in data["benefit_illustration"]:
|
||
if not isinstance(row, dict):
|
||
continue
|
||
gcv = float(row.get("guaranteed_cash_value") or 0)
|
||
rev = float(row.get("reversionary_bonus") or 0)
|
||
term = float(row.get("terminal_dividend") or 0)
|
||
total = float(row.get("total_surrender_value") or 0)
|
||
if total < gcv:
|
||
row["total_surrender_value"] = gcv + rev + term
|
||
|
||
# 推断产品类型
|
||
detected_type = infer_plan_type(data)
|
||
|
||
# 写入缓存
|
||
if self.use_cache:
|
||
self._save_to_cache(abs_path, data)
|
||
|
||
product_name = data.get("product_name", "unknown")
|
||
return ExtractionResult(
|
||
pdf_path=abs_path, product_name=product_name,
|
||
plan_type=detected_type, status="success", data=data,
|
||
usage={"input": response.tokens.get("input", 0), "output": response.tokens.get("output", 0)} if response.tokens else None,
|
||
duration_ms=(time.time() - start) * 1000,
|
||
)
|
||
|
||
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 = data.get("product_name", "unknown")
|
||
plan_type = infer_plan_type(data)
|
||
return ExtractionResult(
|
||
pdf_path=pdf_path, product_name=product_name,
|
||
plan_type=plan_type, status="cached", data=data,
|
||
)
|
||
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 = _extract_pdf_text(abs_path)
|
||
if not text:
|
||
raise ValueError("无法提取 PDF 文本")
|
||
|
||
text = text[:6000]
|
||
|
||
system_prompt = (
|
||
"你是一位保险计划书解析专家。请从以下计划书内容中提取海报所需的关键字段。\n"
|
||
"输出 JSON 格式（不要包含 markdown 代码块标记）：\n"
|
||
'{"age": 35, "gender": "男", "currency": "USD", "sum_assured": 500000, '
|
||
'"premium_term": 5, "annual_premium": 100000, "coverage_period": "终身", '
|
||
'"key_benefits": ["身故赔偿", "全残保障"]}\n'
|
||
"注意：数值用数字，不要带货币符号。"
|
||
)
|
||
|
||
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
|