alert-contract-expiring-hr/scripts/service/html_renderer.py

"""HTML 报告渲染：Jinja2 模板 → 写 output/<task_id>.html。

模板：scripts/templates/task.html.j2
占位替换规则：
- {{TITLE}}, {{TASK_ID}}, {{GENERATED_AT}}
- {{TEACHER_NAME}}, {{EMPLOYEE_ID}}, {{UNIT_NAME}}, {{POSITION}}, {{CONTRACT_TYPE}}
- {{CONTRACT_START}}, {{CONTRACT_END}}, {{DAYS_TO_EXPIRE}}, {{ALERT_LEVEL}}
- {{PERFORMANCE_SUMMARY}}
- {{SUGGESTION_CLASS}}, {{SUGGESTION_LABEL}}, {{SUGGESTION_REASON}}
- {{DEADLINE}}, {{CHECKLIST_BODY}}
"""

from __future__ import annotations

import json
import sqlite3
from datetime import date, datetime, timedelta
from pathlib import Path

from db import repository as repo
from util.constants import ALERT_LEVEL_LABELS, SUGGESTION_LABELS
from util.date_util import parse_iso_date
from util.logging_setup import get_skill_logger
from util.runtime_paths import get_output_dir, get_templates_dir

logger = get_skill_logger()

SUGGESTION_CLASS_MAP = {
    "renew": "renew",
    "observe": "observe",
    "non_renew": "non-renew",
}


def render_task_html(conn: sqlite3.Connection, task_id: str) -> Path:
    """渲染指定 task 为 HTML 文件，返回输出路径。

    不存在 task → 抛 ValueError。
    """
    task = repo.get_task(conn, task_id)
    if task is None:
        raise ValueError(f"task_id={task_id!r} 不存在")

    # 关联 alert 的 days_to_expire / level（用于显示）
    alert = conn.execute(
        "SELECT * FROM alerts WHERE alert_id = ?", (task["alert_id"],)
    ).fetchone() if task["alert_id"] else None

    days_to_expire = alert["days_to_expire"] if alert else None
    alert_level = alert["alert_level"] if alert else "yellow"
    alert_level_label = ALERT_LEVEL_LABELS.get(alert_level, alert_level)

    # 截止日 = 合同到期日 - 30 天（教师备齐材料的最后期限）
    contract_end = parse_iso_date(task["contract_end"])
    deadline = (contract_end - timedelta(days=30)).isoformat()

    # 模板上下文
    checklist_items = json.loads(task["materials_checklist"]) if task["materials_checklist"] else []
    checklist_html = "\n".join(
        f'        <li><span class="box"></span>{item}</li>' for item in checklist_items
    )

    suggestion_class = SUGGESTION_CLASS_MAP.get(task["ai_suggestion"], "observe")
    suggestion_label = SUGGESTION_LABELS.get(task["ai_suggestion"], task["ai_suggestion"])

    # v1.1.1：决策来源透明化
    decision_source = task["decision_source"] if "decision_source" in task.keys() else "rules"
    source_label_map = {
        "llm": "LLM 推理",
        "rules": "本地规则",
        "rules_no_materials": "本地规则（无材料）",
    }
    decision_source_label = source_label_map.get(decision_source, decision_source)

    # v1.1.1：材料计数（来自 materials_content）
    materials_count = conn.execute(
        "SELECT COUNT(*) AS n FROM materials_content "
        "WHERE contract_id = ? AND extracted_text IS NOT NULL",
        (task["contract_id"],),
    ).fetchone()["n"]

    ctx = {
        "TITLE": "续聘准备任务书",
        "TASK_ID": task_id,
        "GENERATED_AT": task["generated_at"],
        "TEACHER_NAME": task["teacher_name"],
        "EMPLOYEE_ID": task["employee_id"],
        "UNIT_NAME": task["unit_name"],
        "POSITION": task["position"],
        "CONTRACT_TYPE": task["contract_type"],
        "CONTRACT_START": task["contract_start"],
        "CONTRACT_END": task["contract_end"],
        "DAYS_TO_EXPIRE": days_to_expire if days_to_expire is not None else "—",
        "ALERT_LEVEL": alert_level_label,
        "PERFORMANCE_SUMMARY": task["performance_summary"],
        "SUGGESTION_CLASS": suggestion_class,
        "SUGGESTION_LABEL": suggestion_label,
        "SUGGESTION_REASON": task["suggestion_reason"],
        "DEADLINE": deadline,
        "CHECKLIST_BODY": checklist_html,
        "DECISION_SOURCE": decision_source_label,
        "MATERIALS_COUNT": materials_count,
    }

    template_path = get_templates_dir() / "task.html.j2"
    if not template_path.exists():
        raise FileNotFoundError(f"模板不存在: {template_path}")

    html = _render_template(template_path, ctx)

    output_dir = get_output_dir()
    output_dir.mkdir(parents=True, exist_ok=True)
    output_path = output_dir / f"{task_id}.html"
    output_path.write_text(html, encoding="utf-8")

    logger.info("html_renderer: %s → %s", task_id, output_path)
    return output_path


def _render_template(template_path: Path, ctx: dict) -> str:
    """用 Jinja2 渲染模板；本地无 jinja2 时降级到占位符替换。"""
    template_text = template_path.read_text(encoding="utf-8")

    try:
        from jinja2 import Template

        return Template(template_text).render(**ctx)
    except ImportError:
        # 降级：纯 str.replace（最简实现）
        for k, v in ctx.items():
            template_text = template_text.replace("{{" + k + "}}", str(v))
        return template_text