4 Commits

Author SHA1 Message Date
chenfuqiang
825b210c21 chore(release): 版本升至 1.0.3(SKILL.md / SKILL_VERSION / CHANGELOG 对齐 POLICY-RELEASE-001)
All checks were successful
技能自动化发布 / release (push) Successful in 6s
2026-09-26 21:09:45 +08:00
chenfuqiang
643dae0ec4 fix: 缺省匹配年度收敛最新年度 + 批量坏文件跳过 + 匹配理由据实化 + 职称档位补全(P1+P2) 2026-09-26 21:09:40 +08:00
chenfuqiang
e219b055b2 chore(release): 版本升至 1.0.2(SKILL.md / SKILL_VERSION / CHANGELOG 对齐 POLICY-RELEASE-001)
All checks were successful
技能自动化发布 / release (push) Successful in 7s
2026-09-24 18:03:00 +08:00
chenfuqiang
4a8613b27e fix: 批量匹配跳过画像待补录教师 + render-html 按行重建 + 金标准 fixtures 业务化(P0+P1)
- run_match 新增 skip_incomplete:批量缺省跳过画像不完整教师并记录 skipped,指定单教师仍 PROFILE_INCOMPLETE;汇总页出"画像待补录"行
- render-html 补 --report-id 行上下文(inputMapping $row.$pk),CLI 容错解析回退最近 10 份
- report_service 日志改容错 _log(),修复未 setup 时单跑测试崩溃的顺序依赖
- fixtures 替换为教师画像/赛道/期望结论金标准样例,新增 sample_tracks.json
- 新增 4 个批量跳过用例(314 项,除 3 个已知 jsonschema 环境失败外全绿)
- tests/README 增补本机无共享 runtime 的测试方式;TUTORIAL/ACTIONS.md 同步表述
2026-09-24 18:01:41 +08:00
19 changed files with 451 additions and 76 deletions

View File

@@ -3,6 +3,20 @@
本文件对应技能市场 **「更新日志」** Tab。
发布 CI 会按当前 tag 版本提取匹配的 `##` 小节,写入该版本的 `changelog`。
## 1.0.3
- 缺省匹配年度收敛为赛道库最新年度:往年赛道不再混入缺省匹配结果(仍可用年度参数指定其它年份)
- 批量匹配容错:input 目录中无法解析的画像文件自动跳过并在任务中告警说明,不再中断整批
- 匹配理由更据实:与赛道需求标签无任何重合时不再显示"存在部分相关性",改为提示补充研究标签
- 职称档位补全(助理教授、工程师系列等);未识别职称遇门槛赛道仍按保守处理并说明原因
## 1.0.2
- 批量匹配健壮性:画像待补录的教师(无研究标签也无论文)改为跳过并记录原因,不再中断整批;单教师指定匹配仍明确提示补录;汇总页单列"画像待补录"
- 「生成建议书」支持按行重建:在"推荐报告"表选中行只重建该报告,未选行/HTML 右键回退重建最近 10 份
- 脱敏金标准样例落地:教师画像、指南赛道、期望匹配结论三份 fixtures 可直接导入;测试说明补充无共享 runtime 的本地跑法
- 修复报告服务日志初始化顺序问题(单测独立运行不再依赖其它用例先初始化)
## 1.0.1
- 首个版本:申报指南"智能赛道推荐"上线

10
DEMO.md
View File

@@ -4,6 +4,11 @@ demo_video_url: ""
演示视频录制后请把地址填入上方 `demo_video_url`。以下是无视频时的图文演示脚本。
## 演示数据准备
- 可直接导入 `tests/fixtures/` 三份脱敏样例:教师画像、指南赛道、期望匹配结论
- 缺省匹配取赛道库**最新年度**赛道:演示前确认"指南赛道"表已录入当年赛道,往年赛道不会混入结果
## 演示一:教师问推荐(对话)
1. 对话任务:"我是张教授,研究钙钛矿、光伏、半导体和界面调控,SCI 一区 3 篇,今年适合报什么?"
@@ -15,3 +20,8 @@ demo_video_url: ""
1. "指南赛道"表批量导入 2027 年各学部赛道;"教师画像"表维护本院教师。
2. 顶栏「赛道匹配」一键匹配;批量模式出汇总页,强推荐名单一目了然。
3. 「生成建议书」重建 HTML 供打印分发。
## 演示三:批量容错(可选)
1. input 目录放入多位教师画像 JSON,混入一个格式损坏的文件。
2. 发起批量匹配:坏文件被跳过并逐条告警说明,其余教师正常出报告与汇总页。

View File

@@ -1,7 +1,7 @@
---
name: 申报赛道推荐助手
description: "高校申报指南智能赛道推荐:维护教师学术画像与指南赛道库,按标签匹配 + 硬门槛过滤生成可解释的 Top 3 赛道推荐与《申报策略建议书》。"
version: 1.0.1
version: 1.0.3
author: 深圳匠厂科技有限公司
metadata:
openclaw:

View File

