feat: 票据合规预审助手首版 v1.0.0（verify-invoice-default）

scripts/service/report_service.py

@@ -0,0 +1,182 @@
+"""《初审意见单》HTML 报告组装。
+
+写入 `{skill_data}/html/report-{report_id}.html`，宿主数据管理侧栏可预览；
+重新生成走固定 Action id `render-html`（见 references/ACTIONS.md）。
+`{{BODY}}` 内数据一律经 html.escape 转义。
+"""
+
+from __future__ import annotations
+
+import html
+import json
+from typing import Any
+
+from service.html_page_service import load_html_template, render_html_placeholders, write_html_page
+from util.timeutil import unix_to_iso
+
+_TEMPLATE_REL = "assets/html/prereview-report.html"
+
+_LEVEL_BADGE = {"pass": "✅", "warn": "⚠️", "reject": "❌"}
+
+
+def _esc(value: Any) -> str:
+    return html.escape(str(value if value is not None else ""), quote=True)
+
+
+def _items_rows(items_json: str) -> str:
+    try:
+        items = json.loads(items_json) if items_json else []
+    except (TypeError, ValueError):
+        items = []
+    if not items:
+        return '<tr><td colspan="3" class="muted">无</td></tr>'
+    rows = []
+    for item in items:
+        badge = _LEVEL_BADGE.get(str(item.get("level") or ""), "•")
+        ref = str(item.get("rule_ref") or "")
+        ref_cell = f'<span class="ref">依据：{_esc(ref)}</span>' if ref else ""
+        rows.append(
+            "<tr>"
+            f'<td class="nowrap">{badge} {_esc(item.get("rule_id"))}</td>'
+            f'<td>{_esc(item.get("name"))}<br/><span class="muted">{_esc(item.get("detail"))}</span>{ref_cell}</td>'
+            f'<td class="nowrap">{_esc(item.get("level"))}</td>'
+            "</tr>"
+        )
+    return "".join(rows)
+
+
+def _info_table(inv: dict[str, Any]) -> str:
+    fields = (
+        ("发票类型", "invoice_type"),
+        ("发票代码", "invoice_code"),
+        ("发票号码", "invoice_number"),
+        ("开票日期", "issue_date"),
+        ("金额（元）", "amount"),
+        ("购买方名称", "buyer_name"),
+        ("购买方税号", "buyer_tax_no"),
+        ("销售方名称", "seller_name"),
+        ("销售方税号", "seller_tax_no"),
+        ("费用类别", "expense_category"),
+        ("消费城市", "city"),
+    )
+    rows = "".join(
+        f"<tr><th>{_esc(label)}</th><td>{_esc(inv.get(key) if inv.get(key) not in (None, '') else '—')}</td></tr>"
+        for label, key in fields
+    )
+    return f'<table class="info">{rows}</table>'
+
+
+def _context_table(context: dict[str, Any]) -> str:
+    fields = (
+        ("报销事由", "purpose"),
+        ("预算项目编号", "project_no"),
+        ("出差开始", "trip_start"),
+        ("出差结束", "trip_end"),
+        ("目的地", "trip_city"),
+        ("职级", "staff_title"),
+    )
+    rows = []
+    for label, key in fields:
+        value = str(context.get(key) or "").strip()
+        if value:
+            rows.append(f"<tr><th>{_esc(label)}</th><td>{_esc(value)}</td></tr>")
+    if not rows:
+        return ""
+    return f'<table class="info">{"".join(rows)}</table>'
+
+
+def build_report_html(
+    *,
+    report: dict[str, Any],
+    invoice: dict[str, Any],
+    context: dict[str, Any],
+) -> str:
+    """组装完整《初审意见单》HTML 文本。"""
+    verdict = str(report.get("verdict") or "")
+    badge = {"通过": "verdict-pass", "待确认": "verdict-warn", "驳回": "verdict-reject"}.get(verdict, "verdict-warn")
+    mark = {"通过": "✅ 通过", "待确认": "⚠️ 待确认", "驳回": "❌ 驳回"}.get(verdict, verdict)
+
+    verify_status = str(report.get("verify_status") or "待查验")
+    verify_note = str(report.get("verify_note") or "")
+    created = unix_to_iso(report.get("created_at"))
+
+    body = f"""
+    <div class="verdict-line">
+      <span class="verdict {badge}">{mark}</span>
+      <span class="muted">批次 {_esc(report.get("batch_no"))} · 报告编号 {_esc(report.get("id"))} · 生成于 {_esc(created or "")}</span>
+    </div>
+    <p class="summary">{_esc(report.get("summary"))}</p>
+    <div class="section"><h2>发票信息</h2>{_info_table(invoice)}</div>
+    <div class="section"><h2>报销上下文</h2>{_context_table(context) or '<p class="muted">未提供差旅/项目上下文。</p>'}</div>
+    <div class="section"><h2>❌ 驳回项（必须整改）</h2><table><tr><th>规则</th><th>说明</th><th>级别</th></tr>{_items_rows(report.get("reject_json") or "")}</table></div>
+    <div class="section"><h2>⚠️ 待确认项（请核实）</h2><table><tr><th>规则</th><th>说明</th><th>级别</th></tr>{_items_rows(report.get("warn_json") or "")}</table></div>
+    <div class="section"><h2>✅ 通过项</h2><table><tr><th>规则</th><th>说明</th><th>级别</th></tr>{_items_rows(report.get("pass_json") or "")}</table></div>
+    <div class="section"><h2>真伪查验</h2>
+      <p>当前状态：<strong>{_esc(verify_status)}</strong>{(" · 备注：" + _esc(verify_note)) if verify_note else ""}</p>
+      <p class="muted">请经办人到全国增值税发票查验平台（inv-veri.chinatax.gov.cn）核验票面真伪后，在数据管理“初审意见”表回填查验结果。</p>
+    </div>
+    <footer>本《初审意见单》由票据合规预审助手自动生成，预审结果不构成最终报销结论，请以财务处人工审核为准。</footer>
+    """
+
+    template = load_html_template(_TEMPLATE_REL)
+    return render_html_placeholders(
+        template,
+        {
+            "TITLE": f"初审意见单 · 发票 {report.get('invoice_number') or ''}",
+            "GENERATED_AT": created or "",
+            "BODY": body,
+        },
+    )
+
+
+def build_report_for_invoice(report_id: int, batch_no: str = "") -> str:
+    """按报告编号从库中取数并写出 HTML，返回文件绝对路径（重建入口用）。"""
+    from db.connection import get_conn
+    from util.logging_config import get_skill_logger
+
+    log = get_skill_logger()
+    log.info("report_render_start report_id=%s stage=render", report_id)
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT * FROM prereview_reports WHERE id = ?", (report_id,))
+        row = cur.fetchone()
+        if not row:
+            raise ValueError(f"初审报告不存在：{report_id}")
+        report = _row_to_dict(cur, "prereview_reports", row)
+        invoice = {}
+        if report.get("invoice_id"):
+            cur.execute("SELECT * FROM invoice_records WHERE id = ?", (report["invoice_id"],))
+            irow = cur.fetchone()
+            if irow:
+                invoice = _row_to_dict(cur, "invoice_records", irow)
+        context: dict[str, Any] = {}
+        try:
+            context = json.loads(invoice.get("trip_context_json") or "{}")
+        except (TypeError, ValueError):
+            context = {}
+        html_text = build_report_html(report=report, invoice=invoice, context=context)
+    finally:
+        conn.close()
+
+    filename = f"report-{report_id}.html"
+    path = write_html_page(filename, html_text)
+    log.info("report_render_done report_id=%s path=%s stage=render", report_id, path)
+    return path
+
+
+def write_report_for_kernel(
+    *,
+    report_id: int,
+    report: dict[str, Any],
+    invoice: dict[str, Any],
+    context: dict[str, Any],
+) -> str:
+    """内核路径：用事务内内存数据直接组装并写出 HTML（不回查未提交的行）。"""
+    html_text = build_report_html(report=report, invoice=invoice, context=context)
+    return write_html_page(f"report-{report_id}.html", html_text)
+
+
+def _row_to_dict(cursor, table_name: str, row) -> dict[str, Any]:
+    columns = [desc[0] for desc in cursor.description]
+    return dict(zip(columns, row))