tools/seed_demo_data.py

"""一键准备演示数据：按依赖顺序跑齐所有 seed 脚本。

## 为什么需要它

演示数据此前散在 10 个脚本里，**顺序有讲究**（账号 → 口令 → 账户 → 行情 → 配置 → 知识），
而且有一个环节（知识库素材）根本没有脚本、靠手工调接口。换台机器接手时没人知道该跑哪些、
按什么顺序跑。本脚本是唯一入口。

## 顺序与依赖

| # | 步骤 | 为什么在这个位置 |
|---|---|---|
| 1 | 账号与权限 | 后面所有步骤都要用到这些账号 |
| 2 | 演示口令 | 依赖 1 建出的账号 |
| 3 | 客户账户与持仓 | 客户页面与下单的前提 |
| 4 | 场内行情 | **下单的硬前置**；必须在演示前跑，行情会过期 |
| 5 | 风控预警样本 | 风控页面要有东西可看 |
| 6 | 投顾演示数据 | 投顾工作台要有方案可看 |
| 7-9 | 各类发布配置 | Agent 工具白名单与提示词，缺了客服/风控会"失败关闭" |
| 10 | 知识库素材 | 客服答得出问题的前提 |

## ⚠️ 两个必须知道的点

1. **最后一步与行情都需要 Agent Worker 才会真正生效**：知识入库只写 MySQL + 投 outbox 事件，
   向量由 Worker 消费事件后写 Milvus；没有 Worker 时现象是"客服照旧答不上"，且**没有报错**。
   所以跑完本脚本**必须**再起 Worker（`start.ps1` 会一起起）。
2. **第 2 步不是幂等的**：`set_user_password.py` 重跑等于**重设密码**（bcrypt 每次加盐不同）。
   这是有意的（改密就该覆盖），但要知道它不是"已存在就跳过"。

## 用法

    python tools/seed_demo_data.py              # 全跑
    python tools/seed_demo_data.py --from 4     # 从第 4 步开始（重跑行情等）
    python tools/seed_demo_data.py --only 4,10  # 只跑指定步骤
    python tools/seed_demo_data.py --list       # 只列步骤
"""

from __future__ import annotations

import argparse
import subprocess
import sys
import time
from pathlib import Path

PROJECT_ROOT = Path(__file__).resolve().parents[1]

if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(errors="replace")  # type: ignore[union-attr]

#: (标题, 脚本相对路径, 备注)
STEPS: tuple[tuple[str, str, str], ...] = (
    ("账号与权限", "tools/seed_test_rbac.py", "五个演示角色 + 权限号段 9001-9046"),
    ("演示口令", "tools/set_user_password.py", "⚠️ 非幂等：重跑等于重设密码"),
    ("客户账户与持仓", "tools/seed_sim_account_demo.py", "客户 9001 开 10 万虚拟资金 + 持仓"),
    ("场内行情", "tools/sync_market_prices.py", "⚠️ 下单硬前置；行情会过期，演示前必跑"),
    ("风控预警样本", "tools/seed_risk_alert_demo_data.py", "三条不同状态的演示预警"),
    ("投顾演示数据", "tools/seed_advisor_demo.py", "投顾工作台要展示的方案与归属"),
    ("风控 Agent 白名单", "tools/publish_risk_agent_config.py", "缺了风控助手工具会失败关闭"),
    ("客服配置白名单", "tools/publish_customer_service_config.py", "缺了客服工具会失败关闭"),
    ("客服闲聊提示词", "tools/publish_chitchat_prompt.py", "话术走发布配置，不硬编码"),
    ("知识库素材", "tools/seed_knowledge_demo.py", "⚠️ 需要 Worker 才会写进 Milvus"),
)


def run_step(index: int, title: str, script: str, note: str, *, dry_run: bool) -> bool:
    path = PROJECT_ROOT / script
    print()
    print("=" * 88)
    print(f"[{index}/{len(STEPS)}] {title}   —— {note}")
    print(f"        {script}")
    print("=" * 88)
    if not path.exists():
        print(f"[跳过] 脚本不存在：{path}")
        return False
    if dry_run:
        print("[dry-run] 未执行")
        return True
    started = time.perf_counter()
    # 刻意**不捕获输出**（继承 stdio）：既不吞掉子脚本的报错，
    # 也避免在受限环境里因 piped stdio 失败。
    result = subprocess.run([sys.executable, str(path)], check=False, cwd=str(PROJECT_ROOT))
    elapsed = time.perf_counter() - started
    ok = result.returncode == 0
    print(f"[{'完成' if ok else '失败'}] {title}  用时 {elapsed:.1f}s  退出码 {result.returncode}")
    return ok


def main() -> int:
    parser = argparse.ArgumentParser(description="一键准备演示数据")
    parser.add_argument("--list", action="store_true", help="只列步骤")
    parser.add_argument("--from", dest="start", type=int, default=1, help="从第几步开始")
    parser.add_argument("--only", default="", help="只跑这些步骤（逗号分隔，如 4,10）")
    parser.add_argument("--dry-run", action="store_true", help="只打印将执行什么")
    args = parser.parse_args()

    if args.list:
        print(f"共 {len(STEPS)} 步：")
        for index, (title, script, note) in enumerate(STEPS, 1):
            print(f"  {index:>2}. {title:<20} {script:<48} {note}")
        return 0

    only = {int(part) for part in args.only.split(",") if part.strip()} if args.only else None

    def wanted(index: int) -> bool:
        return index in only if only is not None else index >= args.start

    selected = [(index, *step) for index, step in enumerate(STEPS, 1) if wanted(index)]
    if not selected:
        print("没有匹配的步骤。")
        return 1

    print(f"准备演示数据：{len(selected)} 步" + ("（dry-run）" if args.dry_run else ""))
    failed: list[str] = []
    for index, title, script, note in selected:
        if not run_step(index, title, script, note, dry_run=args.dry_run):
            failed.append(title)

    print()
    print("=" * 88)
    if failed:
        print(f"有 {len(failed)} 步失败：{'、'.join(failed)}")
        print("先看上面的原始报错；多数失败是外部依赖没起来（MySQL / Redis / Docker）。")
        return 1
    print("全部完成。")
    if not args.dry_run:
        print(
            "\n下一步：\n"
            "  1. 起服务：powershell -ExecutionPolicy Bypass -File start.ps1\n"
            "     （它会把 API 与 **Agent Worker** 一起起 —— 没有 Worker，知识检索不到、\n"
            "       客服对话会一直显示'超时'）\n"
            "  2. 验证：python tools/e2e_smoke_test.py"
        )
    return 0


if __name__ == "__main__":
    sys.exit(main())