Files

576 lines
24 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""任务编排层: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 db.seed_data import KNOWN_RANKS
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 为空")
if len(metrics) > 1:
raise ValueError("verify-metric 仅支持单个申报案例;多个案例请放入 input 目录走 verify-batch-input")
case = dict(metrics[0])
if teacher_name:
case.setdefault("teacherName", teacher_name)
if department:
case.setdefault("department", department)
case.setdefault("applyRank", apply_rank)
final_rank = str(case.get("applyRank") or "").strip()
if final_rank not in KNOWN_RANKS:
message = f"申报级别只接受 {'、'.join(KNOWN_RANKS)}:{final_rank or '(未提供)'}"
log.warning("task_failed task_type=verify_metric error_code=RANK_INVALID stage=verify")
finish(status="failed", message=message, skill=SKILL_SLUG, error_code="RANK_INVALID", stage="verify")
print(f"ERROR:RANK_INVALID {message}")
return 1
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 collect_input_cases, 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()
cases, skipped_files = collect_input_cases(input_dir)
for entry in skipped_files:
log.warning("batch_input_file_skipped file=%s reason=%s stage=batch", entry["file"], entry["reason"])
emit(f"{entry['file']}:{entry['reason']},该文件已跳过", type="warn", skill=SKILL_SLUG, stage="batch")
if not cases:
message = f"input 目录没有可用的申报案例 JSON({len(skipped_files)} 个文件无法解析):{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
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"],
"skipped_files": skipped_files,
"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