group_fqcd_jr/tools/publish_customer_service_config.py

"""发布客服 Agent 的运行期配置：意图工具白名单（走发布状态机）。

两个必须讲清的点：

1. **为什么必须发这一步**：工具白名单是失败关闭的——`ToolExecutor` 拿发布配置里
   `agent_tools` / `customer_service:<intent>` 的 `allowed_tools` 与代码声明的
   `AgentDefinition.allowed_tools` 取交集，缺配置时交集为空、任何工具调用都被拒。
   「Agent 写好了但没发配置」的表现是"客服什么都答不了、一直在引导人工"。

2. **为什么必须继承现有配置项**：`config_release` 是**整版本替换**语义——激活新版本后，
   旧版本的所有配置项都不再生效。若只发布客服自己的白名单，示例 Agent 的
   `fund_query_demo:fund_quote` 会被静默清空。所以发布前先把当前 effective 版本里的
   配置项原样搬进新版本，再追加本次新增项。

用法：python tools/publish_customer_service_config.py
"""

import asyncio
import datetime as dt
import json
import sys
import uuid
from pathlib import Path
from typing import Any

import asyncmy
import httpx
import jwt

from app.core.config import get_settings
from app.main import create_app

ADMIN = "9003"
AGENT_TYPE = "customer_service"
TOOL_NAME = "search_knowledge"
SUITABILITY_TOOL = "check_suitability"
# 画像只读工具：客服的"出口零"（本人风险等级/投资偏好/测评是否过期）走它取权威字段。
# 那个出口复用的是 `faq` 意图 key（见 `customer_service.PROFILE_WHITELIST_INTENT`），
# 所以**必须**把它加进 `faq` 的白名单里；漏了的表现是"问画像一律转人工"，
# 而画像出口本身是对的——这是纯配置缺口（实测复现过）。
PROFILE_TOOL = "query_customer_profile"
# 只有会调用工具的意图才需要白名单；chitchat（模型生成）与 transfer_human（引导人工）
# 都不查知识库。给它们配空白名单反而会掩盖"配置漏配"，因此不发布这两条。
INTENT_TOOLS: dict[str, tuple[str, ...]] = {
    "faq": (TOOL_NAME, PROFILE_TOOL),
    "product_inquiry": (TOOL_NAME,),
    "policy_explain": (TOOL_NAME,),
    # 适当性裁决要两步：先从知识库拿到产品的风险等级，再由底座按档案里的客户等级裁决
    "suitability_check": (TOOL_NAME, SUITABILITY_TOOL),
}


def token(subject: str) -> str:
    settings = get_settings()
    private_key = Path(settings.jwt_private_key_path).read_text(encoding="utf-8")
    now = dt.datetime.now(dt.UTC)
    return jwt.encode(
        {
            "sub": subject, "iss": settings.jwt_issuer, "aud": settings.jwt_audience,
            "exp": now + dt.timedelta(minutes=30), "nbf": now - dt.timedelta(seconds=5),
            "jti": str(uuid.uuid4()),
        },
        private_key,
        algorithm="RS256",
    )


async def active_config_items() -> list[dict[str, Any]]:
    """读取当前生效版本的全部配置项，用于在新版本里原样继承。"""
    settings = get_settings()
    # MYSQL_DSN 形如 mysql+asyncmy://user:pass@host:port/db
    dsn = settings.mysql_dsn.split("://", 1)[1]
    credentials, location = dsn.split("@", 1)
    user, password = credentials.split(":", 1)
    host_port, database = location.split("/", 1)
    host, _, port = host_port.partition(":")
    connection = await asyncmy.connect(
        host=host, port=int(port or 3306), user=user, password=password, db=database
    )
    try:
        cursor = connection.cursor()
        await cursor.execute(
            """
            SELECT i.namespace, i.config_key, i.value_json, i.schema_version
            FROM platform_config_item i
            JOIN config_release r ON r.id = i.release_id
            WHERE r.status = 'active'
            """
        )
        rows = await cursor.fetchall()
    finally:
        connection.close()
    items: list[dict[str, Any]] = []
    for namespace, config_key, value_json, schema_version in rows:
        value = json.loads(value_json) if isinstance(value_json, str) else value_json
        items.append({
            "namespace": namespace,
            "item_key": config_key,
            "value_json": value,
            "schema_version": schema_version,
        })
    return items


