viktor
428cff8c54
Обёртка над ai.reg.cloud. Протокол совместим с OpenAI, но у платформы свои особенности: ключ бывает открыт не ко всем моделям, отказ приходит отдельным кодом, картинки и речь доступны не всем. Что внутри: - синхронный и асинхронный клиенты - каскад моделей на случай, если часть закрыта или занята - ремонт оборванного json из ответов - чистка типографики и отлов иероглифов в русском тексте Claude-Session: https://claude.ai/code/session_01Loz5stgQZH6ofBdMqTxiJu
297 строки
12 KiB
Python
297 строки
12 KiB
Python
"""Клиент к ai.reg.cloud. Синхронный и асинхронный."""
|
||
from __future__ import annotations
|
||
|
||
import asyncio
|
||
import json
|
||
import os
|
||
import re
|
||
import time
|
||
import unicodedata
|
||
from dataclasses import dataclass
|
||
from typing import Any, Iterable
|
||
|
||
import httpx
|
||
|
||
BASE_URL = "https://ai.reg.cloud/v1"
|
||
KEY_ENV = "REGCLOUD_AI_KEY"
|
||
|
||
# Проверено 23.08.2026. Ключи бывают открыты только к части моделей,
|
||
# полный список всегда спрашивайте у самой платформы через models().
|
||
KNOWN_MODELS = (
|
||
"claude-5-sonnet", "claude-4.5-haiku",
|
||
"gpt-5.5", "gpt-5.1-instant", "gpt-5.1-codex-mini", "gpt-5.3-codex", "gpt-oss-120b",
|
||
"gemini-3.1-pro", "gemini-3.5-flash",
|
||
"grok-4.3",
|
||
"qwen3.7-max", "qwen3.6-plus", "qwen3.6-27b", "qwen3.5-397b-a17b", "qwen3-coder", "qwen3-235b-a22b",
|
||
"glm-5.2", "glm-5-turbo",
|
||
"deepseek-v4-pro", "deepseek-v4-flash", "deepseek-r1",
|
||
"minimax-m3", "minimax-m2.7", "kimi-k2.6", "kimi-k2.7-code",
|
||
"mistral-large", "mistral-small", "command-r-plus", "gemma-4-26b-a4b-it",
|
||
)
|
||
|
||
|
||
class RegCloudError(RuntimeError):
|
||
"""Платформа ответила ошибкой."""
|
||
|
||
|
||
class ModelAccessError(RegCloudError):
|
||
"""Ключ не открыт к этой модели. Спросите список через models()."""
|
||
|
||
|
||
@dataclass
|
||
class Usage:
|
||
prompt_tokens: int = 0
|
||
completion_tokens: int = 0
|
||
total_tokens: int = 0
|
||
|
||
@classmethod
|
||
def parse(cls, raw: dict | None) -> "Usage":
|
||
raw = raw or {}
|
||
return cls(int(raw.get("prompt_tokens") or 0),
|
||
int(raw.get("completion_tokens") or 0),
|
||
int(raw.get("total_tokens") or 0))
|
||
|
||
|
||
@dataclass
|
||
class Answer:
|
||
text: str
|
||
model: str
|
||
usage: Usage
|
||
seconds: float
|
||
|
||
def __str__(self) -> str:
|
||
return self.text
|
||
|
||
|
||
class _Common:
|
||
"""Общая часть синхронного и асинхронного клиентов."""
|
||
|
||
def __init__(self, key: str | None = None, base_url: str = BASE_URL,
|
||
timeout: float = 300.0, retries: int = 3):
|
||
self.key = key or os.environ.get(KEY_ENV, "")
|
||
if not self.key:
|
||
raise RegCloudError(f"нет ключа, передайте его явно или положите в {KEY_ENV}")
|
||
self.base_url = base_url.rstrip("/")
|
||
self.timeout = timeout
|
||
self.retries = retries
|
||
|
||
@property
|
||
def _headers(self) -> dict:
|
||
return {"Authorization": f"Bearer {self.key}", "Content-Type": "application/json"}
|
||
|
||
def _payload(self, prompt: str, system: str, model: str, temperature: float,
|
||
max_tokens: int, json_mode: bool, messages: list | None, extra: dict) -> dict:
|
||
if messages is None:
|
||
messages = ([{"role": "system", "content": system}] if system else []) \
|
||
+ [{"role": "user", "content": prompt}]
|
||
body: dict[str, Any] = {"model": model, "messages": messages, "max_tokens": max_tokens}
|
||
# у части моделей температура фиксирована, тогда её лучше не слать вовсе
|
||
if temperature is not None:
|
||
body["temperature"] = temperature
|
||
if json_mode:
|
||
body["response_format"] = {"type": "json_object"}
|
||
body.update(extra)
|
||
return body
|
||
|
||
@staticmethod
|
||
def _raise_for(status: int, text: str, model: str) -> None:
|
||
if status == 403 and "model_access" in text:
|
||
raise ModelAccessError(f"ключ не открыт к модели {model}. {text[:200]}")
|
||
if status == 429:
|
||
raise RegCloudError(f"лимит запросов, {text[:200]}")
|
||
raise RegCloudError(f"платформа ответила {status}: {text[:300]}")
|
||
|
||
@staticmethod
|
||
def _read(data: dict, model: str, started: float) -> Answer:
|
||
choices = data.get("choices") or []
|
||
if not choices:
|
||
raise RegCloudError(f"пустой ответ: {json.dumps(data, ensure_ascii=False)[:200]}")
|
||
text = (choices[0].get("message") or {}).get("content") or ""
|
||
return Answer(text.strip(), data.get("model", model),
|
||
Usage.parse(data.get("usage")), round(time.time() - started, 2))
|
||
|
||
|
||
class RegCloudAI(_Common):
|
||
"""Обычный синхронный клиент."""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self._http = httpx.Client(timeout=self.timeout)
|
||
|
||
def models(self) -> list[str]:
|
||
r = self._http.get(f"{self.base_url}/models", headers=self._headers)
|
||
if r.status_code != 200:
|
||
self._raise_for(r.status_code, r.text, "-")
|
||
return [m["id"] for m in r.json().get("data", [])]
|
||
|
||
def chat(self, prompt: str = "", *, system: str = "", model: str = "claude-5-sonnet",
|
||
temperature: float | None = 0.7, max_tokens: int = 4000,
|
||
json_mode: bool = False, messages: list | None = None, **extra) -> Answer:
|
||
body = self._payload(prompt, system, model, temperature, max_tokens, json_mode, messages, extra)
|
||
started = time.time()
|
||
last = "не пробовали"
|
||
|
||
for attempt in range(self.retries):
|
||
r = self._http.post(f"{self.base_url}/chat/completions", headers=self._headers, json=body)
|
||
if r.status_code == 200:
|
||
return self._read(r.json(), model, started)
|
||
if r.status_code in (401, 403, 404):
|
||
self._raise_for(r.status_code, r.text, model)
|
||
last = f"{r.status_code} {r.text[:200]}"
|
||
time.sleep(3 * (attempt + 1))
|
||
raise RegCloudError(f"{model}: {last}")
|
||
|
||
def ask(self, prompt: str, **kw) -> str:
|
||
"""Короткий путь, когда нужен только текст ответа."""
|
||
return self.chat(prompt, **kw).text
|
||
|
||
def cascade(self, prompt: str, models: Iterable[str], **kw) -> Answer:
|
||
"""Идём по списку моделей, пока какая-нибудь не ответит.
|
||
Удобно, когда часть моделей закрыта для ключа или временно занята."""
|
||
errors = []
|
||
for model in models:
|
||
try:
|
||
return self.chat(prompt, model=model, **kw)
|
||
except RegCloudError as exc:
|
||
errors.append(f"{model}: {str(exc)[:90]}")
|
||
raise RegCloudError("ни одна модель не ответила. " + " | ".join(errors))
|
||
|
||
def close(self) -> None:
|
||
self._http.close()
|
||
|
||
def __enter__(self): return self
|
||
def __exit__(self, *a): self.close()
|
||
|
||
|
||
class AsyncRegCloudAI(_Common):
|
||
"""То же самое, но асинхронно."""
|
||
|
||
def __init__(self, *args, **kwargs):
|
||
super().__init__(*args, **kwargs)
|
||
self._http = httpx.AsyncClient(timeout=self.timeout)
|
||
|
||
async def models(self) -> list[str]:
|
||
r = await self._http.get(f"{self.base_url}/models", headers=self._headers)
|
||
if r.status_code != 200:
|
||
self._raise_for(r.status_code, r.text, "-")
|
||
return [m["id"] for m in r.json().get("data", [])]
|
||
|
||
async def chat(self, prompt: str = "", *, system: str = "", model: str = "claude-5-sonnet",
|
||
temperature: float | None = 0.7, max_tokens: int = 4000,
|
||
json_mode: bool = False, messages: list | None = None, **extra) -> Answer:
|
||
body = self._payload(prompt, system, model, temperature, max_tokens, json_mode, messages, extra)
|
||
started = time.time()
|
||
last = "не пробовали"
|
||
|
||
for attempt in range(self.retries):
|
||
r = await self._http.post(f"{self.base_url}/chat/completions", headers=self._headers, json=body)
|
||
if r.status_code == 200:
|
||
return self._read(r.json(), model, started)
|
||
if r.status_code in (401, 403, 404):
|
||
self._raise_for(r.status_code, r.text, model)
|
||
last = f"{r.status_code} {r.text[:200]}"
|
||
await asyncio.sleep(3 * (attempt + 1))
|
||
raise RegCloudError(f"{model}: {last}")
|
||
|
||
async def ask(self, prompt: str, **kw) -> str:
|
||
return (await self.chat(prompt, **kw)).text
|
||
|
||
async def cascade(self, prompt: str, models: Iterable[str], **kw) -> Answer:
|
||
errors = []
|
||
for model in models:
|
||
try:
|
||
return await self.chat(prompt, model=model, **kw)
|
||
except RegCloudError as exc:
|
||
errors.append(f"{model}: {str(exc)[:90]}")
|
||
raise RegCloudError("ни одна модель не ответила. " + " | ".join(errors))
|
||
|
||
async def close(self) -> None:
|
||
await self._http.aclose()
|
||
|
||
async def __aenter__(self): return self
|
||
async def __aexit__(self, *a): await self.close()
|
||
|
||
|
||
# ---------- разбор ответов ----------
|
||
|
||
def extract_json(text: str) -> Any:
|
||
"""Модели заворачивают json в тройные кавычки и обрывают его на середине.
|
||
Достаём и по возможности чиним."""
|
||
text = text.strip()
|
||
if "```" in text:
|
||
for part in text.split("```")[1:]:
|
||
part = part.strip()
|
||
if part.startswith("json"):
|
||
part = part[4:]
|
||
if part.strip().startswith(("{", "[")):
|
||
text = part.strip()
|
||
break
|
||
start = min((i for i in (text.find("{"), text.find("[")) if i >= 0), default=-1)
|
||
if start > 0:
|
||
text = text[start:]
|
||
try:
|
||
return json.loads(text)
|
||
except json.JSONDecodeError:
|
||
pass
|
||
|
||
stack, in_str, escaped = [], False, False
|
||
cut, cut_stack = None, []
|
||
for i, ch in enumerate(text):
|
||
if in_str:
|
||
if escaped:
|
||
escaped = False
|
||
elif ch == "\\":
|
||
escaped = True
|
||
elif ch == '"':
|
||
in_str = False
|
||
cut, cut_stack = i + 1, list(stack)
|
||
continue
|
||
if ch == '"':
|
||
in_str = True
|
||
elif ch in "{[":
|
||
stack.append("}" if ch == "{" else "]")
|
||
elif ch in "}]":
|
||
if stack and stack[-1] == ch:
|
||
stack.pop()
|
||
cut, cut_stack = i + 1, list(stack)
|
||
|
||
for end, st in ((len(text), stack), (cut, cut_stack)):
|
||
if not end:
|
||
continue
|
||
head = text[:end].rstrip().rstrip(",")
|
||
for _ in range(3):
|
||
try:
|
||
return json.loads(head + "".join(reversed(st)))
|
||
except json.JSONDecodeError:
|
||
comma = max(head.rfind(","), head.rfind("["), head.rfind("{"))
|
||
if comma <= 0:
|
||
break
|
||
head = head[:comma].rstrip().rstrip(",")
|
||
raise RegCloudError(f"не разобрал json: {text[:200]}")
|
||
|
||
|
||
TYPO = {"—": "-", "–": "-", "‑": "-", "−": "-", " ": " ", "": "",
|
||
"“": '"', "”": '"', "«": '"', "»": '"', "‘": "'", "’": "'", "…": "..."}
|
||
|
||
|
||
def normalize(text: str) -> str:
|
||
"""Приводит типографику к простому виду. Длинные тире и ёлочки ломают
|
||
субтитры и синтез речи, поэтому в текстах для озвучки их лучше не держать."""
|
||
for a, b in TYPO.items():
|
||
text = text.replace(a, b)
|
||
return re.sub(r"[ \t]+", " ", text).strip()
|
||
|
||
|
||
def foreign_chars(text: str) -> list[str]:
|
||
"""Дешёвые модели иногда вставляют иероглифы прямо в русский текст.
|
||
Ловим их до того, как текст уйдёт в дело."""
|
||
bad = []
|
||
for ch in text:
|
||
code = ord(ch)
|
||
if code < 128 or ch.isspace() or 0x400 <= code <= 0x4FF:
|
||
continue
|
||
if unicodedata.category(ch) in ("Pd", "Pi", "Pf", "Po", "Sm", "Sc", "So", "Zs", "Mn"):
|
||
continue
|
||
bad.append(ch)
|
||
return sorted(set(bad))
|
© 2026 StackWorks