105 lines
3.8 KiB
Python
105 lines
3.8 KiB
Python
"""Async client for Sorena AI API (OpenAI-compatible chat endpoint)."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import os
|
|
from typing import Any
|
|
|
|
import aiohttp
|
|
|
|
logger = logging.getLogger("sorena-ai")
|
|
|
|
LANG_NAMES: dict[str, str] = {
|
|
"fa": "فارسی",
|
|
"ar": "عربی",
|
|
"en": "انگلیسی",
|
|
"de": "آلمانی",
|
|
"fr": "فرانسوی",
|
|
"es": "اسپانیایی",
|
|
"tr": "ترکی",
|
|
}
|
|
|
|
|
|
class SorenaAiClient:
|
|
def __init__(
|
|
self,
|
|
api_url: str | None = None,
|
|
model: str | None = None,
|
|
temperature: float | None = None,
|
|
timeout: int | None = None,
|
|
bearer_token: str | None = None,
|
|
):
|
|
self.api_url = api_url or os.getenv("SORENA_AI_URL", "https://saeed-taghadosi.ir/ai/demo")
|
|
self.model = model or os.getenv("SORENA_AI_MODEL", "gpt-3.5-turbo-0125")
|
|
self.temperature = float(temperature if temperature is not None else os.getenv("SORENA_AI_TEMPERATURE", "0.3"))
|
|
self.timeout = int(timeout if timeout is not None else os.getenv("SORENA_AI_TIMEOUT", "60"))
|
|
self.bearer_token = bearer_token if bearer_token is not None else os.getenv("SORENA_AI_TOKEN", "")
|
|
|
|
async def chat(self, messages: list[dict[str, str]], temperature: float | None = None) -> str | None:
|
|
payload = {
|
|
"model": self.model,
|
|
"messages": messages,
|
|
"temperature": self.temperature if temperature is None else temperature,
|
|
}
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Authorization": f"Bearer {self.bearer_token}",
|
|
}
|
|
|
|
timeout = aiohttp.ClientTimeout(total=self.timeout)
|
|
try:
|
|
async with aiohttp.ClientSession(timeout=timeout) as session:
|
|
async with session.post(self.api_url, json=payload, headers=headers, ssl=False) as response:
|
|
raw = await response.text()
|
|
if response.status < 200 or response.status >= 300:
|
|
logger.error("Sorena AI HTTP %s: %s", response.status, raw[:300])
|
|
return None
|
|
return self._extract_text(json.loads(raw))
|
|
except Exception:
|
|
logger.exception("Sorena AI request failed")
|
|
return None
|
|
|
|
async def translate(self, text: str, from_lang: str, to_lang: str) -> str | None:
|
|
text = text.strip()
|
|
if not text:
|
|
return ""
|
|
|
|
from_name = LANG_NAMES.get(from_lang, from_lang)
|
|
to_name = LANG_NAMES.get(to_lang, to_lang)
|
|
|
|
system = (
|
|
f"تو یک مترجم حرفهای {from_name} به {to_name} هستی.\n\n"
|
|
"قوانین:\n"
|
|
"- فقط متن ترجمهشده را برگردان، بدون توضیح یا مقدمه\n"
|
|
"- اصطلاحات فنی را دقیق ترجمه کن\n"
|
|
"- اعداد و واحدها را تغییر نده\n"
|
|
"- این متن گفتگوی زنده در جلسه آنلاین است؛ روان و طبیعی ترجمه کن"
|
|
)
|
|
|
|
return await self.chat(
|
|
[
|
|
{"role": "system", "content": system},
|
|
{"role": "user", "content": text},
|
|
],
|
|
temperature=0.3,
|
|
)
|
|
|
|
@staticmethod
|
|
def _extract_text(data: Any) -> str | None:
|
|
if not isinstance(data, dict):
|
|
return None
|
|
|
|
text = data.get("res") or data.get("message") or data.get("response") or data.get("text")
|
|
choices = data.get("choices")
|
|
if isinstance(choices, list) and choices:
|
|
message = choices[0].get("message") if isinstance(choices[0], dict) else None
|
|
if isinstance(message, dict) and message.get("content"):
|
|
text = message["content"]
|
|
|
|
if not text or text == "Error":
|
|
return None
|
|
|
|
return str(text).strip()
|