271 lines
9.6 KiB
Python
271 lines
9.6 KiB
Python
"""发票信息提取器：从票据合规预审页面提取发票数据。
|
||
|
||
根据 sandbox.jc2009.com/industries/campus/smartcampus/app/finance/invoice-precheck 页面 DOM 结构：
|
||
- 报销事由：文本输入框
|
||
- 预算项目：下拉选择框
|
||
- 差旅审批单：下拉选择框
|
||
- 选择发票：列表，每项包含 checkbox、发票号、金额、类型、公司名、日期、购方税号
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import logging
|
||
import re
|
||
from typing import Any, Dict, List, Optional
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
|
||
async def extract_invoice_list(page: Any) -> List[Dict[str, Any]]:
|
||
"""从票据合规预审页面提取发票列表。
|
||
|
||
页面中发票以列表项形式展示，每项包含：
|
||
- Checkbox（勾选状态）
|
||
- 发票号（8位数字）
|
||
- 金额（¥xxx）
|
||
- 类型（transport/lodging 等）
|
||
- 公司名
|
||
- 日期
|
||
- 购方税号
|
||
- 验真状态（"未验真" 标记）
|
||
|
||
Returns:
|
||
发票信息字典列表
|
||
"""
|
||
invoices: List[Dict[str, Any]] = []
|
||
|
||
try:
|
||
# 等待发票区域加载（checkbox 出现即可开始提取）
|
||
try:
|
||
await page.locator('input[type="checkbox"]').first.wait_for(state="attached", timeout=10_000)
|
||
except Exception:
|
||
pass # 超时也继续尝试提取
|
||
|
||
# 方式一：通过 checkbox 定位发票项（最可靠）
|
||
checkboxes = page.locator('input[type="checkbox"]')
|
||
checkbox_count = await checkboxes.count()
|
||
logger.info("found_checkboxes count=%s", checkbox_count)
|
||
|
||
if checkbox_count > 0:
|
||
# 每个 checkbox 对应一个发票项，向上找到父容器
|
||
for i in range(checkbox_count):
|
||
try:
|
||
checkbox = checkboxes.nth(i)
|
||
# 获取 checkbox 所在的父级容器（卡片或行）
|
||
parent = checkbox.locator("xpath=ancestor::div[contains(@class,'card') or contains(@class,'item') or contains(@class,'row') or contains(@class,'invoice')]").first
|
||
if await parent.count() == 0:
|
||
# fallback: 向上找 3 层
|
||
parent = checkbox.locator("xpath=ancestor::div[3]").first
|
||
|
||
text = await parent.inner_text() if await parent.count() > 0 else ""
|
||
is_checked = await checkbox.is_checked()
|
||
|
||
invoice = _parse_invoice_text(text)
|
||
if invoice:
|
||
invoice["selected"] = is_checked
|
||
invoices.append(invoice)
|
||
except Exception as e:
|
||
logger.warning("extract_checkbox_item_failed index=%s error=%s", i, str(e))
|
||
continue
|
||
|
||
# 方式二：如果方式一没提取到，通过文本匹配
|
||
if not invoices:
|
||
logger.info("fallback_to_text_extraction")
|
||
invoices = await _extract_by_text_pattern(page)
|
||
|
||
logger.info("extraction_complete total_invoices=%s", len(invoices))
|
||
return invoices
|
||
|
||
except Exception as e:
|
||
logger.error("extract_invoice_list_failed error=%s", str(e))
|
||
return []
|
||
|
||
|
||
async def _extract_by_text_pattern(page: Any) -> List[Dict[str, Any]]:
|
||
"""通过文本模式匹配提取发票信息（fallback）。"""
|
||
invoices: List[Dict[str, Any]] = []
|
||
|
||
try:
|
||
# 查找包含发票号（8位数字）的元素
|
||
invoice_pattern = page.locator("div, tr, li").filter(
|
||
has_text=re.compile(r"\d{8}.*[¥￥]\s*[\d,]+")
|
||
)
|
||
count = await invoice_pattern.count()
|
||
logger.info("text_pattern_match count=%s", count)
|
||
|
||
for i in range(count):
|
||
try:
|
||
element = invoice_pattern.nth(i)
|
||
text = await element.inner_text()
|
||
invoice = _parse_invoice_text(text)
|
||
if invoice:
|
||
# 检查是否被选中（附近有 checkbox）
|
||
checkbox = element.locator('input[type="checkbox"]')
|
||
if await checkbox.count() > 0:
|
||
invoice["selected"] = await checkbox.first.is_checked()
|
||
else:
|
||
invoice["selected"] = False
|
||
invoices.append(invoice)
|
||
except Exception as e:
|
||
logger.warning("text_pattern_extract_failed index=%s error=%s", i, str(e))
|
||
continue
|
||
|
||
except Exception as e:
|
||
logger.warning("text_pattern_extraction_error error=%s", str(e))
|
||
|
||
return invoices
|
||
|
||
|
||
def _parse_invoice_text(text: str) -> Optional[Dict[str, Any]]:
|
||
"""从发票文本中解析结构化数据。
|
||
|
||
典型格式：
|
||
87654321 · ¥553 · transport
|
||
中国铁路北京局集团有限公司 · 2026-03-12 · 购方税号 121000004000012345
|
||
"""
|
||
if not text or not text.strip():
|
||
return None
|
||
|
||
lines = [ln.strip() for ln in text.splitlines() if ln.strip()]
|
||
full_text = " ".join(lines)
|
||
|
||
# 提取发票号（8位数字）
|
||
invoice_no_match = re.search(r"(\d{8})", full_text)
|
||
invoice_no = invoice_no_match.group(1) if invoice_no_match else ""
|
||
if not invoice_no:
|
||
return None
|
||
|
||
# 提取金额（¥ 后跟数字）
|
||
amount_match = re.search(r"[¥￥]\s*([\d,]+\.?\d*)", full_text)
|
||
amount = 0.0
|
||
if amount_match:
|
||
amount_str = amount_match.group(1).replace(",", "")
|
||
try:
|
||
amount = float(amount_str)
|
||
except ValueError:
|
||
pass
|
||
|
||
# 提取类型（transport/lodging/meal/other）
|
||
invoice_type = ""
|
||
for tp in ["transport", "lodging", "meal", "other"]:
|
||
if tp in full_text.lower():
|
||
invoice_type = tp
|
||
break
|
||
|
||
# 提取公司名（中文公司名，包含常见后缀）
|
||
company_match = re.search(
|
||
r"([\u4e00-\u9fa5]{2,}(?:公司|集团|酒店|商贸|有限|商店|宾馆|饭店))", full_text
|
||
)
|
||
company = company_match.group(1) if company_match else ""
|
||
|
||
# 提取日期（YYYY-MM-DD）
|
||
date_match = re.search(r"(\d{4}-\d{2}-\d{2})", full_text)
|
||
date_str = date_match.group(1) if date_match else ""
|
||
|
||
# 提取购方税号
|
||
tax_no_match = re.search(r"购方税号\s*(\S+)", full_text)
|
||
tax_no = tax_no_match.group(1) if tax_no_match else ""
|
||
if not tax_no:
|
||
# fallback: 查找 15-20 位字母数字混合税号
|
||
tax_no_fallback = re.search(r"([0-9A-Z]{15,20})", full_text)
|
||
if tax_no_fallback:
|
||
tax_no = tax_no_fallback.group(1)
|
||
|
||
# 判断验真状态
|
||
verified = "未验真" not in full_text
|
||
|
||
return {
|
||
"invoice_no": invoice_no,
|
||
"amount": amount,
|
||
"invoice_type": invoice_type,
|
||
"company": company,
|
||
"date": date_str,
|
||
"tax_no": tax_no,
|
||
"verified": verified,
|
||
}
|
||
|
||
|
||
async def extract_invoice_detail(page: Any, invoice_index: int = 0) -> Optional[Dict[str, Any]]:
|
||
"""提取单个发票的详细信息（点击进入详情页后调用）。
|
||
|
||
Args:
|
||
page: Playwright 页面对象
|
||
invoice_index: 发票索引（从0开始）
|
||
|
||
Returns:
|
||
发票详细信息字典
|
||
"""
|
||
try:
|
||
rows = page.locator("tr").filter(has_text=re.compile(r"\d{6,}"))
|
||
count = await rows.count()
|
||
if invoice_index >= count:
|
||
logger.warning("invoice_index_out_of_range index=%s total=%s", invoice_index, count)
|
||
return None
|
||
|
||
row = rows.nth(invoice_index)
|
||
text = await row.inner_text()
|
||
return _parse_invoice_text(text)
|
||
|
||
except Exception as e:
|
||
logger.error("extract_invoice_detail_failed error=%s", str(e))
|
||
return None
|
||
|
||
|
||
async def extract_form_fields(page: Any) -> Dict[str, str]:
|
||
"""提取页面表单字段（报销事由、预算项目、差旅审批单）。"""
|
||
result: Dict[str, str] = {}
|
||
|
||
# 报销事由
|
||
try:
|
||
reason_input = page.get_by_label("报销事由")
|
||
if await reason_input.count() > 0:
|
||
result["reimbursement_reason"] = (await reason_input.input_value()).strip()
|
||
except Exception:
|
||
pass
|
||
|
||
# 预算项目（下拉框选中项）
|
||
try:
|
||
budget_selects = page.locator("select")
|
||
select_count = await budget_selects.count()
|
||
for i in range(select_count):
|
||
sel = budget_selects.nth(i)
|
||
text = await sel.inner_text()
|
||
if "GDST" in text or "乡村振兴" in text or "预算" in text:
|
||
# 获取选中的 option
|
||
selected = sel.locator("option:checked")
|
||
if await selected.count() > 0:
|
||
result["budget_project"] = (await selected.first.inner_text()).strip()
|
||
else:
|
||
result["budget_project"] = text.strip()[:100]
|
||
break
|
||
except Exception:
|
||
pass
|
||
|
||
# 差旅审批单（下拉框选中项）
|
||
try:
|
||
travel_selects = page.locator("select")
|
||
select_count = await travel_selects.count()
|
||
for i in range(select_count):
|
||
sel = travel_selects.nth(i)
|
||
text = await sel.inner_text()
|
||
if "CL-" in text or "差旅" in text or "北京" in text:
|
||
selected = sel.locator("option:checked")
|
||
if await selected.count() > 0:
|
||
result["travel_approval"] = (await selected.first.inner_text()).strip()
|
||
else:
|
||
result["travel_approval"] = text.strip()[:100]
|
||
break
|
||
except Exception:
|
||
pass
|
||
|
||
return result
|
||
|
||
|
||
def format_invoice_data(invoices: List[Dict[str, Any]]) -> Dict[str, Any]:
|
||
"""格式化发票数据为结构化格式。"""
|
||
return {
|
||
"total_count": len(invoices),
|
||
"invoices": invoices,
|
||
"extraction_complete": True,
|
||
}
|