2 Commits
v1.0.2 ... main

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
7 changed files with 177 additions and 29 deletions

View File

@@ -3,6 +3,13 @@
本文件对应技能市场 **「更新日志」** Tab。 本文件对应技能市场 **「更新日志」** Tab。
发布 CI 会按当前 tag 版本提取匹配的 `##` 小节,写入该版本的 `changelog`。 发布 CI 会按当前 tag 版本提取匹配的 `##` 小节,写入该版本的 `changelog`。
## 1.0.3
- 缺省匹配年度收敛为赛道库最新年度:往年赛道不再混入缺省匹配结果(仍可用年度参数指定其它年份)
- 批量匹配容错:input 目录中无法解析的画像文件自动跳过并在任务中告警说明,不再中断整批
- 匹配理由更据实:与赛道需求标签无任何重合时不再显示"存在部分相关性",改为提示补充研究标签
- 职称档位补全(助理教授、工程师系列等);未识别职称遇门槛赛道仍按保守处理并说明原因
## 1.0.2 ## 1.0.2
- 批量匹配健壮性:画像待补录的教师(无研究标签也无论文)改为跳过并记录原因,不再中断整批;单教师指定匹配仍明确提示补录;汇总页单列"画像待补录" - 批量匹配健壮性:画像待补录的教师(无研究标签也无论文)改为跳过并记录原因,不再中断整批;单教师指定匹配仍明确提示补录;汇总页单列"画像待补录"

10
DEMO.md
View File

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

View File

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

View File

@@ -28,19 +28,28 @@ TIER_NOT = "不建议"
_TIER_RANK = {TIER_NOT: 0, TIER_TRY: 1, TIER_STRONG: 2} _TIER_RANK = {TIER_NOT: 0, TIER_TRY: 1, TIER_STRONG: 2}
# 职称档位:数值越大越高职称;门槛 min_title 使用"X及以上"档位 # 职称档位:数值越大越高职称;门槛 min_title 使用"X及以上"档位。
# 助理教授按中级档从审处理(预聘制档次各校不一,宁缺勿滥);
# 未识别职称遇门槛赛道一律剔除并说明原因(审慎非激进)。
_TITLE_RANK = { _TITLE_RANK = {
"助教": 0, "助教": 0,
"初级": 0,
"讲师": 1, "讲师": 1,
"中级": 1,
"助理研究员": 1, "助理研究员": 1,
"助理教授": 1,
"工程师": 1,
"博士后": 1, "博士后": 1,
"副高及以上": 2, "副高及以上": 2,
"副教授": 2, "副教授": 2,
"副研究员": 2, "副研究员": 2,
"高级实验师": 2, "高级实验师": 2,
"高级工程师": 2,
"正高及以上": 3, "正高及以上": 3,
"教授": 3, "教授": 3,
"研究员": 3, "研究员": 3,
"教授级高级工程师": 3,
"正高级工程师": 3,
} }
_STRONG_THRESHOLD = 80 _STRONG_THRESHOLD = 80
@@ -197,7 +206,7 @@ def match_teacher(teacher: dict[str, Any], tracks: list[dict[str, Any]], season:
if min_sci > 0: if min_sci > 0:
reasons.append(f"SCI一区余量:现有 {sci_q1} 篇 / 要求 {min_sci} 篇") reasons.append(f"SCI一区余量:现有 {sci_q1} 篇 / 要求 {min_sci} 篇")
if not reasons: if not reasons:
reasons.append("与赛道方向存在部分相关性") reasons.append("现有研究标签与该赛道需求标签暂无重合,补充研究标签后可能改善匹配")
if match_pct >= _STRONG_THRESHOLD: if match_pct >= _STRONG_THRESHOLD:
tier = TIER_STRONG tier = TIER_STRONG
@@ -346,14 +355,17 @@ def run_match(
def load_tracks(year: Optional[int] = None) -> list[dict[str, Any]]: def load_tracks(year: Optional[int] = None) -> list[dict[str, Any]]:
"""读取赛道库(缺省全部年度;可按年度筛选)。""" """读取赛道库(缺省取赛道库最新年度,与 actions.json 契约一致;显式传 year 按年度筛选)。"""
conn = get_conn() conn = get_conn()
try: try:
cur = conn.cursor() cur = conn.cursor()
if year: if year is None:
cur.execute("SELECT * FROM guide_tracks WHERE year = ? ORDER BY id", (year,)) cur.execute("SELECT MAX(year) FROM guide_tracks")
else: row = cur.fetchone()
cur.execute("SELECT * FROM guide_tracks ORDER BY year DESC, id") 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] columns = [d[0] for d in cur.description]
rows = [dict(zip(columns, row)) for row in cur.fetchall()] rows = [dict(zip(columns, row)) for row in cur.fetchall()]
finally: finally:
@@ -361,6 +373,35 @@ def load_tracks(year: Optional[int] = None) -> list[dict[str, Any]]:
return rows 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]]: def load_teachers(teacher_name: str = "") -> list[dict[str, Any]]:
"""读取教师画像(可按姓名精确筛选)。""" """读取教师画像(可按姓名精确筛选)。"""
conn = get_conn() conn = get_conn()

