"""CLI 入口:argparse 装配与分发(Controller)。""" from __future__ import annotations import argparse import os import sys from typing import List, Optional from content_manager.services import article_service, image_service, video_service from content_manager.services.article_service import resolve_publish_platform from content_manager.util.argparse_zh import ZhArgumentParser def _print_root_usage_zh() -> None: print( """内容管理:请指定资源类型子命令。 python main.py article list python main.py article get 1 python main.py article add --title "标题" --body-file 文章.md python main.py article generate 豆包 搜狐号 RPA降本增效 python main.py image add --file D:\\\\a.png [--title "说明"] python main.py video add --file D:\\\\a.mp4 [--title "说明"] [--duration-ms 120000] 查看完整说明:python main.py -h""" ) def _handle_article_add(args: argparse.Namespace) -> None: if args.body_file: fp = os.path.abspath(args.body_file) try: with open(fp, encoding="utf-8") as f: body = f.read() except OSError as e: print(f"❌ 无法读取正文文件:{fp}\n原因:{e}") sys.exit(1) else: body = args.body or "" article_service.cmd_add(args.title, body, source="manual") def _handle_article_import(args: argparse.Namespace) -> None: article_service.cmd_import_json(args.path) def _handle_article_generate(args: argparse.Namespace) -> None: raw_parts = [str(x).strip() for x in (args.generate_args or []) if str(x).strip()] if not raw_parts: print("❌ 缺少主题或关键词。") print("示例:python main.py article generate 豆包 搜狐号 RPA降本增效") sys.exit(1) platform_guess = resolve_publish_platform(raw_parts[0]) if platform_guess and len(raw_parts) == 1: print("❌ 缺少主题或关键词。") sys.exit(1) if platform_guess and len(raw_parts) >= 2: publish_platform = platform_guess topic = " ".join(raw_parts[1:]).strip() else: publish_platform = "common" topic = " ".join(raw_parts).strip() if not topic: print("❌ 主题或关键词不能为空。") sys.exit(1) article_service.cmd_generate( args.llm_target, topic, publish_platform=publish_platform, title=getattr(args, "title", None), ) def _handle_article_feedback(args: argparse.Namespace) -> None: article_service.cmd_feedback(args.article_id, args.status, args.account_id, args.error_msg) def _handle_article_save_legacy(args: argparse.Namespace) -> None: article_service.cmd_save(args.legacy_id, args.legacy_title, args.legacy_content) def _handle_image_add(args: argparse.Namespace) -> None: image_service.cmd_add(args.file, title=getattr(args, "title", None)) def _handle_image_feedback(args: argparse.Namespace) -> None: image_service.cmd_feedback(args.image_id, args.status, args.account_id, args.error_msg) def _handle_video_add(args: argparse.Namespace) -> None: video_service.cmd_add( args.file, title=getattr(args, "title", None), duration_ms=getattr(args, "duration_ms", None), ) def _handle_video_feedback(args: argparse.Namespace) -> None: video_service.cmd_feedback(args.video_id, args.status, args.account_id, args.error_msg) def build_parser() -> ZhArgumentParser: fmt = argparse.RawDescriptionHelpFormatter p = ZhArgumentParser( prog="main.py", description="内容管理:文章(正文在库内)与图片/视频(文件在数据目录,库内仅存路径)。", epilog="示例见各子命令 -h;一级分组:article / image / video", formatter_class=fmt, ) sub = p.add_subparsers( dest="resource", required=True, metavar="资源类型", help="article 文章 | image 图片 | video 视频", parser_class=ZhArgumentParser, ) # ----- article ----- art = sub.add_parser("article", help="文章:正文与元数据在 SQLite", formatter_class=fmt) art_sub = art.add_subparsers( dest="article_cmd", required=True, metavar="子命令", parser_class=ZhArgumentParser, ) sp = art_sub.add_parser("list", help="列出文章") sp.add_argument("--limit", type=int, default=10) sp.add_argument("--max-chars", type=int, default=50) sp.set_defaults(handler=lambda a: article_service.cmd_list(limit=a.limit, max_chars=a.max_chars)) sp = art_sub.add_parser("get", help="按 id 输出 JSON") sp.add_argument("article_id", metavar="文章id") sp.set_defaults(handler=lambda a: article_service.cmd_get(a.article_id)) sp = art_sub.add_parser( "add", help="新增文章", formatter_class=fmt, epilog="示例:python main.py article add --title \"标题\" --body \"正文\"", ) sp.add_argument("--title", required=True) g = sp.add_mutually_exclusive_group(required=True) g.add_argument("--body-file", metavar="路径") g.add_argument("--body", metavar="正文") sp.set_defaults(handler=_handle_article_add) sp = art_sub.add_parser("import-json", help="从 JSON 批量导入") sp.add_argument("path", metavar="JSON路径") sp.set_defaults(handler=_handle_article_import) sp = art_sub.add_parser("generate", help="调用 llm-manager 生成并入库", formatter_class=fmt) sp.add_argument("llm_target", metavar="大模型目标") sp.add_argument("generate_args", nargs="+", metavar="生成参数") sp.add_argument("--title", metavar="标题", default=None) sp.set_defaults(handler=_handle_article_generate) sp = art_sub.add_parser("prompt-list", help="查看提示词模板") sp.add_argument("platform", nargs="?", default=None, metavar="发布平台") sp.add_argument("--limit", type=int, default=30) sp.set_defaults(handler=lambda a: article_service.cmd_prompt_list(a.platform, a.limit)) sp = art_sub.add_parser("delete", help="删除文章") sp.add_argument("article_id", metavar="文章id") sp.set_defaults(handler=lambda a: article_service.cmd_delete(a.article_id)) sp = art_sub.add_parser("feedback", help="回写发布状态", formatter_class=fmt) sp.add_argument("article_id", metavar="文章id") sp.add_argument("status", metavar="状态") sp.add_argument("account_id", nargs="?", default=None, metavar="账号") sp.add_argument("error_msg", nargs="?", default=None, metavar="错误说明") sp.set_defaults(handler=_handle_article_feedback) sp = art_sub.add_parser("save", help="旧版单行正文保存", formatter_class=fmt) sp.add_argument("legacy_id", metavar="id") sp.add_argument("legacy_title", metavar="标题") sp.add_argument("legacy_content", metavar="正文一行") sp.set_defaults(handler=_handle_article_save_legacy) # ----- image ----- img = sub.add_parser("image", help="图片:文件在数据目录,images 表存相对路径", formatter_class=fmt) img_sub = img.add_subparsers( dest="image_cmd", required=True, metavar="子命令", parser_class=ZhArgumentParser, ) sp = img_sub.add_parser("list", help="列出图片") sp.add_argument("--limit", type=int, default=20) sp.add_argument("--max-chars", type=int, default=80) sp.set_defaults(handler=lambda a: image_service.cmd_list(limit=a.limit, max_chars=a.max_chars)) sp = img_sub.add_parser("get", help="按 id 输出 JSON(含 absolute_path)") sp.add_argument("image_id", metavar="图片id") sp.set_defaults(handler=lambda a: image_service.cmd_get(a.image_id)) sp = img_sub.add_parser("add", help="从本地文件复制入库", formatter_class=fmt) sp.add_argument("--file", required=True, metavar="文件", help="源图片路径") sp.add_argument("--title", default=None, metavar="标题", help="可选说明") sp.set_defaults(handler=_handle_image_add) sp = img_sub.add_parser("delete", help="删除记录与磁盘目录") sp.add_argument("image_id", metavar="图片id") sp.set_defaults(handler=lambda a: image_service.cmd_delete(a.image_id)) sp = img_sub.add_parser("feedback", help="回写状态", formatter_class=fmt) sp.add_argument("image_id", metavar="图片id") sp.add_argument("status", metavar="状态") sp.add_argument("account_id", nargs="?", default=None, metavar="账号") sp.add_argument("error_msg", nargs="?", default=None, metavar="错误说明") sp.set_defaults(handler=_handle_image_feedback) # ----- video ----- vid = sub.add_parser("video", help="视频:文件在数据目录,videos 表存相对路径", formatter_class=fmt) vid_sub = vid.add_subparsers( dest="video_cmd", required=True, metavar="子命令", parser_class=ZhArgumentParser, ) sp = vid_sub.add_parser("list", help="列出视频") sp.add_argument("--limit", type=int, default=20) sp.add_argument("--max-chars", type=int, default=80) sp.set_defaults(handler=lambda a: video_service.cmd_list(limit=a.limit, max_chars=a.max_chars)) sp = vid_sub.add_parser("get", help="按 id 输出 JSON") sp.add_argument("video_id", metavar="视频id") sp.set_defaults(handler=lambda a: video_service.cmd_get(a.video_id)) sp = vid_sub.add_parser("add", help="从本地文件复制入库", formatter_class=fmt) sp.add_argument("--file", required=True, metavar="文件") sp.add_argument("--title", default=None, metavar="标题") sp.add_argument("--duration-ms", type=int, default=None, metavar="毫秒", help="可选时长") sp.set_defaults(handler=_handle_video_add) sp = vid_sub.add_parser("delete", help="删除记录与磁盘目录") sp.add_argument("video_id", metavar="视频id") sp.set_defaults(handler=lambda a: video_service.cmd_delete(a.video_id)) sp = vid_sub.add_parser("feedback", help="回写状态", formatter_class=fmt) sp.add_argument("video_id", metavar="视频id") sp.add_argument("status", metavar="状态") sp.add_argument("account_id", nargs="?", default=None, metavar="账号") sp.add_argument("error_msg", nargs="?", default=None, metavar="错误说明") sp.set_defaults(handler=_handle_video_feedback) return p def main(argv: Optional[List[str]] = None) -> int: argv = argv if argv is not None else sys.argv[1:] if not argv: _print_root_usage_zh() return 1 parser = build_parser() args = parser.parse_args(argv) args.handler(args) return 0