group_fqcd_jr/tools/configure_embedding_endpoint.py

"""配置并激活 embedding 模型端点（走管理 API，不直接写库）。

为什么走 API 而不是 INSERT：端点配置要经过 draft → approved → active 状态机并留下
`interaction_audit`。直接写库会绕过审核与审计，而且 `DatabaseModelGateway` 只认
`status='active'`，手工写错状态会表现为「模型端点未注册或未激活」这种与病因无关的报错。

幂等：脚本先查该 endpoint_code 是否已存在，已存在则跳过创建，只做后续状态推进。

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

import asyncio
import datetime as dt
import sys
import uuid
from pathlib import Path

import asyncmy
import httpx
import jwt

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

ADMIN = "9003"
ENDPOINT_CODE = "qwen-embedding"
MYSQL_DSN_HOST = "127.0.0.1"
MYSQL_USER, MYSQL_PASSWORD, MYSQL_DB = "root", "123456", "jr"


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 endpoint_id_if_exists() -> int | None:
    connection = await asyncmy.connect(
        host=MYSQL_DSN_HOST, port=3306, user=MYSQL_USER, password=MYSQL_PASSWORD, db=MYSQL_DB
    )
    try:
        cursor = connection.cursor()
        await cursor.execute(
            "SELECT id FROM model_endpoint_config WHERE endpoint_code=%s", (ENDPOINT_CODE,)
        )
        row = await cursor.fetchone()
        return int(row[0]) if row else None
    finally:
        connection.close()


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:
    response = await client.get(path, headers=auth)
    return response.headers.get("ETag")


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:
        endpoint_id = await endpoint_id_if_exists()
        if endpoint_id is None:
            created = await post(
                client, "/api/v1/admin/model-endpoints", auth=auth,
                payload={
                    "endpoint_code": ENDPOINT_CODE,
                    "provider": "dashscope",
                    "model_name": "qwen3.7-text-embedding-flash",
                    "base_url": "https://dashscope.aliyuncs.com/compatible-mode/v1",
                    "secret_ref": "env:QWEN_EMBEDDING_API_KEY",
                    "capabilities": ["embedding"],
                    "allowed_data_levels": ["public", "internal"],
                    "context_window": 8192,
                    "timeout_ms": 30000,
                },
            )
            print(f"创建端点：{created.status_code} {created.text[:160]}")
            if created.status_code != 201:
                return 1
            endpoint_id = int(created.json()["data"]["id"])
        else:
            print(f"端点已存在，复用 id={endpoint_id}")

        detail_path = f"/api/v1/admin/model-endpoints/{endpoint_id}"
        current = (await client.get(detail_path, headers=auth)).json()["data"]
        print(f"当前状态：{current['status']}")

        if current["status"] == "draft":
            reviewed = await post(
                client, f"{detail_path}/reviews", auth=auth,
                payload={"decision": "approved", "comment": "embedding 端点配置"},
                if_match=await etag_of(client, detail_path, auth),
            )
            print(f"审核：{reviewed.status_code} {reviewed.text[:160]}")
            if reviewed.status_code not in (200, 201):
                return 1

        current = (await client.get(detail_path, headers=auth)).json()["data"]
        if current["status"] == "approved":
            activated = await post(
                client, f"{detail_path}/activations", auth=auth,
                # activations 的 body 是必填的 EmptyPayload（extra=forbid），
                # 不传 body 会得到 422 "Field required"，因此显式给空对象。
                payload={},
                if_match=await etag_of(client, detail_path, auth),
            )
            print(f"激活：{activated.status_code} {activated.text[:200]}")
            if activated.status_code not in (200, 201):
                return 1

        final = (await client.get(detail_path, headers=auth)).json()["data"]
        print(f"最终状态：{final['status']}  模型={final['model_name']}")
        return 0 if final["status"] == "active" else 1


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