sysops
84 lines
3.0 KiB
Python
84 lines
3.0 KiB
Python
"""Koppel cockpit Herman-chat aan Hermes/Telegram memory."""
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from app.config import settings
|
|
|
|
log = logging.getLogger("cockpit.hermes_memory")
|
|
|
|
_USERNAME_MAP = {
|
|
"aissa": "ceo",
|
|
"aïssa": "ceo",
|
|
"aissa zarioh": "ceo",
|
|
"mo": "cto",
|
|
"mohamed": "cto",
|
|
}
|
|
|
|
|
|
def resolve_hermes_user(
|
|
*,
|
|
username: str | None = None,
|
|
role: str | None = None,
|
|
channel: str | None = None,
|
|
) -> str:
|
|
"""OpenAI-compatible `user` field = Telegram chat_id for Hermes memory."""
|
|
uname = (username or "").strip().lower()
|
|
r = (role or "").strip().lower()
|
|
mapped = _USERNAME_MAP.get(uname)
|
|
if mapped == "ceo" or r in ("ceo", "admin") or uname in ("aissa", "aïssa"):
|
|
return str(settings.HERMES_CEO_CHAT_ID or "8859782446")
|
|
if mapped == "cto" or r == "cto" or uname in ("mo", "mohamed"):
|
|
return str(settings.HERMES_CTO_CHAT_ID or "789036463")
|
|
return str(settings.HERMES_CEO_CHAT_ID or "8859782446")
|
|
|
|
|
|
async def fetch_recent_feed(chat_id: str | int, limit: int | None = None) -> list[dict[str, Any]]:
|
|
limit = limit or int(getattr(settings, "HERMES_MEMORY_LIMIT", 24) or 24)
|
|
base = (settings.HERMES_FEED_URL or "").rstrip("/")
|
|
if not base:
|
|
return []
|
|
try:
|
|
async with httpx.AsyncClient(timeout=12.0) as client:
|
|
r = await client.get(f"{base}/feed", params={"chat_id": int(chat_id), "limit": limit})
|
|
if r.status_code != 200:
|
|
return []
|
|
items = (r.json() or {}).get("items") or []
|
|
return list(reversed(items))
|
|
except Exception as exc:
|
|
log.warning("hermes feed fetch failed: %s", exc)
|
|
return []
|
|
|
|
|
|
def feed_to_messages(items: list[dict[str, Any]], *, max_chars: int = 12000) -> list[dict[str, str]]:
|
|
out: list[dict[str, str]] = []
|
|
total = 0
|
|
for it in items:
|
|
text = (it.get("content_text") or "").strip()
|
|
if not text:
|
|
continue
|
|
role_raw = (it.get("role") or "").lower()
|
|
role = "assistant" if role_raw in ("assistant", "hermes", "herman") else "user"
|
|
if role == "assistant" and (it.get("direction") or "") == "in":
|
|
role = "user"
|
|
entry = {"role": role, "content": text[:2000]}
|
|
total += len(entry["content"])
|
|
if total > max_chars:
|
|
break
|
|
out.append(entry)
|
|
return out
|
|
|
|
|
|
def continuity_system(chat_id: str, display_name: str | None = None) -> str:
|
|
name = display_name or ("Aïssa" if str(chat_id) == str(settings.HERMES_CEO_CHAT_ID) else "de gebruiker")
|
|
return (
|
|
"Je bent Herman / Hermes, dezelfde AI Co-CEO als in Telegram. "
|
|
f"Deze cockpit-chat is gekoppeld aan Telegram-sessie chat_id={chat_id} ({name}). "
|
|
"Gebruik je bestaande Hermes-memory en recente Telegram-geschiedenis. "
|
|
"Doe alsof dit dezelfde doorlopende conversatie is — geen nieuw persoon, geen amnesie. "
|
|
"Antwoord in het Nederlands, bondig en actiegericht."
|
|
)
|