async def post(
    client: httpx.AsyncClient, path: str, *, auth: dict[str, str],
    payload: dict[str, object] | None = None, if_match: str | None = None,
) -> httpx.Response:
    headers = {**auth, "Idempotency-Key": uuid.uuid4().hex}
    if if_match:
        headers["If-Match"] = if_match
    return await client.post(path, json=payload, headers=headers)


async def etag_of(client: httpx.AsyncClient, path: str, auth: dict[str, str]) -> str | None:
    return (await client.get(path, headers=auth)).headers.get("ETag")


SUITABILITY_INTENT = "suitability_check"
# description 与 examples 是**给分类器看的**：意图码本身只是个名字，真正让模型分辨
# "能买吗"和"这个产品是什么"的是这几个例子。所以 examples 全部取客户真实说法。
SUITABILITY_INTENT_SPEC: dict[str, Any] = {
    "intent_name": "投资者适当性判断",
    "description": "客户询问以自己的风险承受能力能否购买某只产品，或询问自身风险等级与产品的匹配情况",
    "examples": [
        "c1客户能买它吗", "我能买这个产品吗", "那它我能买吗",
        "这只基金适合我吗", "我的风险等级能买吗",
    ],
    "confidence_threshold": "0.6000",
}


async def ensure_suitability_intent(client: httpx.AsyncClient, auth: dict[str, str]) -> int:
    """确保 `suitability_check` 意图在运行期生效（返回 0 成功、1 失败）。

    为什么必须做这一步：意图码要三处对齐（见 `customer_service.py` 的注释），而运行期
    只读 `agent_intent_config` 里 status='active' 的行。少了这一行，分类链路看不到这个
    意图，"能买吗"会被分到别的意图里去，客户拿到的就是"C1 的通用规则"而不是结论。
    """
    path = "/api/v1/admin/agent-intent-configs"
    listed = await client.get(f"{path}?limit=100", headers=auth)
    rows = listed.json().get("data", []) if listed.status_code == 200 else []
    existing = next(
        (row for row in rows
         if row.get("agent_type") == AGENT_TYPE and row.get("intent_code") == SUITABILITY_INTENT),
        None,
    )
    if existing is not None and str(existing.get("status")) == "active":
        print(f"[意图配置] id={existing['id']} 已生效，跳过")
        return 0
    if existing is not None and str(existing.get("status")) in {"draft", "approved"}:
        config_id = int(existing["id"])
    else:
        # 没有历史行或历史行已归档：新版本号（该表 agent_type+intent_code+version 唯一）
        version = int(existing.get("version", 0)) + 1 if existing else 1
        created = await post(client, path, auth=auth, payload={
            "agent_type": AGENT_TYPE,
            "intent_code": SUITABILITY_INTENT,
            **SUITABILITY_INTENT_SPEC,
            "allowed_tools": [TOOL_NAME, SUITABILITY_TOOL],
            "version": version,
        })
        if created.status_code != 201:
            print(f"[意图配置] 创建失败：{created.status_code} {created.text[:200]}")
            return 1
        config_id = int(created.json()["data"]["id"])
        print(f"[意图配置] 已创建 id={config_id}（{SUITABILITY_INTENT} v{version}）")

    base = f"{path}/{config_id}"
    # 幂等：重跑脚本时某一步可能已经推进过（"已经审过了"不该 409 让整个脚本失败）
    settled = {"reviews": {"approved", "active"}, "activations": {"active"}}
    for action, payload in (
        ("reviews", {"decision": "approved", "comment": "创建人自审"}),
        # 激活端点要求 body 是对象；传 None 时 httpx 根本不发 body，会被判 422
        ("activations", {}),
    ):
        current = (await client.get(base, headers=auth)).json().get("data", {})
        if str(current.get("status")) in settled[action]:
            print(f"[意图配置] {action} 已在目标状态（{current.get('status')}），跳过")
            continue
        response = await post(
            client, f"{base}/{action}", auth=auth, payload=payload,
            if_match=await etag_of(client, base, auth),
        )
        if response.status_code != 200:
            print(f"[意图配置] {action} 失败：{response.status_code} {response.text[:200]}")
            return 1
    print(f"[意图配置] id={config_id} 已生效（运行期按 status='active' 读取）")
    return 0


