完成差旅发票录入流程
This commit is contained in:
wandering
+208
-36
@@ -1,65 +1,237 @@
|
||||
"""发票提取编排
|
||||
|
||||
串联 PDF 读取 → LLM 提取 → 支付截图匹配 → 分类，生成支付记录列表。
|
||||
统一扫描目录下所有文件（PDF + 图片），通过 LLM 提取结构化数据，
|
||||
根据 LLM 返回的「invoice_type」字段自动分类为发票/支付记录/出差事前申请单。
|
||||
|
||||
对外接口:
|
||||
extract_invoices(directory) -> tuple[list[dict], dict]
|
||||
extract_invoices(directory) -> tuple[list[dict], list[dict], dict]
|
||||
"""
|
||||
|
||||
import json
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
|
||||
from .. import get_logger
|
||||
from .invoice import classify_invoice_batch
|
||||
from .llm_extractor import extract_invoice_from_text
|
||||
from .llm_extractor import extract_document
|
||||
from .matcher import match_invoices_to_cards
|
||||
from .pdf import extract_text_from_pdf, find_pdf_files
|
||||
|
||||
log = get_logger("extractor")
|
||||
|
||||
|
||||
def extract_invoices(
|
||||
directory: str = ".",
|
||||
) -> tuple[list[dict[str, str]], dict[str, list[dict[str, str]]]]:
|
||||
"""扫描目录下所有 PDF,提取发票信息并匹配支付记录
|
||||
def _classify_invoice_batch(
|
||||
invoices: list[dict[str, str]],
|
||||
) -> dict[str, list[dict[str, str]]]:
|
||||
"""按发票类型分组"""
|
||||
travel: list[dict[str, str]] = []
|
||||
general: list[dict[str, str]] = []
|
||||
application: list[dict[str, str]] = []
|
||||
for inv in invoices:
|
||||
inv_type = inv.get("invoice_type", "general")
|
||||
if inv_type == "application":
|
||||
application.append(inv)
|
||||
elif inv_type in ("train", "hotel"):
|
||||
travel.append(inv)
|
||||
else:
|
||||
general.append(inv)
|
||||
return {"travel": travel, "general": general, "application": application}
|
||||
|
||||
|
||||
# JSON 缓存目录（相对于源文件目录）
|
||||
CACHE_DIR_NAME = ".invoice_cache"
|
||||
|
||||
# 支持的文件扩展名
|
||||
SUPPORTED_EXTENSIONS = {".pdf", ".jpg", ".jpeg", ".png", ".webp", ".bmp"}
|
||||
|
||||
|
||||
def _get_cache_dir(source_dir: Path) -> Path:
|
||||
"""获取缓存目录路径"""
|
||||
cache_dir = source_dir / CACHE_DIR_NAME
|
||||
cache_dir.mkdir(exist_ok=True)
|
||||
return cache_dir
|
||||
|
||||
|
||||
def _get_json_path(file_path: Path, cache_dir: Path) -> Path:
|
||||
"""根据文件路径生成对应的 JSON 缓存路径"""
|
||||
return cache_dir / f"{file_path.stem}.json"
|
||||
|
||||
|
||||
def _save_to_cache(file_path: Path, extracted_data: dict[str, Any], cache_dir: Path) -> Path:
|
||||
"""将提取结果保存到 JSON 缓存文件，并记录源文件路径"""
|
||||
json_path = _get_json_path(file_path, cache_dir)
|
||||
cache_data = {
|
||||
"source_file": str(file_path),
|
||||
"source_filename": file_path.name,
|
||||
"extracted_data": extracted_data,
|
||||
}
|
||||
with open(json_path, "w", encoding="utf-8") as f:
|
||||
json.dump(cache_data, f, ensure_ascii=False, indent=2)
|
||||
log.info(f"提取结果已缓存: {json_path.name}")
|
||||
return json_path
|
||||
|
||||
|
||||
def _load_from_cache(json_path: Path) -> dict[str, Any] | None:
|
||||
"""从 JSON 缓存文件加载提取结果"""
|
||||
if not json_path.exists():
|
||||
return None
|
||||
try:
|
||||
with open(json_path, encoding="utf-8") as f:
|
||||
cache_data: dict[str, Any] = json.load(f)
|
||||
result: dict[str, Any] | None = cache_data.get("extracted_data")
|
||||
return result
|
||||
except Exception as e:
|
||||
log.warning(f"读取缓存失败 {json_path.name}: {e}")
|
||||
return None
|
||||
|
||||
|
||||
def _extract_document(file_path: Path, cache_dir: Path) -> dict[str, str] | None:
|
||||
"""提取单个文件的结构化信息，优先使用缓存。
|
||||
|
||||
根据 LLM 返回的「invoice_type」字段自动分类：
|
||||
- "payment" -> 支付记录
|
||||
- "application" -> 申请单
|
||||
- 有 "invoice_number" -> 发票
|
||||
- 其他 -> 无法识别
|
||||
|
||||
Args:
|
||||
file_path: 文件路径（PDF 或图片）。
|
||||
cache_dir: JSON 缓存目录。
|
||||
|
||||
Returns:
|
||||
(payment_records, groups): 支付记录列表和按发票类型分组的字典
|
||||
groups = {'travel': [差旅发票], 'general': [普通发票]}
|
||||
提取结果字典，失败时返回 None。
|
||||
"""
|
||||
pdf_files = find_pdf_files(directory)
|
||||
if not pdf_files:
|
||||
log.warning("未找到 PDF 文件")
|
||||
return [], {"travel": [], "general": []}
|
||||
json_path = _get_json_path(file_path, cache_dir)
|
||||
cached = _load_from_cache(json_path)
|
||||
if cached:
|
||||
cached["_source_file"] = file_path.name
|
||||
log.info(f"使用缓存: {file_path.name}")
|
||||
return cached
|
||||
|
||||
log.info(f"发现 {len(pdf_files)} 个 PDF 文件")
|
||||
log.info(f"使用多模态提取: {file_path.name}")
|
||||
try:
|
||||
result = extract_document(file_path)
|
||||
if result:
|
||||
result["_source_file"] = file_path.name
|
||||
_save_to_cache(file_path, result, cache_dir)
|
||||
return result
|
||||
except Exception as e:
|
||||
log.warning(f"多模态提取失败: {file_path.name} ({e})")
|
||||
|
||||
return None
|
||||
|
||||
|
||||
def _find_all_files(directory: str) -> list[Path]:
|
||||
"""扫描目录下所有支持的文件（PDF + 图片）"""
|
||||
dir_path = Path(directory)
|
||||
files = [f for f in dir_path.iterdir() if f.is_file() and f.suffix.lower() in SUPPORTED_EXTENSIONS]
|
||||
return sorted(files)
|
||||
|
||||
|
||||
def _save_match_result(payment_records: list[dict[str, Any]], cache_dir: Path) -> None:
|
||||
"""将发票与支付记录的匹配结果保存到缓存。"""
|
||||
match_data: dict[str, list[dict[str, Any]]] = {}
|
||||
for record in payment_records:
|
||||
card_source = record.get("_source_file", "")
|
||||
matched = record.get("_matched_invoices", [])
|
||||
card_amount = record.get("card_amount", "")
|
||||
|
||||
if matched:
|
||||
invoice_list = [
|
||||
{
|
||||
"file": inv.get("_source_file", ""),
|
||||
"type": inv.get("invoice_type", ""),
|
||||
"amount": inv.get("total_amount", inv.get("total_amount", "")),
|
||||
}
|
||||
for inv in matched
|
||||
]
|
||||
|
||||
if card_source:
|
||||
match_data[f"{card_source} (¥{card_amount})"] = invoice_list
|
||||
else:
|
||||
key = "__unmatched__"
|
||||
if key not in match_data:
|
||||
match_data[key] = []
|
||||
match_data[key].extend(invoice_list)
|
||||
|
||||
if not match_data:
|
||||
return
|
||||
|
||||
match_path = cache_dir / "match_result.json"
|
||||
with open(match_path, "w", encoding="utf-8") as f:
|
||||
json.dump(match_data, f, ensure_ascii=False, indent=2)
|
||||
log.info(f"匹配结果已缓存: {match_path.name}")
|
||||
|
||||
|
||||
def extract_invoices(
|
||||
directory: str = ".",
|
||||
) -> tuple[list[dict[str, str]], list[dict[str, str]], dict[str, list[dict[str, str]]]]:
|
||||
"""扫描目录下所有文件，提取信息并匹配支付记录
|
||||
|
||||
统一使用 LLM 提取，根据返回的「invoice_type」自动分类：
|
||||
- 发票（有 invoice_number）-> 参与金额匹配
|
||||
- 支付记录（invoice_type="payment"）-> 参与金额匹配
|
||||
- 出差事前申请单 -> 单独存储，不参与匹配
|
||||
|
||||
Returns:
|
||||
(payment_records, applications, groups):
|
||||
- payment_records: 支付记录列表（仅包含真实发票，不含申请单）
|
||||
- applications: 出差事前申请单列表（单独存储，不参与支付匹配）
|
||||
- groups: 按文档类型分组的字典
|
||||
{'travel': [差旅发票], 'general': [普通发票], 'application': [出差事前申请单]}
|
||||
"""
|
||||
source_dir = Path(directory)
|
||||
cache_dir = _get_cache_dir(source_dir)
|
||||
|
||||
all_files = _find_all_files(directory)
|
||||
if not all_files:
|
||||
log.warning("未找到支持的文件")
|
||||
return [], [], {"travel": [], "general": [], "application": []}
|
||||
|
||||
log.info(f"发现 {len(all_files)} 个文件")
|
||||
|
||||
all_invoices = []
|
||||
for pdf_path in pdf_files:
|
||||
text = extract_text_from_pdf(pdf_path)
|
||||
if not text:
|
||||
log.warning(f"未能提取文本: {pdf_path.name}")
|
||||
all_cards = []
|
||||
applications = []
|
||||
|
||||
for file_path in all_files:
|
||||
result = _extract_document(file_path, cache_dir)
|
||||
|
||||
if not result:
|
||||
log.warning(f"未能解析: {file_path.name}")
|
||||
continue
|
||||
|
||||
invoice = extract_invoice_from_text(text, pdf_path.name)
|
||||
inv_type = result.get("invoice_type", "")
|
||||
|
||||
if invoice and invoice.get("发票号码"):
|
||||
all_invoices.append(invoice)
|
||||
log.info(f"[{invoice['发票类型']}] 已解析: {pdf_path.name}")
|
||||
if inv_type == "application":
|
||||
applications.append(result)
|
||||
log.info(f"[{inv_type}] 已解析: {file_path.name}")
|
||||
elif inv_type == "payment":
|
||||
all_cards.append(result)
|
||||
log.info(f"[{inv_type}] 已解析: {file_path.name}")
|
||||
elif result.get("invoice_number"):
|
||||
all_invoices.append(result)
|
||||
log.info(f"[{inv_type}] 已解析: {file_path.name}")
|
||||
else:
|
||||
log.warning(f"未能解析: {pdf_path.name}")
|
||||
log.warning(f"无法分类: {file_path.name} (invoice_type={inv_type})")
|
||||
|
||||
if all_invoices:
|
||||
log.info(f"共处理 {len(all_invoices)} 张发票")
|
||||
else:
|
||||
log.info(f"分类结果: 发票 {len(all_invoices)} 张, 支付记录 {len(all_cards)} 条, 申请单 {len(applications)} 份")
|
||||
|
||||
if not all_invoices:
|
||||
log.warning("未成功解析任何发票")
|
||||
|
||||
# 将支付截图与发票进行金额匹配，返回以支付记录为主键的列表
|
||||
payment_records = match_invoices_to_cards(all_invoices, directory)
|
||||
payment_records = match_invoices_to_cards(all_invoices, all_cards)
|
||||
_save_match_result(payment_records, cache_dir)
|
||||
|
||||
# 从支付记录中还原所有发票用于分类
|
||||
all_invoices_restored: list[dict[str, str]] = []
|
||||
# 构建分类
|
||||
all_documents: list[dict[str, str]] = []
|
||||
for record in payment_records:
|
||||
all_invoices_restored.extend(record.get("_matched_invoices", []))
|
||||
all_documents.extend(record.get("_matched_invoices", []))
|
||||
all_documents.extend(applications)
|
||||
|
||||
groups = classify_invoice_batch(all_invoices_restored)
|
||||
log.info(f"差旅发票: {len(groups['travel'])} 张, 普通发票: {len(groups['general'])} 张")
|
||||
groups = _classify_invoice_batch(all_documents)
|
||||
log.info(
|
||||
f"文档分类: 差旅发票 {len(groups['travel'])} 张, "
|
||||
f"普通发票 {len(groups['general'])} 张, "
|
||||
f"出差事前申请单 {len(groups['application'])} 份"
|
||||
)
|
||||
|
||||
return payment_records, groups
|
||||
return payment_records, applications, groups
|
||||
|
||||
Reference in New Issue
Block a user