feat: 职称评审核验助手首版 v1.0.1（verify-promotion-metric-default）

scripts/cli/app.py

@@ -0,0 +1,197 @@
+"""CLI：argparse 与分发（职称评审核验助手）。
+
+只负责参数解析并转交 service；业务内核在 verify_service / threshold_service，
+勿在此按入口复制逻辑。
+"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+from typing import List, Optional
+
+from service.task_service import (
+    cmd_config_path,
+    cmd_health,
+    cmd_import_thresholds,
+    cmd_init_db,
+    cmd_list_cases,
+    cmd_render_html,
+    cmd_run,
+    cmd_set_threshold,
+    cmd_show_case,
+    cmd_stats,
+    cmd_verify_batch_input,
+    cmd_verify_metric,
+    cmd_version,
+    cmd_view_thresholds,
+)
+from util.config_bootstrap import bootstrap_skill_config
+from util.argparse_zh import ZhArgumentParser
+from util.constants import LOG_LOGGER_NAME, SKILL_SLUG
+from util.logging_config import get_skill_logger, setup_skill_logging
+
+
+def _cli_str_or_none(raw: Optional[str]) -> Optional[str]:
+    if raw is None:
+        return None
+    v = str(raw).strip()
+    return v or None
+
+
+def _cli_int_or_none(raw: Optional[str]) -> Optional[int]:
+    raw = _cli_str_or_none(raw)
+    if raw is None:
+        return None
+    try:
+        return int(float(raw))
+    except ValueError:
+        return None
+
+
+def _handle_run(args: argparse.Namespace) -> int:
+    tail = [str(x).strip() for x in (args.run_tail or []) if str(x).strip()]
+    if len(tail) > 2:
+        print("参数过多。")
+        print("用法：python main.py run [target [input_id]] | run [-t target] [-i input_id]")
+        return 1
+    t_target: Optional[str] = None
+    t_input: Optional[str] = None
+    if len(tail) == 2:
+        t_target, t_input = tail[0], tail[1]
+    elif len(tail) == 1:
+        if tail[0].isdigit():
+            t_input = tail[0]
+        else:
+            t_target = tail[0]
+    target = _cli_str_or_none(getattr(args, "target", None)) or t_target
+    input_id = _cli_str_or_none(getattr(args, "input_id", None)) or t_input
+    return cmd_run(target=target, input_id=input_id)
+
+
+def _handle_set_threshold(args):
+    return cmd_set_threshold(
+        rank=str(args.rank or "").strip(),
+        metric_key=str(args.metric_key or "").strip(),
+        metric_name=_cli_str_or_none(args.metric_name) or "",
+        min_value=float(args.min_value),
+        unit=_cli_str_or_none(args.unit) or "",
+        basis_doc=_cli_str_or_none(args.basis_doc) or "",
+        clause=_cli_str_or_none(args.clause) or "",
+    )
+
+
+def _handle_verify(args):
+    return cmd_verify_metric(
+        teacher_name=str(args.teacher or "").strip(),
+        apply_rank=str(args.apply_rank or "").strip(),
+        metrics_json=str(args.metrics_json or "").strip(),
+        department=_cli_str_or_none(args.department) or "",
+    )
+
+
+def _handle_list_reports(args):
+    return cmd_list_cases(limit=int(args.limit or 20), verdict=_cli_str_or_none(args.verdict) or "")
+
+
+def _handle_render_html(args):
+    report_id = getattr(args, "report_id", None)
+    return cmd_render_html(report_id=int(report_id) if report_id else None)
+
+
+def _print_full_usage() -> None:
+    print("职称评审核验助手（main.py）可用命令：")
+    print("  python main.py verify-metric --teacher 张三 --apply-rank 副教授 --metrics-json <file|内联|-> [--department 学院]")
+    print("  python main.py verify-batch-input                  # 批量核验 input/ 目录申报案例")
+    print("  python main.py view-thresholds [--rank 副教授]")
+    print("  python main.py set-threshold --rank 副教授 --metric-key presiding_provincial --metric-name 主持省部级项目 --min-value 1 [--unit 项] [--basis-doc 文件] [--clause 条款]")
+    print("  python main.py import-thresholds --thresholds-json <file|->")
+    print("  python main.py list-cases [--limit N] [--verdict 硬伤] | show-case --id N | render-html [--report-id N]")
+    print("  python main.py stats")
+    print("  python main.py run [target [input_id]] [-t target] [-i input_id]")
+    print("  python main.py health | config-path | version | init-db")
+
+
+def build_parser() -> ZhArgumentParser:
+    p = ZhArgumentParser(
+        prog="main.py",
+        description="职称评审核验助手：维护各级别刚性门槛库，对申报案例逐项比对生成达标矩阵与红黄绿三色结论，出具《核验报告》。",
+        formatter_class=argparse.RawDescriptionHelpFormatter,
+    )
+    sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
+
+    sp = sub.add_parser("verify-metric", help="职称硬指标核验（主命令）")
+    sp.add_argument("--teacher", required=True, metavar="教师姓名")
+    sp.add_argument("--apply-rank", required=True, metavar="申报级别(副教授/教授)", dest="apply_rank")
+    sp.add_argument("--metrics-json", required=True, metavar="指标JSON(file|内联|-)", dest="metrics_json")
+    sp.add_argument("--department", default=None, metavar="所属学院")
+    sp.set_defaults(handler=_handle_verify)
+
+    sp = sub.add_parser("verify-batch-input", help="批量核验 input/ 目录申报案例")
+    sp.set_defaults(handler=lambda _a: cmd_verify_batch_input())
+
+    sp = sub.add_parser("view-thresholds", help="查看评审门槛库")
+    sp.add_argument("--rank", default=None, metavar="级别筛选")
+    sp.set_defaults(handler=lambda a: cmd_view_thresholds(rank=_cli_str_or_none(a.rank) or ""))
+
+    sp = sub.add_parser("set-threshold", help="录入/更新评审门槛")
+    sp.add_argument("--rank", required=True, metavar="申报级别", dest="rank")
+    sp.add_argument("--metric-key", required=True, metavar="指标键", dest="metric_key")
+    sp.add_argument("--metric-name", default=None, metavar="指标名称", dest="metric_name")
+    sp.add_argument("--min-value", required=True, type=float, metavar="最低要求", dest="min_value")
+    sp.add_argument("--unit", default=None, metavar="单位")
+    sp.add_argument("--basis-doc", default=None, metavar="依据文件", dest="basis_doc")
+    sp.add_argument("--clause", default=None, metavar="条款")
+    sp.set_defaults(handler=_handle_set_threshold)
+
+    sp = sub.add_parser("import-thresholds", help="批量导入门槛")
+    sp.add_argument("--thresholds-json", required=True, metavar="JSON(file|-)", dest="thresholds_json")
+    sp.set_defaults(handler=lambda a: cmd_import_thresholds(str(a.thresholds_json)))
+
+    sp = sub.add_parser("list-cases", help="查询核验案例")
+    sp.add_argument("--limit", default=20, metavar="条数")
+    sp.add_argument("--verdict", default=None, metavar="结论筛选(安全/边缘/硬伤)")
+    sp.set_defaults(handler=_handle_list_reports)
+
+    sp = sub.add_parser("show-case", help="查看核验案例详情")
+    sp.add_argument("--id", required=True, type=int, metavar="案例编号", dest="case_id")
+    sp.set_defaults(handler=lambda a: cmd_show_case(case_id=int(a.case_id)))
+
+    sp = sub.add_parser("render-html", help="重建《核验报告》（缺省最近 10 份）")
+    sp.add_argument("--report-id", default=None, metavar="案例编号", dest="report_id")
+    sp.set_defaults(handler=_handle_render_html)
+
+    sp = sub.add_parser("stats", help="核验统计")
+    sp.set_defaults(handler=lambda _a: cmd_stats())
+
+    sp = sub.add_parser("run", help="任务执行命令骨架")
+    sp.add_argument("--target", "-t", default=None, metavar="目标")
+    sp.add_argument("--input-id", "-i", default=None, metavar="输入id", dest="input_id")
+    sp.add_argument("run_tail", nargs="*", metavar="位置参数")
+    sp.set_defaults(handler=_handle_run)
+
+    sp = sub.add_parser("health", help="健康检查")
+    sp.set_defaults(handler=lambda _a: cmd_health())
+
+    sp = sub.add_parser("config-path", help="输出用户 .env 与模板路径")
+    sp.set_defaults(handler=lambda _a: cmd_config_path())
+
+    sp = sub.add_parser("version", help="版本信息(JSON)")
+    sp.set_defaults(handler=lambda _a: cmd_version())
+
+    sp = sub.add_parser("init-db", help="幂等初始化本地数据库与数据管理元数据")
+    sp.set_defaults(handler=lambda _a: cmd_init_db())
+    return p
+
+
+def main(argv: Optional[List[str]] = None) -> int:
+    argv = argv if argv is not None else sys.argv[1:]
+    bootstrap_skill_config()
+    setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
+    get_skill_logger().info("cli_start argv=%s", sys.argv)
+    if not argv:
+        _print_full_usage()
+        return 1
+    parser = build_parser()
+    args = parser.parse_args(argv)
+    return int(args.handler(args))

scripts/db/connection.py

@@ -0,0 +1,169 @@
+"""SQLite 连接与幂等初始化（职称评审核验助手）。
+
+业务表 2 张：
+- threshold_rules（评审门槛库，可编辑）
+- verify_cases（核验案例，只读，核验自动写入）
+
+表间禁止 SQLite 外键，仅存业务 id（见 references/SCHEMA.md）。
+"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from db.display_metadata import (
+    init_display_metadata,
+    seed_column_metadata_batch,
+    upsert_table_metadata,
+)
+from db.seed_data import seed_all
+from db.timestamp_columns import init_updated_at_trigger
+from util.runtime_paths import get_db_path
+
+
+def get_conn() -> sqlite3.Connection:
+    return sqlite3.connect(get_db_path())
+
+
+_THRESHOLD_RULES_DDL = """
+CREATE TABLE IF NOT EXISTS threshold_rules (
+    id          INTEGER PRIMARY KEY AUTOINCREMENT,
+    rank        TEXT NOT NULL,
+    metric_key  TEXT NOT NULL,
+    metric_name TEXT NOT NULL,
+    min_value   REAL NOT NULL,
+    unit        TEXT,
+    basis_doc   TEXT,
+    clause      TEXT,
+    created_at  INTEGER NOT NULL DEFAULT (unixepoch()),
+    updated_at  INTEGER NOT NULL DEFAULT (unixepoch()),
+    UNIQUE (rank, metric_key)
+)
+"""
+
+_VERIFY_CASES_DDL = """
+CREATE TABLE IF NOT EXISTS verify_cases (
+    id           INTEGER PRIMARY KEY AUTOINCREMENT,
+    batch_no     TEXT NOT NULL,
+    teacher_name TEXT,
+    department   TEXT,
+    apply_rank   TEXT,
+    metrics_json TEXT,
+    verdict      TEXT NOT NULL,
+    matrix_json  TEXT,
+    report_file  TEXT,
+    created_at   INTEGER NOT NULL DEFAULT (unixepoch())
+)
+"""
+
+_VERIFY_CASES_INDEX = """
+CREATE INDEX IF NOT EXISTS idx_verify_cases_batch
+ON verify_cases(batch_no)
+"""
+
+_TS_COLUMN_NAMES = {"created_at", "updated_at"}
+
+
+def _col(column_name: str, display_name: str, description: str, **extra: object) -> dict[str, object]:
+    base: dict[str, object] = {
+        "column_name": column_name,
+        "display_name": display_name,
+        "description": description,
+        "visible": 1,
+        "searchable": 0,
+        "editable": 0,
+    }
+    base.update(extra)
+    if column_name in _TS_COLUMN_NAMES:
+        base["display_type"] = "datetime_unix_seconds"
+    return base
+
+
+_THRESHOLD_RULES_COLUMNS: tuple[dict[str, object], ...] = (
+    _col("id", "编号", "主键", searchable=1),
+    _col("rank", "申报级别", "副教授 / 教授。", searchable=1),
+    _col("metric_key", "指标键", "指标唯一键（presiding_provincial 等）。", searchable=1),
+    _col("metric_name", "指标名称", "如 主持省部级项目。", searchable=1, editable=1),
+    _col("min_value", "最低要求", "刚性门槛数值。", searchable=1, editable=1, display_type="number"),
+    _col("unit", "单位", "项/篇/课时/年 等。", editable=1),
+    _col("basis_doc", "依据文件", "如《专业技术职务评聘办法》。", searchable=1, editable=1),
+    _col("clause", "条款", "门槛出处条款号。", editable=1),
+    _col("created_at", "创建时间", "记录创建时间（Unix 秒级，自动生成）。"),
+    _col("updated_at", "更新时间", "记录最后更新时间（Unix 秒级，自动维护）。"),
+)
+
+_VERIFY_CASES_COLUMNS: tuple[dict[str, object], ...] = (
+    _col("id", "案例编号", "主键，核验记录编号。", searchable=1),
+    _col("batch_no", "核验批次号", "同一次核验请求的批次号。", searchable=1),
+    _col("teacher_name", "教师姓名", "被核验教师。", searchable=1),
+    _col("department", "所属学院", "教师所属学院。", searchable=1),
+    _col("apply_rank", "申报级别", "副教授 / 教授。", searchable=1),
+    _col("metrics_json", "申报指标", "教师实有指标（JSON，默认隐藏）。", visible=0, display_type="textarea"),
+    _col(
+        "verdict",
+        "核验结论",
+        "安全（形式审查通过）/ 边缘（建议补充）/ 硬伤（暂缓提交）。",
+        searchable=1,
+        options_json='[{"value":"安全","label":"🟢 安全"},{"value":"边缘","label":"🟡 边缘"},{"value":"硬伤","label":"🔴 硬伤"}]',
+    ),
+    _col("matrix_json", "达标矩阵", "逐项达标矩阵（JSON）。", visible=0, display_type="textarea"),
+    _col("report_file", "报告文件", "《核验报告》HTML 文件路径。", visible=0),
+    _col("created_at", "创建时间", "记录创建时间（Unix 秒级，自动生成）。"),
+)
+
+
+def _init_threshold_rules_metadata(cursor) -> None:
+    cursor.execute(_THRESHOLD_RULES_DDL)
+    init_updated_at_trigger(cursor, "threshold_rules")
+    upsert_table_metadata(
+        cursor,
+        table_name="threshold_rules",
+        display_name="评审门槛库",
+        description="各级别职称申报的刚性门槛（数值须以人事处最新文件为准）。",
+        grid_help="按 申报级别+指标键 唯一；行内可编辑最低要求与依据条款。",
+        visible=1,
+        readonly=0,
+    )
+    seed_column_metadata_batch(
+        cursor,
+        table_name="threshold_rules",
+        columns=_THRESHOLD_RULES_COLUMNS,
+        valid_column_names=[str(c["column_name"]) for c in _THRESHOLD_RULES_COLUMNS],
+    )
+
+
+def _init_verify_cases_metadata(cursor) -> None:
+    cursor.execute(_VERIFY_CASES_DDL)
+    cursor.execute(_VERIFY_CASES_INDEX)
+    upsert_table_metadata(
+        cursor,
+        table_name="verify_cases",
+        display_name="核验案例",
+        description="每次核验的达标矩阵与三色结论；可重建《核验报告》。",
+        grid_help="🔴硬伤=存在不达门槛项；🟡边缘=达标但余量不足10%；🟢安全=全部超标准10%以上。",
+        visible=1,
+        readonly=1,
+        default_sort_json='[{"column":"id","direction":"desc"}]',
+        default_visible_columns_json='["id","teacher_name","department","apply_rank","verdict","created_at"]',
+    )
+    seed_column_metadata_batch(
+        cursor,
+        table_name="verify_cases",
+        columns=_VERIFY_CASES_COLUMNS,
+        valid_column_names=[str(c["column_name"]) for c in _VERIFY_CASES_COLUMNS],
+    )
+
+
+def init_db() -> dict[str, int]:
+    """幂等创建/迁移本地 SQLite、数据管理元数据与门槛种子。"""
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        init_display_metadata(cur)
+        _init_threshold_rules_metadata(cur)
+        _init_verify_cases_metadata(cur)
+        seeded = seed_all(cur)
+        conn.commit()
+        return seeded
+    finally:
+        conn.close()

scripts/db/display_metadata.py

