551 lines
23 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

"""Python 命令入口。
安装后 ``python -m muse`` 与 CLI 使用一致的参数、退出码和业务结果。
子命令及参数以 --help 输出为准;迁移为显式维护命令,不在启动时自动执行。
未实现或未注入的能力明确失败,不返回伪成功。
"""
from __future__ import annotations
import argparse
import json
import sys
from contextlib import ExitStack
from pathlib import Path
from muse.共享.错误 import Muse错误
from muse.启动 import 应用装配, 构建, 构建默认运行, 自检
from muse.配置 import 读取配置
_退出码_成功 = 0
_退出码_用法错误 = 2
_退出码_业务失败 = 1
def _解析参数(argv: list[str] | None) -> argparse.Namespace:
解析器 = argparse.ArgumentParser(
prog="python -m muse",
description="Muse 应用命令入口",
)
子命令 = 解析器.add_subparsers(dest="命令", required=True)
数据解析 = 子命令.add_parser("数据维护", help="完整备份、离线核对和新空库恢复")
数据动作 = 数据解析.add_subparsers(dest="动作", required=True)
备份解析 = 数据动作.add_parser("备份")
备份解析.add_argument("配置", help="来源维护用途配置")
备份解析.add_argument("目录", type=Path, help="全新备份目录")
核对解析 = 数据动作.add_parser("核对")
核对解析.add_argument("目录", type=Path, help="可离线读取的完整备份目录")
解包解析 = 数据动作.add_parser("解包")
解包解析.add_argument("目录", type=Path, help="完整备份tar文件")
解包解析.add_argument("输出目录", type=Path, help="全新解包目录")
解包解析.add_argument("--预期备份ID", required=True)
恢复解析 = 数据动作.add_parser("恢复")
恢复解析.add_argument("配置", help="明确新空库和新文件根的维护用途配置")
恢复解析.add_argument("目录", type=Path)
恢复解析.add_argument("--预期备份ID", required=True)
for action in ("计划", "推进"):
task_parser = 数据动作.add_parser(action)
task_parser.add_argument("配置", help="具备作者身份的运行用途配置")
task_parser.add_argument("目标", help="完整维护请求JSON文件或点名任务身份")
评测解析 = 子命令.add_parser("行为评测", help="固定场景裁决或显式角色行为评测")
评测解析.add_argument("场景")
评测解析.add_argument("--观察")
评测解析.add_argument("--适配器", choices=["scripted", "role_agent"], default="role_agent")
评测解析.add_argument("--配置")
评测解析.add_argument("--业务配置")
评测解析.add_argument("--动作", choices=["准备", "执行", "报告"])
评测解析.add_argument("--目标")
实验解析 = 子命令.add_parser("评测", help="按维护或评测用途封存数据集和实验条件")
实验解析.add_argument("配置")
实验解析.add_argument(
"动作",
choices=[
"发布数据集",
"发布回放数据集",
"发布方法数据集",
"发布返修数据集",
"发布规则样本",
"验证规则",
"规则验证报告",
"停止规则验证",
"导出规则凭据",
"创建返修比较",
"导出返修证明",
"数据集",
"创建实验",
"实验",
"生成输入",
"启动实验",
"推进实验",
"执行工作面",
"报告",
"取消实验",
"补登记交付",
"发布金标准",
"标定",
"封存标定",
"效果判据",
"封存效果",
"导出启用凭据",
],
)
实验解析.add_argument("目标", help="请求JSON文件或已保存的版本/实验身份")
实验解析.add_argument("--样本", help="生成输入所属的固定样本身份")
实验解析.add_argument("--维护配置", help="导出启用凭据时明确提供同库维护用途配置")
实验解析.add_argument("--来源配置", help="封存返修原文对时明确提供同作者生产用途配置")
经验解析 = 子命令.add_parser("经验", help="具体改进、真实评测关联与目标责任人启停")
经验解析.add_argument("配置")
经验解析.add_argument(
"动作",
choices=[
"提出改进",
"提出规则改进",
"申请规则验证",
"规则启停",
"读取改进",
"申请验证",
"启用",
"停用",
"运行来源",
"记录运行观察",
"运行观察",
"运行观察清单",
"重复问题",
"记录参考挑选",
"挑选信号",
"推断建议",
"复利度量",
],
)
经验解析.add_argument("目标", nargs="?", help="改进身份或完整请求JSON")
经验解析.add_argument("--评测配置", help="申请验证所用的明确同库评测用途配置")
自检解析 = 子命令.add_parser("自检", help="报告装配状态;不连接外部服务")
交付解析 = 子命令.add_parser("交付", help="定稿、导出和人工发布登记")
交付解析.add_argument("配置")
交付解析.add_argument(
"动作",
choices=[
"回看",
"冻结",
"定稿",
"导出",
"推进",
"下载",
"计划",
"连载",
"发布",
"反馈",
"跨章候选",
],
)
交付解析.add_argument("目标", help="作品/定稿/任务身份或请求JSON文件")
交付解析.add_argument("--输出", help="下载到新文件,拒绝覆盖")
自检解析.add_argument("配置", help="应用配置 TOML 路径")
迁移解析 = 子命令.add_parser("迁移", help="执行数据库迁移(显式维护命令)")
迁移解析.add_argument("配置", help="应用配置 TOML 路径")
迁移解析.add_argument("--目标版本", type=int, default=None, help="执行到该版本为止(默认全部)")
服务解析 = 子命令.add_parser("服务", help="启动已配置的作者工作台")
服务解析.add_argument("配置", help="应用配置 TOML 路径")
任务解析 = 子命令.add_parser("任务", help="查看、暂停、取消或恢复持久任务")
任务解析.add_argument("配置", help="作者配置 TOML 路径")
任务解析.add_argument(
"动作",
choices=[
"列表",
"查看",
"事件",
"控制",
"推进",
"预算",
"批准额度",
"调用",
"结算",
"核对调用",
],
)
任务解析.add_argument("目标", nargs="?", help="任务 ID;控制时为具名请求 JSON 文件")
任务解析.add_argument("--游标", type=int, default=0)
任务解析.add_argument("--额度文件", help="批准额度时提供与HTTP相同的完整预算JSON")
创作解析 = 子命令.add_parser("创作", help="发起通用生成并查看候选检查报告")
创作解析.add_argument("配置", help="作者配置 TOML 路径")
创作解析.add_argument("动作", choices=["发起", "报告"])
创作解析.add_argument("目标", nargs="?", help="发起时为请求 JSON 文件;报告时为候选 ID")
创作解析.add_argument("--版本", type=int, default=None, help="报告的目标候选版本")
审校解析 = 子命令.add_parser("审校", help="声音候选、完整审校和只读机器味诊断")
审校解析.add_argument("配置")
审校解析.add_argument(
"动作",
choices=[
"发起",
"报告",
"诊断",
"诊断报告",
"声音",
"声音候选",
"案例",
"案例清单",
"案例采集",
"案例审阅",
"案例决定",
"案例撤回",
"案例规则",
"规则状态",
"规则审阅",
"规则决定",
"声音审阅",
"声音决定",
"模型修订",
"有限返修",
"返修会话",
"返修比较证明",
"返修选择",
"选段定位",
"作者修订",
"修订回放",
"修订报告",
"导入规则",
"规则",
],
)
审校解析.add_argument("目标", nargs="?")
审校解析.add_argument("--离线", action="store_true")
研究解析 = 子命令.add_parser("研究", help="发起资料拆解、查看进度与检查拆书结果")
研究解析.add_argument("配置", help="作者配置 TOML 路径")
研究解析.add_argument("动作", choices=["拆解", "进度", "检查", "分析", "承接", "授权"])
研究解析.add_argument(
"目标", nargs="?", help="拆解为请求 JSON;进度为来源ID;检查/承接为任务ID;分析为分析ID"
)
结构解析 = 子命令.add_parser("结构", help="发布内置种子或查看已发布的固定结构")
结构解析.add_argument("配置", help="应用配置 TOML 路径")
结构解析.add_argument("动作", choices=["发布内置", "查看", "维护"])
结构解析.add_argument("目标", nargs="?", help="结构身份")
结构解析.add_argument("--版本", type=int, default=1)
作品解析 = 子命令.add_parser("作品", help="管理作品档案、章节和人工正文版本")
作品解析.add_argument("配置", help="作者配置 TOML 路径")
作品解析.add_argument(
"动作",
choices=[
"结构",
"查看",
"目录",
"节点目录",
"保存节点",
"进度",
"正文",
"保存档案",
"新增章节",
"重排章节",
"保存正文",
"回退正文",
],
)
作品解析.add_argument("目标", help="查询身份或写入请求 JSON 文件")
作品解析.add_argument("--结构", default="work_core")
作品解析.add_argument("--版本", type=int, default=1)
作品解析.add_argument("--分支", default="main")
审阅解析 = 子命令.add_parser("审阅", help="比较候选、修改候选并提交作者决定")
审阅解析.add_argument("配置", help="作者配置TOML路径")
审阅解析.add_argument("动作", choices=["列表", "查看", "建立候选", "修改候选", "打开", "决定"])
审阅解析.add_argument("目标", help="查询身份或操作请求JSON文件")
审阅解析.add_argument("--分支", default="main")
探索解析 = 子命令.add_parser("探索", help="保存私人想法与检查最终整理稿")
探索解析.add_argument("配置", help="作者配置TOML路径")
探索解析.add_argument("动作", choices=["列表", "查看", "保存", "检查", "选定", "查看选定"])
探索解析.add_argument("目标", nargs="?", help="稿件身份或保存请求JSON文件")
探索解析.add_argument("--预期哈希")
规划解析 = 子命令.add_parser("规划", help="保存规划候选、查看版本并执行明确作者决定")
规划解析.add_argument("配置")
规划解析.add_argument(
"动作",
choices=[
"结构",
"列表",
"查看",
"创建候选",
"编辑候选",
"查看候选",
"打开审阅",
"决定",
"读取细纲",
],
)
规划解析.add_argument("目标")
规划解析.add_argument("--结构", default="outline")
规划解析.add_argument("--版本", type=int, default=1)
规划解析.add_argument("--章节")
配置解析 = 子命令.add_parser("配置", help="保存、查看或验证启用运行配置;不自动启用")
配置解析.add_argument("配置", help="应用连接配置TOML路径")
配置解析.add_argument("动作", choices=["保存", "查看", "验证", "启用", "停用"])
配置解析.add_argument("配置ID")
配置解析.add_argument("版本", nargs="?", default="-", help="停用时填 -")
配置解析.add_argument("--内容", help="运行配置草案TOML或验证启用批准JSON路径")
管理解析 = 子命令.add_parser("管理", help="管理当前配置的受控暂存;不访问其他根")
管理解析.add_argument("配置", help="明确的应用连接配置")
管理解析.add_argument(
"动作", choices=["清理无租约暂存", "查看清理回执", "系统状态", "配置清单", "登记额度"]
)
管理解析.add_argument("对象ID", nargs="?")
return 解析器.parse_args(argv)
def _运行迁移(参数: argparse.Namespace) -> int:
from muse.基础设施.数据库.迁移 import 迁移目标库
已执行 = 迁移目标库(读取配置(参数.配置), 目标版本=参数.目标版本)
if 已执行:
for 项 in 已执行:
print(f"已应用 V{项.版本:04d} {项.文件名}")
else:
print("无待执行迁移(账本已是最新)")
return _退出码_成功
def main(argv: list[str] | None = None) -> int:
参数 = _解析参数(argv)
with ExitStack() as 生命周期:
return _运行命令(参数, 生命周期)
def _运行命令(参数: argparse.Namespace, 生命周期: ExitStack) -> int:
def 取得装配(*, 默认运行: bool = False) -> 应用装配:
配置 = 读取配置(参数.配置)
app = 构建默认运行(配置) if 默认运行 else 构建(配置)
# 私人探索文件操作不借数据库,也不为此读取数据库凭据。
仅探索文件 = 参数.命令 == "探索" and 参数.动作 in {"保存", "列表", "查看", "检查"}
if not 仅探索文件:
生命周期.enter_context(app.生命周期())
return app
try:
# 与HTTP入口相同:只读诊断仍可用,写入前必须匹配完整迁移账本。
只读动作 = {
"列表",
"查看",
"事件",
"报告",
"数据集",
"实验",
"读取改进",
"查看候选",
"读取细纲",
"查看清理回执",
"系统状态",
"配置清单",
"诊断报告",
"声音",
"案例",
"案例清单",
"规则",
"修订回放",
"修订报告",
"选段定位",
"进度",
"检查",
"分析",
"回看",
"定稿",
"下载",
"连载",
}
if (
参数.命令 not in {"自检", "迁移", "服务", "数据维护"}
and not (参数.命令 == "探索" and 参数.动作 in {"保存", "列表", "查看", "检查"})
and not getattr(参数, "离线", False)
and getattr(参数, "动作", None) not in 只读动作
and getattr(参数, "配置", None)
):
from muse.基础设施.数据库.连接 import 数据库工厂
from muse.基础设施.系统运行 import 要求运行兼容
cfg = 读取配置(参数.配置)
要求运行兼容(数据库工厂(cfg.数据库, cfg.运行用途))
if 参数.命令 == "数据维护":
if 参数.动作 in {"计划", "推进"}:
from muse.接入.cli.维护命令 import 运行维护任务命令
result = 运行维护任务命令(取得装配(), 参数.动作, 参数.目标)
print(json.dumps(result, ensure_ascii=False, default=str))
return _退出码_成功
from muse.编排.数据维护 import 执行数据维护
result = 执行数据维护(
参数.动作,
参数.目录,
配置=读取配置(参数.配置) if getattr(参数, "配置", None) else None,
预期备份ID=getattr(参数, "预期备份ID", None),
输出目录=getattr(参数, "输出目录", None),
)
print(json.dumps(result, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "经验":
from muse.接入.cli.经验命令 import 运行经验命令
结果 = 运行经验命令(取得装配(), 参数.动作, 参数.目标, 评测配置=参数.评测配置)
print(json.dumps(结果, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "评测":
from muse.接入.cli.评测命令 import 运行评测命令
结果 = 运行评测命令(
取得装配(),
参数.动作,
参数.目标,
样本=参数.样本,
维护配置=参数.维护配置,
来源配置=参数.来源配置,
)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "规划":
from muse.接入.cli.作品命令 import 运行规划命令
结果 = 运行规划命令(
取得装配(默认运行=True),
参数.动作,
参数.目标,
结构ID=参数.结构,
版本=参数.版本,
章节ID=参数.章节,
)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "探索":
from muse.接入.cli.作品命令 import 运行探索命令
结果 = 运行探索命令(取得装配(), 参数.动作, 参数.目标, 预期哈希=参数.预期哈希)
print(json.dumps(结果, ensure_ascii=False))
return (
_退出码_业务失败
if isinstance(结果, dict) and 结果.get("valid") is False
else _退出码_成功
)
if 参数.命令 == "审阅":
from muse.接入.cli.审阅命令 import 运行审阅命令
结果 = 运行审阅命令(取得装配(), 参数.动作, 参数.目标, 分支=参数.分支)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "管理":
from muse.接入.cli.管理命令 import 运行管理命令
结果 = 运行管理命令(取得装配(), 参数.动作, 参数.对象ID)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "配置":
from muse.接入.cli.管理命令 import 运行配置命令
结果 = 运行配置命令(取得装配(), 参数.动作, 参数.配置ID, 参数.版本, 参数.内容)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "作品":
from muse.接入.cli.作品命令 import 运行作品命令
结果 = 运行作品命令(
取得装配(),
参数.动作,
参数.目标,
结构ID=参数.结构,
版本=参数.版本,
分支=参数.分支,
)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "结构":
from muse.接入.cli.结构命令 import 运行结构命令
结果 = 运行结构命令(取得装配(), 参数.动作, 参数.目标, 参数.版本)
print(json.dumps(结果, ensure_ascii=False))
return _退出码_成功
if 参数.命令 == "任务":
from muse.接入.cli.任务命令 import 运行任务命令
结果 = 运行任务命令(
取得装配(默认运行=True),
参数.动作,
参数.目标,
参数.游标,
额度文件=参数.额度文件,
)
print(json.dumps(结果, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "创作":
from muse.接入.cli.创作命令 import 运行创作命令
结果 = 运行创作命令(取得装配(默认运行=True), 参数.动作, 参数.目标, 版本=参数.版本)
print(json.dumps(结果, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "行为评测":
from muse.接入.cli.评测命令 import 运行行为评测
if 参数.适配器 == "role_agent" and all(
(参数.配置, 参数.业务配置, 参数.动作, 参数.目标)
):
from muse.接入.cli.角色行为命令 import 运行角色行为命令
结果 = 运行角色行为命令(参数.场景, 参数.配置, 参数.业务配置, 参数.动作, 参数.目标)
else:
结果 = 运行行为评测(参数.场景, 参数.观察, 参数.适配器)
print(json.dumps(结果, ensure_ascii=False))
return (
_退出码_业务失败
if 结果.get("failed") or 结果.get("passed") is False
else _退出码_成功
)
if 参数.命令 == "审校":
from muse.接入.cli.审校命令 import 运行审校命令
app = None if 参数.离线 else 取得装配(默认运行=True)
结果 = 运行审校命令(app, 参数.动作, 参数.目标, 离线=参数.离线)
print(json.dumps(结果, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "研究":
from muse.接入.cli.研究命令 import 运行研究命令
结果 = 运行研究命令(取得装配(默认运行=True), 参数.动作, 参数.目标)
print(json.dumps(结果, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "自检":
装配 = 构建(读取配置(参数.配置))
for 键, 值 in 自检(装配).items():
print(f"{键}: {值}")
return _退出码_成功
if 参数.命令 == "交付":
from muse.接入.cli.交付命令 import 运行交付命令
result = 运行交付命令(
取得装配(),
参数.动作,
参数.目标,
output=参数.输出,
)
print(json.dumps(result, ensure_ascii=False, default=str))
return _退出码_成功
if 参数.命令 == "迁移":
return _运行迁移(参数)
if 参数.命令 == "服务":
import uvicorn
from muse.基础设施.可观测性 import 配置日志
from muse.接入.http.应用 import 创建应用
配置 = 读取配置(参数.配置)
应用 = 创建应用(配置)
配置日志()
assert 配置.HTTP is not None
uvicorn.run(应用, host=配置.HTTP.监听地址, port=配置.HTTP.端口, access_log=False)
return _退出码_成功
except Muse错误 as exc:
print(json.dumps(exc.呈现(), ensure_ascii=False), file=sys.stderr)
return _退出码_业务失败
return _退出码_用法错误
if __name__ == "__main__":
raise SystemExit(main())