aiagent/scripts/testing/test_zhini_kefu_6.py

#!/usr/bin/env python3
"""
测试「知你客服6号」Agent：登录 -> 创建执行 -> 轮询直到结束。

用法:
  python test_zhini_kefu_6.py
  python test_zhini_kefu_6.py --base-url http://127.0.0.1:8037
  set PLATFORM_BASE_URL=... && python test_zhini_kefu_6.py

依赖: requests（与项目其他测试脚本一致）
"""
from __future__ import annotations

import argparse
import json
import os
import sys
import time

import requests

# 知你客服6号（本地平台 Agent 管理中的名称对应 ID，若你环境不同请改此处或传 --agent-id）
DEFAULT_AGENT_ID = "2acc84d5-814b-4d61-9703-94a4b117375f"
DEFAULT_MESSAGE = "你好"


def main() -> int:
    if sys.platform == "win32" and hasattr(sys.stdout, "reconfigure"):
        try:
            sys.stdout.reconfigure(encoding="utf-8")
            sys.stderr.reconfigure(encoding="utf-8")
        except Exception:
            pass

    parser = argparse.ArgumentParser(description="测试知你客服6号，发送一条用户消息并打印结果")
    parser.add_argument(
        "--base-url",
        default=os.getenv("PLATFORM_BASE_URL", "http://127.0.0.1:8037"),
        help="平台 API 根地址（默认 http://127.0.0.1:8037）",
    )
    parser.add_argument("--username", default=os.getenv("PLATFORM_USERNAME", "admin"))
    parser.add_argument("--password", default=os.getenv("PLATFORM_PASSWORD", "123456"))
    parser.add_argument("--agent-id", default=os.getenv("ZHINI_6_AGENT_ID", DEFAULT_AGENT_ID))
    parser.add_argument("--message", "-m", default=DEFAULT_MESSAGE, help="用户消息，默认：你好")
    parser.add_argument(
        "--user-id",
        default="script_test_zhini6",
        help="多轮记忆隔离用 user_id（可选）",
    )
    parser.add_argument("--timeout", type=int, default=180, help="轮询最长秒数")
    parser.add_argument("--poll", type=float, default=0.8, help="轮询间隔秒")
    args = parser.parse_args()

    base = args.base_url.rstrip("/")

    print("=" * 60)
    print("知你客服6号 执行测试")
    print("  base_url :", base)
    print("  agent_id :", args.agent_id)
    print("  message  :", args.message)
    print("=" * 60)

    # 1. 登录
    r = requests.post(
        f"{base}/api/v1/auth/login",
        data={"username": args.username, "password": args.password},
        headers={"Content-Type": "application/x-www-form-urlencoded"},
        timeout=15,
    )
    if r.status_code != 200:
        print("登录失败:", r.status_code, r.text[:500], file=sys.stderr)
        return 1
    token = r.json().get("access_token")
    if not token:
        print("登录响应无 access_token", file=sys.stderr)
        return 1

    headers = {
        "Authorization": f"Bearer {token}",
        "Content-Type": "application/json",
    }

    msg = args.message
    body = {
        "agent_id": args.agent_id,
        "input_data": {
            "query": msg,
            "USER_INPUT": msg,
            "user_id": args.user_id,
        },
    }

    # 2. 创建执行
    r2 = requests.post(f"{base}/api/v1/executions", headers=headers, json=body, timeout=30)
    if r2.status_code != 201:
        print("创建执行失败:", r2.status_code, r2.text[:1000], file=sys.stderr)
        return 1

    ex = r2.json()
    eid = ex["id"]
    print("已创建执行:", eid)
    print("初始状态:", ex.get("status"))

    # 3. 轮询
    deadline = time.time() + args.timeout
    status = ex.get("status", "pending")
    while status in ("pending", "running") and time.time() < deadline:
        time.sleep(args.poll)
        rs = requests.get(f"{base}/api/v1/executions/{eid}", headers=headers, timeout=60)
        if rs.status_code != 200:
            print("查询执行失败:", rs.status_code, rs.text[:500], file=sys.stderr)
            return 1
        detail = rs.json()
        status = detail.get("status", status)
        print("  ...", status)

    final = requests.get(f"{base}/api/v1/executions/{eid}", headers=headers, timeout=60).json()
    status = final.get("status")
    print()
    print("最终状态:", status)
    if final.get("error_message"):
        print("错误信息:", final["error_message"])
    od = final.get("output_data")
    if od is not None:
        print("output_data:")
        print(json.dumps(od, ensure_ascii=False, indent=2))
    else:
        print("output_data: null")

    return 0 if status == "completed" else 2


if __name__ == "__main__":
    raise SystemExit(main())