@@ -0,0 +1,324 @@
+"""匠厂数据管理展示元数据（_jiangchang_tables / _jiangchang_columns）。
+
+SQLite 无原生表/字段 COMMENT；中文展示名称通过这两张元数据表提供给宿主「数据管理」页面。
+字段在界面上的顺序由 CREATE TABLE / PRAGMA table_info(cid) 决定，**不**使用 display_order 排序。
+"""
+
+from __future__ import annotations
+
+from typing import Iterable
+
+# 宿主会隐藏这些表，不作为业务表展示（见 jiangchang electron/data-management/types.ts）
+INTERNAL_METADATA_TABLES = frozenset({"_jiangchang_tables", "_jiangchang_columns"})
+LEGACY_METADATA_TABLES = frozenset({"_schema_meta"})
+METADATA_TABLES = INTERNAL_METADATA_TABLES | LEGACY_METADATA_TABLES
+
+DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
+STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})
+
+_JIANGCHANG_TABLES_DDL = """
+CREATE TABLE IF NOT EXISTS _jiangchang_tables (
+    table_name TEXT PRIMARY KEY,
+    display_name TEXT NOT NULL,
+    description TEXT,
+    grid_help TEXT,
+    sort_order INTEGER,
+    visible INTEGER NOT NULL DEFAULT 1,
+    readonly INTEGER NOT NULL DEFAULT 0,
+    default_sort_json TEXT,
+    default_visible_columns_json TEXT
+)
+"""
+
+_JIANGCHANG_COLUMNS_DDL = """
+CREATE TABLE IF NOT EXISTS _jiangchang_columns (
+    table_name TEXT NOT NULL,
+    column_name TEXT NOT NULL,
+    display_name TEXT NOT NULL,
+    description TEXT,
+    grid_help TEXT,
+    display_order INTEGER,
+    visible INTEGER NOT NULL DEFAULT 1,
+    searchable INTEGER NOT NULL DEFAULT 1,
+    editable INTEGER NOT NULL DEFAULT 1,
+    display_type TEXT,
+    options_json TEXT,
+    PRIMARY KEY (table_name, column_name)
+)
+"""
+
+def _pragma_column_names(cursor, table_name: str) -> set[str]:
+    cursor.execute(f"PRAGMA table_info({table_name!r})")
+    return {str(row[1]) for row in cursor.fetchall()}
+
+
+def _ensure_jiangchang_tables_columns(cursor) -> None:
+    """旧库补列：CREATE IF NOT EXISTS 不会给已有表加新字段。"""
+    cols = _pragma_column_names(cursor, "_jiangchang_tables")
+    if "default_sort_json" not in cols:
+        cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN default_sort_json TEXT")
+    if "default_visible_columns_json" not in cols:
+        cursor.execute(
+            "ALTER TABLE _jiangchang_tables ADD COLUMN default_visible_columns_json TEXT"
+        )
+    if "grid_help" not in cols:
+        cursor.execute("ALTER TABLE _jiangchang_tables ADD COLUMN grid_help TEXT")
+
+
+def _ensure_jiangchang_columns_columns(cursor) -> None:
+    cols = _pragma_column_names(cursor, "_jiangchang_columns")
+    if "grid_help" not in cols:
+        cursor.execute("ALTER TABLE _jiangchang_columns ADD COLUMN grid_help TEXT")
+
+
+def init_display_metadata_tables(cursor) -> None:
+    cursor.execute(_JIANGCHANG_TABLES_DDL)
+    cursor.execute(_JIANGCHANG_COLUMNS_DDL)
+    _ensure_jiangchang_tables_columns(cursor)
+    _ensure_jiangchang_columns_columns(cursor)
+
+
+def upsert_table_metadata(
+    cursor,
+    *,
+    table_name: str,
+    display_name: str,
+    description: str | None = None,
+    grid_help: str | None = None,
+    visible: int = 1,
+    readonly: int = 0,
+    default_sort_json: str | None = None,
+    default_visible_columns_json: str | None = None,
+) -> None:
+    """写入表级元数据。
+
+    default_sort_json: 宿主数据管理默认排序，JSON 数组，例如
+    '[{"column":"sent_at","direction":"desc"},{"column":"id","direction":"desc"}]'
+    为空则宿主回退到 updated_at → created_at → 主键。
+
+    default_visible_columns_json: 宿主数据管理默认展示列，JSON 字符串数组，例如
+    '["stat_type","period","stage"]'
+    为空则全显；用户改过「列」后以本机偏好为准（软默认，区别于 visible=0 硬隐藏）。
+
+    grid_help: 数据管理表格上方说明；空则不出横幅。与表单用的 description 独立，禁止互拷。
+    """
+    cursor.execute(
+        """
+        INSERT INTO _jiangchang_tables (
+            table_name, display_name, description, grid_help, visible, readonly,
+            default_sort_json, default_visible_columns_json
+        )
+        VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+        ON CONFLICT(table_name) DO UPDATE SET
+            display_name = excluded.display_name,
+            description = excluded.description,
+            grid_help = excluded.grid_help,
+            visible = excluded.visible,
+            readonly = excluded.readonly,
+            default_sort_json = excluded.default_sort_json,
+            default_visible_columns_json = excluded.default_visible_columns_json
+        """,
+        (
+            table_name,
+            display_name,
+            description,
+            grid_help,
+            visible,
+            readonly,
+            default_sort_json,
+            default_visible_columns_json,
+        ),
+    )
+
+
+def upsert_column_metadata(
+    cursor,
+    *,
+    table_name: str,
+    column_name: str,
+    display_name: str,
+    description: str | None = None,
+    grid_help: str | None = None,
+    visible: int = 1,
+    searchable: int = 1,
+    editable: int = 1,
+    display_type: str | None = None,
+    options_json: str | None = None,
+) -> None:
+    # display_order 保留为 NULL：宿主按 PRAGMA table_info cid 展示字段，不读 display_order。
+    # grid_help: 表头「?」；空则不出。与表单用的 description 独立，禁止互拷。
+    cursor.execute(
+        """
+        INSERT INTO _jiangchang_columns (
+            table_name, column_name, display_name, description, grid_help,
+            display_order, visible, searchable, editable, display_type, options_json
+        )
+        VALUES (?, ?, ?, ?, ?, NULL, ?, ?, ?, ?, ?)
+        ON CONFLICT(table_name, column_name) DO UPDATE SET
+            display_name = excluded.display_name,
+            description = excluded.description,
+            grid_help = excluded.grid_help,
+            display_order = NULL,
+            visible = excluded.visible,
+            searchable = excluded.searchable,
+            editable = excluded.editable,
+            display_type = excluded.display_type,
+            options_json = excluded.options_json
+        """,
+        (
+            table_name,
+            column_name,
+            display_name,
+            description,
+            grid_help,
+            visible,
+            searchable,
+            editable,
+            display_type,
+            options_json,
+        ),
+    )
+
+
+def seed_column_metadata_batch(
+    cursor,
+    *,
+    table_name: str,
+    columns: Iterable[dict[str, object]],
+    valid_column_names: Iterable[str] | None = None,
+) -> None:
+    """批量 upsert 字段元数据；传入 valid_column_names 时会在 upsert 后清理陈旧字段元数据。"""
+    for column in columns:
+        upsert_column_metadata(
+            cursor,
+            table_name=table_name,
+            column_name=str(column["column_name"]),
+            display_name=str(column["display_name"]),
+            description=str(column["description"]) if column.get("description") else None,
+            grid_help=str(column["grid_help"]) if column.get("grid_help") else None,
+            visible=int(column.get("visible", 1)),
+            searchable=int(column.get("searchable", 1)),
+            editable=int(column.get("editable", 1)),
+            display_type=str(column["display_type"]) if column.get("display_type") else None,
+            options_json=str(column["options_json"]) if column.get("options_json") else None,
+        )
+    if valid_column_names is not None:
+        prune_stale_column_metadata(cursor, table_name, valid_column_names)
+
+
+def prune_stale_column_metadata(cursor, table_name: str, valid_column_names: Iterable[str]) -> None:
+    """删除表中已不存在物理字段的元数据行（幂等初始化后不得残留旧字段）。"""
+    valid = set(valid_column_names)
+    cursor.execute(
+        "SELECT column_name FROM _jiangchang_columns WHERE table_name = ?",
+        (table_name,),
+    )
+    for (column_name,) in cursor.fetchall():
+        if str(column_name) not in valid:
+            cursor.execute(
+                "DELETE FROM _jiangchang_columns WHERE table_name = ? AND column_name = ?",
+                (table_name, column_name),
+            )
+
+
+def get_table_readonly(cursor, table_name: str) -> int | None:
+    cursor.execute(
+        "SELECT readonly FROM _jiangchang_tables WHERE table_name = ?",
+        (table_name,),
+    )
+    row = cursor.fetchone()
+    if not row:
+        return None
+    return int(row[0] or 0)
+
+
+def get_column_editable(cursor, table_name: str, column_name: str) -> int | None:
+    cursor.execute(
+        """
+        SELECT editable FROM _jiangchang_columns
+        WHERE table_name = ? AND column_name = ?
+        """,
+        (table_name, column_name),
+    )
+    row = cursor.fetchone()
+    if not row:
+        return None
+    return int(row[0] or 0)
+
+
+def list_editable_business_columns(cursor, table_name: str) -> list[str]:
+    """返回 editable=1 的业务字段名（不含宿主侧主键/generated 等二次过滤）。"""
+    physical = list_physical_column_names(cursor, table_name)
+    cursor.execute(
+        """
+        SELECT column_name, editable
+        FROM _jiangchang_columns
+        WHERE table_name = ?
+        """,
+        (table_name,),
+    )
+    editable_set = {
+        str(column_name)
+        for column_name, editable in cursor.fetchall()
+        if editable is not None and int(editable) == 1
+    }
+    return [name for name in physical if name in editable_set]
+
+
+def table_allows_manual_create(cursor, table_name: str) -> bool:
+    """表级 readonly=0 且至少有一个 editable=1 字段时，宿主才应提供空白新增表单。"""
+    readonly = get_table_readonly(cursor, table_name)
+    if readonly is None or int(readonly) != 0:
+        return False
+    return bool(list_editable_business_columns(cursor, table_name))
+
+
+def init_display_metadata(cursor) -> None:
+    init_display_metadata_tables(cursor)
+
+
+def drop_deprecated_user_table(
+    cursor,
+    table_name: str,
+    *,
+    triggers: Iterable[str] | None = None,
+) -> None:
+    """永久退役用户可见表：删 trigger、DROP 物理表、清理展示元数据（幂等）。
+
+    用于废弃 task_logs 等旧表。禁止只删元数据而保留物理表。
+    见 references/SCHEMA.md「表增删改与退役」。
+    """
+    for trigger_name in triggers or ():
+        cursor.execute(f"DROP TRIGGER IF EXISTS {trigger_name}")
+    default_updated_at = f"{table_name}_set_updated_at"
+    cursor.execute(f"DROP TRIGGER IF EXISTS {default_updated_at}")
+    cursor.execute(f"DROP TABLE IF EXISTS {_quote_identifier(table_name)}")
+    cursor.execute(
+        "DELETE FROM _jiangchang_columns WHERE table_name = ?",
+        (table_name,),
+    )
+    cursor.execute(
+        "DELETE FROM _jiangchang_tables WHERE table_name = ?",
+        (table_name,),
+    )
+
+
+def iter_user_tables(cursor) -> Iterable[str]:
+    cursor.execute(
+        "SELECT name FROM sqlite_master WHERE type='table' AND name NOT LIKE 'sqlite_%' ORDER BY name"
+    )
+    for (name,) in cursor.fetchall():
+        if name in METADATA_TABLES:
+            continue
+        yield str(name)
+
+
+def _quote_identifier(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def list_physical_column_names(cursor, table_name: str) -> list[str]:
+    cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
+    rows = cursor.fetchall()
+    rows.sort(key=lambda row: int(row[0]))
+    return [str(row[1]) for row in rows]

scripts/db/display_metadata_validator.py

@@ -0,0 +1,592 @@
+"""数据管理展示元数据校验（模板开发自检，不依赖宿主）。"""
+
+from __future__ import annotations
+
+import json
+import os
+import re
+import sqlite3
+from dataclasses import dataclass, field
+
+from db.display_metadata import (
+    DATETIME_UNIX_SECONDS,
+    METADATA_TABLES,
+    STANDARD_TIMESTAMP_COLUMNS,
+    iter_user_tables,
+    list_editable_business_columns,
+    list_physical_column_names,
+    table_allows_manual_create,
+)
+from db.timestamp_columns import has_updated_at_trigger
+from util.constants import SKILL_SLUG
+from util.slug_naming import LEGACY_EXEMPT_SLUGS, validate_slug_semantics
+
+_TEMPLATE_PLACEHOLDER_SLUGS = frozenset({"your-skill-slug", "your_skill_slug"})
+_ENGLISH_PLACEHOLDER_NAME_PATTERNS = (
+    re.compile(r"^my\s+skill$", re.I),
+    re.compile(r"^your[-_\s]?skill", re.I),
+    re.compile(r"^skill[-_\s]?template$", re.I),
+    re.compile(r"^[a-z0-9_-]+$"),
+)
+_KEBAB_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
+_CJK_RE = re.compile(r"[\u4e00-\u9fff]")
+_FK_DDL_PATTERN = re.compile(r"\bFOREIGN\s+KEY\b|\bREFERENCES\b", re.I)
+
+
+@dataclass
+class ValidationReport:
+    errors: list[str] = field(default_factory=list)
+    warnings: list[str] = field(default_factory=list)
+
+    @property
+    def ok(self) -> bool:
+        return not self.errors
+
+
+def _has_cjk(text: str) -> bool:
+    return bool(_CJK_RE.search(text))
+
+
+def _looks_like_raw_physical_name(display_name: str, physical_name: str) -> bool:
+    if not display_name.strip():
+        return True
+    if display_name.strip().lower() == physical_name.strip().lower():
+        return True
+    compact = display_name.replace(" ", "").replace("_", "").lower()
+    if compact == physical_name.replace("_", "").lower():
+        return True
+    return False
+
+
+def _parse_frontmatter_field(text: str, field_name: str) -> str:
+    parts = text.split("---", 2)
+    if len(parts) < 2:
+        return ""
+    for line in parts[1].splitlines():
+        match = re.match(rf"^{re.escape(field_name)}:\s*(.+)\s*$", line.strip())
+        if match:
+            return match.group(1).strip().strip("\"'")
+    return ""
+
+
+def _parse_openclaw_slug(text: str) -> str:
+    parts = text.split("---", 2)
+    if len(parts) < 2:
+        return ""
+    lines = parts[1].splitlines()
+    for index, line in enumerate(lines):
+        if re.match(r"^\s+openclaw:\s*$", line):
+            for inner in lines[index + 1 :]:
+                if inner.strip() and inner[0] not in (" ", "\t"):
+                    break
+                match = re.match(r"^(\s+)slug:\s*(.+)\s*$", inner)
+                if match:
+                    return match.group(2).strip().strip("\"'")
+    return ""
+
+
+def parse_skill_frontmatter(skill_root: str) -> dict[str, str]:
+    path = os.path.join(skill_root, "SKILL.md")
+    with open(path, encoding="utf-8") as f:
+        text = f.read()
+    return {
+        "name": _parse_frontmatter_field(text, "name"),
+        "slug": _parse_openclaw_slug(text),
+    }
+
+
+def validate_skill_frontmatter(skill_root: str, *, allow_template_placeholders: bool = True) -> ValidationReport:
+    report = ValidationReport()
+    data = parse_skill_frontmatter(skill_root)
+    name = str(data.get("name") or "").strip()
+    slug = str(data.get("slug") or "").strip()
+
+    if not name:
+        report.errors.append("SKILL.md frontmatter.name 缺失")
+    elif not _has_cjk(name):
+        if allow_template_placeholders and slug in _TEMPLATE_PLACEHOLDER_SLUGS:
+            report.warnings.append("SKILL.md frontmatter.name 建议使用清晰的中文业务名称（模板占位可暂保留）")
+        else:
+            report.errors.append("SKILL.md frontmatter.name 应使用用户可见的中文业务名称")
+    else:
+        for pattern in _ENGLISH_PLACEHOLDER_NAME_PATTERNS:
+            if pattern.fullmatch(name):
+                if allow_template_placeholders and slug in _TEMPLATE_PLACEHOLDER_SLUGS:
+                    report.warnings.append(f"SKILL.md frontmatter.name 仍为占位风格：{name!r}")
+                else:
+                    report.errors.append(f"SKILL.md frontmatter.name 不应为英文占位名：{name!r}")
+                break
+
+    if not slug:
+        report.errors.append("SKILL.md metadata.openclaw.slug 缺失")
+    elif not _KEBAB_SLUG.fullmatch(slug):
+        report.errors.append(f"metadata.openclaw.slug 必须为 kebab-case：{slug!r}")
+    elif slug in _TEMPLATE_PLACEHOLDER_SLUGS and not allow_template_placeholders:
+        report.errors.append(f"metadata.openclaw.slug 仍为模板占位：{slug!r}")
+
+    constants_slug = SKILL_SLUG.strip()
+    if slug and constants_slug and slug != constants_slug:
+        report.warnings.append(
+            f"SKILL.md slug ({slug!r}) 与 util.constants.SKILL_SLUG ({constants_slug!r}) 不一致"
+        )
+
+    if slug and slug not in LEGACY_EXEMPT_SLUGS and _KEBAB_SLUG.fullmatch(slug):
+        naming_errors, naming_warnings = validate_slug_semantics(
+            slug,
+            strict=not allow_template_placeholders,
+        )
+        report.errors.extend(naming_errors)
+        report.warnings.extend(naming_warnings)
+
+    return report
+
+
+def _quote_identifier(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def _table_column_info(cursor, table_name: str) -> dict[str, dict[str, object]]:
+    cursor.execute(f"PRAGMA table_info({_quote_identifier(table_name)})")
+    info: dict[str, dict[str, object]] = {}
+    for row in cursor.fetchall():
+        info[str(row[1])] = {
+            "cid": int(row[0]),
+            "name": str(row[1]),
+            "type": str(row[2] or ""),
+            "notnull": int(row[3] or 0),
+            "dflt_value": row[4],
+            "pk": int(row[5] or 0),
+        }
+    return info
+
+
+def _table_column_xinfo(cursor, table_name: str) -> dict[str, dict[str, object]]:
+    """PRAGMA table_xinfo 可识别 generated/hidden 字段（SQLite 3.31+）。"""
+    cursor.execute(f"PRAGMA table_xinfo({_quote_identifier(table_name)})")
+    info: dict[str, dict[str, object]] = {}
+    for row in cursor.fetchall():
+        hidden = int(row[6] or 0) if len(row) > 6 else 0
+        info[str(row[1])] = {
+            "cid": int(row[0]),
+            "name": str(row[1]),
+            "type": str(row[2] or ""),
+            "notnull": int(row[3] or 0),
+            "dflt_value": row[4],
+            "pk": int(row[5] or 0),
+            "hidden": hidden,
+        }
+    return info
+
+
+def _is_generated_column(xinfo: dict[str, object]) -> bool:
+    hidden = int(xinfo.get("hidden") or 0)
+    return hidden in (2, 3)
+
+
+def _is_blob_column(column_info: dict[str, object]) -> bool:
+    return "BLOB" in str(column_info.get("type") or "").upper()
+
+
+def _default_uses_unixepoch(default_value: object) -> bool:
+    if default_value is None:
+        return False
+    text = str(default_value).strip().lower()
+    return "unixepoch" in text
+
+
+def validate_options_json(options_json: str | None) -> tuple[list[str], list[str]]:
+    """校验 options_json：静态枚举数组，或 lookup 对象。"""
+    errors: list[str] = []
+    warnings: list[str] = []
+    if options_json is None or str(options_json).strip() == "":
+        return errors, warnings
+    try:
+        parsed = json.loads(str(options_json))
+    except json.JSONDecodeError as exc:
+        errors.append(f"options_json 不是合法 JSON：{exc}")
+        return errors, warnings
+
+    if isinstance(parsed, dict):
+        for key in ("ref_table", "value_column", "label_column"):
+            if not str(parsed.get(key) or "").strip():
+                errors.append(f"options_json lookup 必须包含非空 {key}")
+        ident = re.compile(r"^[a-zA-Z_][a-zA-Z0-9_]*$")
+        for key in ("ref_table", "value_column", "label_column", "filter_column"):
+            raw = parsed.get(key)
+            if raw is None or raw == "":
+                continue
+            if not ident.match(str(raw)):
+                errors.append(f"options_json.{key} 必须是安全标识符，当前为 {raw!r}")
+        if parsed.get("filter_column") not in (None, "") and "filter_value" not in parsed:
+            warnings.append("options_json lookup 声明了 filter_column 但未提供 filter_value")
+        if "multiple" in parsed and not isinstance(parsed.get("multiple"), bool):
+            errors.append("options_json lookup.multiple 必须是布尔值")
+        return errors, warnings
+
+    if not isinstance(parsed, list):
+        errors.append("options_json 必须是 JSON 数组（静态枚举）或对象（lookup）")
+        return errors, warnings
+    for index, item in enumerate(parsed):
+        if not isinstance(item, dict):
+            errors.append(f"options_json[{index}] 必须是对象")
+            continue
+        if "value" not in item or "label" not in item:
+            errors.append(f"options_json[{index}] 必须包含 value 与 label")
+    return errors, warnings
+
+
+def validate_filter_search_metadata(
+    cursor,
+    table_name: str,
+) -> tuple[list[str], list[str]]:
+    """对可见但未开启 searchable 的枚举 / boolean 列给出软建议。"""
+    warnings: list[str] = []
+    cursor.execute(
+        """
+        SELECT column_name, visible, searchable, display_type, options_json
+        FROM _jiangchang_columns
+        WHERE table_name = ?
+        """,
+        (table_name,),
+    )
+    for column_name, visible, searchable, display_type, options_json in cursor.fetchall():
+        if visible is not None and int(visible) == 0:
+            continue
+        if searchable is not None and int(searchable) != 0:
+            continue
+
+        display_type_norm = str(display_type or "").strip()
+        is_boolean = display_type_norm == "boolean"
+
+        has_enum = False
+        if options_json is not None and str(options_json).strip():
+            opt_errors, _ = validate_options_json(options_json)
+            if not opt_errors:
+                try:
+                    parsed = json.loads(str(options_json))
+                except json.JSONDecodeError:
+                    parsed = None
+                has_enum = isinstance(parsed, list) and len(parsed) > 0
+
+        if not has_enum and not is_boolean:
+            continue
+
+        warnings.append(
+            f"{table_name}.{column_name}: 枚举/状态类列建议 searchable=1，"
+            "以便宿主列筛选与按中文 label 全局搜索（见 references/SCHEMA.md）"
+        )
+
+    return [], warnings
+
+
+def validate_column_permissions(
+    cursor,
+    table_name: str,
+    *,
+    physical_columns: list[str] | None = None,
+) -> tuple[list[str], list[str]]:
+    """校验常见系统字段 editable=0，并识别技术上可写但不适合手工新增的表。"""
+    errors: list[str] = []
+    warnings: list[str] = []
+    ordered = physical_columns or list_physical_column_names(cursor, table_name)
+    column_info = _table_column_info(cursor, table_name)
+    column_xinfo = _table_column_xinfo(cursor, table_name)
+
+    cursor.execute(
+        """
+        SELECT column_name, editable, options_json
+        FROM _jiangchang_columns
+        WHERE table_name = ?
+        """,
+        (table_name,),
+    )
+    meta_by_column = {
+        str(row[0]): {"editable": row[1], "options_json": row[2]}
+        for row in cursor.fetchall()
+    }
+
+    for column_name, meta in meta_by_column.items():
+        physical = column_info.get(column_name) or {}
+        xinfo = column_xinfo.get(column_name) or physical
+        if not physical and column_name not in column_xinfo:
+            continue
+
+        pk = int(physical.get("pk") or xinfo.get("pk") or 0)
+        editable = meta["editable"]
+        if pk and editable is not None and int(editable) != 0:
+            errors.append(f"{table_name}.{column_name} 是主键，editable 应为 0")
+
+        if column_name == "id" and editable is not None and int(editable) != 0:
+            errors.append(f"{table_name}.id 的 editable 应为 0（自增主键）")
+
+        if _is_generated_column(xinfo) and editable is not None and int(editable) != 0:
+            errors.append(f"{table_name}.{column_name} 是 generated 字段，editable 应为 0")
+
+        if _is_blob_column(physical or xinfo) and editable is not None and int(editable) != 0:
+            errors.append(f"{table_name}.{column_name} 是 BLOB 字段，editable 应为 0")
+
+        opt_errors, _ = validate_options_json(meta.get("options_json"))
+        for item in opt_errors:
+            errors.append(f"{table_name}.{column_name}: {item}")
+
+    cursor.execute(
+        "SELECT readonly FROM _jiangchang_tables WHERE table_name = ?",
+        (table_name,),
+    )
+    readonly_row = cursor.fetchone()
+    readonly = int(readonly_row[0]) if readonly_row else None
+    editable_columns = list_editable_business_columns(cursor, table_name)
+    if readonly == 0 and not editable_columns:
+        warnings.append(
+            f"{table_name}: 表级 readonly=0 但所有字段 editable=0，"
+            "宿主不应打开空白新增表单；数据应由技能业务流程产生"
+        )
+    if readonly == 1 and editable_columns:
+        warnings.append(
+            f"{table_name}: 表级 readonly=1 但存在 editable=1 字段，"
+            "宿主应以表级只读为准"
+        )
+
+    return errors, warnings
+
+
+def _expected_timestamp_display_name(column_name: str) -> str:
+    if column_name == "created_at":
+        return "创建时间"
+    if column_name == "updated_at":
+        return "更新时间"
+    return ""
+
+
+def validate_standard_timestamp_columns(
+    cursor,
+    table_name: str,
+    *,
+    physical_columns: list[str] | None = None,
+) -> tuple[list[str], list[str]]:
+    errors: list[str] = []
+    warnings: list[str] = []
+    ordered = physical_columns or list_physical_column_names(cursor, table_name)
+    column_info = _table_column_info(cursor, table_name)
+
+    for column_name in STANDARD_TIMESTAMP_COLUMNS:
+        if column_name not in ordered:
+            continue
+
+        physical = column_info.get(column_name)
+        if not physical:
+            errors.append(f"{table_name}.{column_name} 缺少物理字段定义")
+            continue
+
+        if "INT" not in str(physical["type"]).upper():
+            errors.append(
+                f"{table_name}.{column_name} 字段类型应为 INTEGER（Unix 秒级时间戳），当前为 {physical['type']!r}"
+            )
+
+        if column_name == "created_at" and not _default_uses_unixepoch(physical["dflt_value"]):
+            errors.append(
+                f"{table_name}.created_at 应设置数据库默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}"
+            )
+        if column_name == "updated_at" and not _default_uses_unixepoch(physical["dflt_value"]):
+            errors.append(
+                f"{table_name}.updated_at 应设置数据库初始默认值 DEFAULT (unixepoch())，当前默认值为 {physical['dflt_value']!r}"
+            )
+
+        cursor.execute(
+            """
+            SELECT display_name, display_type, editable, searchable, display_order
+            FROM _jiangchang_columns
+            WHERE table_name = ? AND column_name = ?
+            """,
+            (table_name, column_name),
+        )
+        meta_row = cursor.fetchone()
+        if not meta_row:
+            errors.append(f"{table_name}.{column_name} 缺少 _jiangchang_columns 元数据")
+            continue
+
+        display_name, display_type, editable, searchable, display_order = meta_row
+        expected_name = _expected_timestamp_display_name(column_name)
+        if str(display_name or "").strip() != expected_name:
+            errors.append(
+                f"{table_name}.{column_name} 的 display_name 应为 {expected_name!r}，当前为 {display_name!r}"
+            )
+        if str(display_type or "").strip() != DATETIME_UNIX_SECONDS:
+            errors.append(
+                f"{table_name}.{column_name} 的 display_type 应为 {DATETIME_UNIX_SECONDS!r}，当前为 {display_type!r}"
+            )
+        if editable is None or int(editable) != 0:
+            errors.append(f"{table_name}.{column_name} 的 editable 应为 0（系统时间字段不可编辑）")
+        if display_order is not None:
+            errors.append(f"{table_name}.{column_name} 的 display_order 应为 NULL")
+        if searchable is not None and int(searchable) != 0:
+            warnings.append(f"{table_name}.{column_name} 的时间字段通常不必开启 searchable")
+
+    if "created_at" in ordered and "updated_at" in ordered:
+        if ordered.index("updated_at") < ordered.index("created_at"):
+            warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前")
+
+    if "updated_at" in ordered and not has_updated_at_trigger(cursor, table_name):
+        errors.append(
+            f"{table_name}.updated_at 缺少自动维护 trigger（期望 {table_name}_set_updated_at）"
+        )
+
+    return errors, warnings
+
+
+def column_order_warnings(table_name: str, ordered_names: list[str]) -> list[str]:
+    warnings: list[str] = []
+    if "id" in ordered_names and ordered_names.index("id") != 0:
+        warnings.append(f"{table_name}: 字段 id 通常应定义在 CREATE TABLE 的第一列")
+    if "created_at" in ordered_names:
+        idx = ordered_names.index("created_at")
+        trailing = ordered_names[idx + 1 :]
+        if trailing and trailing != ["updated_at"]:
+            warnings.append(
+                f"{table_name}: created_at 之后通常只应跟 updated_at，不应再放置其他业务字段"
+            )
+    if "updated_at" in ordered_names and "created_at" in ordered_names:
+        if ordered_names.index("updated_at") < ordered_names.index("created_at"):
+            warnings.append(f"{table_name}: updated_at 不应出现在 created_at 之前")
+    return warnings
+
+
+def _table_exists(cursor, table_name: str) -> bool:
+    cursor.execute(
+        "SELECT 1 FROM sqlite_master WHERE type = 'table' AND name = ? LIMIT 1",
+        (table_name,),
+    )
+    return cursor.fetchone() is not None
+
+
+def validate_no_sqlite_foreign_keys(cursor) -> tuple[list[str], list[str]]:
+    """业务表 DDL 不得含 FOREIGN KEY / REFERENCES（关联走 repository/service）。"""
+    errors: list[str] = []
+    cursor.execute(
+        """
+        SELECT name, sql FROM sqlite_master
+        WHERE type = 'table' AND name NOT LIKE 'sqlite_%'
+        """
+    )
+    for table_name, sql in cursor.fetchall():
+        if table_name in METADATA_TABLES:
+            continue
+        ddl = str(sql or "")
+        if ddl and _FK_DDL_PATTERN.search(ddl):
+            errors.append(
+                f"业务表 {table_name!r} 的 DDL 含 SQLite FOREIGN KEY/REFERENCES；"
+                "请改为业务 id 列 + repository/service 逻辑关联（见 references/SCHEMA.md）"
+            )
+    return errors, []
+
+
+def validate_display_metadata(conn: sqlite3.Connection, *, check_column_order: bool = True) -> ValidationReport:
+    report = ValidationReport()
+    cur = conn.cursor()
+
+    has_tables_meta = _table_exists(cur, "_jiangchang_tables")
+    has_columns_meta = _table_exists(cur, "_jiangchang_columns")
+    if not has_tables_meta:
+        report.errors.append("缺少数据管理元数据表 _jiangchang_tables")
+    if not has_columns_meta:
+        report.errors.append("缺少数据管理元数据表 _jiangchang_columns")
+    if not has_tables_meta or not has_columns_meta:
+        return report
+
+    fk_errors, _ = validate_no_sqlite_foreign_keys(cur)
+    report.errors.extend(fk_errors)
+
+    for table_name in iter_user_tables(cur):
+        cur.execute(
+            "SELECT display_name FROM _jiangchang_tables WHERE table_name = ?",
+            (table_name,),
+        )
+        row = cur.fetchone()
+        if not row or not str(row[0] or "").strip():
+            report.errors.append(f"业务表 {table_name!r} 缺少 _jiangchang_tables.display_name")
+        elif not _has_cjk(str(row[0])):
+            report.errors.append(f"业务表 {table_name!r} 的 display_name 应使用中文业务名称")
+
+        physical_columns = list_physical_column_names(cur, table_name)
+        if check_column_order:
+            report.warnings.extend(column_order_warnings(table_name, physical_columns))
+
+        for column_name in physical_columns:
+            cur.execute(
+                """
+                SELECT display_name, visible
+                FROM _jiangchang_columns
+                WHERE table_name = ? AND column_name = ?
+                """,
+                (table_name, column_name),
+            )
+            col_row = cur.fetchone()
+            if not col_row:
+                report.errors.append(
+                    f"字段 {table_name}.{column_name} 缺少 _jiangchang_columns 元数据"
+                )
+                continue
+            display_name = str(col_row[0] or "").strip()
+            visible = col_row[1] is None or int(col_row[1]) != 0
+            if visible and not display_name:
+                report.errors.append(f"字段 {table_name}.{column_name} 的 display_name 为空")
+            elif visible and _looks_like_raw_physical_name(display_name, column_name):
+                report.errors.append(
+                    f"字段 {table_name}.{column_name} 的 display_name 不应直接等于英文物理名：{display_name!r}"
+                )
+            elif visible and not _has_cjk(display_name):
+                report.errors.append(
+                    f"字段 {table_name}.{column_name} 的 display_name 应使用中文：{display_name!r}"
+                )
+
+        ts_errors, ts_warnings = validate_standard_timestamp_columns(
+            cur,
+            table_name,
+            physical_columns=physical_columns,
+        )
+        report.errors.extend(ts_errors)
+        report.warnings.extend(ts_warnings)
+
+        perm_errors, perm_warnings = validate_column_permissions(
+            cur,
+            table_name,
+            physical_columns=physical_columns,
+        )
+        report.errors.extend(perm_errors)
+        report.warnings.extend(perm_warnings)
+
+        _, filter_warnings = validate_filter_search_metadata(cur, table_name)
+        report.warnings.extend(filter_warnings)
+
+    cur.execute("SELECT table_name, column_name FROM _jiangchang_columns")
+    for table_name, column_name in cur.fetchall():
+        if table_name in METADATA_TABLES:
+            continue
+        cur.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?",
+            (table_name,),
+        )
+        if not cur.fetchone():
+            report.errors.append(f"元数据引用了不存在的表：{table_name!r}")
+            continue
+        physical = list_physical_column_names(cur, str(table_name))
+        if str(column_name) not in physical:
+            report.errors.append(f"元数据引用了不存在的字段：{table_name}.{column_name}")
+
+    cur.execute("SELECT table_name FROM _jiangchang_tables")
+    for (table_name,) in cur.fetchall():
+        if table_name in METADATA_TABLES:
+            continue
+        cur.execute(
+            "SELECT 1 FROM sqlite_master WHERE type='table' AND name = ?",
+            (table_name,),
+        )
+        if not cur.fetchone():
+            report.errors.append(f"元数据引用了不存在的表：{table_name!r}")
+
+    return report
+
+
+def validate_template_database(conn: sqlite3.Connection) -> ValidationReport:
+    """模板自检入口：校验展示元数据完整性，并扫描业务表 DDL 是否含 SQLite 外键。"""
+    return validate_display_metadata(conn)

scripts/db/seed_data.py

@@ -0,0 +1,58 @@
+"""业务种子数据：职称评审刚性门槛（示例，上线前由人事处核准维护）。"""
+
+from __future__ import annotations
+
+import sqlite3
+
+from util.timeutil import now_unix
+
+_BASIS = "《专业技术职务评聘办法》"
+
+# (rank, metric_key, metric_name, min_value, unit, basis_doc, clause)
+_SEED_THRESHOLDS: tuple[tuple[str, str, str, float, str, str, str], ...] = (
+    ("副教授", "presiding_provincial", "主持省部级项目", 1.0, "项", _BASIS, "第X条"),
+    ("副教授", "core_papers", "核心期刊论文", 3.0, "篇", _BASIS, "第X条"),
+    ("副教授", "teaching_hours", "年均教学工作量", 160.0, "课时", _BASIS, "第X条"),
+    ("副教授", "current_title_years", "任现职年限", 5.0, "年", _BASIS, "第X条"),
+    ("教授", "presiding_provincial", "主持省部级项目", 2.0, "项", _BASIS, "第X条"),
+    ("教授", "core_papers", "核心期刊论文", 6.0, "篇", _BASIS, "第X条"),
+    ("教授", "monographs", "专著/教材", 1.0, "部", _BASIS, "第X条"),
+    ("教授", "current_title_years", "任现职年限", 5.0, "年", _BASIS, "第X条"),
+)
+
+_KNOWN_METRIC_KEYS = (
+    "presiding_provincial",
+    "core_papers",
+    "teaching_hours",
+    "current_title_years",
+    "monographs",
+    "awards",
+    "degree",
+)
+_KNOWN_RANKS = ("副教授", "教授")
+
+
+def seed_thresholds(cursor: sqlite3.Cursor) -> int:
+    """幂等写入门槛种子，返回新增条数。"""
+    inserted = 0
+    for rank, metric_key, metric_name, min_value, unit, basis_doc, clause in _SEED_THRESHOLDS:
+        cursor.execute(
+            "SELECT COUNT(*) FROM threshold_rules WHERE rank = ? AND metric_key = ?",
+            (rank, metric_key),
+        )
+        if int(cursor.fetchone()[0]) > 0:
+            continue
+        cursor.execute(
+            """
+            INSERT INTO threshold_rules
+                (rank, metric_key, metric_name, min_value, unit, basis_doc, clause, created_at, updated_at)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+            """,
+            (rank, metric_key, metric_name, min_value, unit, basis_doc, clause, now_unix(), now_unix()),
+        )
+        inserted += 1
+    return inserted
+
+
+def seed_all(cursor: sqlite3.Cursor) -> dict[str, int]:
+    return {"threshold_rules": seed_thresholds(cursor)}

scripts/db/timestamp_columns.py

@@ -0,0 +1,50 @@
+"""标准时间字段：Unix 秒级时间戳、默认值与 updated_at 自动维护。
+
+业务表若包含 created_at / updated_at，在 CREATE TABLE 后调用
+``init_updated_at_trigger(cursor, table_name)``。模板默认不建业务表。
+"""
+
+from __future__ import annotations
+
+STANDARD_TIMESTAMP_COLUMNS = frozenset({"created_at", "updated_at"})
+DATETIME_UNIX_SECONDS = "datetime_unix_seconds"
+
+
+def updated_at_trigger_name(table_name: str) -> str:
+    return f"{table_name}_set_updated_at"
+
+
+def _quote_identifier(name: str) -> str:
+    return '"' + name.replace('"', '""') + '"'
+
+
+def init_updated_at_trigger(cursor, table_name: str) -> None:
+    """幂等创建 {table}_set_updated_at：业务字段 UPDATE 后刷新 updated_at。"""
+    trigger = updated_at_trigger_name(table_name)
+    quoted = _quote_identifier(table_name)
+    quoted_trigger = _quote_identifier(trigger)
+    cursor.execute(
+        f"""
+CREATE TRIGGER IF NOT EXISTS {quoted_trigger}
+AFTER UPDATE ON {quoted}
+FOR EACH ROW
+BEGIN
+    UPDATE {quoted}
+    SET updated_at = unixepoch()
+    WHERE id = NEW.id AND updated_at = OLD.updated_at;
+END
+"""
+    )
+
+
+def has_updated_at_trigger(cursor, table_name: str) -> bool:
+    trigger_name = updated_at_trigger_name(table_name)
+    cursor.execute(
+        """
+        SELECT 1 FROM sqlite_master
+        WHERE type = 'trigger' AND name = ? AND tbl_name = ?
+        LIMIT 1
+        """,
+        (trigger_name, table_name),
+    )
+    return cursor.fetchone() is not None

scripts/main.py

@@ -0,0 +1,38 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+skill-template CLI 入口。
+
+复制为新技能后，请修改注释与常量，但保留当前分层结构：
+cli（argv）→ service（业务编排）→ db（持久化）→ util（通用工具）。
+
+jiangchang_skill_core 来自宿主共享 Python runtime 安装的 jiangchang-platform-kit，
+不在技能仓库内 vendored。
+"""
+from __future__ import annotations
+
+import os
+import sys
+
+if sys.platform == "win32":
+    import io
+
+    sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding="utf-8", errors="replace")
+    sys.stderr = io.TextIOWrapper(sys.stderr.buffer, encoding="utf-8", errors="replace")
+
+_scripts_dir = os.path.dirname(os.path.abspath(__file__))
+if _scripts_dir not in sys.path:
+    sys.path.insert(0, _scripts_dir)
+
+from jiangchang_skill_core.runtime_env import apply_cli_local_defaults
+
+apply_cli_local_defaults()
+
+from util.config_bootstrap import bootstrap_skill_config
+
+bootstrap_skill_config()
+
+from cli.app import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

scripts/service/entitlement_service.py

@@ -0,0 +1,46 @@
+"""在线鉴权（可选模板）。"""
+
+from __future__ import annotations
+
+from typing import Tuple
+
+import requests
+from jiangchang_skill_core import config
+
+
+def check_entitlement(skill_slug: str) -> Tuple[bool, str]:
+    auth_base = (config.get("JIANGCHANG_AUTH_BASE_URL") or "").strip().rstrip("/")
+    if not auth_base:
+        return True, ""
+    user_id = (config.get("JIANGCHANG_USER_ID") or "").strip()
+    if not user_id:
+        return False, "鉴权失败：缺少用户身份（JIANGCHANG_USER_ID）"
+
+    auth_api_key = (config.get("JIANGCHANG_AUTH_API_KEY") or "").strip()
+    timeout = int((config.get("JIANGCHANG_AUTH_TIMEOUT_SECONDS") or "5").strip())
+    headers = {"Content-Type": "application/json"}
+    if auth_api_key:
+        headers["Authorization"] = f"Bearer {auth_api_key}"
+    payload = {
+        "user_id": user_id,
+        "skill_slug": skill_slug,
+        "trace_id": (config.get("JIANGCHANG_TRACE_ID") or "").strip(),
+        "context": {"entry": "main.py"},
+    }
+    try:
+        res = requests.post(f"{auth_base}/api/entitlements/check", json=payload, headers=headers, timeout=timeout)
+    except requests.RequestException as exc:
+        return False, f"鉴权请求失败：{exc}"
+    if res.status_code != 200:
+        return False, f"鉴权服务异常：HTTP {res.status_code}"
+    try:
+        body = res.json()
+    except ValueError:
+        return False, "鉴权服务异常：返回非 JSON"
+    code = body.get("code")
+    data = body.get("data") or {}
+    if code != 200:
+        return False, str(body.get("msg") or "鉴权失败")
+    if not data.get("allow", False):
+        return False, str(data.get("reason") or "未购买或已过期")
+    return True, ""

scripts/service/html_page_service.py

@@ -0,0 +1,78 @@
+"""技能 HTML 报告页：写入 `{skill_data}/html/*.html`（宿主扫描预览）。
+
+契约要点（权威说明见 `references/ACTIONS.md`「HTML 页面报告」）：
+
+- 目录：仅一层 `html/*.html`（非递归）
+- 重新生成 Action id 必须为 `render-html`（宿主硬编码）
+- 页面布局底线见 `references/ACTIONS.md`「宿主预览布局」；模板壳 `assets/html/example-report.html`
+- 成功后建议 `steps: refreshCatalog` 以重扫侧栏
+- 本模块只提供路径/模板/原子写；业务 KPI/表格由各技能自行填充
+"""
+
+from __future__ import annotations
+
+import os
+import re
+from typing import Mapping
+
+from util.logging_config import get_skill_logger
+from util.runtime_paths import get_html_dir, get_skill_root
+
+# Host Data Management looks up this exact Skill Action id for「重新生成」.
+HOST_HTML_RENDER_ACTION_ID = "render-html"
+
+_UNSAFE_NAME = re.compile(r"[\\/]|^\.\.|\.\.$")
+
+
+def _log():
+    try:
+        return get_skill_logger()
+    except RuntimeError:
+        return None
+
+
+def assert_html_filename(filename: str) -> str:
+    name = (filename or "").strip()
+    if not name.lower().endswith(".html"):
+        raise ValueError(f"html filename must end with .html: {filename!r}")
+    if _UNSAFE_NAME.search(name) or "/" in name or "\\" in name:
+        raise ValueError(f"html filename must be a basename only: {filename!r}")
+    if len(name) > 200:
+        raise ValueError(f"html filename too long: {filename!r}")
+    return name
+
+
+def html_page_path(filename: str, *, create_dir: bool = True) -> str:
+    return os.path.join(get_html_dir(create=create_dir), assert_html_filename(filename))
+
+
+def load_html_template(relative_under_skill_root: str) -> str:
+    """Load UTF-8 template from skill root (e.g. ``assets/html/example-report.html``)."""
+    rel = (relative_under_skill_root or "").strip().replace("\\", "/")
+    if not rel or rel.startswith("/") or ".." in rel.split("/"):
+        raise ValueError(f"invalid template path: {relative_under_skill_root!r}")
+    path = os.path.join(get_skill_root(), *rel.split("/"))
+    with open(path, encoding="utf-8") as f:
+        return f.read()
+
+
+def render_html_placeholders(template: str, mapping: Mapping[str, str]) -> str:
+    """Replace ``{{KEY}}`` tokens. Unknown keys are left as-is."""
+    page = template
+    for key, value in mapping.items():
+        page = page.replace("{{" + str(key) + "}}", str(value))
+    return page
+
+
+def write_html_page(filename: str, html: str) -> str:
+    """Atomically write UTF-8 HTML under ``html/``. Returns absolute path."""
+    dest = html_page_path(filename, create_dir=True)
+    tmp = dest + ".tmp"
+    text = html if isinstance(html, str) else str(html)
+    with open(tmp, "w", encoding="utf-8", newline="\n") as f:
+        f.write(text)
+    os.replace(tmp, dest)
+    log = _log()
+    if log:
+        log.info("html_page_written path=%s bytes=%s", dest, len(text.encode("utf-8")))
+    return dest

scripts/service/report_service.py

@@ -0,0 +1,184 @@
+"""《核验报告》HTML 组装。
+
+- 每案例一份 `report-{case_id}.html`（三色结论 + 达标矩阵 + 补充材料清单）
+- 批量核验另生成汇总页 `brief-{batch_no}.html`
+写入 `{skill_data}/html/`，宿主数据管理侧栏可预览；重建走固定 Action id `render-html`。
+"""
+
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from service.html_page_service import load_html_template, render_html_placeholders, write_html_page
+from util.logging_config import get_skill_logger
+
+
+def _log():
+    try:
+        return get_skill_logger()
+    except RuntimeError:
+        return None
+
+_TEMPLATE_REL = "assets/html/verify-report.html"
+
+_VERDICT_BADGE = {"安全": "tier-strong", "边缘": "tier-try", "硬伤": "tier-strong"}
+_COLOR_BADGE = {"green": "tier-strong", "yellow": "tier-try", "red": "tag-red", "gray": "tier-not"}
+_COLOR_MARK = {"green": "🟢 达标", "yellow": "🟡 边缘", "red": "🔴 硬伤", "gray": "⚪ 待核"}
+
+
+def _esc(value: Any) -> str:
+    import html
+
+    return html.escape(str(value if value is not None else "—"), quote=True)
+
+
+def _verdict_badge(verdict: str) -> str:
+    return {"安全": "tier-strong", "边缘": "tier-try", "硬伤": "tag-red"}.get(verdict, "tier-not")
+
+
+def _matrix_rows(matrix: list[dict[str, Any]]) -> str:
+    rows = []
+    for item in matrix:
+        color = str(item.get("color") or "gray")
+        badge = _COLOR_BADGE.get(color, "tier-not")
+        required = item.get("required")
+        required_text = "—" if required is None else f"{required:g}"
+        margin = item.get("margin")
+        margin_text = "—" if margin is None else f"{margin * 100:.1f}%"
+        rows.append(
+            "<tr>"
+            f"<td>{_esc(item.get('metric'))}</td>"
+            f"<td class='nowrap'>{_esc(required_text)}</td>"
+            f"<td class='nowrap'>{_esc(item.get('actual'))}</td>"
+            f"<td class='nowrap'>{_esc(margin_text)}</td>"
+            f"<td class='nowrap'><span class='tag {badge}'>{_COLOR_MARK.get(color, color)}</span></td>"
+            f"<td class='muted'>{_esc(item.get('basis'))}</td>"
+            "</tr>"
+        )
+    if not rows:
+        rows.append('<tr><td colspan="6" class="muted">无</td></tr>')
+    return "".join(rows)
+
+
+def _advice_list(matrix: list[dict[str, Any]]) -> str:
+    advices = [item.get("advice") for item in matrix if item.get("advice")]
+    if not advices:
+        return '<p class="muted">无补充事项。</p>'
+    items = "".join(f"<li>{_esc(a)}</li>" for a in advices)
+    return f"<ol class='reasons'>{items}</ol>"
+
+
+def build_verify_html(result: dict[str, Any], batch_no: str = "") -> str:
+    verdict = str(result.get("verdict") or "")
+    mark = {"安全": "🟢 安全", "边缘": "🟡 边缘", "硬伤": "🔴 硬伤"}.get(verdict, verdict)
+    body = f"""
+    <div class="verdict-line">
+      <span class="tag {_verdict_badge(verdict)}">{mark}</span>
+      <span class="muted">{_esc(result.get("department") or "")} · 申报 {_esc(result.get("apply_rank") or "—")} · 批次 {_esc(batch_no or result.get("batch_no"))}</span>
+    </div>
+    <p class="para">{_esc(result.get("summary"))}</p>
+    <div class="section"><h2>硬指标达标矩阵</h2>
+      <table>
+        <tr><th>指标</th><th>最低要求</th><th>实有</th><th>余量</th><th>判定</th><th>依据</th></tr>
+        {_matrix_rows(result.get("matrix") or [])}
+      </table>
+    </div>
+    <div class="section"><h2>补充材料清单</h2>{_advice_list(result.get("matrix") or [])}</div>
+    <footer>核验仅针对硬指标形式审查，不构成评审结论；门槛以人事处最新文件为准，是否通过评审由评委会决定。</footer>
+    """
+    template = load_html_template(_TEMPLATE_REL)
+    return render_html_placeholders(
+        template,
+        {
+            "TITLE": f"职称申报核验报告 · {result.get('teacher_name') or ''}",
+            "GENERATED_AT": "",
+            "BODY": body,
+        },
+    )
+
+
+def write_report_for_kernel(*, result: dict[str, Any], batch_no: str) -> str:
+    html_text = build_verify_html(result, batch_no)
+    log = _log()
+    path = write_html_page(f"report-{result['case_id']}.html", html_text)
+    if log:
+        log.info("verify_report_written case_id=%s path=%s stage=report", result["case_id"], path)
+    return path
+
+
+def build_report_for_case_id(case_id: int) -> str:
+    """按案例 id 从库中取数并重写 HTML（重建入口用）。"""
+    from db.connection import get_conn
+
+    log = _log()
+    if log:
+        log.info("report_render_start case_id=%s stage=render", case_id)
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT * FROM verify_cases WHERE id = ?", (case_id,))
+        row = cur.fetchone()
+        if not row:
+            raise ValueError(f"核验案例不存在：{case_id}")
+        columns = [d[0] for d in cur.description]
+        record = dict(zip(columns, row))
+    finally:
+        conn.close()
+    result = dict(record)
+    try:
+        result["matrix"] = json.loads(record.get("matrix_json") or "[]")
+    except (TypeError, ValueError):
+        result["matrix"] = []
+    html_text = build_verify_html(result)
+    path = write_html_page(f"report-{case_id}.html", html_text)
+    log.info("report_render_done case_id=%s path=%s stage=render", case_id, path)
+    return path
+
+
+def build_batch_brief(*, batch_no: str, results: list[dict[str, Any]]) -> str:
+    rows = []
+    for result in results:
+        verdict = str(result.get("verdict") or "")
+        badge = _verdict_badge(verdict)
+        red_items = [m for m in result.get("matrix") or [] if m.get("color") == "red"]
+        red_text = "、".join(str(m.get("metric")) for m in red_items) if red_items else "—"
+        rows.append(
+            "<tr>"
+            f"<td>{_esc(result.get('teacher_name'))}</td>"
+            f"<td class='muted'>{_esc(result.get('department'))}</td>"
+            f"<td class='nowrap'>{_esc(result.get('apply_rank'))}</td>"
+            f"<td class='nowrap'><span class='tag {badge}'>{_esc(verdict)}</span></td>"
+            f"<td class='muted'>{_esc(red_text)}</td>"
+            "</tr>"
+        )
+    if not rows:
+        rows.append('<tr><td colspan="5" class="muted">本期无核验结果。</td></tr>')
+
+    counts: dict[str, int] = {}
+    for result in results:
+        counts[str(result.get("verdict"))] = counts.get(str(result.get("verdict")), 0) + 1
+    chips = " ".join(
+        f"<span class='tag {_verdict_badge(v)}'>{_esc(v)} × {n}</span>" for v, n in sorted(counts.items())
+    )
+
+    body = f"""
+    <div class="verdict-line">
+      {chips or '<span class="muted">无</span>'}
+      <span class="muted">批次 {_esc(batch_no)} · 案例 {_esc(len(results))} 个</span>
+    </div>
+    <div class="section"><h2>核验结论一览</h2>
+      <table><tr><th>教师</th><th>学院</th><th>申报级别</th><th>结论</th><th>硬伤项</th></tr>{''.join(rows)}</table>
+    </div>
+    <footer>汇总由批量核验自动生成，供师资科排序复核；核验不构成评审结论。</footer>
+    """
+    template = load_html_template(_TEMPLATE_REL)
+    page = render_html_placeholders(
+        template,
+        {"TITLE": "职称核验汇总", "GENERATED_AT": "", "BODY": body},
+    )
+    path = write_html_page(f"brief-{batch_no}.html", page)
+    log = _log()
+    if log:
+        log.info("verify_brief_written batch_no=%s path=%s stage=report", batch_no, path)
+    return path

scripts/service/sibling_bridge.py

@@ -0,0 +1,57 @@
+"""兄弟技能 CLI 调用工具。
+
+通过子进程调用同级其他 skill 的 main.py，并解析 JSON 输出。
+具体调用哪些兄弟技能，由复制后的业务 skill 在 service 层按需决定。
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from typing import Any, Dict, List, Optional
+
+from util.logging_config import subprocess_env_with_trace
+from util.runtime_paths import get_skills_root
+
+
+def call_sibling_json(skill_slug: str, args: List[str]) -> Optional[Dict[str, Any]]:
+    """通用兄弟技能调用器。
+
+    Args:
+        skill_slug: 兄弟技能的 slug（即同级目录名）。
+        args: 传给 main.py 的命令行参数列表。
+
+    Returns:
+        若兄弟技能输出可解析的 JSON 对象则返回 dict；否则返回 None。
+    """
+    script_path = os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")
+    if not os.path.isfile(script_path):
+        return None
+    proc = subprocess.run(
+        [sys.executable, script_path, *args],
+        capture_output=True,
+        text=True,
+        encoding="utf-8",
+        errors="replace",
+        env=subprocess_env_with_trace(),
+    )
+    raw = (proc.stdout or "").strip()
+    if not raw or raw.startswith("ERROR"):
+        return None
+    try:
+        data = json.loads(raw)
+    except json.JSONDecodeError:
+        return None
+    return data if isinstance(data, dict) else None
+
+
+def get_sibling_main_path(skill_slug: str) -> str:
+    """获取兄弟技能的 main.py 绝对路径。
+
+    使用示例（在 task_service.py 中）：
+        from service.sibling_bridge import get_sibling_main_path, call_sibling_json
+        result = call_sibling_json("account-manager", ["list", "--limit", "10"])
+    """
+    return os.path.join(get_skills_root(), skill_slug, "scripts", "main.py")

scripts/service/task_run_support.py

@@ -0,0 +1,84 @@
+"""任务运行辅助：录屏信息组装与 CLI 输出（通用模板，无业务逻辑）。"""
+
+from __future__ import annotations
+
+from typing import Any, Dict, Optional
+
+
+def build_video_info(
+    video_summary: dict[str, Any],
+    compose_context_before: Optional[dict[str, Any]] = None,
+    compose_context_after: Optional[dict[str, Any]] = None,
+) -> dict[str, Any]:
+    """将 RpaVideoSession.summary() 转为 finish() / CLI 可用的 video 块。"""
+    return {
+        "enabled": video_summary.get("enabled", False),
+        "path": video_summary.get("path"),
+        "capture_path": video_summary.get("capture_path"),
+        "record_log_path": video_summary.get("record_log_path"),
+        "warnings": list(video_summary.get("warnings") or []),
+        "voiceover_path": video_summary.get("voiceover_path"),
+        "music_path": video_summary.get("music_path"),
+        "audio_warnings": list(video_summary.get("audio_warnings") or []),
+        "compose_context": {
+            **(compose_context_before or {}),
+            "after_compose": compose_context_after or {},
+        },
+    }
+
+
+def merge_video_into_result_summary(
+    payload: dict[str, Any],
+    video_info: Optional[dict[str, Any]],
+) -> dict[str, Any]:
+    """把 video artifact 字段写入 finish() 结果对象（video / video_path 等）。"""
+    if not video_info:
+        return payload
+    payload = dict(payload)
+    payload["video"] = video_info
+    payload["video_path"] = video_info.get("path")
+    payload["raw_video"] = video_info.get("capture_path")
+    payload["video_log"] = video_info.get("record_log_path")
+    payload["video_warnings"] = list(video_info.get("warnings") or [])
+    payload["voiceover_path"] = video_info.get("voiceover_path")
+    payload["music_path"] = video_info.get("music_path")
+    payload["audio_warnings"] = list(video_info.get("audio_warnings") or [])
+    return payload
+
+
+def _print_video_summary(video_block: Optional[dict[str, Any]]) -> None:
+    """任务完成后在 CLI 打印录屏路径、日志与诊断信息。"""
+    if not isinstance(video_block, dict) or not video_block.get("enabled"):
+        return
+    record_log = video_block.get("record_log_path")
+    if record_log:
+        print(f"录屏日志：{record_log}")
+    path = video_block.get("path")
+    if path:
+        print(f"录屏路径：{path}")
+    else:
+        print("录屏路径：未生成")
+        warnings = video_block.get("warnings") or []
+        if warnings:
+            print(f"视频诊断：{'; '.join(str(w) for w in warnings)}")
+        capture = video_block.get("capture_path")
+        if capture:
+            print(f"原始录屏路径：{capture}")
+    voiceover = video_block.get("voiceover_path")
+    if voiceover:
+        print(f"旁白路径：{voiceover}")
+    music = video_block.get("music_path")
+    if music:
+        print(f"背景音乐：{music}")
+    audio_warnings = video_block.get("audio_warnings") or []
+    if audio_warnings:
+        print(f"音频诊断：{'; '.join(str(w) for w in audio_warnings)}")
+    compose_ctx = video_block.get("compose_context")
+    if isinstance(compose_ctx, dict):
+        print(
+            "视频合成上下文："
+            f"media_assets_root={compose_ctx.get('media_assets_root', '')}; "
+            f"ffmpeg_path={compose_ctx.get('ffmpeg_path', '')}; "
+            f"background_music_issue={compose_ctx.get('background_music_issue') or ''}; "
+            f"background_music_sample_path={compose_ctx.get('background_music_sample_path', '')}"
+        )

scripts/service/task_service.py

@@ -0,0 +1,570 @@
+"""任务编排层：job_context / emit / finish、鉴权与错误转换。
+
+领域业务在 ``service/verify_service.py``（核验内核）、
+``service/threshold_service.py``（门槛/案例/统计）、
+``service/report_service.py``（HTML 简报），由本模块调用；
+多入口（CLI / Action / 对话 / cron）复用同一内核，禁止按入口分叉业务。
+"""
+
+from __future__ import annotations
+
+import asyncio
+import json
+import os
+import uuid
+from typing import Any, Optional
+
+from jiangchang_skill_core import collect_runtime_diagnostics, config, format_runtime_health_lines
+from jiangchang_skill_core.activity import JobStopped, emit, finish, job_context
+
+from jiangchang_skill_core.rpa.video_session import RpaVideoSession
+from service.entitlement_service import check_entitlement
+from service.task_run_support import (
+    _print_video_summary,
+    build_video_info,
+    merge_video_into_result_summary,
+)
+from util.constants import LOG_LOGGER_NAME, PLATFORM_KIT_MIN_VERSION, SKILL_SLUG, SKILL_VERSION
+from util.logging_config import get_skill_logger, setup_skill_logging
+from util.runtime_paths import get_db_path, get_skill_root
+from db.connection import init_db
+
+
+def _get_task_logger():
+    try:
+        return get_skill_logger()
+    except RuntimeError:
+        setup_skill_logging(SKILL_SLUG, LOG_LOGGER_NAME)
+        return get_skill_logger()
+
+
+def _print_json(payload: dict[str, Any]) -> None:
+    print(json.dumps(payload, ensure_ascii=False))
+
+
+# ---------------------------------------------------------------------------
+# 模板演示入口（保留：满足 POLICY-RPA-003 录屏集成与 CLI run 契约；不承载本技能业务）
+# ---------------------------------------------------------------------------
+
+async def _run_template_demo(target: Optional[str], input_id: Optional[str]) -> tuple[int, dict[str, Any]]:
+    """最小 RpaVideoSession 示范：不启动浏览器、不执行业务，仅演示录屏包裹范式。"""
+    batch_id = uuid.uuid4().hex[:12]
+    data_dir = get_skill_root()
+
+    async with RpaVideoSession(
+        skill_slug=SKILL_SLUG,
+        skill_data_dir=data_dir,
+        batch_id=batch_id,
+        title="开始执行示例任务",
+        closing_title="示例任务执行完成",
+    ) as video:
+        video.add_step("准备执行示例任务")
+        if target:
+            video.add_step(f"接收目标参数：{target}")
+        if input_id:
+            video.add_step(f"接收输入标识：{input_id}")
+        video.add_step("示例任务执行完成")
+
+    video_summary = video.summary()
+    video_info = build_video_info(video_summary, {}, {})
+    return 1, video_info
+
+
+def cmd_run(target: Optional[str] = None, input_id: Optional[str] = None) -> int:
+    """模板演示入口：本技能主能力请使用 broadcast 等业务命令。"""
+    log = _get_task_logger()
+    task_type = "demo"
+
+    with job_context(skill=SKILL_SLUG):
+        log.info(
+            "task_start task_type=%s target_id=%s input_id=%s stage=run",
+            task_type,
+            target,
+            input_id,
+        )
+        emit("开始处理任务", skill=SKILL_SLUG, stage="run")
+        try:
+            ok, reason = check_entitlement(SKILL_SLUG)
+            if not ok:
+                log.warning(
+                    "entitlement_failed task_type=%s stage=auth error_code=ENTITLEMENT_DENIED",
+                    task_type,
+                )
+                emit("鉴权失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="auth")
+                finish(
+                    status="failed",
+                    message=reason,
+                    skill=SKILL_SLUG,
+                    error_code="ENTITLEMENT_DENIED",
+                    stage="auth",
+                )
+                print(f"ERROR:ENTITLEMENT_DENIED {reason}")
+                return 1
+
+            log.info("entitlement_ok task_type=%s stage=auth", task_type)
+            log.info("rpa_start stage=demo_prepare task_type=%s", task_type)
+            log.info("task_run_demo_start target_id=%s input_id=%s stage=demo_prepare", target, input_id)
+            rc, video_info = asyncio.run(_run_template_demo(target, input_id))
+            log.info("rpa_done stage=demo_prepare status=ok")
+            log.info(
+                "task_run_demo_done target_id=%s input_id=%s stage=demo_finish status=failed",
+                target,
+                input_id,
+            )
+            _print_video_summary(video_info)
+
+            summary_payload = merge_video_into_result_summary(
+                {
+                    "template_demo": True,
+                    "target": target,
+                    "input_id": input_id,
+                },
+                video_info,
+            )
+
+            log.warning(
+                "task_failed task_type=%s stage=demo_finish error_code=TEMPLATE_STUB",
+                task_type,
+            )
+            finish(
+                status="failed",
+                message="模板演示入口未承载本技能业务",
+                skill=SKILL_SLUG,
+                error_code="TEMPLATE_STUB",
+                **summary_payload,
+            )
+            print("ERROR:TEMPLATE_STUB 这是模板演示入口，请使用 verify-metric 命令执行职称硬指标核验。")
+            return rc
+        except JobStopped:
+            raise
+        except Exception:
+            log.exception(
+                "task_failed task_type=%s target_id=%s input_id=%s",
+                task_type,
+                target,
+                input_id,
+            )
+            emit("任务失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="run")
+            finish(
+                status="failed",
+                message="任务执行异常，详见统一日志",
+                skill=SKILL_SLUG,
+                stage="run",
+            )
+            return 1
+
+
+# ---------------------------------------------------------------------------
+# 基础命令
+# ---------------------------------------------------------------------------
+
+def cmd_config_path() -> int:
+    example_path = os.path.join(get_skill_root(), ".env.example")
+    env_path = config.get_env_file_path() or ""
+    _print_json(
+        {
+            "skill": SKILL_SLUG,
+            "env_path": env_path,
+            "example_path": os.path.abspath(example_path),
+        }
+    )
+    return 0
+
+
+def cmd_health() -> int:
+    runtime = collect_runtime_diagnostics(
+        skill_slug=SKILL_SLUG,
+        platform_kit_min_version=PLATFORM_KIT_MIN_VERSION,
+        skill_root=get_skill_root(),
+    )
+    example_path = os.path.join(get_skill_root(), ".env.example")
+    env_path = config.get_env_file_path() or ""
+    env_exists = bool(env_path and os.path.isfile(env_path))
+
+    health_status = "failed" if runtime.has_fatal_issues else "ok"
+    lines = [
+        f"{SKILL_SLUG} health: {health_status}",
+        *format_runtime_health_lines(runtime),
+        f"env_path: {env_path}",
+        f"env_exists: {env_exists}",
+        f"example_path: {os.path.abspath(example_path)}",
+    ]
+    for line in lines:
+        print(line)
+    return 1 if runtime.has_fatal_issues else 0
+
+
+def cmd_version() -> int:
+    _print_json({"version": SKILL_VERSION, "skill": SKILL_SLUG})
+    return 0
+
+
+def cmd_init_db() -> int:
+    """幂等创建/迁移本地 SQLite 与数据管理元数据（宿主 install/upgrade 可调用）。"""
+    log = _get_task_logger()
+    try:
+        init_db()
+    except Exception as exc:  # noqa: BLE001 — CLI 边界，统一成可机读失败
+        print(
+            json.dumps(
+                {"ok": False, "skill": SKILL_SLUG, "error": str(exc)},
+                ensure_ascii=False,
+            )
+        )
+        return 1
+    _print_json({"ok": True, "skill": SKILL_SLUG, "db_path": get_db_path()})
+    return 0
+
+
+# ---------------------------------------------------------------------------
+# 门槛库维护
+# ---------------------------------------------------------------------------
+
+def _parse_json_arg(raw: str, label: str) -> list[dict[str, Any]]:
+    import sys
+
+    text = (raw or "").strip()
+    if not text:
+        raise ValueError(f"缺少导入 JSON（{label}）")
+    if text == "-":
+        text = sys.stdin.read().strip()
+    elif text.startswith(("[", "{")):
+        pass
+    elif os.path.isfile(text):
+        with open(text, encoding="utf-8") as f:
+            text = f.read().strip()
+    else:
+        raise ValueError(f"{label} 不是有效文件路径或 JSON：{text[:80]}")
+    try:
+        payload = json.loads(text)
+    except json.JSONDecodeError as exc:
+        raise ValueError(f"导入 JSON 解析失败：{exc}") from exc
+    if isinstance(payload, dict):
+        for key in ("thresholds", "cases", "items"):
+            if isinstance(payload.get(key), list):
+                payload = payload[key]
+                break
+        else:
+            payload = [payload]
+    if not isinstance(payload, list):
+        raise ValueError(f"{label} 应为对象数组")
+    return payload
+
+
+def cmd_set_threshold(
+    *,
+    rank: str,
+    metric_key: str,
+    metric_name: str = "",
+    min_value: float,
+    unit: str = "",
+    basis_doc: str = "",
+    clause: str = "",
+) -> int:
+    log = _get_task_logger()
+    from service.threshold_service import upsert_threshold
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=set_threshold rank=%s metric=%s stage=threshold", rank, metric_key)
+        emit("开始更新评审门槛", skill=SKILL_SLUG, stage="threshold")
+        try:
+            ok, reason = check_entitlement(SKILL_SLUG)
+            if not ok:
+                log.warning("entitlement_failed task_type=set_threshold stage=auth")
+                finish(status="failed", message=reason, skill=SKILL_SLUG, error_code="ENTITLEMENT_DENIED", stage="auth")
+                print(f"ERROR:ENTITLEMENT_DENIED {reason}")
+                return 1
+            payload = upsert_threshold({
+                "rank": rank,
+                "metric_key": metric_key,
+                "metric_name": metric_name,
+                "min_value": float(min_value),
+                "unit": unit,
+                "basis_doc": basis_doc,
+                "clause": clause,
+            })
+            _print_json(payload)
+            finish(status="success", message=f"门槛已更新：{rank}/{metric_key}", skill=SKILL_SLUG, stage="threshold")
+            return 0
+        except ValueError as exc:
+            log.warning("task_failed task_type=set_threshold error_code=THRESHOLD_INVALID detail=%s", exc)
+            finish(status="failed", message=str(exc), skill=SKILL_SLUG, error_code="THRESHOLD_INVALID", stage="threshold")
+            print(f"ERROR:THRESHOLD_INVALID {exc}")
+            return 1
+        except JobStopped:
+            raise
+        except Exception:
+            log.exception("task_failed task_type=set_threshold stage=threshold")
+            finish(status="failed", message="门槛更新异常，详见统一日志", skill=SKILL_SLUG, stage="threshold")
+            return 1
+
+
+def cmd_import_thresholds(thresholds_json: str) -> int:
+    log = _get_task_logger()
+    from service.threshold_service import import_thresholds
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=import_thresholds stage=threshold")
+        emit("开始批量导入门槛", skill=SKILL_SLUG, stage="threshold")
+        try:
+            ok, reason = check_entitlement(SKILL_SLUG)
+            if not ok:
+                log.warning("entitlement_failed task_type=import_thresholds stage=auth")
+                finish(status="failed", message=reason, skill=SKILL_SLUG, error_code="ENTITLEMENT_DENIED", stage="auth")
+                print(f"ERROR:ENTITLEMENT_DENIED {reason}")
+                return 1
+            rows = _parse_json_arg(thresholds_json, "--thresholds-json")
+            payload = import_thresholds(rows)
+            _print_json(payload)
+            finish(status="success", message=f"导入完成：新增 {payload['created']}，更新 {payload['updated']}", skill=SKILL_SLUG, stage="threshold")
+            return 0
+        except ValueError as exc:
+            log.warning("task_failed task_type=import_thresholds error_code=THRESHOLD_INVALID detail=%s", exc)
+            finish(status="failed", message=str(exc), skill=SKILL_SLUG, error_code="THRESHOLD_INVALID", stage="threshold")
+            print(f"ERROR:THRESHOLD_INVALID {exc}")
+            return 1
+        except JobStopped:
+            raise
+        except Exception:
+            log.exception("task_failed task_type=import_thresholds stage=threshold")
+            finish(status="failed", message="门槛导入异常，详见统一日志", skill=SKILL_SLUG, stage="threshold")
+            return 1
+
+
+def cmd_view_thresholds(rank: str = "") -> int:
+    log = _get_task_logger()
+    from service.threshold_service import list_thresholds
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=view_thresholds rank=%s", rank)
+        try:
+            thresholds = list_thresholds(rank=rank)
+            _print_json({"ok": True, "count": len(thresholds), "thresholds": thresholds})
+            return 0
+        except Exception:
+            log.exception("task_failed task_type=view_thresholds")
+            print("ERROR:QUERY_FAILED 门槛查询异常，详见统一日志")
+            return 1
+
+
+# ---------------------------------------------------------------------------
+# 核验主命令
+# ---------------------------------------------------------------------------
+
+def cmd_verify_metric(
+    *,
+    teacher_name: str,
+    apply_rank: str,
+    metrics_json: str,
+    department: str = "",
+) -> int:
+    """职称硬指标核验（主命令，同步）。"""
+    log = _get_task_logger()
+    from service.verify_service import run_verify
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=verify_metric teacher=%s rank=%s stage=verify", teacher_name, apply_rank)
+        emit("开始职称硬指标核验", skill=SKILL_SLUG, stage="verify")
+        try:
+            ok, reason = check_entitlement(SKILL_SLUG)
+            if not ok:
+                log.warning("entitlement_failed task_type=verify_metric stage=auth")
+                finish(status="failed", message=reason, skill=SKILL_SLUG, error_code="ENTITLEMENT_DENIED", stage="auth")
+                print(f"ERROR:ENTITLEMENT_DENIED {reason}")
+                return 1
+
+            metrics = _parse_json_arg(metrics_json, "--metrics-json")
+            if not metrics:
+                raise ValueError("申报指标 JSON 为空")
+            case = metrics[0] if len(metrics) == 1 else dict(metrics[0])
+            if teacher_name:
+                case.setdefault("teacherName", teacher_name)
+            if department:
+                case.setdefault("department", department)
+            case.setdefault("applyRank", apply_rank)
+
+            result = run_verify([case])
+            brief = result["reports"][0]
+            _print_json({
+                "ok": True,
+                "batch_no": result["batch_no"],
+                "case_id": brief["case_id"],
+                "teacher_name": brief["teacher_name"],
+                "verdict": brief["verdict"],
+                "summary": brief["summary"],
+                "matrix": brief["matrix"],
+                "report_file": brief["report_file"],
+                "disclaimer": result["disclaimer"],
+            })
+            finish(
+                status="success",
+                message=f"核验完成：{brief['teacher_name']} 结论 {brief['verdict']}",
+                skill=SKILL_SLUG,
+                stage="verify",
+                batch_no=result["batch_no"],
+            )
+            return 0
+        except JobStopped:
+            raise
+        except ValueError as exc:
+            log.warning("task_failed task_type=verify_metric error_code=METRIC_INVALID detail=%s", exc)
+            finish(status="failed", message=str(exc), skill=SKILL_SLUG, error_code="METRIC_INVALID", stage="verify")
+            print(f"ERROR:METRIC_INVALID {exc}")
+            return 1
+        except Exception:
+            log.exception("task_failed task_type=verify_metric stage=verify")
+            emit("核验失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="verify")
+            finish(status="failed", message="核验执行异常，详见统一日志", skill=SKILL_SLUG, stage="verify")
+            return 1
+
+
+def cmd_verify_batch_input() -> int:
+    """批量核验（async Action 入口）：核验 input/ 目录中的申报案例 JSON。"""
+    log = _get_task_logger()
+    from service.verify_service import run_verify
+    from util.runtime_paths import get_input_dir
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=verify_batch stage=batch")
+        emit("开始批量核验", skill=SKILL_SLUG, stage="batch")
+        try:
+            ok, reason = check_entitlement(SKILL_SLUG)
+            if not ok:
+                log.warning("entitlement_failed task_type=verify_batch stage=auth")
+                finish(status="failed", message=reason, skill=SKILL_SLUG, error_code="ENTITLEMENT_DENIED", stage="auth")
+                print(f"ERROR:ENTITLEMENT_DENIED {reason}")
+                return 1
+
+            input_dir = get_input_dir()
+            files = sorted(
+                os.path.join(input_dir, name)
+                for name in os.listdir(input_dir)
+                if name.lower().endswith(".json")
+            )
+            if not files:
+                message = f"input 目录没有申报案例 JSON：{input_dir}"
+                log.warning("task_failed task_type=verify_batch error_code=METRIC_INVALID stage=batch")
+                finish(status="failed", message=message, skill=SKILL_SLUG, error_code="METRIC_INVALID", stage="batch")
+                print(f"ERROR:METRIC_INVALID {message}")
+                return 1
+
+            cases = []
+            for path in files:
+                with open(path, encoding="utf-8") as f:
+                    payload = json.load(f)
+                if isinstance(payload, dict):
+                    payload = payload.get("cases") or [payload]
+                cases.extend(payload)
+
+            result = run_verify(cases)
+            for index, brief in enumerate(result["reports"], start=1):
+                emit(
+                    f"[{index}/{len(result['reports'])}] {brief['teacher_name']}：{brief['verdict']}",
+                    skill=SKILL_SLUG,
+                    stage="batch",
+                )
+            _print_json({
+                "ok": True,
+                "batch_no": result["batch_no"],
+                "count": result["count"],
+                "verdict_counts": result["verdict_counts"],
+                "brief_file": result["brief_file"],
+            })
+            finish(status="success", message=f"批量核验完成 {result['count']} 个案例", skill=SKILL_SLUG, stage="batch", batch_no=result["batch_no"])
+            return 0
+        except JobStopped:
+            raise
+        except Exception:
+            log.exception("task_failed task_type=verify_batch stage=batch")
+            emit("批量核验失败，已记录诊断信息", type="warn", skill=SKILL_SLUG, stage="batch")
+            finish(status="failed", message="批量核验执行异常，详见统一日志", skill=SKILL_SLUG, stage="batch")
+            return 1
+
+
+# ---------------------------------------------------------------------------
+# 案例与统计
+# ---------------------------------------------------------------------------
+
+def cmd_list_cases(limit: int = 20, verdict: str = "") -> int:
+    log = _get_task_logger()
+    from service.threshold_service import list_cases
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=list_cases limit=%s verdict=%s", limit, verdict)
+        try:
+            cases = list_cases(limit=limit, verdict=verdict)
+            _print_json({"ok": True, "count": len(cases), "cases": cases})
+            return 0
+        except Exception:
+            log.exception("task_failed task_type=list_cases")
+            print("ERROR:QUERY_FAILED 核验案例查询异常，详见统一日志")
+            return 1
+
+
+def cmd_show_case(case_id: int) -> int:
+    log = _get_task_logger()
+    from service.threshold_service import get_case
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=show_case case_id=%s", case_id)
+        try:
+            case = get_case(int(case_id))
+            _print_json({"ok": True, "case": case})
+            return 0
+        except ValueError as exc:
+            log.warning("task_failed task_type=show_case error_code=CASE_NOT_FOUND detail=%s", exc)
+            print(f"ERROR:CASE_NOT_FOUND {exc}")
+            return 1
+        except Exception:
+            log.exception("task_failed task_type=show_case")
+            print("ERROR:QUERY_FAILED 核验案例查询异常，详见统一日志")
+            return 1
+
+
+def cmd_render_html(report_id: Optional[int] = None) -> int:
+    log = _get_task_logger()
+    from service.report_service import build_report_for_case_id
+    from service.threshold_service import list_cases
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=render_html report_id=%s stage=render", report_id)
+        try:
+            if report_id:
+                targets = [int(report_id)]
+            else:
+                targets = [c["id"] for c in list_cases(limit=10)]
+            if not targets:
+                log.warning("task_failed task_type=render_html error_code=CASE_NOT_FOUND stage=render")
+                finish(status="failed", message="暂无可重建的核验案例", skill=SKILL_SLUG, error_code="CASE_NOT_FOUND", stage="render")
+                print("ERROR:CASE_NOT_FOUND 暂无可重建的核验案例")
+                return 1
+            written = [{"case_id": cid, "path": build_report_for_case_id(cid)} for cid in targets]
+            _print_json({"ok": True, "count": len(written), "reports": written})
+            finish(status="success", message=f"已重建 {len(written)} 份核验报告", skill=SKILL_SLUG, stage="render")
+            return 0
+        except JobStopped:
+            raise
+        except ValueError as exc:
+            log.warning("task_failed task_type=render_html error_code=CASE_NOT_FOUND detail=%s", exc)
+            finish(status="failed", message=str(exc), skill=SKILL_SLUG, error_code="CASE_NOT_FOUND", stage="render")
+            print(f"ERROR:CASE_NOT_FOUND {exc}")
+            return 1
+        except Exception:
+            log.exception("task_failed task_type=render_html stage=render")
+            finish(status="failed", message="报告重建异常，详见统一日志", skill=SKILL_SLUG, stage="render")
+            return 1
+
+
+def cmd_stats() -> int:
+    log = _get_task_logger()
+    from service.threshold_service import stats
+
+    with job_context(skill=SKILL_SLUG):
+        log.info("task_start task_type=stats")
+        try:
+            _print_json(stats())
+            return 0
+        except Exception:
+            log.exception("task_failed task_type=stats")
+            print("ERROR:QUERY_FAILED 统计查询异常，详见统一日志")
+            return 1

scripts/service/threshold_service.py

@@ -0,0 +1,188 @@
+"""评审门槛库 / 核验案例 的维护与查询。"""
+
+from __future__ import annotations
+
+import json
+from typing import Any, Optional
+
+from db.connection import get_conn, init_db
+from service.verify_service import _METRIC_ALIASES, _snake, norm_metrics
+from util.logging_config import get_skill_logger
+from util.timeutil import now_unix
+
+_KNOWN_RANKS = ("副教授", "教授")
+
+
+def _log():
+    try:
+        return get_skill_logger()
+    except RuntimeError:
+        return None
+
+
+def upsert_threshold(fields: dict[str, Any]) -> dict[str, Any]:
+    """按（级别+指标键）幂等写入门槛。"""
+    normalized = norm_metrics({str(k): v for k, v in fields.items()})
+    rank = str(normalized.get("rank") or fields.get("rank") or "").strip()
+    metric_key = str(normalized.get("metricKey") or fields.get("metric_key") or fields.get("metricKey") or "").strip()
+    metric_name = str(normalized.get("metricName") or fields.get("metric_name") or "").strip()
+    rank = rank if rank in _KNOWN_RANKS else (str(fields.get("rank") or "").strip() or rank)
+    if rank not in _KNOWN_RANKS:
+        raise ValueError(f"申报级别只接受 {'、'.join(_KNOWN_RANKS)}：{rank}")
+    if not metric_key:
+        raise ValueError("指标键（metric_key）不能为空")
+    min_value = fields.get("min_value", fields.get("minValue"))
+    try:
+        min_value = float(min_value)
+    except (TypeError, ValueError) as exc:
+        raise ValueError(f"最低要求必须为数字：{min_value}") from exc
+    if min_value < 0:
+        raise ValueError(f"最低要求必须非负：{min_value}")
+
+    metric_name = metric_name or str(_METRIC_ALIASES.get(metric_key, (metric_key,))[0])
+    values = {
+        "rank": rank,
+        "metric_key": metric_key,
+        "metric_name": metric_name,
+        "min_value": min_value,
+        "unit": str(fields.get("unit") or "").strip(),
+        "basis_doc": str(fields.get("basis_doc") or fields.get("basisDoc") or "").strip(),
+        "clause": str(fields.get("clause") or "").strip(),
+    }
+
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT id FROM threshold_rules WHERE rank = ? AND metric_key = ?", (rank, metric_key))
+        row = cur.fetchone()
+        if row:
+            threshold_id = int(row[0])
+            cur.execute(
+                """
+                UPDATE threshold_rules
+                SET metric_name = ?, min_value = ?, unit = ?, basis_doc = ?, clause = ?, updated_at = ?
+                WHERE id = ?
+                """,
+                (values["metric_name"], values["min_value"], values["unit"], values["basis_doc"], values["clause"], now_unix(), threshold_id),
+            )
+            action = "updated"
+        else:
+            cur.execute(
+                """
+                INSERT INTO threshold_rules
+                    (rank, metric_key, metric_name, min_value, unit, basis_doc, clause, created_at, updated_at)
+                VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)
+                """,
+                (values["rank"], values["metric_key"], values["metric_name"], values["min_value"], values["unit"], values["basis_doc"], values["clause"], now_unix(), now_unix()),
+            )
+            threshold_id = int(cur.lastrowid)
+            action = "created"
+        conn.commit()
+    finally:
+        conn.close()
+    log = _log()
+    if log:
+        log.info("threshold_upserted rank=%s metric=%s action=%s stage=threshold", rank, metric_key, action)
+    return {"ok": True, "rank": rank, "metric_key": metric_key, "id": threshold_id, "action": action}
+
+
+def import_thresholds(rows: list[dict[str, Any]]) -> dict[str, Any]:
+    if not isinstance(rows, list) or not rows:
+        raise ValueError("导入内容应为非空 JSON 数组（门槛）")
+    if len(rows) > 200:
+        raise ValueError(f"单次导入条数过多（{len(rows)}），上限 200 条")
+    results = [upsert_threshold(row) for row in rows]
+    return {
+        "ok": True,
+        "count": len(results),
+        "created": sum(1 for r in results if r["action"] == "created"),
+        "updated": sum(1 for r in results if r["action"] == "updated"),
+        "items": results,
+    }
+
+
+def list_thresholds(rank: str = "") -> list[dict[str, Any]]:
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        if rank:
+            cur.execute("SELECT * FROM threshold_rules WHERE rank = ? ORDER BY id", (rank,))
+        else:
+            cur.execute("SELECT * FROM threshold_rules ORDER BY rank, id")
+        columns = [d[0] for d in cur.description]
+        rows = [dict(zip(columns, row)) for row in cur.fetchall()]
+    finally:
+        conn.close()
+    return rows
+
+
+def list_cases(limit: int = 20, verdict: str = "") -> list[dict[str, Any]]:
+    init_db()
+    limit = max(1, min(int(limit), 200))
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        if verdict:
+            cur.execute(
+                "SELECT id, batch_no, teacher_name, department, apply_rank, verdict, report_file, created_at FROM verify_cases WHERE verdict = ? ORDER BY id DESC LIMIT ?",
+                (verdict, limit),
+            )
+        else:
+            cur.execute(
+                "SELECT id, batch_no, teacher_name, department, apply_rank, verdict, report_file, created_at FROM verify_cases ORDER BY id DESC LIMIT ?",
+                (limit,),
+            )
+        columns = [d[0] for d in cur.description]
+        rows = [dict(zip(columns, row)) for row in cur.fetchall()]
+    finally:
+        conn.close()
+    return rows
+
+
+def get_case(case_id: int) -> dict[str, Any]:
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT * FROM verify_cases WHERE id = ?", (case_id,))
+        row = cur.fetchone()
+        if not row:
+            raise ValueError(f"核验案例不存在：{case_id}")
+        columns = [d[0] for d in cur.description]
+        case = dict(zip(columns, row))
+    finally:
+        conn.close()
+    for key in ("matrix_json", "metrics_json"):
+        try:
+            case[key[:-5]] = json.loads(case.get(key) or "[]")
+        except (TypeError, ValueError):
+            case[key[:-5]] = [] if key == "matrix_json" else {}
+    return case
+
+
+def stats() -> dict[str, Any]:
+    init_db()
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute("SELECT verdict, COUNT(*) FROM verify_cases GROUP BY verdict")
+        by_verdict = {str(r[0]): int(r[1]) for r in cur.fetchall()}
+        cur.execute("SELECT COUNT(*) FROM threshold_rules")
+        threshold_total = int(cur.fetchone()[0])
+        cur.execute("SELECT COUNT(*) FROM verify_cases")
+        case_total = int(cur.fetchone()[0])
+        cur.execute(
+            "SELECT teacher_name, department, apply_rank FROM verify_cases WHERE verdict = '硬伤' ORDER BY id DESC LIMIT 5"
+        )
+        red_cases = [{"teacher_name": r[0], "department": r[1], "apply_rank": r[2]} for r in cur.fetchall()]
+    finally:
+        conn.close()
+    return {
+        "ok": True,
+        "case_total": case_total,
+        "by_verdict": by_verdict,
+        "threshold_total": threshold_total,
+        "red_cases": red_cases,
+    }

scripts/service/verify_service.py

@@ -0,0 +1,288 @@
+"""职称硬指标核验内核（唯一业务内核）。
+
+核验规则（对齐《核心建设内容：各部门"Skill"深度需求规格书》人事处 Skill 001）：
+  R1 逐项比对：实有 < 最低要求 → 红（硬伤，附依据条款）
+  R2 边缘识别：达标但余量 < 10% → 黄（建议补充说明或再积累）
+  R3 安全判定：全部已核项 ≥ 要求 × 1.10 → 绿
+  R4 数据待核：指标缺失 → 灰（不判不达标，提示补材料）
+  R5 综合结论：任一红 → 硬伤（暂缓提交）；有黄/灰 → 边缘（建议补充）；全绿 → 安全
+  R6 基础条件：学历/任现职年限与门槛同口径比对
+  V1 门槛库完备性：指标无对应门槛 → 请人事处补充配置
+
+核验仅针对硬指标形式审查，不构成评审结论。
+"""
+
+from __future__ import annotations
+
+import json
+import uuid
+from typing import Any, Optional
+
+from db.connection import get_conn, init_db
+from util.logging_config import get_skill_logger
+from util.timeutil import now_unix
+
+VERDICT_SAFE = "安全"
+VERDICT_BORDER = "边缘"
+VERDICT_RED = "硬伤"
+
+_COLOR_PASS = "green"
+_COLOR_BORDER = "yellow"
+_COLOR_RED = "red"
+_COLOR_PENDING = "gray"
+
+_SAFE_MARGIN = 0.10
+
+_METRIC_ALIASES = {
+    "teacherName": ("teacher_name", "姓名", "教师姓名"),
+    "department": ("department", "学院", "所属学院"),
+    "applyRank": ("apply_rank", "申报级别", "申报职称"),
+    "presidingProvincial": ("presiding_provincial", "主持省部级项目", "省部级项目"),
+    "corePapers": ("core_papers", "核心期刊论文", "核心论文"),
+    "teachingHours": ("teaching_hours", "年均教学工作量", "教学工作量"),
+    "currentTitleYears": ("current_title_years", "任现职年限", "现职称年限"),
+    "monographs": ("monographs", "专著/教材", "专著"),
+    "awards": ("awards", "教学/科研获奖", "获奖"),
+    "degree": ("degree", "学历学位", "学历"),
+}
+
+
+def _log():
+    try:
+        return get_skill_logger()
+    except RuntimeError:
+        return None
+
+
+def _camel(name: str) -> str:
+    parts = str(name).split("_")
+    if len(parts) <= 1:
+        return str(name)
+    head, *rest = parts
+    return head + "".join(p[:1].upper() + p[1:] for p in rest if p)
+
+
+def _snake(name: str) -> str:
+    return "".join("_" + ch.lower() if ch.isupper() else ch for ch in str(name)).lstrip("_")
+
+
+def norm_metrics(raw: dict[str, Any]) -> dict[str, Any]:
+    """归一化申报指标键（兼容 camelCase / snake_case / 中文键，canonical 同时提供两种形式）。"""
+    out: dict[str, Any] = {}
+    for key, value in raw.items():
+        out.setdefault(str(key), value)
+        for canonical, names in _METRIC_ALIASES.items():
+            if str(key) in names or _camel(str(key)) == canonical:
+                out.setdefault(canonical, value)
+                out.setdefault(_snake(canonical), value)
+    return out
+
+
+def _to_float(value: Any) -> Optional[float]:
+    if value is None or value == "":
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
+
+
+def load_thresholds(rank: str) -> dict[str, dict[str, Any]]:
+    """读取某级别的门槛（metric_key → 门槛行）。"""
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        cur.execute(
+            "SELECT metric_key, metric_name, min_value, unit, basis_doc, clause FROM threshold_rules WHERE rank = ?",
+            (rank,),
+        )
+        thresholds = {}
+        for metric_key, metric_name, min_value, unit, basis_doc, clause in cur.fetchall():
+            thresholds[str(metric_key)] = {
+                "metric_name": metric_name,
+                "min_value": float(min_value),
+                "unit": unit or "",
+                "basis_doc": basis_doc or "",
+                "clause": clause or "",
+            }
+    finally:
+        conn.close()
+    return thresholds
+
+
+def verify_metrics(metrics_raw: dict[str, Any], thresholds: dict[str, dict[str, Any]]) -> dict[str, Any]:
+    """对单个申报案例执行矩阵计算（纯计算，不落库）。"""
+    metrics = norm_metrics(metrics_raw)
+    teacher_name = str(metrics.get("teacherName") or "").strip() or "未命名申报人"
+    apply_rank = str(metrics.get("applyRank") or "").strip()
+
+    matrix: list[dict[str, Any]] = []
+    has_red = False
+    has_yellow_or_gray = False
+
+    for metric_key, threshold in thresholds.items():
+        actual_raw = metrics.get(metric_key)
+        required = float(threshold["min_value"])
+        unit = threshold["unit"]
+        basis = threshold["basis_doc"] + (threshold["clause"] or "")
+        display_actual = metrics.get(metric_key)
+
+        actual = _to_float(actual_raw)
+        if actual is None:
+            has_yellow_or_gray = True
+            matrix.append({
+                "metric": threshold["metric_name"],
+                "metric_key": metric_key,
+                "required": required,
+                "actual": display_actual,
+                "margin": None,
+                "color": _COLOR_PENDING,
+                "basis": basis,
+                "advice": f"数据待核：请补充{threshold['metric_name']}证明材料后重新核验（R4）。",
+            })
+            continue
+
+        margin = round((actual - required) / required, 4) if required > 0 else None
+        if actual < required:
+            has_red = True
+            advice = f"硬伤：{threshold['metric_name']}实有 {actual:g} {unit}，未达最低要求 {required:g} {unit}，须补足后方可提交"
+            color = _COLOR_RED
+        elif margin is not None and margin < _SAFE_MARGIN:
+            has_yellow_or_gray = True
+            advice = f"边缘项：仅超线 {margin * 100:.1f}%，建议补充说明或继续积累（R2）"
+            color = _COLOR_BORDER
+        else:
+            advice = ""
+            color = _COLOR_PASS
+        matrix.append({
+            "metric": threshold["metric_name"],
+            "metric_key": metric_key,
+            "required": required,
+            "actual": actual,
+            "margin": margin,
+            "color": color,
+            "basis": basis,
+            "advice": f"{advice}（依据：{basis}）" if advice else "",
+        })
+
+    # V1：申报指标中存在但门槛库未配置的键（排除标识字段与其别名形式）
+    identity_keys = {"teacher_name", "teacherName", "department", "apply_rank", "applyRank"}
+    known_keys = set(thresholds.keys()) | set(_METRIC_ALIASES.keys())
+    unconfigured = []
+    for key in metrics:
+        if key in identity_keys or key in known_keys:
+            continue
+        if _snake(key) in known_keys or _camel(key) in known_keys:
+            continue
+        if metrics.get(key) in (None, ""):
+            continue
+        if any(key in names for names in _METRIC_ALIASES.values()):
+            continue
+        unconfigured.append(str(key))
+    for key in unconfigured:
+        has_yellow_or_gray = True
+        matrix.append({
+            "metric": str(_METRIC_ALIASES.get(key, (key,))[0]) if key in _METRIC_ALIASES else key,
+            "metric_key": key,
+            "required": None,
+            "actual": metrics.get(key),
+            "margin": None,
+            "color": _COLOR_PENDING,
+            "basis": "",
+            "advice": "门槛库未配置该指标的最低要求，请人事处补充（V1）。",
+        })
+
+    if has_red:
+        verdict = VERDICT_RED
+        summary = "存在不达门槛的硬伤项，暂缓提交；请按矩阵补足材料后重新核验。"
+    elif has_yellow_or_gray:
+        verdict = VERDICT_BORDER
+        summary = "无硬伤，但存在边缘项或待核数据，建议补充说明后提交。"
+    else:
+        verdict = VERDICT_SAFE
+        summary = "全部硬指标超标准 10% 以上，形式审查通过。"
+
+    return {
+        "teacher_name": teacher_name,
+        "department": str(metrics.get("department") or ""),
+        "apply_rank": apply_rank,
+        "verdict": verdict,
+        "summary": summary,
+        "matrix": matrix,
+    }
+
+
+def run_verify(cases: list[dict[str, Any]], *, batch_no: str = "") -> dict[str, Any]:
+    """核验内核：逐案例矩阵计算 → 写库 → 生成《核验报告》。"""
+    from service.report_service import write_report_for_kernel, build_batch_brief
+
+    log = _log()
+    init_db()
+    if not cases:
+        raise ValueError("没有可核验的申报案例")
+    if len(cases) > 200:
+        raise ValueError(f"单批核验案例数过多（{len(cases)}），上限 200 个")
+
+    batch_no = batch_no or now_unix().to_bytes(5, "big").hex() + uuid.uuid4().hex[:6]
+    if log:
+        log.info("verify_kernel_start batch_no=%s count=%s stage=kernel", batch_no, len(cases))
+
+    results = []
+    conn = get_conn()
+    try:
+        cur = conn.cursor()
+        for case in cases:
+            if not isinstance(case, dict):
+                raise ValueError("申报案例必须是 JSON 对象")
+            result = verify_metrics(case, load_thresholds(str(case.get("applyRank") or case.get("apply_rank") or "").strip()))
+            cur.execute(
+                """
+                INSERT INTO verify_cases
+                    (batch_no, teacher_name, department, apply_rank, metrics_json, verdict, matrix_json)
+                VALUES (?, ?, ?, ?, ?, ?, ?)
+                """,
+                (
+                    batch_no,
+                    result["teacher_name"],
+                    result["department"],
+                    result["apply_rank"],
+                    json.dumps(case, ensure_ascii=False),
+                    result["verdict"],
+                    json.dumps(result["matrix"], ensure_ascii=False),
+                ),
+            )
+            case_id = int(cur.lastrowid)
+            result["case_id"] = case_id
+            result["batch_no"] = batch_no
+            report_file = write_report_for_kernel(result=result, batch_no=batch_no)
+            cur.execute("UPDATE verify_cases SET report_file = ? WHERE id = ?", (report_file, case_id))
+            result["report_file"] = report_file
+            results.append(result)
+        conn.commit()
+    finally:
+        conn.close()
+
+    verdict_counts: dict[str, int] = {}
+    for result in results:
+        verdict_counts[result["verdict"]] = verdict_counts.get(result["verdict"], 0) + 1
+    brief_file = build_batch_brief(batch_no=batch_no, results=results)
+
+    payload = {
+        "ok": True,
+        "batch_no": batch_no,
+        "count": len(results),
+        "verdict_counts": verdict_counts,
+        "brief_file": brief_file,
+        "disclaimer": "核验仅针对硬指标形式审查，不构成评审结论；门槛以人事处最新文件为准。",
+        "reports": results,
+    }
+    if log:
+        log.info(
+            "verify_kernel_done batch_no=%s count=%s red=%s yellow=%s green=%s stage=kernel",
+            batch_no,
+            len(results),
+            verdict_counts.get(VERDICT_RED, 0),
+            verdict_counts.get(VERDICT_BORDER, 0),
+            verdict_counts.get(VERDICT_SAFE, 0),
+        )
+    return payload

scripts/util/argparse_zh.py

@@ -0,0 +1,12 @@
+"""argparse 中文错误说明。"""
+
+from __future__ import annotations
+
+import argparse
+import sys
+
+
+class ZhArgumentParser(argparse.ArgumentParser):
+    def error(self, message: str) -> None:
+        print(f"参数错误：{message}\n请执行：python main.py -h 查看帮助", file=sys.stderr)
+        self.exit(2)

scripts/util/config_bootstrap.py

@@ -0,0 +1,19 @@
+"""技能配置初始化：.env.example 落盘与用户 .env 缺失项合并。"""
+
+from __future__ import annotations
+
+import os
+
+from jiangchang_skill_core import config
+
+from util.constants import SKILL_SLUG, SKILL_VERSION
+from util.runtime_paths import get_skill_root
+
+
+def bootstrap_skill_config() -> str:
+    """确保用户数据目录 .env 存在，并追加 .env.example 中的新配置项。"""
+    example_path = os.path.join(get_skill_root(), ".env.example")
+    env_path = config.ensure_env_file(SKILL_SLUG, example_path)
+    comment = f"{SKILL_SLUG} v{SKILL_VERSION}" if SKILL_VERSION else None
+    config.merge_missing_env_keys(example_path, env_path, comment_skill=comment)
+    return env_path

scripts/util/constants.py

@@ -0,0 +1,6 @@
+"""技能标识、版本与平台公共库约束（职称评审核验助手）。"""
+
+SKILL_SLUG = "verify-promotion-metric-default"
+SKILL_VERSION = "1.0.1"
+LOG_LOGGER_NAME = "openclaw.skill.verify_promotion_metric_default"
+PLATFORM_KIT_MIN_VERSION = "1.2.11"

scripts/util/logging_config.py

@@ -0,0 +1,21 @@
+"""Re-export unified logging (implementation: jiangchang_skill_core.unified_logging)."""
+
+from jiangchang_skill_core.unified_logging import (
+    attach_unified_file_handler,
+    ensure_trace_for_process,
+    get_skill_log_file_path,
+    get_skill_logger,
+    get_unified_logs_dir,
+    setup_skill_logging,
+    subprocess_env_with_trace,
+)
+
+__all__ = [
+    "attach_unified_file_handler",
+    "ensure_trace_for_process",
+    "get_skill_log_file_path",
+    "get_skill_logger",
+    "get_unified_logs_dir",
+    "setup_skill_logging",
+    "subprocess_env_with_trace",
+]

scripts/util/runtime_paths.py

@@ -0,0 +1,155 @@
+"""数据根、技能目录、兄弟技能根路径与用户数据子目录解析。"""
+
+from __future__ import annotations
+
+import os
+
+from jiangchang_skill_core.runtime_env import get_data_root, get_sibling_skills_root, get_user_id
+
+from util.constants import SKILL_SLUG
+
+_SCRIPTS_DIR = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
+
+# 用户可见的上行/下行目录（数据管理「输入/输出」文件浏览器）
+STANDARD_SUBDIR_INPUT = "input"
+STANDARD_SUBDIR_OUTPUT = "output"
+
+# 宿主扫描的 HTML 报告页（数据管理侧栏 htmlPages，非 input/output 文件浏览器）
+STANDARD_SUBDIR_HTML = "html"
+
+# 内部/可选目录（不上侧栏）
+STANDARD_SUBDIR_UPLOADS = "uploads"
+STANDARD_SUBDIR_CACHE = "cache"
+STANDARD_SUBDIR_TEMP = "temp"
+STANDARD_SUBDIR_RPA_ARTIFACTS = "rpa-artifacts"
+STANDARD_SUBDIR_VIDEOS = "videos"
+
+CONFIG_KEY_INPUT_DIR = "SKILL_INPUT_DIR"
+CONFIG_KEY_OUTPUT_DIR = "SKILL_OUTPUT_DIR"
+CONFIG_KEY_UPLOAD_DIR = "SKILL_UPLOAD_DIR"
+CONFIG_KEY_CACHE_DIR = "SKILL_CACHE_DIR"
+CONFIG_KEY_TEMP_DIR = "SKILL_TEMP_DIR"
+
+
+def get_skill_root() -> str:
+    return os.path.dirname(_SCRIPTS_DIR)
+
+
+def get_openclaw_root() -> str:
+    return get_sibling_skills_root(_SCRIPTS_DIR)
+
+
+def get_skills_root() -> str:
+    return get_sibling_skills_root(_SCRIPTS_DIR)
+
+
+def get_skill_data_dir() -> str:
+    path = os.path.join(get_data_root(), get_user_id(), SKILL_SLUG)
+    os.makedirs(path, exist_ok=True)
+    return path
+
+
+def get_db_path(filename: str | None = None) -> str:
+    name = filename or f"{SKILL_SLUG}.db"
+    return os.path.join(get_skill_data_dir(), name)
+
+
+def _normalize_subdir(value: str) -> str:
+    cleaned = value.strip().strip("/\\")
+    if not cleaned or cleaned.startswith(".."):
+        raise ValueError(f"invalid skill data subdir: {value!r}")
+    return cleaned.replace("/", os.sep).replace("\\", os.sep)
+
+
+def resolve_data_path(
+    config_key: str | None,
+    default_subdir: str,
+    *,
+    create: bool = True,
+) -> str:
+    """解析技能拥有路径：空配置 → 数据目录下 default_subdir；相对 → 相对数据目录；绝对 → 原样。"""
+    from jiangchang_skill_core import config
+
+    base = get_skill_data_dir()
+    raw = (config.get(config_key) or "").strip() if config_key else ""
+    if not raw:
+        path = os.path.join(base, _normalize_subdir(default_subdir))
+    elif os.path.isabs(raw):
+        path = os.path.abspath(os.path.expanduser(raw))
+    else:
+        path = os.path.join(base, _normalize_subdir(raw))
+
+    if create:
+        os.makedirs(path, exist_ok=True)
+    return path
+
+
+def resolve_input_path(raw_path: str, *, create_parent: bool = False) -> str:
+    """解析 CLI 输入路径：绝对路径原样；相对路径相对 {skill_data_dir}/input/。"""
+    text = (raw_path or "").strip()
+    if not text:
+        return text
+    expanded = os.path.expanduser(text)
+    if os.path.isabs(expanded):
+        return os.path.abspath(expanded)
+    input_dir = get_input_dir(create=create_parent)
+    return os.path.abspath(os.path.join(input_dir, os.path.normpath(expanded)))
+
+
+def get_input_dir(*, create: bool = True) -> str:
+    """上行：用户/Agent 放入的待处理文件（清单、CSV、产品表等）。"""
+    return resolve_data_path(CONFIG_KEY_INPUT_DIR, STANDARD_SUBDIR_INPUT, create=create)
+
+
+def get_output_dir(*, create: bool = True) -> str:
+    """下行：技能/宿主交给用户拿走的文件（视频、报表、导出 CSV 等）。"""
+    return resolve_data_path(CONFIG_KEY_OUTPUT_DIR, STANDARD_SUBDIR_OUTPUT, create=create)
+
+
+def get_uploads_dir(*, create: bool = True) -> str:
+    return resolve_data_path(CONFIG_KEY_UPLOAD_DIR, STANDARD_SUBDIR_UPLOADS, create=create)
+
+
+def get_cache_dir(*, create: bool = True) -> str:
+    return resolve_data_path(CONFIG_KEY_CACHE_DIR, STANDARD_SUBDIR_CACHE, create=create)
+
+
+def get_temp_dir(*, create: bool = True) -> str:
+    return resolve_data_path(CONFIG_KEY_TEMP_DIR, STANDARD_SUBDIR_TEMP, create=create)
+
+
+def get_rpa_artifacts_dir(batch_id: str, *, create: bool = True) -> str:
+    batch = (batch_id or "").strip() or "default"
+    path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_RPA_ARTIFACTS, batch)
+    if create:
+        os.makedirs(path, exist_ok=True)
+    return path
+
+
+def get_videos_dir(*, create: bool = True) -> str:
+    path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_VIDEOS)
+    if create:
+        os.makedirs(path, exist_ok=True)
+    return path
+
+
+def get_html_dir(*, create: bool = True) -> str:
+    """HTML 报告页目录：宿主扫描 `{skill_data}/html/*.html`（一层、非递归）。"""
+    path = os.path.join(get_skill_data_dir(), STANDARD_SUBDIR_HTML)
+    if create:
+        os.makedirs(path, exist_ok=True)
+    return path
+
+
+def list_resolved_data_paths() -> dict[str, str]:
+    """供 health / config-path 输出已解析的数据子目录（不创建目录）。"""
+    return {
+        "skill_data_dir": get_skill_data_dir(),
+        "input_dir": get_input_dir(create=False),
+        "output_dir": get_output_dir(create=False),
+        "html_dir": get_html_dir(create=False),
+        "uploads_dir": get_uploads_dir(create=False),
+        "cache_dir": get_cache_dir(create=False),
+        "temp_dir": get_temp_dir(create=False),
+        "videos_dir": get_videos_dir(create=False),
+    }

scripts/util/slug_naming.py

@@ -0,0 +1,227 @@
+"""技能 slug 语义校验（verb-noun-platform 规范，见 development/NAMING.md）。"""
+
+from __future__ import annotations
+
+import os
+import re
+
+_KEBAB_SLUG = re.compile(r"^[a-z0-9]+(?:-[a-z0-9]+)*$")
+_MAX_SLUG_LENGTH = 48
+_MIN_SEGMENTS = 3
+_MAX_SEGMENTS = 5
+_MAX_NOUN_SEGMENTS = 3
+
+_SKILL_ROOT = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+_NAMING_DIR = os.path.join(_SKILL_ROOT, "assets", "naming")
+
+_BUILTIN_VERBS = frozenset(
+    {
+        "scrape",
+        "receive",
+        "download",
+        "export",
+        "add",
+        "fill",
+        "enter",
+        "bind",
+        "withdraw",
+        "reconcile",
+        "verify",
+        "submit",
+        "query",
+        "process",
+        "reply",
+        "generate",
+        "draft",
+        "review",
+        "confirm",
+        "track",
+        "upload",
+        "match",
+        "triage",
+        "ship",
+        "reprice",
+    }
+)
+_BUILTIN_PLATFORMS = frozenset(
+    {
+        "amazon",
+        "shopee",
+        "xinghang",
+        "alibaba",
+        "kingdee",
+        "icbc",
+        "pingpong",
+        "worldfirst",
+        "paypal",
+        "lianlian",
+        "single-window",
+        "export-rebate",
+        "etax",
+    }
+)
+_BUILTIN_SCOPES = frozenset({"all", "batch", "multi", "default"})
+
+LEGACY_EXEMPT_SLUGS: frozenset[str] = frozenset(
+    {
+        "account-manager",
+        "skill-template",
+        "your-skill-slug",
+        "your_skill_slug",
+    }
+)
+
+
+def load_wordlist(filename: str) -> frozenset[str]:
+    """从 assets/naming/{filename} 加载词表；文件不存在时 fallback 内置集合。"""
+    fallbacks = {
+        "verbs.txt": _BUILTIN_VERBS,
+        "platforms.txt": _BUILTIN_PLATFORMS,
+        "scopes.txt": _BUILTIN_SCOPES,
+    }
+    fallback = fallbacks.get(filename, frozenset())
+    path = os.path.join(_NAMING_DIR, filename)
+    if not os.path.isfile(path):
+        return fallback
+    words: set[str] = set()
+    with open(path, encoding="utf-8") as f:
+        for line in f:
+            word = line.strip()
+            if word and not word.startswith("#"):
+                words.add(word)
+    return frozenset(words) if words else fallback
+
+
+def parse_slug_segments(slug: str) -> list[str]:
+    """按 '-' 分割；非法字符由上层 kebab 校验处理。"""
+    slug = slug.strip()
+    if not slug:
+        return []
+    return slug.split("-")
+
+
+def _match_platform_suffix(
+    segments: list[str],
+    platforms: frozenset[str],
+) -> tuple[str, int] | None:
+    """返回 (platform_name, platform_part_count) 或 None。"""
+    for platform in sorted(platforms, key=lambda item: (-item.count("-"), item)):
+        parts = platform.split("-")
+        part_count = len(parts)
+        if len(segments) < part_count + 2:
+            continue
+        if segments[-part_count:] == parts:
+            return platform, part_count
+    return None
+
+
+def classify_slug_form(segments: list[str], *, slug: str = "") -> str:
+    """
+    返回: 'legacy_exempt' | 'standard' | 'scope' | 'invalid'
+
+    - legacy_exempt: slug in LEGACY_EXEMPT_SLUGS
+    - standard: 末段匹配 platforms, 首段 in verbs, 3<=len<=5
+    - scope: 末段 in scopes, 首段 in verbs, 3<=len<=5
+    - invalid: 其他
+    """
+    if slug in LEGACY_EXEMPT_SLUGS:
+        return "legacy_exempt"
+
+    if not segments or not (_MIN_SEGMENTS <= len(segments) <= _MAX_SEGMENTS):
+        return "invalid"
+
+    verbs = load_wordlist("verbs.txt")
+    platforms = load_wordlist("platforms.txt")
+    scopes = load_wordlist("scopes.txt")
+
+    if segments[0] not in verbs:
+        return "invalid"
+
+    if segments[-1] in scopes:
+        return "scope"
+
+    if _match_platform_suffix(segments, platforms) is not None:
+        return "standard"
+
+    return "invalid"
+
+
+def validate_slug_semantics(
+    slug: str,
+    *,
+    strict: bool = True,
+) -> tuple[list[str], list[str]]:
+    """返回 (errors, warnings)。"""
+    slug = slug.strip()
+    if slug in LEGACY_EXEMPT_SLUGS:
+        return [], []
+
+    errors: list[str] = []
+    warnings: list[str] = []
+
+    if not _KEBAB_SLUG.fullmatch(slug):
+        msg = f"slug 必须为 kebab-case（小写字母、数字、连字符）：{slug!r}"
+        (errors if strict else warnings).append(msg)
+        return errors, warnings
+
+    if len(slug) > _MAX_SLUG_LENGTH:
+        msg = f"slug 长度不得超过 {_MAX_SLUG_LENGTH} 字符（当前 {len(slug)}）：{slug!r}"
+        (errors if strict else warnings).append(msg)
+
+    segments = parse_slug_segments(slug)
+    segment_count = len(segments)
+
+    if segment_count < _MIN_SEGMENTS:
+        msg = (
+            f"slug 段数应为 {_MIN_SEGMENTS}～{_MAX_SEGMENTS}（verb-noun-platform），"
+            f"当前为 {segment_count} 段：{slug!r}；新技能禁止 2 段无平台形态（如 reconcile-finance）"
+        )
+        (errors if strict else warnings).append(msg)
+        return errors, warnings
+
+    if segment_count > _MAX_SEGMENTS:
+        msg = f"slug 段数不得超过 {_MAX_SEGMENTS}（当前 {segment_count}）：{slug!r}"
+        (errors if strict else warnings).append(msg)
+
+    verbs = load_wordlist("verbs.txt")
+    platforms = load_wordlist("platforms.txt")
+    scopes = load_wordlist("scopes.txt")
+
+    if segments[0] in platforms and segments[0] not in verbs:
+        msg = f"平台名不应放在 slug 最前：{slug!r}（应为 verb-noun-platform）"
+        (errors if strict else warnings).append(msg)
+
+    if segments[0] not in verbs:
+        msg = f"slug 首段应为动词白名单中的词：{segments[0]!r}（slug={slug!r}）"
+        (errors if strict else warnings).append(msg)
+
+    form = classify_slug_form(segments, slug=slug)
+    if form == "scope":
+        if segments[-1] not in scopes:
+            msg = f"slug 末段应为 scope 白名单中的词：{segments[-1]!r}（slug={slug!r}）"
+            (errors if strict else warnings).append(msg)
+        noun_count = segment_count - 2
+    elif form == "standard":
+        platform_match = _match_platform_suffix(segments, platforms)
+        if platform_match is None:
+            msg = f"slug 末段应匹配平台白名单：{slug!r}"
+            (errors if strict else warnings).append(msg)
+            noun_count = 0
+        else:
+            _platform_name, platform_part_count = platform_match
+            noun_count = segment_count - 1 - platform_part_count
+    else:
+        if segment_count >= _MIN_SEGMENTS:
+            if segments[-1] not in scopes:
+                platform_match = _match_platform_suffix(segments, platforms)
+                if platform_match is None:
+                    msg = f"slug 末段应匹配平台或 scope 白名单：{slug!r}"
+                    (errors if strict else warnings).append(msg)
+        noun_count = 0
+
+    if noun_count > _MAX_NOUN_SEGMENTS:
+        warnings.append(
+            f"slug 中间名词段建议不超过 {_MAX_NOUN_SEGMENTS} 个词（当前 {noun_count}）：{slug!r}"
+        )
+
+    return errors, warnings

scripts/util/timeutil.py

@@ -0,0 +1,20 @@
+"""时间戳与 ISO 展示。"""
+
+from __future__ import annotations
+
+import time
+from datetime import datetime
+from typing import Optional
+
+
+def now_unix() -> int:
+    return int(time.time())
+
+
+def unix_to_iso(ts: Optional[int]) -> Optional[str]:
+    if ts is None:
+        return None
+    try:
+        return datetime.fromtimestamp(int(ts)).isoformat(timespec="seconds")
+    except (ValueError, OSError, OverflowError):
+        return None