"""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 ") 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 ") 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))