204 lines
7.3 KiB
Python
204 lines
7.3 KiB
Python
"""ИИ-слой: перевод абузных постов на русский и саммари подборки.
|
||
|
||
Работает через любой OpenAI-совместимый эндпоинт. Ключ берётся из env
|
||
(OMNI_AI_KEY / OMNI_AI_BASE / OMNI_AI_MODEL) либо из конфига Hermes
|
||
(custom_providers с base_url, содержащим reformboss). Ключ никогда не пишется
|
||
в файлы репозитория и не логируется.
|
||
|
||
Если ключа нет — функции возвращают None, вызывающий код молча падает на
|
||
исходный текст. Никаких обязательных зависимостей: только stdlib.
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import os
|
||
import sqlite3
|
||
import time
|
||
import urllib.error
|
||
import urllib.request
|
||
|
||
DEFAULT_BASE = "https://api.reformboss.com/v1"
|
||
DEFAULT_MODEL = "gpt-6.1-sol"
|
||
FALLBACK_MODELS = ("glm-5.3", "qwen3.8-flash")
|
||
TIMEOUT = 120
|
||
|
||
_CFG = {"base": None, "key": None, "model": None}
|
||
|
||
|
||
def _cfg_path():
|
||
"""Путь к конфигу Hermes. expanduser на MSYS иногда отдаёт '~', поэтому перебираем."""
|
||
import getpass
|
||
env = os.environ.get("OMNI_HERMES_CFG")
|
||
if env and os.path.exists(env):
|
||
return env
|
||
cands = []
|
||
try:
|
||
user = getpass.getuser()
|
||
except Exception:
|
||
user = os.environ.get("USERNAME") or ""
|
||
roots = [os.environ.get("USERPROFILE"), os.environ.get("HOME")]
|
||
if user:
|
||
roots.append(os.path.join("C:/Users", user))
|
||
for root in roots:
|
||
if root and not root.startswith("~"):
|
||
cands.append(os.path.join(root.replace("\\", "/"), "AppData", "Local", "hermes", "config.yaml"))
|
||
try:
|
||
cands.append(os.path.join(os.path.expanduser("~"), "AppData", "Local", "hermes", "config.yaml"))
|
||
except Exception:
|
||
pass
|
||
for c in cands:
|
||
if os.path.exists(c):
|
||
return c
|
||
return cands[0] if cands else ""
|
||
|
||
|
||
def _hermes_cfg():
|
||
p = _cfg_path()
|
||
try:
|
||
import yaml # есть в системе, но не обязателен
|
||
except Exception:
|
||
return None, None
|
||
try:
|
||
d = yaml.safe_load(open(p, encoding="utf-8")) or {}
|
||
except Exception as e:
|
||
if os.environ.get("OMNI_AI_DEBUG"):
|
||
print("ai cfg read fail:", type(e).__name__, e)
|
||
return None, None
|
||
for pr in d.get("custom_providers") or []:
|
||
base = str(pr.get("base_url") or "")
|
||
if "reformboss" in base and pr.get("api_key"):
|
||
return base.rstrip("/"), pr["api_key"]
|
||
return None, None
|
||
|
||
|
||
def cfg():
|
||
if _CFG["base"]:
|
||
return _CFG
|
||
base = os.environ.get("OMNI_AI_BASE")
|
||
key = os.environ.get("OMNI_AI_KEY")
|
||
model = os.environ.get("OMNI_AI_MODEL")
|
||
if not (base and key):
|
||
hb, hk = _hermes_cfg()
|
||
base = base or hb
|
||
key = key or hk
|
||
_CFG["base"] = (base or DEFAULT_BASE).rstrip("/")
|
||
_CFG["key"] = key
|
||
_CFG["model"] = model or DEFAULT_MODEL
|
||
return _CFG
|
||
|
||
|
||
def available():
|
||
return bool(cfg()["key"])
|
||
|
||
|
||
def _once(model, messages, max_tokens, temperature):
|
||
c = cfg()
|
||
body = json.dumps({"model": model, "messages": messages,
|
||
"max_tokens": max_tokens, "temperature": temperature}).encode()
|
||
req = urllib.request.Request(c["base"] + "/chat/completions", data=body,
|
||
headers={"Authorization": "Bearer " + c["key"],
|
||
"Content-Type": "application/json"})
|
||
with urllib.request.urlopen(req, timeout=TIMEOUT) as r:
|
||
d = json.loads(r.read().decode())
|
||
return (d["choices"][0]["message"].get("content") or "").strip()
|
||
|
||
|
||
def chat(messages, model=None, max_tokens=700, temperature=0.2):
|
||
"""Запрос с ретраями и сменой модели: 503/429 на одной модели не роняют перевод."""
|
||
c = cfg()
|
||
if not c["key"]:
|
||
return None
|
||
chain = [model or c["model"]] + [m for m in FALLBACK_MODELS if m != (model or c["model"])]
|
||
for m in chain:
|
||
for attempt in range(2):
|
||
try:
|
||
out = _once(m, messages, max_tokens, temperature)
|
||
if out:
|
||
return out
|
||
except Exception:
|
||
if attempt == 0: # после последней попытки спим в пустоту
|
||
time.sleep(1.5)
|
||
return None
|
||
|
||
|
||
# --- кэш ---------------------------------------------------------------------
|
||
def _db():
|
||
from .config import DB
|
||
c = sqlite3.connect(str(DB), timeout=30)
|
||
c.execute("CREATE TABLE IF NOT EXISTS ai_cache (k TEXT PRIMARY KEY, v TEXT, ts INTEGER)")
|
||
c.commit()
|
||
return c
|
||
|
||
|
||
def _cache_get(k):
|
||
try:
|
||
c = _db()
|
||
r = c.execute("SELECT v FROM ai_cache WHERE k=?", (k,)).fetchone()
|
||
c.close()
|
||
return r[0] if r else None
|
||
except Exception:
|
||
return None
|
||
|
||
|
||
def _cache_put(k, v):
|
||
try:
|
||
c = _db()
|
||
c.execute("INSERT OR REPLACE INTO ai_cache(k, v, ts) VALUES(?,?,?)", (k, v, int(time.time())))
|
||
c.commit()
|
||
c.close()
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def translate(text, lang="ru"):
|
||
"""Перевод короткого поста/заголовка. Возвращает строку или None."""
|
||
text = (text or "").strip()
|
||
if not text or not available():
|
||
return None
|
||
if _looks_ru(text):
|
||
return text
|
||
key = "tr:" + hashlib.sha1((lang + text).encode()).hexdigest()
|
||
hit = _cache_get(key)
|
||
if hit:
|
||
return hit
|
||
out = chat([
|
||
{"role": "system", "content":
|
||
"Переводчик. Переведи текст на русский. Сохрани смысл, названия сервисов и ссылки "
|
||
"не переводи. Без пояснений, только перевод, максимум 200 символов."},
|
||
{"role": "user", "content": text[:1200]},
|
||
], max_tokens=300)
|
||
if out:
|
||
out = out.strip().strip('"')
|
||
_cache_put(key, out)
|
||
return out or None
|
||
|
||
|
||
def summarize(posts, tag=None):
|
||
"""Саммари подборки абузов: 3-5 коротких пунктов, что раздают и где."""
|
||
posts = [p for p in (posts or []) if p][:12]
|
||
if not posts or not available():
|
||
return None
|
||
key = "sum:" + hashlib.sha1(("|".join(posts[:12]) + str(tag)).encode()).hexdigest()
|
||
hit = _cache_get(key)
|
||
if hit:
|
||
return hit
|
||
body = "\n".join("- (%s) %s" % (p[:180], "") for p in posts)
|
||
out = chat([
|
||
{"role": "system", "content":
|
||
"Ты аналитик. По списку постов с форумов сделай саммари на русском: 3-5 пунктов, "
|
||
"каждый до 120 символов, по делу — что предлагают, платно или халява, где. "
|
||
"Без вступлений, без воды, без markdown-заголовков."},
|
||
{"role": "user", "content": body[:4000]},
|
||
], max_tokens=600)
|
||
if out:
|
||
_cache_put(key, out)
|
||
return out or None
|
||
|
||
|
||
def _looks_ru(text):
|
||
letters = [ch for ch in text if ch.isalpha()]
|
||
if not letters:
|
||
return True
|
||
ru = sum(1 for ch in letters if "а" <= ch.lower() <= "я" or ch.lower() in "ё")
|
||
return ru / len(letters) > 0.45 |