278 lines
12 KiB
Python
278 lines
12 KiB
Python
"""CLI：argparse 与分发（票据合规预审助手）。
|
||
|
||
只负责参数解析并转交 service；业务内核在 prereview_service / standards_service，
|
||
勿在此按入口复制逻辑。
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import sys
|
||
from typing import List, Optional
|
||
|
||
from service.task_service import (
|
||
cmd_add_whitelist,
|
||
cmd_config_path,
|
||
cmd_health,
|
||
cmd_init_db,
|
||
cmd_list_holidays,
|
||
cmd_list_reports,
|
||
cmd_list_whitelist,
|
||
cmd_mark_verified,
|
||
cmd_prereview,
|
||
cmd_prereview_batch_input,
|
||
cmd_remove_whitelist,
|
||
cmd_render_html,
|
||
cmd_run,
|
||
cmd_set_standard,
|
||
cmd_show_report,
|
||
cmd_stats,
|
||
cmd_version,
|
||
cmd_view_standards,
|
||
)
|
||
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 _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_prereview(args: argparse.Namespace) -> int:
|
||
return cmd_prereview(
|
||
invoice_json=str(args.invoice_json or "").strip(),
|
||
purpose=_cli_str_or_none(args.purpose) or "",
|
||
project_no=_cli_str_or_none(args.project_no) or "",
|
||
trip_start=_cli_str_or_none(args.trip_start) or "",
|
||
trip_end=_cli_str_or_none(args.trip_end) or "",
|
||
trip_city=_cli_str_or_none(args.trip_city) or "",
|
||
staff_title=_cli_str_or_none(args.staff_title) 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_list_reports(args: argparse.Namespace) -> int:
|
||
limit = int(args.limit or 20)
|
||
return cmd_list_reports(limit=limit, verdict=_cli_str_or_none(args.verdict))
|
||
|
||
|
||
def _handle_show_report(args: argparse.Namespace) -> int:
|
||
return cmd_show_report(report_id=int(args.report_id))
|
||
|
||
|
||
def _handle_mark_verified(args: argparse.Namespace) -> int:
|
||
return cmd_mark_verified(
|
||
report_id=int(args.report_id),
|
||
result=str(args.result or "").strip(),
|
||
note=_cli_str_or_none(args.note) or "",
|
||
)
|
||
|
||
|
||
def _handle_view_standards(args: argparse.Namespace) -> int:
|
||
return cmd_view_standards(category=_cli_str_or_none(args.category))
|
||
|
||
|
||
def _handle_set_standard(args: argparse.Namespace) -> int:
|
||
return cmd_set_standard(
|
||
category=str(args.category or "").strip(),
|
||
city_tier=str(args.city_tier or "").strip(),
|
||
staff_title=str(args.staff_title or "").strip(),
|
||
max_amount=float(args.max_amount),
|
||
unit=_cli_str_or_none(args.unit) or "",
|
||
basis_doc=_cli_str_or_none(args.basis_doc) or "",
|
||
clause=_cli_str_or_none(args.clause) or "",
|
||
city_example=_cli_str_or_none(args.city_example) or "",
|
||
)
|
||
|
||
|
||
def _handle_add_whitelist(args: argparse.Namespace) -> int:
|
||
return cmd_add_whitelist(
|
||
tax_no=str(args.tax_no or "").strip(),
|
||
org_name=str(args.org_name or "").strip(),
|
||
note=_cli_str_or_none(args.note) or "",
|
||
)
|
||
|
||
|
||
def _handle_remove_whitelist(args: argparse.Namespace) -> int:
|
||
return cmd_remove_whitelist(tax_no=str(args.tax_no or "").strip())
|
||
|
||
|
||
def _handle_list_holidays(args: argparse.Namespace) -> int:
|
||
return cmd_list_holidays(year=_cli_str_or_none(args.year))
|
||
|
||
|
||
def _print_full_usage() -> None:
|
||
print("票据合规预审助手（main.py）可用命令：")
|
||
print(" python main.py prereview --invoice-json <file|-> [--purpose 文本] [--project-no 编号]")
|
||
print(" [--trip-start YYYY-MM-DD] [--trip-end YYYY-MM-DD]")
|
||
print(" [--trip-city 城市] [--staff-title 职级]")
|
||
print(" python main.py prereview-batch-input # 批量预审 input/ 目录 JSON")
|
||
print(" python main.py render-html [--report-id N] # 重建初审意见单（缺省最近 10 份）")
|
||
print(" python main.py list-reports [--limit N] [--verdict 通过|待确认|驳回]")
|
||
print(" python main.py show-report --id N")
|
||
print(" python main.py mark-verified --report-id N --result pass|fail [--note 备注]")
|
||
print(" python main.py view-standards [--category 住宿费|交通费]")
|
||
print(" python main.py set-standard --category C --city-tier T --staff-title J --max-amount N")
|
||
print(" python main.py list-whitelist")
|
||
print(" python main.py add-whitelist --tax-no X --name Y [--note 备注]")
|
||
print(" python main.py remove-whitelist --tax-no X")
|
||
print(" python main.py list-holidays [--year YYYY]")
|
||
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)
|
||
|
||
# prereview
|
||
sp = sub.add_parser("prereview", help="票据合规预审（主命令）")
|
||
sp.add_argument("--invoice-json", required=True, metavar="发票JSON(文件|内联|-)", dest="invoice_json")
|
||
sp.add_argument("--purpose", default="", metavar="报销事由")
|
||
sp.add_argument("--project-no", default="", metavar="预算项目编号", dest="project_no")
|
||
sp.add_argument("--trip-start", default="", metavar="出差开始 YYYY-MM-DD", dest="trip_start")
|
||
sp.add_argument("--trip-end", default="", metavar="出差结束 YYYY-MM-DD", dest="trip_end")
|
||
sp.add_argument("--trip-city", default="", metavar="目的地城市", dest="trip_city")
|
||
sp.add_argument("--staff-title", default="", metavar="职级(副高及以上/中级及以下)", dest="staff_title")
|
||
sp.set_defaults(handler=_handle_prereview)
|
||
|
||
# prereview-batch-input
|
||
sp = sub.add_parser("prereview-batch-input", help="批量预审 input/ 目录下的发票 JSON 文件")
|
||
sp.set_defaults(handler=lambda _a: cmd_prereview_batch_input())
|
||
|
||
# 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)
|
||
|
||
# list-reports
|
||
sp = sub.add_parser("list-reports", help="查询初审历史")
|
||
sp.add_argument("--limit", default=20, metavar="条数")
|
||
sp.add_argument("--verdict", default=None, metavar="结论筛选")
|
||
sp.set_defaults(handler=_handle_list_reports)
|
||
|
||
# show-report
|
||
sp = sub.add_parser("show-report", help="查看一份初审报告详情")
|
||
sp.add_argument("--report-id", required=True, metavar="报告编号", dest="report_id")
|
||
sp.set_defaults(handler=_handle_show_report)
|
||
|
||
# mark-verified
|
||
sp = sub.add_parser("mark-verified", help="回填人工查验结果（pass/fail）")
|
||
sp.add_argument("--report-id", required=True, metavar="报告编号", dest="report_id")
|
||
sp.add_argument("--result", required=True, metavar="pass|fail")
|
||
sp.add_argument("--note", default=None, metavar="查验备注")
|
||
sp.set_defaults(handler=_handle_mark_verified)
|
||
|
||
# view-standards
|
||
sp = sub.add_parser("view-standards", help="查看经费标准库")
|
||
sp.add_argument("--category", default=None, metavar="费用类别")
|
||
sp.set_defaults(handler=_handle_view_standards)
|
||
|
||
# set-standard
|
||
sp = sub.add_parser("set-standard", help="新增/更新经费标准")
|
||
sp.add_argument("--category", required=True, metavar="费用类别")
|
||
sp.add_argument("--city-tier", required=True, metavar="城市等级", dest="city_tier")
|
||
sp.add_argument("--staff-title", required=True, metavar="职级", dest="staff_title")
|
||
sp.add_argument("--max-amount", required=True, type=float, metavar="标准值", dest="max_amount")
|
||
sp.add_argument("--unit", default=None, metavar="单位")
|
||
sp.add_argument("--basis-doc", default=None, metavar="依据文件", dest="basis_doc")
|
||
sp.add_argument("--clause", default=None, metavar="条款")
|
||
sp.add_argument("--city-example", default=None, metavar="城市示例(逗号分隔)", dest="city_example")
|
||
sp.set_defaults(handler=_handle_set_standard)
|
||
|
||
# whitelist
|
||
sp = sub.add_parser("list-whitelist", help="查看税号白名单")
|
||
sp.set_defaults(handler=lambda _a: cmd_list_whitelist())
|
||
|
||
sp = sub.add_parser("add-whitelist", help="新增税号白名单")
|
||
sp.add_argument("--tax-no", required=True, metavar="税号", dest="tax_no")
|
||
sp.add_argument("--name", required=True, metavar="单位名称", dest="org_name")
|
||
sp.add_argument("--note", default=None, metavar="备注")
|
||
sp.set_defaults(handler=_handle_add_whitelist)
|
||
|
||
sp = sub.add_parser("remove-whitelist", help="删除税号白名单")
|
||
sp.add_argument("--tax-no", required=True, metavar="税号", dest="tax_no")
|
||
sp.set_defaults(handler=_handle_remove_whitelist)
|
||
|
||
# holidays
|
||
sp = sub.add_parser("list-holidays", help="查看节假日表")
|
||
sp.add_argument("--year", default=None, metavar="年份")
|
||
sp.set_defaults(handler=_handle_list_holidays)
|
||
|
||
# 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))
|