183 lines
7.3 KiB
Python
183 lines
7.3 KiB
Python
"""《初审意见单》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))
|