View File

@@ -478,7 +478,10 @@ def cmd_match(*, teacher: str = "", year: Optional[int] = None) -> int:
print(f"ERROR:TEACHER_NOT_FOUND {message}") print(f"ERROR:TEACHER_NOT_FOUND {message}")
return 1 return 1
if not tracks: 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") 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") finish(status="failed", message=message, skill=SKILL_SLUG, error_code="TRACK_EMPTY", stage="match")
print(f"ERROR:TRACK_EMPTY {message}") print(f"ERROR:TRACK_EMPTY {message}")
@@ -529,9 +532,13 @@ def cmd_match(*, teacher: str = "", year: Optional[int] = None) -> int:
def cmd_match_batch_input() -> int: def cmd_match_batch_input() -> int:
"""批量匹配(async Action 入口):匹配 input/ 目录中的教师画像 JSON 与库内赛道。""" """批量匹配(async Action 入口):匹配 input/ 目录中的教师画像 JSON 与库内赛道。
坏文件(解析失败/非对象数组)跳过并告警,不中断整批;画像不完整教师同样跳过并记录(E1)。
缺省匹配赛道库最新年度赛道。
"""
log = _get_task_logger() 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 from util.runtime_paths import get_input_dir
with job_context(skill=SKILL_SLUG): with job_context(skill=SKILL_SLUG):
@@ -546,26 +553,17 @@ def cmd_match_batch_input() -> int:
return 1 return 1
input_dir = get_input_dir() input_dir = get_input_dir()
files = sorted( teachers, skipped_files = collect_input_teachers(input_dir)
os.path.join(input_dir, name) for entry in skipped_files:
for name in os.listdir(input_dir) log.warning("batch_input_file_skipped file=%s reason=%s stage=batch", entry["file"], entry["reason"])
if name.lower().endswith(".json") emit(f"{entry['file']}:{entry['reason']},该文件已跳过", type="warn", skill=SKILL_SLUG, stage="batch")
) if not teachers:
if not files: message = f"input 目录没有可用的教师画像 JSON({len(skipped_files)} 个文件无法解析):{input_dir}"
message = f"input 目录没有教师画像 JSON:{input_dir}"
log.warning("task_failed task_type=match_batch error_code=TEACHER_NOT_FOUND stage=batch") 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") finish(status="failed", message=message, skill=SKILL_SLUG, error_code="TEACHER_NOT_FOUND", stage="batch")
print(f"ERROR:TEACHER_NOT_FOUND {message}") print(f"ERROR:TEACHER_NOT_FOUND {message}")
return 1 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() tracks = load_tracks()
if not tracks: if not tracks:
message = "指南赛道库为空:请先录入赛道(add-track / import-tracks)" message = "指南赛道库为空:请先录入赛道(add-track / import-tracks)"
@@ -597,6 +595,7 @@ def cmd_match_batch_input() -> int:
"batch_no": result["batch_no"], "batch_no": result["batch_no"],
"count": result["count"], "count": result["count"],
"skipped": result.get("skipped") or [], "skipped": result.get("skipped") or [],
"skipped_files": skipped_files,
"tier_counts": result["tier_counts"], "tier_counts": result["tier_counts"],
"brief_file": result["brief_file"], "brief_file": result["brief_file"],
} }

View File

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

View File

@@ -2,6 +2,9 @@
"""赛道匹配内核金标准用例(离线、确定性)。""" """赛道匹配内核金标准用例(离线、确定性)。"""
from __future__ import annotations from __future__ import annotations
import json
import os
import tempfile
import unittest import unittest
from _support import IsolatedDataRoot from _support import IsolatedDataRoot
@@ -80,13 +83,29 @@ class TestScoring(MatchRuleTestBase):
self.assertEqual(top["match_pct"], 100) self.assertEqual(top["match_pct"], 100)
self.assertEqual(top["tier"], "强烈推荐") 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="")) result = self._match(_teacher(researchTags="钙钛矿,光伏", recentPapers=""))
entry = next(e for e in result["top3"] if e["track"] == "面上项目·光伏材料") 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["match_pct"], 47)
self.assertEqual(entry["tier"], "不建议") 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: def test_t1_strong_boundary(self) -> None:
# 恰好 80% 属强烈推荐 # 恰好 80% 属强烈推荐
result = self._match(_teacher(researchTags="钙钛矿,光伏,半导体", recentPapers="")) result = self._match(_teacher(researchTags="钙钛矿,光伏,半导体", recentPapers=""))
@@ -210,5 +229,77 @@ class TestBatchSkipIncomplete(MatchRuleTestBase):
run_match([_teacher(researchTags="", recentPapers="")], _tracks(), "2027", skip_incomplete=False) 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__": if __name__ == "__main__":
unittest.main() unittest.main()