775 lines
32 KiB
Python
775 lines
32 KiB
Python
"""票据合规预审规则引擎（唯一业务内核）。
|
||
|
||
四步审核（对齐《核心建设内容：各部门"Skill"深度需求规格书》财务处 Skill 001）：
|
||
Step 1 字段完整性校验（F1-F4）
|
||
Step 2 三方交叉验真（V1 真伪状态 / V2 购买方税号白名单 / V3 日期逻辑）
|
||
Step 3 经费标准硬约束（S1 住宿费上限 / S2 交通等级 / S3 项目归属）
|
||
Step 4 风控标记（R1 连号 / R2 同商家集中开票 / R3 节假日开票）
|
||
|
||
每条规则输出 {rule_id, name, level, detail, rule_ref}；level ∈ pass/warn/reject。
|
||
总判定：任一 reject → 驳回；有 warn → 待确认；否则 通过。
|
||
预审结果仅作提示，不构成最终报销结论；真伪以人工/官方查验为准。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import json
|
||
import re
|
||
import uuid
|
||
from datetime import date, datetime, timedelta
|
||
from typing import Any, Optional
|
||
|
||
from db.connection import get_conn, init_db
|
||
from util.cny_amount import parse_cny_upper
|
||
from util.logging_config import get_skill_logger
|
||
from util.timeutil import now_unix
|
||
|
||
_VERDICT_PASS = "通过"
|
||
_VERDICT_WARN = "待确认"
|
||
_VERDICT_REJECT = "驳回"
|
||
|
||
_DATE_PATTERNS = ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%Y年%m月%d日")
|
||
|
||
# 发票号码已知位数：数电票 20 位；增值税普/专票 8 位；旧版 10/12 位
|
||
_KNOWN_NUMBER_LENGTHS = {8, 10, 12, 20}
|
||
|
||
_INVOICE_KEY_ALIASES = {
|
||
"invoiceType": ("invoice_type", "发票类型", "type"),
|
||
"invoiceCode": ("invoice_code", "发票代码"),
|
||
"invoiceNumber": ("invoice_number", "invoice_no", "发票号码"),
|
||
"issueDate": ("issue_date", "开票日期", "date"),
|
||
"amount": ("amount", "金额", "total_amount", "价税合计"),
|
||
"amountInWords": ("amount_in_words", "大写金额", "amount_words"),
|
||
"taxRate": ("tax_rate", "税率"),
|
||
"buyerName": ("buyer_name", "购买方名称", "buyer"),
|
||
"buyerTaxNo": ("buyer_tax_no", "购买方税号", "buyer_tax_id"),
|
||
"sellerName": ("seller_name", "销售方名称", "seller"),
|
||
"sellerTaxNo": ("seller_tax_no", "销售方税号", "seller_tax_id"),
|
||
"expenseCategory": ("expense_category", "category", "费用类别"),
|
||
"city": ("city", "城市", "消费城市"),
|
||
"days": ("days", "天数", "住宿天数"),
|
||
"ticketClass": ("ticket_class", "座位等级", "席别", "舱位等级"),
|
||
"manualVerifyResult": ("manual_verify_result", "查验结果", "人工查验结果"),
|
||
}
|
||
|
||
_TICKET_CLASS_LEVELS = (
|
||
(("商务座", "头等舱"), 3),
|
||
(("一等座", "软卧", "公务舱", "一等舱"), 2),
|
||
(("二等座", "硬座", "硬卧", "软座", "经济舱"), 1),
|
||
)
|
||
|
||
_CHECK_PLATFORM_HINT = "全国增值税发票查验平台（inv-veri.chinatax.gov.cn）"
|
||
|
||
|
||
def _log():
|
||
try:
|
||
return get_skill_logger()
|
||
except RuntimeError:
|
||
return None
|
||
|
||
|
||
def _norm_invoice(raw: dict[str, Any]) -> dict[str, Any]:
|
||
"""将输入发票字段归一化为 camelCase 键（兼容 snake_case 与常见中文键）。"""
|
||
out: dict[str, Any] = {}
|
||
for key, value in raw.items():
|
||
camel = _camel(key)
|
||
if camel in _INVOICE_KEY_ALIASES:
|
||
out.setdefault(camel, value)
|
||
continue
|
||
matched = False
|
||
for canonical, aliases in _INVOICE_KEY_ALIASES.items():
|
||
if key in aliases:
|
||
out.setdefault(canonical, value)
|
||
matched = True
|
||
break
|
||
if not matched:
|
||
out.setdefault(camel, value)
|
||
return out
|
||
|
||
|
||
def _camel(name: str) -> str:
|
||
parts = re.split(r"[\s_]+", str(name).strip())
|
||
if len(parts) <= 1:
|
||
return str(name)
|
||
head, *rest = parts
|
||
return head + "".join(p[:1].upper() + p[1:] for p in rest)
|
||
|
||
|
||
def _to_float(value: Any) -> Optional[float]:
|
||
if value is None or value == "":
|
||
return None
|
||
if isinstance(value, (int, float)):
|
||
return float(value)
|
||
text = str(value).strip()
|
||
text = re.sub(r"[¥￥元人民币\s,，]", "", text)
|
||
try:
|
||
return float(text)
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def _parse_date(value: Any) -> Optional[date]:
|
||
if value is None or value == "":
|
||
return None
|
||
if isinstance(value, date):
|
||
return value
|
||
text = str(value).strip()
|
||
for pattern in _DATE_PATTERNS:
|
||
try:
|
||
return datetime.strptime(text, pattern).date()
|
||
except ValueError:
|
||
continue
|
||
return None
|
||
|
||
|
||
def _fmt_date(value: Optional[date]) -> str:
|
||
return value.isoformat() if value else ""
|
||
|
||
|
||
def _load_whitelist_taxnos() -> set[str]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT tax_no FROM taxid_whitelist")
|
||
return {str(row[0]).strip().upper() for row in cur.fetchall()}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _load_holiday_dates() -> set[str]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute("SELECT holiday_date FROM holiday_calendar")
|
||
return {str(row[0]) for row in cur.fetchall()}
|
||
finally:
|
||
conn.close()
|
||
|
||
|
||
def _resolve_city_tier(city: str) -> Optional[tuple[str, str]]:
|
||
"""按城市名归档城市等级，返回 (city_tier, basis)。"""
|
||
if not city:
|
||
return None
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"SELECT city_tier, city_example FROM expense_standards WHERE category = '住宿费' GROUP BY city_tier, city_example"
|
||
)
|
||
rows = cur.fetchall()
|
||
finally:
|
||
conn.close()
|
||
for city_tier, city_example in rows:
|
||
names = [n.strip() for n in str(city_example or "").split(",") if n.strip()]
|
||
if city in names:
|
||
return city_tier, "标准库城市示例"
|
||
return None
|
||
|
||
|
||
def _lookup_standard(category: str, city_tier: Optional[str], staff_title: Optional[str]) -> Optional[dict[str, Any]]:
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
sql = "SELECT city_tier, staff_title, max_amount, unit, basis_doc, clause FROM expense_standards WHERE category = ?"
|
||
params: list[Any] = [category]
|
||
if city_tier:
|
||
sql += " AND city_tier = ?"
|
||
params.append(city_tier)
|
||
if staff_title:
|
||
sql += " AND staff_title = ?"
|
||
params.append(staff_title)
|
||
cur.execute(sql + " LIMIT 1", params)
|
||
row = cur.fetchone()
|
||
finally:
|
||
conn.close()
|
||
if not row:
|
||
return None
|
||
return {
|
||
"city_tier": row[0],
|
||
"staff_title": row[1],
|
||
"max_amount": float(row[2]),
|
||
"unit": row[3] or "",
|
||
"basis_doc": row[4] or "",
|
||
"clause": row[5] or "",
|
||
}
|
||
|
||
|
||
def _ref(basis_doc: str, clause: str) -> str:
|
||
if basis_doc and clause:
|
||
return f"{basis_doc}{clause}"
|
||
return basis_doc or clause or ""
|
||
|
||
|
||
def _item(rule_id: str, name: str, level: str, detail: str, rule_ref: str = "") -> dict[str, str]:
|
||
return {"rule_id": rule_id, "name": name, "level": level, "detail": detail, "rule_ref": rule_ref}
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Step 1 字段完整性校验
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _check_field_completeness(inv: dict[str, Any], today: date) -> list[dict[str, str]]:
|
||
items: list[dict[str, str]] = []
|
||
missing = [
|
||
label
|
||
for key, label in (
|
||
("invoiceNumber", "发票号码"),
|
||
("issueDate", "开票日期"),
|
||
("amount", "金额"),
|
||
("sellerName", "销售方名称"),
|
||
)
|
||
if inv.get(key) in (None, "")
|
||
]
|
||
if missing:
|
||
items.append(
|
||
_item(
|
||
"F1",
|
||
"必填字段齐全",
|
||
"warn",
|
||
f"以下关键字段缺失或为空：{'、'.join(missing)}。请重新拍摄/补录后再提交报销。",
|
||
"票据预审·字段校验",
|
||
)
|
||
)
|
||
|
||
number = str(inv.get("invoiceNumber") or "").strip().replace(" ", "")
|
||
if number and not number.isdigit():
|
||
items.append(_item("F2", "发票号码格式", "warn", f"发票号码含非数字字符：{number}", "票据预审·字段校验"))
|
||
elif number and len(number) not in _KNOWN_NUMBER_LENGTHS:
|
||
items.append(
|
||
_item(
|
||
"F2",
|
||
"发票号码格式",
|
||
"warn",
|
||
f"发票号码 {len(number)} 位，不属于已知位数（数电票 20 位、普/专票 8 位、旧版 10/12 位），疑似异常，请人工核验票面。",
|
||
"票据预审·字段校验",
|
||
)
|
||
)
|
||
|
||
issue_date = _parse_date(inv.get("issueDate"))
|
||
raw_date = str(inv.get("issueDate") or "").strip()
|
||
if raw_date and issue_date is None:
|
||
items.append(_item("F3", "开票日期合法性", "reject", f"开票日期无法解析：{raw_date}（应为 YYYY-MM-DD）", "票据预审·字段校验"))
|
||
elif issue_date and issue_date > today:
|
||
items.append(_item("F3", "开票日期合法性", "reject", f"开票日期 {issue_date.isoformat()} 晚于今日，票面异常。", "票据预审·字段校验"))
|
||
elif issue_date:
|
||
items.append(_item("F3", "开票日期合法性", "pass", f"开票日期 {issue_date.isoformat()} 合法。", "票据预审·字段校验"))
|
||
|
||
amount = _to_float(inv.get("amount"))
|
||
words = str(inv.get("amountInWords") or "").strip()
|
||
if amount is not None and words:
|
||
words_value = parse_cny_upper(words)
|
||
if words_value is None:
|
||
items.append(_item("F4", "大小写金额一致", "warn", f"大写金额“{words}”无法解析，请人工核对大小写金额。", "票据预审·字段校验"))
|
||
elif abs(words_value - amount) > 0.01:
|
||
items.append(
|
||
_item(
|
||
"F4",
|
||
"大小写金额一致",
|
||
"reject",
|
||
f"大写金额 {words_value:.2f} 元与小写金额 {amount:.2f} 元不一致，票面异常。",
|
||
"票据预审·字段校验",
|
||
)
|
||
)
|
||
else:
|
||
items.append(_item("F4", "大小写金额一致", "pass", f"大小写金额一致（{amount:.2f} 元）。", "票据预审·字段校验"))
|
||
return items
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Step 2 三方交叉验真
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _check_cross_verification(
|
||
inv: dict[str, Any],
|
||
context: dict[str, Any],
|
||
whitelist: set[str],
|
||
today: date,
|
||
) -> list[dict[str, str]]:
|
||
items: list[dict[str, str]] = []
|
||
|
||
verify = str(inv.get("manualVerifyResult") or "").strip().lower()
|
||
if verify in ("fail", "false", "不符", "假票", "查验不符"):
|
||
items.append(_item("V1", "发票真伪", "reject", "人工查验结果为“不符”，判定为问题票据。", _CHECK_PLATFORM_HINT))
|
||
elif verify in ("pass", "true", "通过", "真票", "查验通过"):
|
||
items.append(_item("V1", "发票真伪", "pass", "人工查验结果为“通过”。", _CHECK_PLATFORM_HINT))
|
||
else:
|
||
items.append(
|
||
_item(
|
||
"V1",
|
||
"发票真伪",
|
||
"warn",
|
||
f"尚未查验真伪：请经办人到{_CHECK_PLATFORM_HINT}录入票面信息核验，并在本报告回填查验结果。",
|
||
_CHECK_PLATFORM_HINT,
|
||
)
|
||
)
|
||
|
||
buyer_tax_no = str(inv.get("buyerTaxNo") or "").strip()
|
||
if not buyer_tax_no:
|
||
items.append(_item("V2", "购买方税号白名单", "warn", "票面缺少购买方税号，无法进行白名单比对，请补录抬头信息。", "票据预审·税号白名单"))
|
||
elif buyer_tax_no.strip().upper() in whitelist:
|
||
buyer_name = str(inv.get("buyerName") or "").strip()
|
||
items.append(_item("V2", "购买方税号白名单", "pass", f"购买方税号在我校准予报销白名单内{('（' + buyer_name + '）') if buyer_name else ''}。", "票据预审·税号白名单"))
|
||
else:
|
||
items.append(
|
||
_item(
|
||
"V2",
|
||
"购买方税号白名单",
|
||
"reject",
|
||
f"购买方税号 {buyer_tax_no} 不在我校准予报销税号白名单内，抬头开错，须换开后方可报销。",
|
||
"票据预审·税号白名单",
|
||
)
|
||
)
|
||
|
||
issue_date = _parse_date(inv.get("issueDate"))
|
||
trip_start = _parse_date(context.get("trip_start"))
|
||
trip_end = _parse_date(context.get("trip_end"))
|
||
if issue_date and trip_start:
|
||
window_end = trip_end or trip_start
|
||
if issue_date < trip_start:
|
||
items.append(
|
||
_item(
|
||
"V3",
|
||
"日期逻辑（先审批后出行）",
|
||
"warn",
|
||
f"开票日期 {issue_date.isoformat()} 早于出差开始日 {trip_start.isoformat()}，请核实行程与审批单（先审批后出行）。",
|
||
"票据预审·日期逻辑",
|
||
)
|
||
)
|
||
elif issue_date > window_end + timedelta(days=1):
|
||
items.append(
|
||
_item(
|
||
"V3",
|
||
"日期逻辑（先审批后出行）",
|
||
"warn",
|
||
f"开票日期 {issue_date.isoformat()} 晚于出差结束日 {window_end.isoformat()}（含 1 天宽限），请核实行程。",
|
||
"票据预审·日期逻辑",
|
||
)
|
||
)
|
||
else:
|
||
items.append(
|
||
_item(
|
||
"V3",
|
||
"日期逻辑（先审批后出行）",
|
||
"pass",
|
||
f"开票日期落在出差区间 [{trip_start.isoformat()}, {(trip_end or trip_start).isoformat()}] 内。",
|
||
"票据预审·日期逻辑",
|
||
)
|
||
)
|
||
return items
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Step 3 经费标准硬约束
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _check_expense_standards(inv: dict[str, Any], context: dict[str, Any]) -> list[dict[str, str]]:
|
||
items: list[dict[str, str]] = []
|
||
category = str(inv.get("expenseCategory") or "").strip()
|
||
amount = _to_float(inv.get("amount"))
|
||
staff_title = str(context.get("staff_title") or "").strip()
|
||
|
||
if "住宿" in category and amount is not None:
|
||
city = str(context.get("trip_city") or inv.get("city") or "").strip()
|
||
tier_info = _resolve_city_tier(city) if city else None
|
||
if not city:
|
||
items.append(
|
||
_item(
|
||
"S1",
|
||
"住宿费标准比对",
|
||
"warn",
|
||
"未提供消费城市，无法匹配住宿费标准，请补录城市后重新预审或人工对照《差旅费管理办法》。",
|
||
_ref("《差旅费管理办法》", "第十条"),
|
||
)
|
||
)
|
||
elif not tier_info:
|
||
items.append(
|
||
_item(
|
||
"S1",
|
||
"住宿费标准比对",
|
||
"warn",
|
||
f"城市“{city}”不在标准库城市等级中，无法匹配住宿费标准，请人工对照《差旅费管理办法》或请管理员维护该城市。",
|
||
_ref("《差旅费管理办法》", "第十条"),
|
||
)
|
||
)
|
||
else:
|
||
standard = _lookup_standard("住宿费", tier_info[0], staff_title or None)
|
||
days = _to_float(inv.get("days")) or 1.0
|
||
days = max(days, 1.0)
|
||
if not standard:
|
||
items.append(
|
||
_item(
|
||
"S1",
|
||
"住宿费标准比对",
|
||
"warn",
|
||
f"未匹配到“{tier_info[0]}/{staff_title or '未提供职级'}”的住宿费标准，请人工对照《差旅费管理办法》。",
|
||
_ref("《差旅费管理办法》", "第十条"),
|
||
)
|
||
)
|
||
else:
|
||
per_day = amount / days
|
||
limit = standard["max_amount"]
|
||
ref = _ref(standard["basis_doc"], standard["clause"])
|
||
if per_day > limit + 0.001:
|
||
items.append(
|
||
_item(
|
||
"S1",
|
||
"住宿费标准比对",
|
||
"reject",
|
||
(
|
||
f"住宿费折合 {per_day:.2f} 元/人/天，超过“{standard['city_tier']}/{standard['staff_title']}”"
|
||
f"上限 {limit:.2f} 元/人/天，超标 {per_day - limit:.2f} 元。"
|
||
),
|
||
ref,
|
||
)
|
||
)
|
||
else:
|
||
items.append(
|
||
_item(
|
||
"S1",
|
||
"住宿费标准比对",
|
||
"pass",
|
||
f"住宿费折合 {per_day:.2f} 元/人/天，未超“{standard['city_tier']}/{standard['staff_title']}”上限 {limit:.2f} 元/人/天。",
|
||
ref,
|
||
)
|
||
)
|
||
|
||
if "交通" in category or "出租" in str(inv.get("invoiceType") or ""):
|
||
ticket_class = str(inv.get("ticketClass") or "").strip()
|
||
standard = _lookup_standard("交通费", None, staff_title or None)
|
||
if ticket_class and standard:
|
||
class_level = _ticket_class_level(ticket_class)
|
||
ref = _ref(standard["basis_doc"], standard["clause"])
|
||
if class_level is None:
|
||
items.append(_item("S2", "交通席位等级", "warn", f"席位/舱位等级“{ticket_class}”无法识别，请人工对照职级可乘等级。", ref))
|
||
elif class_level > standard["max_amount"]:
|
||
items.append(
|
||
_item(
|
||
"S2",
|
||
"交通席位等级",
|
||
"warn",
|
||
f"“{ticket_class}”等级高于“{standard['staff_title']}”可乘等级（代码 {standard['max_amount']:.0f}），请确认是否符合职级规定或已获审批。",
|
||
ref,
|
||
)
|
||
)
|
||
else:
|
||
items.append(_item("S2", "交通席位等级", "pass", f"“{ticket_class}”符合“{standard['staff_title']}”可乘等级。", ref))
|
||
|
||
project_no = str(context.get("project_no") or "").strip()
|
||
if project_no:
|
||
items.append(
|
||
_item(
|
||
"S3",
|
||
"项目经费归属",
|
||
"pass",
|
||
f"本笔将计入预算项目 {project_no}；项目余额校验待财务系统对接后启用。",
|
||
"票据预审·项目归属",
|
||
)
|
||
)
|
||
return items
|
||
|
||
|
||
def _ticket_class_level(ticket_class: str) -> Optional[int]:
|
||
for keywords, level in _TICKET_CLASS_LEVELS:
|
||
if any(k in ticket_class for k in keywords):
|
||
return level
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# Step 4 风控标记
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _check_holiday(inv: dict[str, Any], holidays: set[str]) -> list[dict[str, str]]:
|
||
items: list[dict[str, str]] = []
|
||
issue_date = _parse_date(inv.get("issueDate"))
|
||
if not issue_date:
|
||
return items
|
||
iso = issue_date.isoformat()
|
||
if iso in holidays:
|
||
items.append(
|
||
_item("R3", "节假日开票", "warn", f"开票日期 {iso} 为法定节假日，请核实消费真实性。", "票据预审·风控标记")
|
||
)
|
||
elif issue_date.weekday() >= 5:
|
||
items.append(
|
||
_item("R3", "节假日开票", "warn", f"开票日期 {iso} 为周末（{'周六' if issue_date.weekday() == 5 else '周日'}），请核实消费真实性。", "票据预审·风控标记")
|
||
)
|
||
return items
|
||
|
||
|
||
def _check_batch_risks(invoices: list[dict[str, Any]]) -> dict[int, list[dict[str, str]]]:
|
||
"""R1 连号发票 + R2 同商家集中开票（批内）。返回 invoice index → items。"""
|
||
extra: dict[int, list[dict[str, str]]] = {}
|
||
|
||
# R1：同发票代码且号码连续（差值 ≤ 2）
|
||
numeric_by_code: dict[str, list[tuple[int, int]]] = {}
|
||
for idx, inv in enumerate(invoices):
|
||
code = str(inv.get("invoiceCode") or "").strip() or "-"
|
||
number = str(inv.get("invoiceNumber") or "").strip()
|
||
if number.isdigit():
|
||
numeric_by_code.setdefault(code, []).append((idx, int(number)))
|
||
for _code, entries in numeric_by_code.items():
|
||
for idx, number in entries:
|
||
peers = [n for j, n in entries if j != idx and abs(n - number) <= 2]
|
||
if peers:
|
||
extra.setdefault(idx, []).append(
|
||
_item(
|
||
"R1",
|
||
"连号发票",
|
||
"warn",
|
||
f"与本批发票号码连号（相邻号码：{', '.join(str(p) for p in sorted(set(peers)))}），请核实是否拆分消费。",
|
||
"票据预审·风控标记",
|
||
)
|
||
)
|
||
|
||
# R2：同商家集中开票（批内按销方税号，缺税号回退销方名称）
|
||
sellers: dict[str, list[int]] = {}
|
||
for idx, inv in enumerate(invoices):
|
||
key = str(inv.get("sellerTaxNo") or "").strip().upper() or str(inv.get("sellerName") or "").strip()
|
||
if key:
|
||
sellers.setdefault(key, []).append(idx)
|
||
for _key, idx_list in sellers.items():
|
||
if len(idx_list) >= 3:
|
||
sample_name = str(invoices[idx_list[0]].get("sellerName") or "同一销售方").strip()
|
||
for idx in idx_list:
|
||
extra.setdefault(idx, []).append(
|
||
_item(
|
||
"R2",
|
||
"同商家集中开票",
|
||
"warn",
|
||
f"本批有 {len(idx_list)} 张发票来自“{sample_name}”，存在集中开票情形，请核实消费真实性。",
|
||
"票据预审·风控标记",
|
||
)
|
||
)
|
||
return extra
|
||
|
||
|
||
def _check_seller_history(inv: dict[str, Any]) -> Optional[dict[str, str]]:
|
||
"""R2 历史扩展：该销方近 30 天（含本张）累计 ≥ 5 张 → 黄灯。"""
|
||
seller_tax_no = str(inv.get("sellerTaxNo") or "").strip()
|
||
if not seller_tax_no:
|
||
return None
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
cur.execute(
|
||
"""
|
||
SELECT COUNT(*) FROM invoice_records
|
||
WHERE seller_tax_no = ? AND created_at >= ?
|
||
""",
|
||
(seller_tax_no, now_unix() - 30 * 24 * 3600),
|
||
)
|
||
history = int(cur.fetchone()[0])
|
||
finally:
|
||
conn.close()
|
||
if history + 1 >= 5:
|
||
seller_name = str(inv.get("sellerName") or seller_tax_no).strip()
|
||
return _item(
|
||
"R2",
|
||
"同商家集中开票",
|
||
"warn",
|
||
f"销售方“{seller_name}”近 30 天累计开票 {history + 1} 张（含本张），存在集中开票情形，请核实消费真实性。",
|
||
"票据预审·风控标记",
|
||
)
|
||
return None
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 内核入口
|
||
# ---------------------------------------------------------------------------
|
||
|
||
def _decide(items: list[dict[str, str]]) -> str:
|
||
levels = {i["level"] for i in items}
|
||
if "reject" in levels:
|
||
return _VERDICT_REJECT
|
||
if "warn" in levels:
|
||
return _VERDICT_WARN
|
||
return _VERDICT_PASS
|
||
|
||
|
||
def _summarize(verdict: str, invoice_number: str, items: list[dict[str, str]]) -> str:
|
||
rejects = [i for i in items if i["level"] == "reject"]
|
||
warns = [i for i in items if i["level"] == "warn"]
|
||
if verdict == _VERDICT_REJECT:
|
||
return f"发票 {invoice_number} 预审驳回：{'；'.join(i['name'] for i in rejects[:3])}"
|
||
if verdict == _VERDICT_WARN:
|
||
return f"发票 {invoice_number} 待确认：{'；'.join(i['name'] for i in warns[:3])}"
|
||
return f"发票 {invoice_number} 预审通过，全部检查项通过。"
|
||
|
||
|
||
def _validate_input(invoices: list[dict[str, Any]]) -> None:
|
||
if not invoices:
|
||
raise ValueError("发票字段为空：请提供至少一张发票的结构化字段 JSON")
|
||
if len(invoices) > 50:
|
||
raise ValueError(f"单批发票数量过多（{len(invoices)} 张），单批上限 50 张")
|
||
for idx, inv in enumerate(invoices):
|
||
if not isinstance(inv, dict):
|
||
raise ValueError(f"第 {idx + 1} 张发票字段不是 JSON 对象")
|
||
|
||
|
||
def run_prereview(invoices_raw: list[dict[str, Any]], context: dict[str, Any]) -> dict[str, Any]:
|
||
"""预审内核：四步审核 → 写库 → 生成《初审意见单》HTML。"""
|
||
from service.report_service import write_report_for_kernel
|
||
|
||
log = _log()
|
||
init_db()
|
||
invoices = [_norm_invoice(inv) for inv in invoices_raw]
|
||
_validate_input(invoices)
|
||
|
||
batch_no = now_unix().to_bytes(5, "big").hex() + uuid.uuid4().hex[:6]
|
||
today = date.today()
|
||
whitelist = _load_whitelist_taxnos()
|
||
holidays = _load_holiday_dates()
|
||
|
||
if log:
|
||
log.info("prereview_kernel_start batch_no=%s count=%s stage=kernel", batch_no, len(invoices))
|
||
|
||
batch_extra = _check_batch_risks(invoices)
|
||
results: list[dict[str, Any]] = []
|
||
|
||
conn = get_conn()
|
||
try:
|
||
cur = conn.cursor()
|
||
for idx, inv in enumerate(invoices):
|
||
items: list[dict[str, str]] = []
|
||
items.extend(_check_field_completeness(inv, today))
|
||
items.extend(_check_cross_verification(inv, context, whitelist, today))
|
||
items.extend(_check_expense_standards(inv, context))
|
||
items.extend(_check_holiday(inv, holidays))
|
||
items.extend(batch_extra.get(idx, []))
|
||
|
||
seller_key = str(inv.get("sellerTaxNo") or "").strip()
|
||
if seller_key:
|
||
history_item = _check_seller_history(inv)
|
||
if history_item and not any(i["rule_id"] == "R2" for i in items):
|
||
items.append(history_item)
|
||
|
||
verdict = _decide(items)
|
||
invoice_number = str(inv.get("invoiceNumber") or "").strip() or f"未提供-{idx + 1}"
|
||
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO invoice_records
|
||
(batch_no, invoice_type, invoice_code, invoice_number, issue_date, amount,
|
||
amount_in_words, tax_rate, buyer_name, buyer_tax_no, seller_name, seller_tax_no,
|
||
expense_category, city, days, purpose, project_no, trip_context_json)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
batch_no,
|
||
str(inv.get("invoiceType") or ""),
|
||
str(inv.get("invoiceCode") or ""),
|
||
invoice_number,
|
||
_fmt_date(_parse_date(inv.get("issueDate"))),
|
||
_to_float(inv.get("amount")),
|
||
str(inv.get("amountInWords") or ""),
|
||
_to_float(inv.get("taxRate")),
|
||
str(inv.get("buyerName") or ""),
|
||
str(inv.get("buyerTaxNo") or ""),
|
||
str(inv.get("sellerName") or ""),
|
||
str(inv.get("sellerTaxNo") or ""),
|
||
str(inv.get("expenseCategory") or ""),
|
||
str(inv.get("city") or ""),
|
||
int(_to_float(inv.get("days")) or 0),
|
||
str(context.get("purpose") or ""),
|
||
str(context.get("project_no") or ""),
|
||
json.dumps(context, ensure_ascii=False),
|
||
),
|
||
)
|
||
invoice_id = int(cur.lastrowid)
|
||
|
||
summary = _summarize(verdict, invoice_number, items)
|
||
cur.execute(
|
||
"""
|
||
INSERT INTO prereview_reports
|
||
(batch_no, invoice_id, invoice_number, seller_name, amount, verdict,
|
||
pass_json, warn_json, reject_json, summary)
|
||
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
|
||
""",
|
||
(
|
||
batch_no,
|
||
invoice_id,
|
||
invoice_number,
|
||
str(inv.get("sellerName") or ""),
|
||
_to_float(inv.get("amount")),
|
||
verdict,
|
||
json.dumps([i for i in items if i["level"] == "pass"], ensure_ascii=False),
|
||
json.dumps([i for i in items if i["level"] == "warn"], ensure_ascii=False),
|
||
json.dumps([i for i in items if i["level"] == "reject"], ensure_ascii=False),
|
||
summary,
|
||
),
|
||
)
|
||
report_id = int(cur.lastrowid)
|
||
|
||
report_view = {
|
||
"id": report_id,
|
||
"batch_no": batch_no,
|
||
"invoice_number": invoice_number,
|
||
"seller_name": str(inv.get("sellerName") or ""),
|
||
"amount": _to_float(inv.get("amount")),
|
||
"verdict": verdict,
|
||
"summary": summary,
|
||
"verify_status": "待查验",
|
||
"created_at": now_unix(),
|
||
"pass_json": json.dumps([i for i in items if i["level"] == "pass"], ensure_ascii=False),
|
||
"warn_json": json.dumps([i for i in items if i["level"] == "warn"], ensure_ascii=False),
|
||
"reject_json": json.dumps([i for i in items if i["level"] == "reject"], ensure_ascii=False),
|
||
}
|
||
invoice_view = {
|
||
"invoice_type": str(inv.get("invoiceType") or ""),
|
||
"invoice_code": str(inv.get("invoiceCode") or ""),
|
||
"invoice_number": invoice_number,
|
||
"issue_date": _fmt_date(_parse_date(inv.get("issueDate"))),
|
||
"amount": _to_float(inv.get("amount")),
|
||
"buyer_name": str(inv.get("buyerName") or ""),
|
||
"buyer_tax_no": str(inv.get("buyerTaxNo") or ""),
|
||
"seller_name": str(inv.get("sellerName") or ""),
|
||
"seller_tax_no": str(inv.get("sellerTaxNo") or ""),
|
||
"expense_category": str(inv.get("expenseCategory") or ""),
|
||
"city": str(inv.get("city") or ""),
|
||
}
|
||
report_file = write_report_for_kernel(
|
||
report_id=report_id,
|
||
report=report_view,
|
||
invoice=invoice_view,
|
||
context=context,
|
||
)
|
||
cur.execute(
|
||
"UPDATE prereview_reports SET report_file = ? WHERE id = ?",
|
||
(report_file, report_id),
|
||
)
|
||
|
||
results.append(
|
||
{
|
||
"report_id": report_id,
|
||
"batch_no": batch_no,
|
||
"invoice_id": invoice_id,
|
||
"invoice_number": invoice_number,
|
||
"verdict": verdict,
|
||
"summary": summary,
|
||
"items": items,
|
||
"report_file": report_file,
|
||
}
|
||
)
|
||
conn.commit()
|
||
finally:
|
||
conn.close()
|
||
|
||
counts = {v: sum(1 for r in results if r["verdict"] == v) for v in (_VERDICT_PASS, _VERDICT_WARN, _VERDICT_REJECT)}
|
||
payload = {
|
||
"ok": True,
|
||
"batch_no": batch_no,
|
||
"count": len(results),
|
||
"verdict_counts": counts,
|
||
"disclaimer": "预审结果仅作提示，不构成最终报销结论；发票真伪以官方查验结果为准。",
|
||
"reports": results,
|
||
}
|
||
if log:
|
||
log.info(
|
||
"prereview_kernel_done batch_no=%s pass=%s warn=%s reject=%s stage=kernel",
|
||
batch_no,
|
||
counts[_VERDICT_PASS],
|
||
counts[_VERDICT_WARN],
|
||
counts[_VERDICT_REJECT],
|
||
)
|
||
return payload
|