250 lines
11 KiB
Python
250 lines
11 KiB
Python
"""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 <file|-> # 批量 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))
|