@@ -23,7 +23,7 @@
1. 在"教师画像"表点顶栏「赛道匹配」;可只填某位教师,缺省匹配全部
2. 提交后稍等,切到"推荐报告"表查看每人的 Top 3 与可匹配赛道数
3. 点顶栏「生成建议书」重建 HTML;左侧报告区打开 `report-编号` 查看 Top 3 卡片、分层清单与剔除原因
3. 点顶栏「生成建议书」重建 HTML(在"推荐报告"表选中某一行则只重建该报告,未选中则重建最近 10 份);左侧报告区打开 `report-编号` 查看 Top 3 卡片、分层清单与剔除原因
## 入口二:对话任务
@@ -67,3 +67,4 @@
- **所有赛道都是"不建议"**:研究标签与赛道需求标签重合度低;请教师补充/细化标签后重新匹配
- **看不到建议书**:先在"推荐报告"表点一次「生成建议书」,再到左侧 HTML 报告区打开
- **批量匹配没有结果**:确认 input 目录里的 JSON 是教师对象数组,且赛道库不为空
- **批量匹配少了某位老师**:该教师画像不完整(研究标签与论文都没有)会被跳过并记录原因,补录标签后重新匹配即可

View File

@@ -329,17 +329,31 @@
{
"id": "render-html",
"label": "生成建议书",
"description": "重建《申报策略建议书》HTML(缺省最近 10 份)。",
"description": "重建《申报策略建议书》HTML(未选择报告行时重建最近 10 份)。",
"placements": ["toolbar", "skill-detail", "agent"],
"executionProfile": "sync",
"interaction": "direct",
"bind": {
"tables": ["match_reports"]
"tables": ["match_reports"],
"inputMapping": {
"id": "$row.$pk"
}
},
"inputSchema": {
"type": "object",
"properties": {
"id": {
"type": "number",
"title": "报告编号",
"description": "重建该报告的建议书。未选择行时留空,重建最近 10 份。",
"readOnly": true
}
}
},
"entrypoint": {
"type": "cli",
"command": "render-html",
"args": []
"args": ["--report-id", "{{id}}"]
},
"steps": [
{

View File

@@ -777,6 +777,12 @@ HINT: <next step>
完整 JSON 形状见 [`../assets/examples/render-html.action.example.json`](../assets/examples/render-html.action.example.json)。最小页面壳见 [`../assets/html/example-report.html`](../assets/html/example-report.html)(已含数据管理 iframe 预览的布局基线;复制后可按需扩展 `{{BODY}}` 内结构)。
**本技能的行上下文写法**:`render-html` 的 `entrypoint.args` 为 `["--report-id", "{{id}}"]`,行 id 经 `bind.inputMapping: { "id": "$row.$pk" }` 注入、并在 `inputSchema.properties.id` 声明(模板参数必须声明,见上文规则)。行为约定:
- 数据管理在"推荐报告"表**选中某行**点「生成建议书」→ 只重建该报告(`report-{id}.html`)。
- 从 **HTML 右键重新生成 / skill-detail / 对话** 触发时**无行上下文**:宿主省略空的可选数字字段或占位未解析;CLI(`scripts/cli/app.py` `_handle_render_html`)对 `--report-id` 为 `""` / 非数字 / 缺值一律回退为"重建最近 10 份",不会报错。
- 因此 `--report-id` 在 argparse 上是可选值(`nargs="?"`),这是对宿主省略行为的容错,勿删。
### 宿主预览布局
数据管理在 **iframe** 内预览 `html/*.html`;宿主 iframe 已占满右侧预览区,**技能页不应再人为收窄正文**(尤其数据管理窗口通常较窄)。

View File

@@ -109,8 +109,10 @@ def _handle_list_reports(args: argparse.Namespace) -> int:
def _handle_render_html(args: argparse.Namespace) -> int:
report_id = getattr(args, "report_id", None)
return cmd_render_html(report_id=int(report_id) if report_id else None)
# 宿主行上下文可能给空串或未解析占位符(如 {{id}}):解析失败一律回退
# 缺省行为(重建最近 10 份),不让 args 变体把渲染打崩。
report_id = _cli_int_or_none(getattr(args, "report_id", None))
return cmd_render_html(report_id=report_id)
def _print_full_usage() -> None:
@@ -195,7 +197,7 @@ def build_parser() -> ZhArgumentParser:
sp.set_defaults(handler=lambda a: cmd_show_report(report_id=int(a.report_id)))
sp = sub.add_parser("render-html", help="重建《申报策略建议书》(缺省最近 10 份)")
sp.add_argument("--report-id", default=None, metavar="报告编号", dest="report_id")
sp.add_argument("--report-id", default=None, nargs="?", metavar="报告编号", dest="report_id")
sp.set_defaults(handler=_handle_render_html)
sp = sub.add_parser("stats", help="匹配统计")

View File

@@ -28,19 +28,28 @@ TIER_NOT = "不建议"
_TIER_RANK = {TIER_NOT: 0, TIER_TRY: 1, TIER_STRONG: 2}
# 职称档位:数值越大越高职称;门槛 min_title 使用"X及以上"档位
# 职称档位:数值越大越高职称;门槛 min_title 使用"X及以上"档位。
# 助理教授按中级档从审处理(预聘制档次各校不一,宁缺勿滥);
# 未识别职称遇门槛赛道一律剔除并说明原因(审慎非激进)。
_TITLE_RANK = {
"助教": 0,
"初级": 0,
"讲师": 1,
"中级": 1,
"助理研究员": 1,
"助理教授": 1,
"工程师": 1,
"博士后": 1,
"副高及以上": 2,
"副教授": 2,
"副研究员": 2,
"高级实验师": 2,
"高级工程师": 2,
"正高及以上": 3,
"教授": 3,
"研究员": 3,
"教授级高级工程师": 3,
"正高级工程师": 3,
}
_STRONG_THRESHOLD = 80
@@ -197,7 +206,7 @@ def match_teacher(teacher: dict[str, Any], tracks: list[dict[str, Any]], season:
if min_sci > 0:
reasons.append(f"SCI一区余量:现有 {sci_q1} 篇 / 要求 {min_sci} 篇")
if not reasons:
reasons.append("与赛道方向存在部分相关性")
reasons.append("现有研究标签与该赛道需求标签暂无重合,补充研究标签后可能改善匹配")
if match_pct >= _STRONG_THRESHOLD:
tier = TIER_STRONG
@@ -247,8 +256,20 @@ def _write_report(result: dict[str, Any], batch_no: str) -> str:
return write_report_for_kernel(result=result, batch_no=batch_no)
def run_match(teachers_raw: list[dict[str, Any]], tracks: list[dict[str, Any]], season: str, *, batch_no: str = "") -> dict[str, Any]:
"""匹配内核:逐教师匹配 → 写库 → 生成建议书(多人时另出汇总页)。"""
def run_match(
teachers_raw: list[dict[str, Any]],
tracks: list[dict[str, Any]],
season: str,
*,
batch_no: str = "",
skip_incomplete: bool = True,
) -> dict[str, Any]:
"""匹配内核:逐教师匹配 → 写库 → 生成建议书(多人时另出汇总页)。
skip_incomplete=True(批量缺省):画像不完整(E1)的教师被跳过并记入
``skipped``,不中断批次;skip_incomplete=False(指定单教师):直接抛
ValueError,由 CLI 映射 PROFILE_INCOMPLETE。
"""
from service.report_service import build_batch_brief
log = _log()
@@ -263,11 +284,21 @@ def run_match(teachers_raw: list[dict[str, Any]], tracks: list[dict[str, Any]],
log.info("match_kernel_start batch_no=%s teachers=%s tracks=%s season=%s stage=kernel", batch_no, len(teachers_raw), len(tracks), season)
results = []
skipped: list[dict[str, str]] = []
conn = get_conn()
try:
cur = conn.cursor()
for teacher_raw in teachers_raw:
result = match_teacher(teacher_raw, tracks, season)
try:
result = match_teacher(teacher_raw, tracks, season)
except ValueError as exc:
if not skip_incomplete:
raise
raw_name = str((teacher_raw or {}).get("teacherName") or (teacher_raw or {}).get("teacher_name") or "").strip() or "未命名教师"
skipped.append({"teacher_name": raw_name, "reason": str(exc)})
if log:
log.warning("match_teacher_skipped teacher=%s reason=%s stage=kernel", raw_name, exc)
continue
cur.execute(
"""
INSERT INTO match_reports
@@ -296,7 +327,7 @@ def run_match(teachers_raw: list[dict[str, Any]], tracks: list[dict[str, Any]],
finally:
conn.close()
brief_file = build_batch_brief(batch_no=batch_no, results=results, season=season)
brief_file = build_batch_brief(batch_no=batch_no, results=results, season=season, skipped=skipped)
tier_counts: dict[str, int] = {}
for result in results:
for tier, entries in result["tiers"].items():
@@ -307,25 +338,34 @@ def run_match(teachers_raw: list[dict[str, Any]], tracks: list[dict[str, Any]],
"batch_no": batch_no,
"season": season,
"count": len(results),
"skipped": skipped,
"tier_counts": tier_counts,
"brief_file": brief_file,
"disclaimer": "推荐仅供参考,申报决策请结合个人规划与学院统筹。",
"reports": results,
}
if log:
log.info("match_kernel_done batch_no=%s count=%s stage=kernel", batch_no, len(results))
log.info(
"match_kernel_done batch_no=%s count=%s skipped=%s stage=kernel",
batch_no,
len(results),
len(skipped),
)
return payload
def load_tracks(year: Optional[int] = None) -> list[dict[str, Any]]:
"""读取赛道库(缺省全部年度;可按年度筛选)。"""
"""读取赛道库(缺省取赛道库最新年度,与 actions.json 契约一致;显式传 year 按年度筛选)。"""
conn = get_conn()
try:
cur = conn.cursor()
if year:
cur.execute("SELECT * FROM guide_tracks WHERE year = ? ORDER BY id", (year,))
else:
cur.execute("SELECT * FROM guide_tracks ORDER BY year DESC, id")
if year is None:
cur.execute("SELECT MAX(year) FROM guide_tracks")
row = cur.fetchone()
if not row or row[0] is None:
return []
year = int(row[0])
cur.execute("SELECT * FROM guide_tracks WHERE year = ? ORDER BY id", (year,))
columns = [d[0] for d in cur.description]
rows = [dict(zip(columns, row)) for row in cur.fetchall()]
finally:
@@ -333,6 +373,35 @@ def load_tracks(year: Optional[int] = None) -> list[dict[str, Any]]:
return rows
def collect_input_teachers(input_dir: str) -> tuple[list[dict[str, Any]], list[dict[str, str]]]:
"""读取 input/ 目录下的教师画像 JSON;坏文件跳过并记录原因,不中断整批。
返回 (教师列表, 跳过文件列表);跳过文件形如 {"file": 名称, "reason": 原因}。
"""
import os
files = sorted(
os.path.join(input_dir, name)
for name in os.listdir(input_dir)
if name.lower().endswith(".json")
)
teachers: list[dict[str, Any]] = []
skipped_files: list[dict[str, str]] = []
for path in files:
name = os.path.basename(path)
try:
with open(path, encoding="utf-8") as f:
payload = json.load(f)
if isinstance(payload, dict):
payload = payload.get("teachers") or [payload]
if not isinstance(payload, list):
raise ValueError("内容不是教师对象数组")
teachers.extend(item for item in payload if isinstance(item, dict))
except (OSError, ValueError) as exc:
skipped_files.append({"file": name, "reason": str(exc)})
return teachers, skipped_files
def load_teachers(teacher_name: str = "") -> list[dict[str, Any]]:
"""读取教师画像(可按姓名精确筛选)。"""
conn = get_conn()

View File

@@ -18,6 +18,14 @@ _TEMPLATE_REL = "assets/html/match-report.html"
_TIER_BADGE = {"强烈推荐": "tier-strong", "可尝试": "tier-try", "不建议": "tier-not"}
def _log():
"""容错取 logger:未 setup_skill_logging(如单测直调 service)时返回 None。"""
try:
return get_skill_logger()
except RuntimeError:
return None
def _esc(value: Any) -> str:
return html_escape(str(value if value is not None else "—"))
@@ -106,9 +114,10 @@ def build_match_html(result: dict[str, Any], batch_no: str = "") -> str:
def write_report_for_kernel(*, result: dict[str, Any], batch_no: str) -> str:
html_text = build_match_html(result, batch_no)
log = get_skill_logger()
log = _log()
path = write_html_page(f"report-{result['report_id']}.html", html_text)
log.info("match_report_written report_id=%s path=%s stage=report", result["report_id"], path)
if log:
log.info("match_report_written report_id=%s path=%s stage=report", result["report_id"], path)
return path
@@ -116,8 +125,9 @@ def build_report_for_report_id(report_id: int) -> str:
"""按报告 id 从库中取数并重写 HTML(重建入口用)。"""
from db.connection import get_conn
log = get_skill_logger()
log.info("report_render_start report_id=%s stage=render", report_id)
log = _log()
if log:
log.info("report_render_start report_id=%s stage=render", report_id)
conn = get_conn()
try:
cur = conn.cursor()
@@ -139,13 +149,20 @@ def build_report_for_report_id(report_id: int) -> str:
result["top3"] = result.get("top") or []
html_text = build_match_html(result)
path = write_html_page(f"report-{report_id}.html", html_text)
log.info("report_render_done report_id=%s path=%s stage=render", report_id, path)
if log:
log.info("report_render_done report_id=%s path=%s stage=render", report_id, path)
return path
def build_batch_brief(*, batch_no: str, results: list[dict[str, Any]], season: str) -> str:
"""批量匹配汇总页:各教师 Top 1 一览。"""
log = get_skill_logger()
def build_batch_brief(
*,
batch_no: str,
results: list[dict[str, Any]],
season: str,
skipped: list[dict[str, Any]] | None = None,
) -> str:
"""批量匹配汇总页:各教师 Top 1 一览;被跳过(画像待补录)教师单列。"""
log = _log()
rows = []
for result in results:
top3 = result.get("top3") or []
@@ -163,6 +180,13 @@ def build_batch_brief(*, batch_no: str, results: list[dict[str, Any]], season: s
)
else:
rows.append(f"<tr><td>{_esc(result.get('teacher_name'))}</td><td colspan='4' class='muted'>无可推荐赛道(画像待补全)</td></tr>")
for entry in skipped or []:
rows.append(
"<tr>"
f"<td>{_esc(entry.get('teacher_name'))}</td>"
f"<td colspan='4' class='muted'>画像待补录,本期未匹配({_esc(entry.get('reason'))})</td>"
"</tr>"
)
if not rows:
rows.append('<tr><td colspan="5" class="muted">本期无匹配结果。</td></tr>')
@@ -181,5 +205,6 @@ def build_batch_brief(*, batch_no: str, results: list[dict[str, Any]], season: s
{"TITLE": "申报赛道匹配汇总", "GENERATED_AT": str(season), "BODY": body},
)
path = write_html_page(f"brief-{batch_no}.html", page)
log.info("match_brief_written batch_no=%s path=%s stage=report", batch_no, path)
if log:
log.info("match_brief_written batch_no=%s path=%s stage=report", batch_no, path)
return path

View File

@@ -478,14 +478,19 @@ def cmd_match(*, teacher: str = "", year: Optional[int] = None) -> int:
print(f"ERROR:TEACHER_NOT_FOUND {message}")
return 1
if not tracks:
message = "指南赛道库为空:请先录入赛道(add-track / import-tracks)"
if year:
message = f"指南赛道库中没有 {year} 年度赛道:请核对年度或先录入(add-track / import-tracks)"
else:
message = "指南赛道库为空:请先录入赛道(add-track / import-tracks)"
log.warning("task_failed task_type=match error_code=TRACK_EMPTY stage=match")
finish(status="failed", message=message, skill=SKILL_SLUG, error_code="TRACK_EMPTY", stage="match")
print(f"ERROR:TRACK_EMPTY {message}")
return 1
season = str(year or tracks[0].get("year") or "")
result = run_match(teachers, tracks, season)
# 指定教师 → 严格模式,画像不完整直接 PROFILE_INCOMPLETE;
# 缺省全量 → 跳过画像待补录教师并记录,不中断批次(E1)。
result = run_match(teachers, tracks, season, skip_incomplete=not teacher)
payload = {k: v for k, v in result.items() if k != "reports"}
payload["reports_brief"] = [
{
@@ -499,9 +504,14 @@ def cmd_match(*, teacher: str = "", year: Optional[int] = None) -> int:
for r in result["reports"]
]
_print_json(payload)
skipped_count = len(result.get("skipped") or [])
message = f"匹配完成 {result['count']} 名教师,建议书已生成"
if skipped_count:
message += f";{skipped_count} 名教师画像待补录已跳过"
log.info("match_skipped count=%s stage=match", skipped_count)
finish(
status="success",
message=f"匹配完成 {result['count']} 名教师,建议书已生成",
message=message,
skill=SKILL_SLUG,
stage="match",
batch_no=result["batch_no"],
@@ -522,9 +532,13 @@ def cmd_match(*, teacher: str = "", year: Optional[int] = None) -> int:
def cmd_match_batch_input() -> int:
"""批量匹配(async Action 入口):匹配 input/ 目录中的教师画像 JSON 与库内赛道。"""
"""批量匹配(async Action 入口):匹配 input/ 目录中的教师画像 JSON 与库内赛道。
坏文件(解析失败/非对象数组)跳过并告警,不中断整批;画像不完整教师同样跳过并记录(E1)。
缺省匹配赛道库最新年度赛道。
"""
log = _get_task_logger()
from service.match_service import load_tracks, run_match
from service.match_service import collect_input_teachers, load_tracks, run_match
from util.runtime_paths import get_input_dir
with job_context(skill=SKILL_SLUG):
@@ -539,26 +553,17 @@ def cmd_match_batch_input() -> int:
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}"
teachers, skipped_files = collect_input_teachers(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 teachers:
message = f"input 目录没有可用的教师画像 JSON({len(skipped_files)} 个文件无法解析):{input_dir}"
log.warning("task_failed task_type=match_batch error_code=TEACHER_NOT_FOUND stage=batch")
finish(status="failed", message=message, skill=SKILL_SLUG, error_code="TEACHER_NOT_FOUND", stage="batch")
print(f"ERROR:TEACHER_NOT_FOUND {message}")
return 1
teachers = []
for path in files:
with open(path, encoding="utf-8") as f:
payload = json.load(f)
if isinstance(payload, dict):
payload = payload.get("teachers") or [payload]
teachers.extend(payload)
tracks = load_tracks()
if not tracks:
message = "指南赛道库为空:请先录入赛道(add-track / import-tracks)"
@@ -568,7 +573,7 @@ def cmd_match_batch_input() -> int:
return 1
season = str(tracks[0].get("year") or "")
result = run_match(teachers, tracks, season)
result = run_match(teachers, tracks, season, skip_incomplete=True)
for index, report in enumerate(result["reports"], start=1):
top = (report.get("top3") or [{}])[0]
emit(
@@ -577,16 +582,28 @@ def cmd_match_batch_input() -> int:
skill=SKILL_SLUG,
stage="batch",
)
for entry in result.get("skipped") or []:
emit(
f"{entry.get('teacher_name')}:画像待补录,本期已跳过({entry.get('reason')})",
type="warn",
skill=SKILL_SLUG,
stage="batch",
)
_print_json(
{
"ok": True,
"batch_no": result["batch_no"],
"count": result["count"],
"skipped": result.get("skipped") or [],
"skipped_files": skipped_files,
"tier_counts": result["tier_counts"],
"brief_file": result["brief_file"],
}
)
finish(status="success", message=f"批量匹配完成 {result['count']} 名教师", skill=SKILL_SLUG, stage="batch", batch_no=result["batch_no"])
message = f"批量匹配完成 {result['count']} 名教师"
if result.get("skipped"):
message += f",{len(result['skipped'])} 名画像待补录已跳过"
finish(status="success", message=message, skill=SKILL_SLUG, stage="batch", batch_no=result["batch_no"])
return 0
except JobStopped:
raise

View File

@@ -1,6 +1,6 @@
"""技能标识、版本与平台公共库约束(申报赛道推荐助手)。"""
SKILL_SLUG = "match-guide-track-default"
SKILL_VERSION = "1.0.1"
SKILL_VERSION = "1.0.3"
LOG_LOGGER_NAME = "openclaw.skill.match_guide_track_default"
PLATFORM_KIT_MIN_VERSION = "1.2.11"

View File

@@ -103,6 +103,20 @@ python tests/run_tests.py test_cli_smoke
- ``tests/.gitignore`` 忽略 ``__pycache__``、``.pytest_cache``、``*.pyc``、``desktop/artifacts/``、``integration/artifacts/``、``.env`` 等。
- 推荐运行前设置 ``PYTHONDONTWRITEBYTECODE=1`` 减少 ``.pyc``。
### 1.5 本机无共享 runtime 时怎么跑(本技能开发机约定)
宿主共享 runtime(``{JIANGCHANG_DATA_ROOT}/python-runtime/.venv``)只在宿主环境存在;纯开发机上 ``import jiangchang_skill_core`` 会直接 ``ModuleNotFoundError``(整套测试 60+ errors)。**不安装任何包**的跑法:
```powershell
$env:PYTHONPATH = "D:\jiangchang\huashang\jiangchang-platform-kit\jiangchang-platform-kit\src"
python tests\run_tests.py -v
```
已知环境性失败(不修、不装包):``test_actions_manifest.py`` 中 3 项依赖 ``jsonschema`` 的 schema 校验
(``test_schema_allows_all_placement_execution_profile_combinations`` / ``test_schema_allows_bind_label_column`` /
``test_schema_rejects_toolbar_without_bind``)会以 ``jsonschema is required`` 失败。**除这 3 项外须全部通过**;
接入宿主共享 runtime 后自动恢复。``requirements.txt`` 保持为空——平台库由宿主 runtime 提供,技能不声明平台依赖。
---
## 2. 可选:Service / Golden(``tests/samples/*.sample``)

View File

@@ -12,8 +12,9 @@
| 文件 | 用途 |
|------|------|
| `sample_request.json` | 通用请求体样例,供 ``tests/samples/test_golden_cases.py.sample`` 演示读取。 |
| `expected_response.json` | 与上配套的期望摘要字段(业务复制后按域替换)。 |
| `sample_request.json` | 教师画像金标准样例(脱敏,JSON 数组),可直接 `import-teachers --teachers-json` 消费;供 `tests/samples/test_golden_cases.py.sample` 演示读取。 |
| `sample_tracks.json` | 指南赛道金标准样例(脱敏,JSON 数组),可直接 `import-tracks --tracks-json` 消费;与上配套覆盖 H1/H2/S1~S3/T1 场景。 |
| `expected_response.json` | 与上两份配套的期望匹配结论(Top 赛道/分层/得分/剔除项,业务复制后按域替换)。 |
| `adapter_profiles.sample.yaml` | 五类运行方式(mock / simulator_api / simulator_rpa / real_api / real_rpa)的配置表达样例。 |
---

View File

@@ -1,6 +1,22 @@
{
"success": true,
"data": {
"status": "ok"
}
"season": "2027",
"teacher_count": 2,
"track_count": 3,
"expectations": [
{
"teacherName": "张教授",
"topTrack": "面上项目·光伏材料",
"tier": "强烈推荐",
"matchPct": 100,
"reasonsContains": ["命中需求标签", "优先资助方向命中"]
},
{
"teacherName": "王讲师",
"topTrack": "青年项目·农村经济",
"tier": "可尝试",
"matchPct": 70,
"excludedContains": ["面上项目·光伏材料", "重点项目·量子计算"]
}
],
"disclaimerContains": "仅供参考"
}

View File

@@ -1,11 +1,22 @@
{
"request_id": "req_test_001",
"tenant_id": "tenant_demo",
"actor": "tester",
"parameters": {
"input_text": "demo input",
"options": {
"dry_run": true
}
[
{
"teacherName": "张教授",
"department": "化材学院",
"title": "教授",
"researchTags": "钙钛矿,光伏,半导体,界面调控,稳定性",
"recentPapers": "钙钛矿电池界面钝化研究",
"projectHistory": "主持省自然科学基金",
"sciQ1Count": 3,
"corePaperCount": 8
},
{
"teacherName": "王讲师",
"department": "经管学院",
"title": "讲师",
"researchTags": "乡村振兴,农村经济",
"recentPapers": "县域乡村产业发展研究",
"projectHistory": "参与市社科基金",
"sciQ1Count": 0,
"corePaperCount": 1
}
}
]

38
tests/fixtures/sample_tracks.json vendored Normal file
View File

@@ -0,0 +1,38 @@
[
{
"year": 2027,
"agency": "数理学部",
"programName": "面上项目",
"trackName": "光伏材料",
"trackTags": "钙钛矿,光伏,半导体",
"priorityKeywords": "界面调控,稳定性",
"minTitle": "副高及以上",
"minSciQ1": 0,
"quota": 10,
"sourceDoc": "2027年度申报指南(样例)"
},
{
"year": 2027,
"agency": "管理学部",
"programName": "青年项目",
"trackName": "农村经济",
"trackTags": "乡村振兴,农村经济",
"priorityKeywords": "数字经济",
"minTitle": "",
"minSciQ1": 0,
"quota": 8,
"sourceDoc": "2027年度申报指南(样例)"
},
{
"year": 2027,
"agency": "信息学部",
"programName": "重点项目",
"trackName": "量子计算",
"trackTags": "量子计算,量子信息",
"priorityKeywords": "",
"minTitle": "正高及以上",
"minSciQ1": 5,
"quota": 3,
"sourceDoc": "2027年度申报指南(样例)"
}
]

View File

@@ -40,7 +40,7 @@
- **计算类**:费用、税费、分摊、账单字段。
- **校验类**:合规、字段一致性、状态机规则。
推荐配合根目录下的 **`tests/fixtures/sample_request.json`** 与 **`tests/fixtures/expected_response.json`**(复制技能后按域替换字段,保持脱敏)。
推荐配合根目录下的 **`tests/fixtures/sample_request.json`**、**`tests/fixtures/sample_tracks.json`** 与 **`tests/fixtures/expected_response.json`**(复制技能后按域替换字段,保持脱敏)。
### 5. `test_logging_contract.py.sample`

View File

@@ -26,11 +26,14 @@ class TestGoldenCasesSample(unittest.TestCase):
req = json.load(f)
with open(exp_path, encoding="utf-8") as f:
exp = json.load(f)
self.assertEqual(req["request_id"], "req_test_001")
self.assertTrue(req["parameters"]["options"]["dry_run"])
# 示意:out = run(req); 然后 assert out["success"] == exp["success"] 等关键字段
self.assertTrue(exp["success"])
self.assertEqual(exp["data"]["status"], "ok")
# 本技能:sample_request = 教师画像数组(import-teachers 可直接消费)
self.assertIsInstance(req, list)
self.assertEqual(req[0]["teacherName"], "张教授")
# 示意:run_match(req, tracks, exp["season"]) 后断言关键字段
self.assertEqual(exp["season"], "2027")
top = exp["expectations"][0]
self.assertEqual(top["tier"], "强烈推荐")
self.assertEqual(top["matchPct"], 100)
if __name__ == "__main__":

View File

@@ -2,6 +2,9 @@
"""赛道匹配内核金标准用例(离线、确定性)。"""
from __future__ import annotations
import json
import os
import tempfile
import unittest
from _support import IsolatedDataRoot
@@ -80,13 +83,29 @@ class TestScoring(MatchRuleTestBase):
self.assertEqual(top["match_pct"], 100)
self.assertEqual(top["tier"], "强烈推荐")
def test_t1_try_band(self) -> None:
def test_coverage_only_below_try_band(self) -> None:
result = self._match(_teacher(researchTags="钙钛矿,光伏", recentPapers=""))
entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料")
# coverage 2/3 → 0.7*0.667=0.4667 → 47% 不建议? + 无 priority → 47%;二维材料 coverage 1/3=0.333→23%
# 仅覆盖率 2/3 → 0.7×0.6667=46.7% → 47%,低于 50% 可尝试档 → 不建议
self.assertEqual(entry["match_pct"], 47)
self.assertEqual(entry["tier"], "不建议")
def test_t1_try_band_with_supplement_hint(self) -> None:
result = self._match(_teacher(researchTags="钙钛矿,光伏,界面调控"))
entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料")
# 覆盖率 2/3 → 0.7×0.6667=46.7% + 优先方向 1 命中 0.15 = 61.7% → 62%,落在可尝试档
self.assertEqual(entry["match_pct"], 62)
self.assertEqual(entry["tier"], "可尝试")
self.assertTrue(any("建议补充" in r for r in entry["reasons"]))
def test_zero_match_reason_is_factual(self) -> None:
result = self._match(_teacher(researchTags="量子计算", recentPapers="", projectHistory=""))
entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料")
reasons_text = "".join(entry["reasons"])
self.assertEqual(entry["match_pct"], 0)
self.assertIn("暂无重合", reasons_text)
self.assertNotIn("部分相关性", reasons_text)
def test_t1_strong_boundary(self) -> None:
# 恰好 80% 属强烈推荐
result = self._match(_teacher(researchTags="钙钛矿,光伏,半导体", recentPapers=""))
@@ -167,5 +186,120 @@ class TestKernelAndPersistence(MatchRuleTestBase):
self.assertEqual(payload["tracks_by_year"].get("2027"), 4)
class TestBatchSkipIncomplete(MatchRuleTestBase):
"""批量缺省跳过画像不完整教师(E1),指定单教师保持严格失败。"""
def test_batch_skips_incomplete_teacher(self) -> None:
from service.match_service import run_match
incomplete = _teacher(teacherName="王讲师", researchTags="", recentPapers="")
payload = run_match([_teacher(), incomplete], _tracks(), "2027")
self.assertEqual(payload["ok"], True)
self.assertEqual(payload["count"], 1)
self.assertEqual(len(payload["skipped"]), 1)
skipped = payload["skipped"][0]
self.assertEqual(skipped["teacher_name"], "王讲师")
self.assertIn("E1", skipped["reason"])
# 正常教师仍有报告
self.assertEqual(len(payload["reports"]), 1)
self.assertEqual(payload["reports"][0]["teacher_name"], "张教授")
def test_batch_brief_marks_skipped_teacher(self) -> None:
from service.match_service import run_match
incomplete = _teacher(teacherName="王讲师", researchTags="", recentPapers="")
payload = run_match([_teacher(), incomplete], _tracks(), "2027")
with open(payload["brief_file"], encoding="utf-8") as f:
html_text = f.read()
self.assertIn("画像待补录", html_text)
self.assertIn("王讲师", html_text)
def test_all_skipped_still_returns_success_payload(self) -> None:
from service.match_service import run_match
payload = run_match([_teacher(researchTags="", recentPapers="")], _tracks(), "2027")
self.assertEqual(payload["ok"], True)
self.assertEqual(payload["count"], 0)
self.assertEqual(len(payload["skipped"]), 1)
def test_strict_mode_raises_for_incomplete(self) -> None:
from service.match_service import run_match
with self.assertRaises(ValueError):
run_match([_teacher(researchTags="", recentPapers="")], _tracks(), "2027", skip_incomplete=False)
class TestTrackYearScope(MatchRuleTestBase):
"""缺省取赛道库最新年度(actions.json 契约);显式年度不受影响。"""
def _seed_two_years(self) -> None:
from service.profile_service import upsert_track
upsert_track({"year": 2026, "track_name": "旧方向", "program_name": "面上项目", "track_tags": "钙钛矿,光伏"})
upsert_track({"year": 2027, "track_name": "新方向", "program_name": "面上项目", "track_tags": "量子计算"})
def test_load_tracks_defaults_to_latest_year(self) -> None:
from service.match_service import load_tracks
self._seed_two_years()
self.assertEqual({t["year"] for t in load_tracks()}, {2027})
def test_load_tracks_explicit_year(self) -> None:
from service.match_service import load_tracks
self._seed_two_years()
self.assertEqual({t["year"] for t in load_tracks(year=2026)}, {2026})
def test_load_tracks_empty_library(self) -> None:
from service.match_service import load_tracks
self.assertEqual(load_tracks(), [])
def test_kernel_default_matches_latest_year_only(self) -> None:
from service.match_service import load_teachers, load_tracks, run_match
from service.profile_service import upsert_teacher
self._seed_two_years()
upsert_teacher({"teacher_name": "张教授", "title": "教授", "research_tags": "量子计算,量子信息"})
payload = run_match(load_teachers(), load_tracks(), "2027")
report = payload["reports"][0]
all_tracks = [e["track"] for entries in report["tiers"].values() for e in entries]
all_tracks += [e["track"] for e in report["top3"]]
self.assertIn("面上项目·新方向", all_tracks)
self.assertNotIn("面上项目·旧方向", all_tracks)
class TestBatchInputCollection(MatchRuleTestBase):
"""input/ 目录收集:坏文件跳过并记录原因,不中断整批。"""
def test_collect_input_teachers_skips_bad_files(self) -> None:
from service.match_service import collect_input_teachers
with tempfile.TemporaryDirectory() as tmp:
with open(os.path.join(tmp, "a.json"), "w", encoding="utf-8") as f:
json.dump([_teacher(), _teacher(teacherName="李教授")], f, ensure_ascii=False)
with open(os.path.join(tmp, "single.json"), "w", encoding="utf-8") as f:
json.dump({"teacherName": "王教授"}, f, ensure_ascii=False)
with open(os.path.join(tmp, "bad.json"), "w", encoding="utf-8") as f:
f.write("{not-json")
with open(os.path.join(tmp, "scalar.json"), "w", encoding="utf-8") as f:
json.dump("just-a-string", f, ensure_ascii=False)
teachers, skipped = collect_input_teachers(tmp)
names = [t.get("teacherName") for t in teachers]
self.assertEqual(names, ["张教授", "李教授", "王教授"])
self.assertEqual({e["file"] for e in skipped}, {"bad.json", "scalar.json"})
self.assertTrue(all(e["reason"] for e in skipped))
def test_collect_input_teachers_empty_dir(self) -> None:
from service.match_service import collect_input_teachers
with tempfile.TemporaryDirectory() as tmp:
teachers, skipped = collect_input_teachers(tmp)
self.assertEqual(teachers, [])
self.assertEqual(skipped, [])
if __name__ == "__main__":
unittest.main()