async def main() -> int:
    app = create_app()
    auth = {"Authorization": f"Bearer {token(ADMIN)}"}
    async with httpx.AsyncClient(
        transport=httpx.ASGITransport(app=app), base_url="http://test", timeout=60
    ) as client:
        if await ensure_suitability_intent(client, auth) != 0:
            return 1

        inherited = await active_config_items()
        print(f"当前生效版本的配置项：{len(inherited)} 条（将原样继承）")
        for item in inherited:
            print(f"  · {item['namespace']} / {item['item_key']}")

        new_items = [
            {
                "namespace": "agent_tools",
                "item_key": f"{AGENT_TYPE}:{intent}",
                "value_json": {"allowed_tools": list(tools)},
                "schema_version": "1",
            }
            for intent, tools in INTENT_TOOLS.items()
        ]
        # **同 key 的继承项必须被本次新定义覆盖**，不能原样搬过去。
        # 血泪教训（实测）：上一版发布的是 `faq = ["query_knowledge", ...]`，而那个工具
        # 已随"客服检索改为 search_knowledge"从代码上限移除；原样继承会让 admin 服务的
        # 子集校验（白名单 ⊆ 代码限定的 allowed_tools）直接 422 拒绝整次发布，
        # 报错是"配置超出 Agent 工具上限"，看不出是继承造成的。
        inherited_only = [
            item for item in inherited
            if (str(item["namespace"]), str(item["item_key"])) not in {
                ("agent_tools", f"{AGENT_TYPE}:{intent}") for intent in INTENT_TOOLS
            }
        ]
        dropped = [item for item in inherited if item not in inherited_only]
        for item in dropped:
            print(f"  [覆盖] {item['namespace']}/{item['item_key']} 将由本次定义替换"
                  f"（原值 {item['value_json']}）")
        pending = [
            item for item in new_items
            if (str(item["namespace"]), str(item["item_key"])) not in
            {(str(i["namespace"]), str(i["item_key"])) for i in inherited_only}
        ]
        if not pending:
            print("客服白名单已存在于当前生效版本，无需发布")
            return 0

        created = await post(client, "/api/v1/admin/config-releases", auth=auth, payload={
            "release_no": f"cs-tools-{uuid.uuid4().hex[:12]}",
            "title": "客服 Agent 意图工具白名单",
            "change_summary": "新增 faq/product_inquiry/policy_explain 的知识检索白名单，并继承既有配置项",
        })
        if created.status_code != 201:
            print(f"创建发布版本失败：{created.status_code} {created.text[:200]}")
            return 1
        release_id = int(created.json()["data"]["id"])
        print(f"\n发布版本 id={release_id}")

        base = f"/api/v1/admin/config-releases/{release_id}/platform-config-items"
        for item in [*inherited_only, *pending]:
            response = await post(client, base, auth=auth, payload=item)
            mark = "继承" if item in inherited_only else "新增"
            print(f"  [{mark}] {item['namespace']}/{item['item_key']} → {response.status_code}")
            if response.status_code != 201:
                print(f"    失败：{response.text[:200]}")
                return 1

        release_base = f"/api/v1/admin/config-releases/{release_id}"
        submitted = await post(
            client, f"{release_base}/validations", auth=auth, payload={},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"\n提交复核：{submitted.status_code}")
        reviewed = await post(
            client, f"{release_base}/reviews", auth=auth,
            payload={"decision": "approved", "comment": "客服工具白名单"},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"审核：{reviewed.status_code}")
        activated = await post(
            client, f"{release_base}/activations", auth=auth, payload={},
            if_match=await etag_of(client, release_base, auth),
        )
        print(f"激活：{activated.status_code}")
        if activated.status_code not in (200, 201):
            print(f"  失败：{activated.text[:200]}")
            return 1
        print(f"最终状态：{activated.json()['data']['status']}")

        remaining = await active_config_items()
        print(f"\n激活后生效版本配置项：{len(remaining)} 条")
        for item in remaining:
            print(f"  · {item['namespace']} / {item['item_key']} = {item['value_json']}")
        return 0


sys.exit(asyncio.run(main()))