"""CLI：argparse 与分发（项目经费进度播报助手）。 只负责参数解析并转交 service；业务内核在 fund_service / project_service， 勿在此按入口复制逻辑。 """ from __future__ import annotations import argparse import sys from typing import List, Optional from service.task_service import ( cmd_add_project, cmd_broadcast, cmd_broadcast_monthly, cmd_config_path, cmd_health, cmd_import_projects, cmd_init_db, cmd_list_projects, cmd_list_reports, cmd_render_html, cmd_run, cmd_set_rule, cmd_show_report, cmd_stats, cmd_version, cmd_view_rules, ) 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_float_or_none(raw: Optional[str]) -> Optional[float]: raw = _cli_str_or_none(raw) if raw is None: return None try: return 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] pick_t = _cli_str_or_none(getattr(args, "target", None)) pick_i = _cli_str_or_none(getattr(args, "input_id", None)) target = pick_t or t_target input_id = pick_i or t_input return cmd_run(target=target, input_id=input_id) def _handle_add_project(args: argparse.Namespace) -> int: return cmd_add_project( project_no=str(args.project_no or "").strip(), name=_cli_str_or_none(args.name) or "", project_type=_cli_str_or_none(args.project_type) or "", pi=_cli_str_or_none(args.pi) or "", department=_cli_str_or_none(args.department) or "", total_approved=_cli_float_or_none(args.total_approved), total_received=_cli_float_or_none(args.total_received), total_spent=_cli_float_or_none(args.total_spent), frozen=_cli_float_or_none(args.frozen), frozen_since=_cli_str_or_none(args.frozen_since) or "", start=_cli_str_or_none(args.start) or "", end=_cli_str_or_none(args.end) or "", status=_cli_str_or_none(args.status) or "", remark=_cli_str_or_none(args.remark) or "", ) def _handle_broadcast(args: argparse.Namespace) -> int: return cmd_broadcast( project_no=_cli_str_or_none(args.project_no) or "", pi=_cli_str_or_none(args.pi) or "", as_of=_cli_str_or_none(args.as_of) or "", include_all_status=bool(args.all_status), ) def _handle_list_reports(args: argparse.Namespace) -> int: return cmd_list_reports(limit=int(args.limit or 20), status_tag=_cli_str_or_none(args.status_tag) or "") 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) def _handle_set_rule(args: argparse.Namespace) -> int: return cmd_set_rule(rule_key=str(args.rule_key or "").strip(), value=float(args.value)) def _print_full_usage() -> None: print("项目经费进度播报助手（main.py）可用命令：") print(" python main.py add-project --project-no X --name Y [--type 纵向] [--pi 张三] [--department 学院]") print(" [--total-approved N] [--total-received N] [--total-spent N]") print(" [--frozen N] [--frozen-since DATE] [--start DATE] [--end DATE] [--status 在研]") print(" python main.py import-projects --projects-json # 批量 upsert 台账") print(" python main.py list-projects [--pi 张三] [--status 在研]") print(" python main.py broadcast [--project-no X] [--pi 张三] [--as-of YYYY-MM-DD] [--all-status]") print(" python main.py broadcast-monthly # 月度播报（定时 Action 入口）") print(" python main.py list-reports [--limit N] [--status-tag 严重滞后]") print(" python main.py show-report --id N") print(" python main.py render-html [--report-id N]") print(" python main.py view-rules") print(" python main.py set-rule --rule-key lag_threshold --value 0.15") print(" python main.py stats") print(" python main.py run [target [input_id]] [-t target] [-i input_id]") print(" python main.py health") print(" python main.py config-path") print(" python main.py version") print(" python main.py init-db") def build_parser() -> ZhArgumentParser: p = ZhArgumentParser( prog="main.py", description="项目经费进度播报助手：按台账计算实际支出率与序时进度，识别严重滞后/支出异常偏快，生成三段式播报与《经费执行简报》。", formatter_class=argparse.RawDescriptionHelpFormatter, ) sub = p.add_subparsers(dest="cmd", required=True, parser_class=ZhArgumentParser) # add-project sp = sub.add_parser("add-project", help="录入/更新项目台账（按项目编号幂等）") sp.add_argument("--project-no", required=True, metavar="项目编号", dest="project_no") sp.add_argument("--name", default=None, metavar="项目名称") sp.add_argument("--type", default=None, metavar="项目类别(纵向/横向/其他)", dest="project_type") sp.add_argument("--pi", default=None, metavar="负责人") sp.add_argument("--department", default=None, metavar="所属学院") sp.add_argument("--total-approved", default=None, metavar="总立项经费(元)", dest="total_approved") sp.add_argument("--total-received", default=None, metavar="已到账(元)", dest="total_received") sp.add_argument("--total-spent", default=None, metavar="已报销(元)", dest="total_spent") sp.add_argument("--frozen", default=None, metavar="暂付款(元)") sp.add_argument("--frozen-since", default=None, metavar="暂付款最早发生日", dest="frozen_since") sp.add_argument("--start", default=None, metavar="执行期开始 YYYY-MM-DD") sp.add_argument("--end", default=None, metavar="执行期结束 YYYY-MM-DD") sp.add_argument("--status", default=None, metavar="状态(在研/已结题/暂缓)") sp.add_argument("--remark", default=None, metavar="备注") sp.set_defaults(handler=_handle_add_project) # import-projects sp = sub.add_parser("import-projects", help="批量导入项目台账（JSON 文件/内联/stdin）") sp.add_argument("--projects-json", required=True, metavar="JSON(file|-)", dest="projects_json") sp.set_defaults(handler=lambda a: cmd_import_projects(str(a.projects_json))) # list-projects sp = sub.add_parser("list-projects", help="查询项目台账") sp.add_argument("--pi", default=None, metavar="负责人筛选") sp.add_argument("--status", default=None, metavar="状态筛选") sp.set_defaults(handler=lambda a: cmd_list_projects(pi=_cli_str_or_none(a.pi) or "", status=_cli_str_or_none(a.status) or "")) # broadcast sp = sub.add_parser("broadcast", help="经费进度播报（主命令，缺省全部在研项目）") sp.add_argument("--project-no", default=None, metavar="指定项目编号", dest="project_no") sp.add_argument("--pi", default=None, metavar="按负责人筛选") sp.add_argument("--as-of", default=None, metavar="评估基准日 YYYY-MM-DD", dest="as_of") sp.add_argument("--all-status", action="store_true", help="包含非在研项目") sp.set_defaults(handler=_handle_broadcast) # broadcast-monthly sp = sub.add_parser("broadcast-monthly", help="月度播报（async Action 入口，含汇总简报）") sp.set_defaults(handler=lambda _a: cmd_broadcast_monthly()) # list-reports sp = sub.add_parser("list-reports", help="查询播报历史") sp.add_argument("--limit", default=20, metavar="条数") sp.add_argument("--status-tag", default=None, metavar="状态筛选", dest="status_tag") sp.set_defaults(handler=_handle_list_reports) # show-report 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))) # render-html sp = sub.add_parser("render-html", help="重建《经费执行简报》HTML（缺省最近 10 份）") sp.add_argument("--report-id", default=None, metavar="记录编号", dest="report_id") sp.set_defaults(handler=_handle_render_html) # rules sp = sub.add_parser("view-rules", help="查看预警阈值") sp.set_defaults(handler=lambda _a: cmd_view_rules()) sp = sub.add_parser("set-rule", help="更新预警阈值") sp.add_argument("--rule-key", required=True, metavar="规则键", dest="rule_key") sp.add_argument("--value", required=True, type=float, metavar="新阈值") sp.set_defaults(handler=_handle_set_rule) # stats 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))