4a8613b27e
- 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 同步表述
237 lines
10 KiB
Python
237 lines
10 KiB
Python
"""CLI：argparse 与分发（申报赛道推荐助手）。
|
||
|
||
只负责参数解析并转交 service；业务内核在 match_service / profile_service，
|
||
勿在此按入口复制逻辑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from typing import List, Optional
|
||
|
||
from service.task_service import (
|
||
cmd_add_teacher,
|
||
cmd_add_track,
|
||
cmd_config_path,
|
||
cmd_health,
|
||
cmd_import_teachers,
|
||
cmd_import_tracks,
|
||
cmd_init_db,
|
||
cmd_list_reports,
|
||
cmd_list_teachers,
|
||
cmd_list_tracks,
|
||
cmd_match,
|
||
cmd_match_batch_input,
|
||
cmd_render_html,
|
||
cmd_run,
|
||
cmd_show_report,
|
||
cmd_stats,
|
||
cmd_version,
|
||
)
|
||
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_add_teacher(args: argparse.Namespace) -> int:
|
||
return cmd_add_teacher(
|
||
name=str(args.name or "").strip(),
|
||
department=_cli_str_or_none(args.department) or "",
|
||
title=_cli_str_or_none(args.title) or "",
|
||
tags=_cli_str_or_none(args.tags) or "",
|
||
papers=_cli_str_or_none(args.papers) or "",
|
||
projects=_cli_str_or_none(args.projects) or "",
|
||
sci_q1=_cli_int_or_none(args.sci_q1),
|
||
core_papers=_cli_int_or_none(args.core_papers),
|
||
)
|
||
|
||
|
||
def _handle_add_track(args: argparse.Namespace) -> int:
|
||
return cmd_add_track(
|
||
year=int(args.year),
|
||
track_name=str(args.track_name or "").strip(),
|
||
program=_cli_str_or_none(args.program) or "",
|
||
agency=_cli_str_or_none(args.agency) or "",
|
||
tags=_cli_str_or_none(args.tags) or "",
|
||
priority=_cli_str_or_none(args.priority) or "",
|
||
min_title=_cli_str_or_none(args.min_title) or "",
|
||
min_sci_q1=_cli_int_or_none(args.min_sci_q1),
|
||
quota=_cli_int_or_none(args.quota),
|
||
source=_cli_str_or_none(args.source) or "",
|
||
)
|
||
|
||
|
||
def _handle_match(args: argparse.Namespace) -> int:
|
||
return cmd_match(teacher=_cli_str_or_none(args.teacher) or "", year=_cli_int_or_none(args.year))
|
||
|
||
|
||
def _handle_list_reports(args: argparse.Namespace) -> int:
|
||
return cmd_list_reports(limit=int(args.limit or 20))
|
||
|
||
|
||
def _handle_render_html(args: argparse.Namespace) -> int:
|
||
# 宿主行上下文可能给空串或未解析占位符（如 {{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:
|
||
print("申报赛道推荐助手（main.py）可用命令：")
|
||
print(" python main.py add-teacher --name X [--department 学院] [--title 副教授] [--tags 标签1,标签2]")
|
||
print(" [--papers 论文1;论文2] [--projects 项目1;项目2] [--sci-q1 N] [--core-papers N]")
|
||
print(" python main.py import-teachers --teachers-json <file|->")
|
||
print(" python main.py add-track --year YYYY --track-name X --track-tags 标签1,标签2 [--program 面上项目]")
|
||
print(" [--agency 数理学部] [--priority 关键词1,关键词2] [--min-title 副高及以上] [--min-sci-q1 N] [--quota N]")
|
||
print(" python main.py import-tracks --tracks-json <file|->")
|
||
print(" python main.py list-teachers [--department 学院] | list-tracks [--year YYYY]")
|
||
print(" python main.py match [--teacher X] [--year YYYY] # 主命令：Top 3 推荐 + 建议书")
|
||
print(" python main.py match-batch-input # 批量匹配 input/ 目录画像")
|
||
print(" python main.py list-reports [--limit N] | show-report --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="申报赛道推荐助手：维护教师学术画像与指南赛道库，标签匹配 + 硬门槛过滤，生成可解释的 Top 3 赛道推荐与《申报策略建议书》。",
|
||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||
)
|
||
sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser)
|
||
|
||
sp = sub.add_parser("add-teacher", help="录入/更新教师画像（按姓名幂等）")
|
||
sp.add_argument("--name", required=True, metavar="教师姓名")
|
||
sp.add_argument("--department", default=None, metavar="所属学院")
|
||
sp.add_argument("--title", default=None, metavar="职称")
|
||
sp.add_argument("--tags", default=None, metavar="研究标签(逗号分隔)")
|
||
sp.add_argument("--papers", default=None, metavar="近3年论文(分号分隔)")
|
||
sp.add_argument("--projects", default=None, metavar="项目经历(分号分隔)")
|
||
sp.add_argument("--sci-q1", default=None, metavar="SCI一区篇数", dest="sci_q1")
|
||
sp.add_argument("--core-papers", default=None, metavar="核心论文篇数", dest="core_papers")
|
||
sp.set_defaults(handler=_handle_add_teacher)
|
||
|
||
sp = sub.add_parser("import-teachers", help="批量导入教师画像")
|
||
sp.add_argument("--teachers-json", required=True, metavar="JSON(file|-)", dest="teachers_json")
|
||
sp.set_defaults(handler=lambda a: cmd_import_teachers(str(a.teachers_json)))
|
||
|
||
sp = sub.add_parser("add-track", help="录入/更新指南赛道")
|
||
sp.add_argument("--year", required=True, metavar="指南年度")
|
||
sp.add_argument("--track-name", required=True, metavar="赛道名称", dest="track_name")
|
||
sp.add_argument("--program", default=None, metavar="项目类型")
|
||
sp.add_argument("--agency", default=None, metavar="学部/归口")
|
||
sp.add_argument("--tags", default=None, metavar="需求标签(逗号分隔)")
|
||
sp.add_argument("--priority", default=None, metavar="优先方向关键词(逗号分隔)")
|
||
sp.add_argument("--min-title", default=None, metavar="职称门槛", dest="min_title")
|
||
sp.add_argument("--min-sci-q1", default=None, metavar="SCI一区门槛", dest="min_sci_q1")
|
||
sp.add_argument("--quota", default=None, metavar="拟资助数")
|
||
sp.add_argument("--source", default=None, metavar="指南来源")
|
||
sp.set_defaults(handler=_handle_add_track)
|
||
|
||
sp = sub.add_parser("import-tracks", help="批量导入指南赛道")
|
||
sp.add_argument("--tracks-json", required=True, metavar="JSON(file|-)", dest="tracks_json")
|
||
sp.set_defaults(handler=lambda a: cmd_import_tracks(str(a.tracks_json)))
|
||
|
||
sp = sub.add_parser("list-teachers", help="查询教师画像")
|
||
sp.add_argument("--department", default=None, metavar="学院筛选")
|
||
sp.set_defaults(handler=lambda a: cmd_list_teachers(department=_cli_str_or_none(a.department) or ""))
|
||
|
||
sp = sub.add_parser("list-tracks", help="查询指南赛道")
|
||
sp.add_argument("--year", default=None, metavar="年度筛选")
|
||
sp.set_defaults(handler=lambda a: cmd_list_tracks(year=_cli_int_or_none(a.year)))
|
||
|
||
sp = sub.add_parser("match", help="申报赛道匹配（主命令，缺省全部教师）")
|
||
sp.add_argument("--teacher", default=None, metavar="指定教师姓名")
|
||
sp.add_argument("--year", default=None, metavar="指南年度")
|
||
sp.set_defaults(handler=_handle_match)
|
||
|
||
sp = sub.add_parser("match-batch-input", help="批量匹配 input/ 目录教师画像")
|
||
sp.set_defaults(handler=lambda _a: cmd_match_batch_input())
|
||
|
||
sp = sub.add_parser("list-reports", help="查询推荐报告历史")
|
||
sp.add_argument("--limit", default=20, metavar="条数")
|
||
sp.set_defaults(handler=_handle_list_reports)
|
||
|
||
sp = sub.add_parser("show-report", help="查看一份推荐报告详情")
|
||
sp.add_argument("--id", required=True, type=int, metavar="报告编号", dest="report_id")
|
||
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, nargs="?", 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))
|