"""Parse Xfyun's three-part credential from ProviderCredential.api_key.""" from __future__ import annotations import json from dataclasses import dataclass @dataclass(frozen=True) class XfyunCredential: app_id: str api_key: str api_secret: str def parse_xfyun_credential(value: str) -> XfyunCredential: """Accept JSON in the existing api_key column. Example: {"appId":"...","apiKey":"...","apiSecret":"..."} """ try: payload = json.loads(value) except json.JSONDecodeError as exc: raise ValueError( 'Xfyun API Key must be JSON: {"appId":"...","apiKey":"...","apiSecret":"..."}' ) from exc if not isinstance(payload, dict): raise ValueError("Xfyun API Key JSON must be an object") credential = XfyunCredential( app_id=str(payload.get("appId") or payload.get("app_id") or "").strip(), api_key=str(payload.get("apiKey") or payload.get("api_key") or "").strip(), api_secret=str( payload.get("apiSecret") or payload.get("api_secret") or "" ).strip(), ) if not credential.app_id or not credential.api_key or not credential.api_secret: raise ValueError("Xfyun API Key JSON requires appId, apiKey, and apiSecret") return credential def websocket_url(value: str, default: str) -> str: url = (value or default).strip() if url.startswith("https://"): return f"wss://{url.removeprefix('https://')}" if url.startswith("http://"): return f"ws://{url.removeprefix('http://')}" return url def is_super_tts(model_id: str, api_url: str) -> bool: model = model_id.lower().replace("-", "_") return "super" in model or "/private/" in api_url.lower() def xfyun_language(value: str) -> str: normalized = (value or "zh_cn").lower().replace("-", "_") return {"zh": "zh_cn", "en": "en_us"}.get(normalized, normalized) def xfyun_speed(value: float) -> int: """Reuse the existing OpenAI-style speed field where 1.0 means normal.""" return max(0, min(100, round(value * 50 if value <= 4 else value)))