"""Console IA — oai.by-cyberdev.com FastAPI : agrégateur de fournisseurs de modèles IA (gratuits). - Catégorie "sans clé" : le serveur utilise son pool interne. - Catégorie "API externe (clé)" : l'utilisateur fournit sa clé (API EXT). Déployable sur O2Switch via Passenger (a2wsgi.ASGIMiddleware). """ import base64 import hashlib import json import os import time import uuid from pathlib import Path from typing import List, Optional import requests from dotenv import load_dotenv from fastapi import FastAPI, HTTPException, Request from fastapi.middleware.cors import CORSMiddleware from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, StreamingResponse from fastapi.staticfiles import StaticFiles from pydantic import BaseModel, Field from app import oauth, providers # --------------------------------------------------------------------------- # Config # --------------------------------------------------------------------------- BASE_DIR = Path(__file__).resolve().parent load_dotenv(BASE_DIR / ".env") AUTH_PASSWORD = os.getenv("AUTH_PASSWORD", "change-me") AUTH_SECRET = os.getenv("AUTH_SECRET", "change-me-secret") ADMIN_USERNAME = os.getenv("ADMIN_USERNAME") or os.getenv("AUTH_USERNAME", "admin") PUBLIC_HOST = os.getenv("PUBLIC_HOST", "oai.by-cyberdev.com") USER_DAILY_QUOTA = int(os.getenv("USER_DAILY_QUOTA", "120") or "120") SERVER_KEYS = { # clés détenues par le serveur -> utilisées par la catégorie "sans clé" "openrouter": os.getenv("OPENROUTER_API_KEY", ""), } # Timeout HTTP : (connexion, lecture). 180 s pour laisser respirer les modèles lents/free. HTTP_TIMEOUT = int(os.getenv("HTTP_TIMEOUT", "180")) # Chiffrement léger (stdlib only) pour les clés utilisateur stockées en BDD _KEY = hashlib.sha256(AUTH_SECRET.encode()).digest() def _xor_crypt(data: bytes) -> bytes: return bytes(b ^ _KEY[i % len(_KEY)] for i, b in enumerate(data)) def encrypt_key(plain: str) -> str: return base64.b64encode(_xor_crypt(plain.encode())).decode() def decrypt_key(b64: str) -> str: return _xor_crypt(base64.b64decode(b64)).decode() # --------------------------------------------------------------------------- # Session / Stockage (BDD MySQL optionnelle, repli fichier JSON) # --------------------------------------------------------------------------- def _mysql_conn(): if not os.getenv("DB_USER") or not os.getenv("DB_NAME"): return None try: import pymysql import socket old = socket.getdefaulttimeout() socket.setdefaulttimeout(3) try: conn = pymysql.connect( host=os.getenv("DB_HOST", "localhost"), user=os.getenv("DB_USER", ""), password=os.getenv("DB_PASSWORD", ""), database=os.getenv("DB_NAME", ""), charset="utf8mb4", cursorclass=pymysql.cursors.DictCursor, connect_timeout=3, read_timeout=5, ) finally: socket.setdefaulttimeout(old) return conn except Exception: return None def mysql_available() -> bool: conn = _mysql_conn() if conn is None: return False conn.close() return True def _ensure_column(cur, table: str, column: str, definition: str) -> None: cur.execute( """ SELECT COUNT(*) AS n FROM INFORMATION_SCHEMA.COLUMNS WHERE TABLE_SCHEMA=DATABASE() AND TABLE_NAME=%s AND COLUMN_NAME=%s """, (table, column), ) row = cur.fetchone() or {} if int(row.get("n", 0)) == 0: cur.execute(f"ALTER TABLE {table} ADD COLUMN {column} {definition}") def init_db(): """Crée les tables si MySQL est configuré. Repli = fichier.""" conn = _mysql_conn() if conn is None: return try: with conn.cursor() as cur: cur.execute(""" CREATE TABLE IF NOT EXISTS user_keys ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, provider VARCHAR(32) NOT NULL, key_cipher TEXT NOT NULL, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, UNIQUE KEY uq (username, provider) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS chat_logs ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, provider VARCHAR(32) NOT NULL, model VARCHAR(128), prompt TEXT, reply TEXT, latency_ms INT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS oauth_states ( state VARCHAR(128) NOT NULL PRIMARY KEY, provider VARCHAR(32) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_created (created_at) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS projects ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, name VARCHAR(120) NOT NULL, provider VARCHAR(32) NOT NULL DEFAULT '', model VARCHAR(128) NOT NULL DEFAULT '', system TEXT, share_token VARCHAR(48) NULL, share_enabled TINYINT(1) NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user (username), INDEX idx_project_share (share_token) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS discussions ( id INT AUTO_INCREMENT PRIMARY KEY, username VARCHAR(64) NOT NULL, project_id INT NULL, title VARCHAR(160) NOT NULL DEFAULT 'Nouvelle discussion', share_token VARCHAR(48) NULL, share_enabled TINYINT(1) NOT NULL DEFAULT 0, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP, INDEX idx_user (username), INDEX idx_share (share_token) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS messages ( id INT AUTO_INCREMENT PRIMARY KEY, discussion_id INT NOT NULL, username VARCHAR(64) NOT NULL, role VARCHAR(16) NOT NULL, content MEDIUMTEXT, meta VARCHAR(255), created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_disc (discussion_id) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) cur.execute(""" CREATE TABLE IF NOT EXISTS sessions ( token VARCHAR(64) NOT NULL PRIMARY KEY, username VARCHAR(64) NOT NULL, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, INDEX idx_user (username) ) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 """) _ensure_column(cur, "projects", "provider", "VARCHAR(32) NOT NULL DEFAULT ''") _ensure_column(cur, "projects", "model", "VARCHAR(128) NOT NULL DEFAULT ''") _ensure_column(cur, "projects", "system", "TEXT") _ensure_column(cur, "projects", "share_token", "VARCHAR(48) NULL") _ensure_column(cur, "projects", "share_enabled", "TINYINT(1) NOT NULL DEFAULT 0") _ensure_column(cur, "discussions", "project_id", "INT NULL") _ensure_column(cur, "discussions", "share_token", "VARCHAR(48) NULL") _ensure_column(cur, "discussions", "share_enabled", "TINYINT(1) NOT NULL DEFAULT 0") _ensure_column(cur, "messages", "meta", "VARCHAR(255)") conn.commit() finally: conn.close() def _keys_file() -> Path: p = BASE_DIR / "app" / "data" / "user_keys.json" p.parent.mkdir(parents=True, exist_ok=True) return p def _load_keys_file() -> dict: p = _keys_file() if p.exists(): try: return json.loads(p.read_text()) except Exception: return {} return {} def _save_keys_file(data: dict): _keys_file().write_text(json.dumps(data, indent=2)) def _store_file() -> Path: p = BASE_DIR / "app" / "data" / "store.json" p.parent.mkdir(parents=True, exist_ok=True) return p def _empty_store() -> dict: return { "seq": {"projects": 1, "discussions": 1, "messages": 1, "chat_logs": 1}, "projects": [], "discussions": [], "messages": [], "chat_logs": [], } def _load_store() -> dict: p = _store_file() if not p.exists(): return _empty_store() try: data = json.loads(p.read_text()) except Exception: data = {} store = _empty_store() store.update({k: data.get(k, store[k]) for k in store if k != "seq"}) store["seq"].update(data.get("seq", {})) return store def _save_store(store: dict) -> None: _store_file().write_text(json.dumps(store, indent=2, ensure_ascii=False)) def _next_store_id(store: dict, bucket: str) -> int: current = int(store.setdefault("seq", {}).get(bucket, 1)) store["seq"][bucket] = current + 1 return current def _now_iso() -> str: return time.strftime("%Y-%m-%d %H:%M:%S") def _find_owned(rows: list, rid: int, username: str) -> Optional[dict]: return next((r for r in rows if int(r.get("id", 0)) == int(rid) and r.get("username") == username), None) def _provider_quota_limit(provider_id: str) -> int: env = "PROVIDER_DAILY_QUOTA_" + "".join( c if c.isalnum() else "_" for c in provider_id.upper() ) return int(os.getenv(env, str(USER_DAILY_QUOTA)) or "0") def _usage_snapshot(username: str) -> dict: today = time.strftime("%Y-%m-%d") counts: dict = {} conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( """ SELECT provider, COUNT(*) AS n FROM chat_logs WHERE username=%s AND DATE(created_at)=CURDATE() GROUP BY provider """, (username,), ) for row in cur.fetchall(): counts[row["provider"]] = int(row["n"]) finally: conn.close() else: for row in _load_store().get("chat_logs", []): if row.get("username") == username and str(row.get("created_at", "")).startswith(today): counts[row.get("provider", "")] = counts.get(row.get("provider", ""), 0) + 1 used = sum(counts.values()) remaining = max(USER_DAILY_QUOTA - used, 0) if USER_DAILY_QUOTA > 0 else None provider_rows = [] for p in providers.all_providers(): limit = _provider_quota_limit(p.id) p_used = counts.get(p.id, 0) provider_rows.append({ "id": p.id, "name": p.name, "logo": p.logo, "used": p_used, "limit": limit, "remaining": max(limit - p_used, 0) if limit > 0 else None, }) return { "daily": { "used": used, "limit": USER_DAILY_QUOTA, "remaining": remaining, }, "providers": provider_rows, } def _check_quota(username: Optional[str], provider_id: str) -> None: if not username: return snap = _usage_snapshot(username) daily = snap["daily"] if daily["limit"] > 0 and daily["used"] >= daily["limit"]: raise HTTPException(429, "Quota quotidien utilisateur atteint.") prow = next((p for p in snap["providers"] if p["id"] == provider_id), None) if prow and prow["limit"] > 0 and prow["used"] >= prow["limit"]: raise HTTPException(429, f"Quota quotidien atteint pour {prow['name']}.") def _public_url(path: str) -> str: return f"https://{PUBLIC_HOST}{path}" def save_user_key(username: str, provider: str, plain_key: str) -> None: cipher = encrypt_key(plain_key) conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( "INSERT INTO user_keys (username, provider, key_cipher) " "VALUES (%s,%s,%s) ON DUPLICATE KEY UPDATE key_cipher=%s", (username, provider, cipher, cipher), ) conn.commit() return finally: conn.close() data = _load_keys_file() data.setdefault(username, {})[provider] = cipher _save_keys_file(data) def get_user_key(username: str, provider: str) -> Optional[str]: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( "SELECT key_cipher FROM user_keys " "WHERE username=%s AND provider=%s", (username, provider), ) row = cur.fetchone() if row: return decrypt_key(row["key_cipher"]) return None finally: conn.close() data = _load_keys_file() return decrypt_key(data.get(username, {}).get(provider, "")) if data.get( username, {}).get(provider) else None def delete_user_key(username: str, provider: str) -> None: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( "DELETE FROM user_keys WHERE username=%s AND provider=%s", (username, provider), ) conn.commit() return finally: conn.close() data = _load_keys_file() data.setdefault(username, {}).pop(provider, None) _save_keys_file(data) def log_chat(username, provider, model, prompt, reply, latency_ms): conn = _mysql_conn() if conn is None: store = _load_store() store.setdefault("chat_logs", []).append({ "id": _next_store_id(store, "chat_logs"), "username": username, "provider": provider, "model": model, "prompt": prompt[:2000], "reply": (reply or "")[:2000], "latency_ms": int(latency_ms), "created_at": _now_iso(), }) _save_store(store) return try: with conn.cursor() as cur: cur.execute( "INSERT INTO chat_logs (username, provider, model, prompt, reply, latency_ms) " "VALUES (%s,%s,%s,%s,%s,%s)", (username, provider, model, prompt[:2000], (reply or "")[:2000], int(latency_ms)), ) conn.commit() finally: conn.close() # --------------------------------------------------------------------------- # Sessions web (cookie simple) # --------------------------------------------------------------------------- SESSIONS: dict = {} # token -> username (memory fallback si MySQL indisponible) # Repli mémoire si la table sessions n'existe pas / MySQL down _sessions_fallback: dict = {} def _session_store(token: str, username: str) -> None: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( "INSERT INTO sessions (token, username) VALUES (%s,%s) " "ON DUPLICATE KEY UPDATE username=%s, created_at=CURRENT_TIMESTAMP", (token, username, username), ) cur.execute("DELETE FROM sessions WHERE created_at < NOW() - INTERVAL 8 DAY") conn.commit() return finally: conn.close() SESSIONS[token] = username _sessions_fallback[token] = username def _session_take_user(token: str) -> Optional[str]: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute("SELECT username FROM sessions WHERE token=%s", (token,)) row = cur.fetchone() return row["username"] if row else None finally: conn.close() return _sessions_fallback.get(token) or SESSIONS.get(token) def _session_drop(token: str) -> None: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute("DELETE FROM sessions WHERE token=%s", (token,)) conn.commit() return finally: conn.close() SESSIONS.pop(token, None) _sessions_fallback.pop(token, None) def _new_session(username: str) -> str: token = uuid.uuid4().hex _session_store(token, username) return token def _current_user(request: Request) -> Optional[str]: token = request.cookies.get("oai_session") if not token: return None return _session_take_user(token) # Repli mémoire pour le state OAuth si MySQL est indisponible _oauth_state_fallback: dict = {} def _oauth_state_store(state: str, provider: str) -> None: conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute( "INSERT INTO oauth_states (state, provider) VALUES (%s,%s) " "ON DUPLICATE KEY UPDATE provider=%s, created_at=CURRENT_TIMESTAMP", (state, provider, provider), ) cur.execute("DELETE FROM oauth_states WHERE created_at < NOW() - INTERVAL 15 MINUTE") conn.commit() return finally: conn.close() _oauth_state_fallback[state] = provider def _oauth_state_take(state: str) -> Optional[str]: """Retourne le provider associé au state (et le consomme une seule fois).""" conn = _mysql_conn() if conn is not None: try: with conn.cursor() as cur: cur.execute("SELECT provider FROM oauth_states WHERE state=%s", (state,)) row = cur.fetchone() cur.execute("DELETE FROM oauth_states WHERE state=%s", (state,)) conn.commit() return row["provider"] if row else None finally: conn.close() return _oauth_state_fallback.pop(state, None) # --------------------------------------------------------------------------- # App # --------------------------------------------------------------------------- app = FastAPI( title="Console IA — oai.by-cyberdev.com", description=( "Agrégateur de modèles IA gratuits. " "Catégorie SANS CLÉ (pool serveur) + catégorie API EXTERNE (clé utilisateur)." ), version="1.0.0", ) app.add_middleware( CORSMiddleware, allow_origins=["*"], allow_credentials=True, allow_methods=["*"], allow_headers=["*"], ) init_db() oauth.init_users_table() # --------------------------------------------------------------------------- # Modèles Pydantic # --------------------------------------------------------------------------- class LoginBody(BaseModel): username: str password: str class RegisterBody(BaseModel): username: str email: Optional[str] = None password: str class ChatBody(BaseModel): provider: str model: str message: str api_key: Optional[str] = None # pour la catégorie "API EXT", sinon déjà stockée system: Optional[str] = None max_tokens: Optional[int] = None # borne la sortie (latence maîtrisée) class TarotCardIn(BaseModel): id: int name: str class TarotUserIn(BaseModel): firstName: str = "" dateOfBirth: str = "" zodiacSign: str = "" class TarotPredictionBody(BaseModel): cards: List[TarotCardIn] = Field(min_length=1, max_length=22) spread: str = "five" language: str = "fr" user: TarotUserIn = TarotUserIn() pack: str = "" class TestKeyBody(BaseModel): provider: str api_key: str class SaveKeyBody(BaseModel): provider: str api_key: str class DeleteKeyBody(BaseModel): provider: str class ProjectBody(BaseModel): name: str provider: Optional[str] = "" model: Optional[str] = "" system: Optional[str] = "" class ProjectPatch(BaseModel): name: Optional[str] = None provider: Optional[str] = None model: Optional[str] = None system: Optional[str] = None class DiscussionBody(BaseModel): title: Optional[str] = "Nouvelle discussion" project_id: Optional[int] = None class DiscussionPatch(BaseModel): title: Optional[str] = None project_id: Optional[int] = None class MessageBody(BaseModel): role: str content: str meta: Optional[str] = None # --------------------------------------------------------------------------- # Helpers HTTP # --------------------------------------------------------------------------- def _resolve_api_key(provider_id: str, username: str, inline_key: Optional[str]) -> Optional[str]: """Ordre : clé inline (formulaire) > clé stockée utilisateur > clé serveur.""" if inline_key and inline_key.strip(): return inline_key.strip() if username: k = get_user_key(username, provider_id) if k: return k return SERVER_KEYS.get(provider_id) or None def chat_completion(provider_id: str, api_key: str, model: str, message: str, system: Optional[str] = None, max_tokens: Optional[int] = None): """Appel compatible OpenAI. Retourne (texte, sources_used).""" prov = providers.get_provider(provider_id) if not prov: raise HTTPException(404, "Fournisseur inconnu") # Endpoint chat/completions url = prov.base_url.rstrip("/") + "/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } if provider_id == "openrouter": headers["HTTP-Referer"] = f"https://{PUBLIC_HOST}" headers["X-Title"] = "Console IA" messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": message}) payload = { "model": model, "messages": messages, "temperature": 0.7, } if max_tokens: payload["max_tokens"] = int(max_tokens) t0 = time.time() try: resp = requests.post(url, json=payload, headers=headers, timeout=(15, HTTP_TIMEOUT)) except requests.exceptions.Timeout: raise HTTPException(504, "Délai dépassé (fournisseur lent)") except requests.exceptions.ConnectionError: raise HTTPException(502, "Impossible de joindre le fournisseur") latency_ms = int((time.time() - t0) * 1000) if resp.status_code == 401: raise HTTPException(401, "Clé API invalide (401). Vérifie le champ clé.") if resp.status_code == 404: body = resp.text[:300] raise HTTPException(404, f"Modèle introuvable (404). {body}") if resp.status_code != 200: body = resp.text[:300] raise HTTPException(resp.status_code, f"Erreur du fournisseur ({resp.status_code}): {body}") try: data = resp.json() text = data["choices"][0]["message"]["content"] except Exception: raise HTTPException(502, "Réponse illisible du fournisseur") return text.strip(), latency_ms def chat_stream(provider_id: str, api_key: str, model: str, message: str, system: Optional[str] = None, max_tokens: Optional[int] = None): """Appel compatible OpenAI en mode stream. Générateur qui renvoie les deltas de texte. Lève HTTPException si le fournisseur répond autre chose que des chunks SSE. """ prov = providers.get_provider(provider_id) if not prov: raise HTTPException(404, "Fournisseur inconnu") url = prov.base_url.rstrip("/") + "/chat/completions" headers = { "Authorization": f"Bearer {api_key}", "Content-Type": "application/json", } if provider_id == "openrouter": headers["HTTP-Referer"] = f"https://{PUBLIC_HOST}" headers["X-Title"] = "Console IA" messages = [] if system: messages.append({"role": "system", "content": system}) messages.append({"role": "user", "content": message}) payload = { "model": model, "messages": messages, "temperature": 0.7, "stream": True, } if max_tokens: payload["max_tokens"] = int(max_tokens) try: resp = requests.post(url, json=payload, headers=headers, stream=True, timeout=(15, HTTP_TIMEOUT)) except requests.exceptions.Timeout: raise HTTPException(504, "Délai dépassé (fournisseur lent)") except requests.exceptions.ConnectionError: raise HTTPException(502, "Impossible de joindre le fournisseur") if resp.status_code != 200: body = resp.text[:300] if resp.status_code == 401: raise HTTPException(401, "Clé API invalide (401). Vérifie le champ clé.") if resp.status_code == 404: raise HTTPException(404, f"Modèle introuvable (404). {body}") raise HTTPException(resp.status_code, f"Erreur du fournisseur ({resp.status_code}): {body}") # Décodage robuste du flux SSE. On NE découpe PAS les lignes avec # decode_unicode=True : un caractère UTF-8 multi-octets (ex. "é" = 2 octets) # coupé à cheval sur deux chunks SSE serait déformé (bug "é"/"û"). # On accumule les octets bruts et on ne décode que ce qui est complet. pending = b"" for chunk in resp.iter_content(chunk_size=4096): if not chunk: continue pending += chunk # Traite les lignes complètes terminées par \n (une seule ligne SSE peut # contenir un \n final ; le JSON "data: ..." ne contient pas de \n interne). while b"\n" in pending: line, pending = pending.split(b"\n", 1) raw = line.strip() if not raw: continue if raw.startswith(b"data: "): raw = raw[6:] if raw == b"[DONE]": resp.close() yield None return try: import json as _json chunk_json = _json.loads(raw.decode("utf-8")) delta = (chunk_json.get("choices") or [{}])[0].get("delta") or {} content = delta.get("content") if content: yield content except Exception: continue # Dernière ligne résiduelle sans \n final if pending: raw = pending.strip() if raw.startswith(b"data: "): raw = raw[6:] if raw and raw != b"[DONE]": try: import json as _json chunk_json = _json.loads(raw.decode("utf-8")) delta = (chunk_json.get("choices") or [{}])[0].get("delta") or {} content = delta.get("content") if content: yield content except Exception: pass resp.close() yield None # marqueur de fin # --------------------------------------------------------------------------- # Routes WEB # --------------------------------------------------------------------------- @app.get("/", response_class=HTMLResponse, include_in_schema=False) async def index(): path = BASE_DIR / "app" / "templates" / "index.html" if path.exists(): return HTMLResponse(path.read_text(encoding="utf-8")) return HTMLResponse("

Console IA

index.html manquant

") # --------------------------------------------------------------------------- # Routes AUTH (comptes + OAuth Google/GitHub/GitLab) # --------------------------------------------------------------------------- @app.post("/v1/login") async def login(body: LoginBody): # admin local (config) puis utilisateurs en base if body.username == ADMIN_USERNAME and body.password == AUTH_PASSWORD: token = _new_session("admin") resp = JSONResponse({"ok": True, "username": "admin"}) resp.set_cookie("oai_session", token, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 7) return resp user = oauth.verify_local(body.username.strip(), body.password) if not user: raise HTTPException(401, "Identifiants incorrects") token = _new_session(user["username"]) resp = JSONResponse({"ok": True, "username": user["username"]}) resp.set_cookie("oai_session", token, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 7) return resp @app.post("/v1/register") async def register(body: RegisterBody): username = body.username.strip() if len(username) < 3: raise HTTPException(400, "Nom d'utilisateur trop court (3+ caractères).") if len(body.password) < 6: raise HTTPException(400, "Mot de passe trop court (6+ caractères).") if body.email and "@" not in body.email: raise HTTPException(400, "Email invalide.") if username == ADMIN_USERNAME: raise HTTPException(400, "Ce nom d'utilisateur est réservé.") if oauth.list_username_taken(username): raise HTTPException(409, "Ce nom d'utilisateur existe déjà.") oauth.create_local_user(username, body.email, body.password) token = _new_session(username) resp = JSONResponse({"ok": True, "username": username}) resp.set_cookie("oai_session", token, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 7) return resp @app.post("/v1/logout") async def logout(request: Request): token = request.cookies.get("oai_session") if token: _session_drop(token) resp = JSONResponse({"ok": True}) resp.delete_cookie("oai_session") return resp @app.get("/v1/me") async def me(request: Request): user = _current_user(request) return {"authenticated": user is not None, "username": user, "oauth": {p: bool(c["client_id"]) for p, c in oauth.oauth_config().items()}} # --- OAuth : Google / GitHub / GitLab --- @app.get("/v1/auth/{provider}/login") async def oauth_login(provider: str, request: Request): base = str(request.base_url).rstrip("/") if provider not in oauth.oauth_config(): raise HTTPException(404, "Provider inconnu") if not oauth.oauth_config()[provider]["client_id"]: raise HTTPException(503, "Ce provider n'est pas configuré (pas de client_id).") state = oauth.make_new_state() _oauth_state_store(state, provider) url = oauth.build_authorize_url(provider, base, state) return RedirectResponse(url) @app.get("/v1/auth/{provider}/callback") async def oauth_callback(provider: str, code: str = None, state: str = None, error: str = None): if error: raise HTTPException(400, f"OAuth refusé: {error}") if not code or not state: raise HTTPException(400, "Paramètres manquants (code/state).") expected_provider = _oauth_state_take(state) if expected_provider != provider or expected_provider is None: raise HTTPException(400, "État OAuth invalide.") base_url = f"https://{PUBLIC_HOST}" try: profile = oauth.exchange_code(provider, code, base_url) except Exception as e: raise HTTPException(502, f"Échec OAuth: {e}") user = oauth.upsert_oauth_user(provider, profile["provider_id"], profile["username"], profile["email"], profile["avatar"]) token = _new_session(user["username"]) resp = RedirectResponse(base_url + "/") resp.set_cookie("oai_session", token, httponly=True, samesite="lax", max_age=60 * 60 * 24 * 7) return resp # --------------------------------------------------------------------------- # Routes PROJETS / DISCUSSIONS / MESSAGES (persistance MySQL) # --------------------------------------------------------------------------- def _require_user(request: Request) -> str: user = _current_user(request) if not user: raise HTTPException(401, "Non connecté") return user def _fmt_disc(row: dict, extra: dict = None) -> dict: out = { "id": row["id"], "title": row["title"], "project_id": row["project_id"], "share_enabled": bool(row["share_enabled"]), "created_at": str(row["created_at"]), "updated_at": str(row["updated_at"]), } if extra: out.update(extra) return out @app.get("/v1/projects") async def list_projects(request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: rows = [p for p in _load_store().get("projects", []) if p.get("username") == user] rows.sort(key=lambda p: p.get("updated_at", ""), reverse=True) return {"projects": rows} try: with conn.cursor() as cur: cur.execute("SELECT * FROM projects WHERE username=%s ORDER BY updated_at DESC", (user,)) rows = cur.fetchall() return {"projects": rows} finally: conn.close() @app.post("/v1/projects") async def create_project(body: ProjectBody, request: Request): user = _require_user(request) name = (body.name or "").strip() if not name: raise HTTPException(400, "Nom de projet requis") conn = _mysql_conn() if conn is None: store = _load_store() rid = _next_store_id(store, "projects") now = _now_iso() store.setdefault("projects", []).append({ "id": rid, "username": user, "name": name, "provider": body.provider or "", "model": body.model or "", "system": body.system or "", "share_token": None, "share_enabled": 0, "created_at": now, "updated_at": now, }) _save_store(store) return JSONResponse({"ok": True, "id": rid}) try: with conn.cursor() as cur: cur.execute( "INSERT INTO projects (username, name, provider, model, system) " "VALUES (%s,%s,%s,%s,%s)", (user, name, body.provider or "", body.model or "", body.system or ""), ) rid = cur.lastrowid conn.commit() return JSONResponse({"ok": True, "id": rid}) finally: conn.close() @app.put("/v1/projects/{pid}") async def update_project(pid: int, body: ProjectPatch, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("projects", []), pid, user) if not row: raise HTTPException(404, "Projet introuvable") for col, val in (("name", body.name), ("provider", body.provider), ("model", body.model), ("system", body.system)): if val is not None: row[col] = val.strip() if col == "name" else val row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: fields, args = [], [] for col, val in (("name", body.name), ("provider", body.provider), ("model", body.model), ("system", body.system)): if val is not None: fields.append(f"{col}=%s") args.append(val) if not fields: return JSONResponse({"ok": True}) args.extend([pid, user]) with conn.cursor() as cur: cur.execute( f"UPDATE projects SET {', '.join(fields)} WHERE id=%s AND username=%s", args, ) if cur.rowcount == 0: raise HTTPException(404, "Projet introuvable") conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.delete("/v1/projects/{pid}") async def delete_project(pid: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() before = len(store.get("projects", [])) store["projects"] = [ p for p in store.get("projects", []) if not (int(p.get("id", 0)) == int(pid) and p.get("username") == user) ] if len(store["projects"]) == before: raise HTTPException(404, "Projet introuvable") for disc in store.get("discussions", []): if disc.get("username") == user and int(disc.get("project_id") or 0) == int(pid): disc["project_id"] = None disc["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: # on détache les discussions du projet (sans les supprimer) cur.execute("UPDATE discussions SET project_id=NULL WHERE project_id=%s AND username=%s", (pid, user)) cur.execute("DELETE FROM projects WHERE id=%s AND username=%s", (pid, user)) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.post("/v1/projects/{pid}/share") async def share_project(pid: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("projects", []), pid, user) if not row: raise HTTPException(404, "Projet introuvable") token = row.get("share_token") or (uuid.uuid4().hex + uuid.uuid4().hex[:12]) row["share_token"] = token row["share_enabled"] = 1 row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True, "token": token, "url": _public_url(f"/project/{token}")}) try: with conn.cursor() as cur: cur.execute("SELECT id, share_token FROM projects WHERE id=%s AND username=%s", (pid, user)) row = cur.fetchone() if not row: raise HTTPException(404, "Projet introuvable") token = row.get("share_token") or (uuid.uuid4().hex + uuid.uuid4().hex[:12]) cur.execute( "UPDATE projects SET share_token=%s, share_enabled=1 WHERE id=%s AND username=%s", (token, pid, user), ) conn.commit() return JSONResponse({"ok": True, "token": token, "url": _public_url(f"/project/{token}")}) finally: conn.close() @app.delete("/v1/projects/{pid}/share") async def unshare_project(pid: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("projects", []), pid, user) if not row: raise HTTPException(404, "Projet introuvable") row["share_enabled"] = 0 row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute( "UPDATE projects SET share_enabled=0 WHERE id=%s AND username=%s", (pid, user), ) if cur.rowcount == 0: raise HTTPException(404, "Projet introuvable") conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.get("/v1/discussions") async def list_discussions(request: Request, project_id: int = None): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() rows = [d for d in store.get("discussions", []) if d.get("username") == user and (project_id is None or int(d.get("project_id") or 0) == int(project_id))] out = [] for d in rows: msg_count = sum( 1 for m in store.get("messages", []) if int(m.get("discussion_id", 0)) == int(d.get("id", 0)) and m.get("username") == user ) out.append(_fmt_disc(d, {"msg_count": msg_count})) out.sort(key=lambda d: d.get("updated_at", ""), reverse=True) return {"discussions": out} try: with conn.cursor() as cur: if project_id is None: cur.execute( "SELECT d.*, " "(SELECT COUNT(*) FROM messages m WHERE m.discussion_id=d.id) AS msg_count " "FROM discussions d WHERE d.username=%s ORDER BY d.updated_at DESC", (user,), ) else: cur.execute( "SELECT d.*, " "(SELECT COUNT(*) FROM messages m WHERE m.discussion_id=d.id) AS msg_count " "FROM discussions d WHERE d.username=%s AND d.project_id=%s " "ORDER BY d.updated_at DESC", (user, project_id), ) rows = cur.fetchall() out = [] for r in rows: d = dict(r) out.append(_fmt_disc(d, {"msg_count": d.get("msg_count", 0)})) return {"discussions": out} finally: conn.close() @app.post("/v1/discussions") async def create_discussion(body: DiscussionBody, request: Request): user = _require_user(request) title = (body.title or "Nouvelle discussion").strip()[:160] conn = _mysql_conn() if conn is None: store = _load_store() rid = _next_store_id(store, "discussions") now = _now_iso() store.setdefault("discussions", []).append({ "id": rid, "username": user, "project_id": body.project_id if body.project_id else None, "title": title, "share_token": None, "share_enabled": 0, "created_at": now, "updated_at": now, }) _save_store(store) return JSONResponse({"ok": True, "id": rid, "title": title}) try: with conn.cursor() as cur: cur.execute( "INSERT INTO discussions (username, project_id, title) VALUES (%s,%s,%s)", (user, body.project_id if body.project_id else None, title), ) rid = cur.lastrowid conn.commit() return JSONResponse({"ok": True, "id": rid, "title": title}) finally: conn.close() @app.get("/v1/discussions/{did}") async def get_discussion(did: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("discussions", []), did, user) if not row: raise HTTPException(404, "Discussion introuvable") proj = None if row.get("project_id"): proj = _find_owned(store.get("projects", []), row["project_id"], user) return {**_fmt_disc(row), "project": proj} try: with conn.cursor() as cur: cur.execute("SELECT * FROM discussions WHERE id=%s AND username=%s", (did, user)) row = cur.fetchone() if not row: raise HTTPException(404, "Discussion introuvable") proj = None if row["project_id"]: cur.execute("SELECT * FROM projects WHERE id=%s AND username=%s", (row["project_id"], user)) proj = cur.fetchone() return {**_fmt_disc(dict(row)), "project": proj} finally: conn.close() @app.put("/v1/discussions/{did}") async def update_discussion(did: int, body: DiscussionPatch, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("discussions", []), did, user) if not row: raise HTTPException(404, "Discussion introuvable") if body.title is not None: row["title"] = body.title.strip()[:160] if body.project_id is not None: row["project_id"] = body.project_id if body.project_id > 0 else None row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute("SELECT id FROM discussions WHERE id=%s AND username=%s", (did, user)) if not cur.fetchone(): raise HTTPException(404, "Discussion introuvable") if body.title is not None: cur.execute("UPDATE discussions SET title=%s WHERE id=%s", (body.title.strip()[:160], did)) if body.project_id is not None: cur.execute("UPDATE discussions SET project_id=%s WHERE id=%s", (body.project_id if body.project_id > 0 else None, did)) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.delete("/v1/discussions/{did}") async def delete_discussion(did: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("discussions", []), did, user) if not row: raise HTTPException(404, "Discussion introuvable") store["messages"] = [ m for m in store.get("messages", []) if not (int(m.get("discussion_id", 0)) == int(did) and m.get("username") == user) ] store["discussions"] = [ d for d in store.get("discussions", []) if not (int(d.get("id", 0)) == int(did) and d.get("username") == user) ] _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute("DELETE FROM messages WHERE discussion_id=%s AND username=%s", (did, user)) cur.execute("DELETE FROM discussions WHERE id=%s AND username=%s", (did, user)) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.get("/v1/discussions/{did}/messages") async def list_messages(did: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() if not _find_owned(store.get("discussions", []), did, user): raise HTTPException(404, "Discussion introuvable") rows = [ { "id": m.get("id"), "role": m.get("role"), "content": m.get("content"), "meta": m.get("meta"), "created_at": m.get("created_at"), } for m in store.get("messages", []) if int(m.get("discussion_id", 0)) == int(did) and m.get("username") == user ] rows.sort(key=lambda m: int(m.get("id", 0))) return {"messages": rows} try: with conn.cursor() as cur: cur.execute("SELECT id FROM discussions WHERE id=%s AND username=%s", (did, user)) if not cur.fetchone(): raise HTTPException(404, "Discussion introuvable") cur.execute( "SELECT id, role, content, meta, created_at FROM messages " "WHERE discussion_id=%s AND username=%s ORDER BY id ASC", (did, user), ) rows = cur.fetchall() return {"messages": rows} finally: conn.close() @app.post("/v1/discussions/{did}/messages") async def add_message(did: int, body: MessageBody, request: Request): user = _require_user(request) role = (body.role or "").strip() if role not in ("user", "assistant", "note"): raise HTTPException(400, "Rôle invalide") conn = _mysql_conn() if conn is None: store = _load_store() disc = _find_owned(store.get("discussions", []), did, user) if not disc: raise HTTPException(404, "Discussion introuvable") mid = _next_store_id(store, "messages") store.setdefault("messages", []).append({ "id": mid, "discussion_id": did, "username": user, "role": role, "content": body.content, "meta": body.meta, "created_at": _now_iso(), }) if role == "user" and disc.get("title") == "Nouvelle discussion": disc["title"] = body.content.strip()[:60] disc["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True, "id": mid}) try: with conn.cursor() as cur: cur.execute("SELECT id FROM discussions WHERE id=%s AND username=%s", (did, user)) if not cur.fetchone(): raise HTTPException(404, "Discussion introuvable") cur.execute( "INSERT INTO messages (discussion_id, username, role, content, meta) " "VALUES (%s,%s,%s,%s,%s)", (did, user, role, body.content, body.meta), ) mid = cur.lastrowid # met à jour le titre auto si c'est le 1er message utilisateur if role == "user": cur.execute( "UPDATE discussions SET title=%s WHERE id=%s AND title='Nouvelle discussion'", (body.content.strip()[:60], did), ) conn.commit() return JSONResponse({"ok": True, "id": mid}) finally: conn.close() class MessagesReplace(BaseModel): messages: list # list[{role, content, meta?}] @app.put("/v1/discussions/{did}/messages") async def replace_messages(did: int, body: MessagesReplace, request: Request): """Remplace tout le contenu de la discussion (sync complet côté front).""" user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() disc = _find_owned(store.get("discussions", []), did, user) if not disc: raise HTTPException(404, "Discussion introuvable") store["messages"] = [ m for m in store.get("messages", []) if not (int(m.get("discussion_id", 0)) == int(did) and m.get("username") == user) ] for m in body.messages or []: role = (m.get("role") or "").strip() if role not in ("user", "assistant", "note"): role = "note" store.setdefault("messages", []).append({ "id": _next_store_id(store, "messages"), "discussion_id": did, "username": user, "role": role, "content": m.get("content", ""), "meta": m.get("meta"), "created_at": _now_iso(), }) disc["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute("SELECT id FROM discussions WHERE id=%s AND username=%s", (did, user)) if not cur.fetchone(): raise HTTPException(404, "Discussion introuvable") cur.execute("DELETE FROM messages WHERE discussion_id=%s AND username=%s", (did, user)) for m in body.messages or []: role = (m.get("role") or "").strip() if role not in ("user", "assistant", "note"): role = "note" cur.execute( "INSERT INTO messages (discussion_id, username, role, content, meta) " "VALUES (%s,%s,%s,%s,%s)", (did, user, role, m.get("content", ""), m.get("meta")), ) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.delete("/v1/discussions/{did}/messages/{mid}") async def delete_message(did: int, mid: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() before = len(store.get("messages", [])) store["messages"] = [ m for m in store.get("messages", []) if not ( int(m.get("id", 0)) == int(mid) and int(m.get("discussion_id", 0)) == int(did) and m.get("username") == user ) ] if len(store["messages"]) == before: raise HTTPException(404, "Message introuvable") disc = _find_owned(store.get("discussions", []), did, user) if disc: disc["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute( "DELETE FROM messages WHERE id=%s AND discussion_id=%s AND username=%s", (mid, did, user), ) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() # --- Partage public (lecture seule) --- @app.post("/v1/discussions/{did}/share") async def share_discussion(did: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("discussions", []), did, user) if not row: raise HTTPException(404, "Discussion introuvable") token = row.get("share_token") or (uuid.uuid4().hex + uuid.uuid4().hex[:12]) row["share_token"] = token row["share_enabled"] = 1 row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True, "token": token, "url": _public_url(f"/share/{token}")}) try: with conn.cursor() as cur: cur.execute("SELECT id FROM discussions WHERE id=%s AND username=%s", (did, user)) if not cur.fetchone(): raise HTTPException(404, "Discussion introuvable") cur.execute("SELECT share_token FROM discussions WHERE id=%s", (did,)) tok = cur.fetchone() if tok and tok["share_token"]: token = tok["share_token"] else: token = uuid.uuid4().hex + uuid.uuid4().hex[:12] cur.execute("UPDATE discussions SET share_token=%s, share_enabled=1 WHERE id=%s", (token, did)) conn.commit() return JSONResponse({"ok": True, "token": token, "url": _public_url(f"/share/{token}")}) finally: conn.close() @app.delete("/v1/discussions/{did}/share") async def unshare_discussion(did: int, request: Request): user = _require_user(request) conn = _mysql_conn() if conn is None: store = _load_store() row = _find_owned(store.get("discussions", []), did, user) if not row: raise HTTPException(404, "Discussion introuvable") row["share_enabled"] = 0 row["updated_at"] = _now_iso() _save_store(store) return JSONResponse({"ok": True}) try: with conn.cursor() as cur: cur.execute( "UPDATE discussions SET share_enabled=0 WHERE id=%s AND username=%s", (did, user), ) conn.commit() return JSONResponse({"ok": True}) finally: conn.close() @app.get("/v1/share/{token}") async def shared_discussion(token: str): """Lecture seule, publique. Retourne le contenu de la discussion partagée.""" conn = _mysql_conn() if conn is None: store = _load_store() row = next((d for d in store.get("discussions", []) if d.get("share_token") == token and d.get("share_enabled")), None) if not row: raise HTTPException(404, "Partage introuvable ou désactivé") msgs = [ { "role": m.get("role"), "content": m.get("content"), "meta": m.get("meta"), "created_at": m.get("created_at"), } for m in store.get("messages", []) if int(m.get("discussion_id", 0)) == int(row.get("id", 0)) ] msgs.sort(key=lambda m: str(m.get("created_at", ""))) return { "title": row["title"], "author": row["username"], "created_at": str(row["created_at"]), "messages": msgs, } try: with conn.cursor() as cur: cur.execute( "SELECT d.id, d.title, d.username, d.created_at FROM discussions d " "WHERE d.share_token=%s AND d.share_enabled=1", (token,), ) row = cur.fetchone() if not row: raise HTTPException(404, "Partage introuvable ou désactivé") cur.execute( "SELECT role, content, meta, created_at FROM messages " "WHERE discussion_id=%s ORDER BY id ASC", (row["id"],), ) msgs = cur.fetchall() return { "title": row["title"], "author": row["username"], "created_at": str(row["created_at"]), "messages": msgs, } finally: conn.close() @app.get("/v1/share/project/{token}") async def shared_project(token: str): """Lecture seule, publique. Retourne le gabarit d'un projet partagé.""" conn = _mysql_conn() if conn is None: row = next((p for p in _load_store().get("projects", []) if p.get("share_token") == token and p.get("share_enabled")), None) if not row: raise HTTPException(404, "Projet partagé introuvable ou désactivé") return { "name": row["name"], "author": row["username"], "provider": row.get("provider", ""), "model": row.get("model", ""), "system": row.get("system", ""), "created_at": str(row.get("created_at", "")), } try: with conn.cursor() as cur: cur.execute( "SELECT name, username, provider, model, system, created_at " "FROM projects WHERE share_token=%s AND share_enabled=1", (token,), ) row = cur.fetchone() if not row: raise HTTPException(404, "Projet partagé introuvable ou désactivé") return { "name": row["name"], "author": row["username"], "provider": row.get("provider", ""), "model": row.get("model", ""), "system": row.get("system", ""), "created_at": str(row.get("created_at", "")), } finally: conn.close() @app.get("/project/{token}") async def shared_project_page(token: str): """Page publique en lecture seule d'un projet partagé.""" page = """ Projet partagé — Console IA
""" return HTMLResponse(page.replace("__TOKEN__", token)) @app.get("/share/{token}") async def shared_page(token: str): """Page publique en lecture seule d'une discussion partagée.""" page = """ Discussion partagée — Console IA
""" return HTMLResponse(page.replace("__TOKEN__", token)) # --------------------------------------------------------------------------- # Routes FOURNISSEURS / CATALOGUE # --------------------------------------------------------------------------- @app.get("/v1/catalog") async def catalog(request: Request): """Renvoie la liste des fournisseurs triée par catégorie.""" scan = _load_scan() no_key = [ { "id": p.id, "name": p.name, "logo": p.logo, "note": p.note, "models": [{"id": m[0], "label": m[1]} for m in p.models] + _scan_free_extra(p.id, scan), } for p in providers.NO_KEY_PROVIDERS ] user_key = [ { "id": p.id, "name": p.name, "logo": p.logo, "note": p.note, "models": [{"id": m[0], "label": m[1]} for m in p.models] + _scan_paid_extra(p.id, scan), "hasKey": bool(_current_user(request) and get_user_key(_current_user(request), p.id)), } for p in providers.USER_KEY_PROVIDERS ] return {"categories": {"no_key": {"title": "Modèles gratuits — sans saisie de clé", "providers": no_key}, "user_key": {"title": "API externes — avec clé (API EXT)", "providers": user_key}}, "scan_at": scan.get("generated_at")} # Cache mémoire du scan (rafraîchi toutes les 10 min) : évite de relire le disque # à chaque requête mais reflète les nouvelles découvertes sans redémarrer. _scan_cache: dict = {"data": None, "ts": 0.0} def _load_scan(): import time as _t p = Path(__file__).resolve().parent / "app" / "data" / "catalog_scan.json" if not p.exists(): return {} now = _t.time() if _scan_cache["data"] is None or now - _scan_cache["ts"] > 600: try: _scan_cache["data"] = json.loads(p.read_text()) except Exception: _scan_cache["data"] = {} _scan_cache["ts"] = now return _scan_cache["data"] def _scan_free_extra(prov_id: str, scan: dict): """Modèles :free découverts par le scan, ajoutés au pool sans clé (OpenRouter).""" if prov_id != "openrouter": return [] free = (scan.get("providers", {}).get(prov_id, {}) or {}).get("free") or [] curated = {m[0] for p in providers.NO_KEY_PROVIDERS for m in p.models} out = [] for mid in free: if mid in curated: continue nice = mid.split("/")[-1].replace(":free", "").replace("-", " ") out.append({"id": mid, "label": f"{nice} :free (auto)", "auto": True}) return out def _scan_paid_extra(prov_id: str, scan: dict): """Modèles payants/externes découverts par le scan, ajoutés au provider.""" paid = (scan.get("providers", {}).get(prov_id, {}) or {}).get("paid") or [] curated = {m[0] for p in providers.USER_KEY_PROVIDERS for m in p.models} out = [] for mid in paid[:40]: # limite raisonnable par provider if mid in curated: continue nice = mid.split("/")[-1].replace("-", " ") out.append({"id": mid, "label": f"{nice} (auto)", "auto": True}) return out def _sse_payload(payload: dict) -> str: return "data: " + json.dumps(payload, ensure_ascii=False) + "\n\n" # --------------------------------------------------------------------------- # Routes CHAT # --------------------------------------------------------------------------- @app.post("/v1/chat") async def chat(body: ChatBody, request: Request): user = _current_user(request) prov = providers.get_provider(body.provider) if not prov: raise HTTPException(404, "Fournisseur inconnu") if not body.message.strip(): raise HTTPException(400, "Message vide") _check_quota(user, prov.id) if prov.needs_user_key: if not user: raise HTTPException(401, "Connecte-toi pour utiliser une API externe") api_key = _resolve_api_key(prov.id, user, body.api_key) if not api_key: raise HTTPException( 400, "Aucune clé pour ce fournisseur. Saisis-la dans le menu API EXT.") else: api_key = SERVER_KEYS.get(prov.id) if not api_key: raise HTTPException( 503, "Clé serveur non configurée pour ce fournisseur.") text, latency_ms = chat_completion(prov.id, api_key, body.model, body.message, body.system, body.max_tokens) if user: log_chat(user, prov.id, body.model, body.message, text, latency_ms) return { "provider": prov.id, "model": body.model, "reply": text, "latency_ms": latency_ms, } @app.post("/v1/predictions/tarot") async def tarot_prediction(body: TarotPredictionBody, request: Request): """Prédiction structurée de tirage de tarot (JSON par carte). Reçoit les ids/noms des cartes + profil utilisateur, interroge le modèle (pool OpenRouter, clé serveur) et renvoie un JSON déterministe : {"cards": [{"id": 1, "text": "..."}, ...], "summary": "..."} Les apps mobiles (iOS/Android) n'ont plus à parser du Markdown. """ user = _current_user(request) prov = providers.get_provider("openrouter") if not prov: raise HTTPException(404, "Fournisseur inconnu") api_key = SERVER_KEYS.get(prov.id) if not api_key: raise HTTPException(503, "Clé serveur non configurée pour ce fournisseur.") _check_quota(user, prov.id) message, system = build_tarot_prediction_prompt(body) model = "deepseek/deepseek-v4-flash" data: dict = {} latency_ms = 0 for _attempt in range(2): text, latency_ms = chat_completion(prov.id, api_key, model, message, system, max_tokens=2200) data = _extract_json_object(text) n_ok = sum(1 for c in (data.get("cards", []) or []) if str(c.get("interpretation", "") or "").strip()) if n_ok >= len(body.cards): break req_ids = [c.id for c in body.cards] cards_in = data.get("cards", []) or [] # Le modèle renvoie les cartes DANS L'ORDRE du tirage (Carte 1..N) mais # recopie parfois mal les ids (doublons / renumérotation). On privilégie # donc l'ordre des positions : réponse[i] => carte de la requête req_ids[i]. by_order = len(cards_in) >= len(req_ids) cards_out = [] used_ids = set() for i, c in enumerate(cards_in): if by_order: cid = req_ids[i] if i < len(req_ids) else 0 else: try: cid = int(c.get("id", 0)) except (TypeError, ValueError): cid = 0 if cid not in req_ids or cid in used_ids: cid = req_ids[i] if i < len(req_ids) else 0 def _txt(*keys): for k in keys: v = str(c.get(k, "") or "").strip() if v: return v return "" if cid and cid not in used_ids: if not _txt("interpretation", "text"): continue cards_out.append({ "id": cid, "long_description": _txt("long_description", "text"), "signification": _txt("signification", "text"), "interpretation": _txt("interpretation", "text"), "astro_message": _txt("astro_message", "text"), }) used_ids.add(cid) summary = str(data.get("summary", "") or "").strip() return { "cards": cards_out, "summary": summary, "provider": prov.id, "model": model, "latency_ms": latency_ms, } _SPREAD_LABELS = { "five": { "fr": ("tirage à cinq cartes", "Carte 1 = Situation passée, Carte 2 = Situation actuelle, " "Carte 3 = Défis, Carte 4 = Conseil, Carte 5 = Résultat probable."), "en": ("five-card spread", "Card 1 = Past situation, Card 2 = Current situation, " "Card 3 = Challenges, Card 4 = Advice, Card 5 = Likely outcome."), "ar": ("توزيع بخمس بطاقات", "البطاقة 1 = الموقف الماضي، البطاقة 2 = الموقف الحالي، " "البطاقة 3 = التحديات، البطاقة 4 = نصيحة، البطاقة 5 = النتيجة المحتملة."), "zh": ("五张牌阵", "牌1 = 过去的情况，牌2 = 当前情况，牌3 = 挑战，牌4 = 建议，牌5 = 可能的结果。"), }, "three": { "fr": ("tirage à trois cartes", "Carte 1 = Passé, Carte 2 = Présent, Carte 3 = Avenir probable."), "en": ("three-card spread", "Card 1 = Past, Card 2 = Present, Card 3 = Likely future."), "ar": ("توزيع بثلاث بطاقات", "البطاقة 1 = الماضي، البطاقة 2 = الحاضر، البطاقة 3 = المستقبل المحتمل."), "zh": ("三张牌阵", "牌1 = 过去，牌2 = 现在，牌3 = 可能的未来。"), }, "one": { "fr": ("tirage d'une seule carte", "Cette carte unique révèle l'énergie dominante de la situation."), "en": ("single card draw", "This single card reveals the dominant energy of the situation."), "ar": ("سحب بطاقة واحدة", "تكشف هذه البطاقة الوحيدة الطاقة السائدة في الموقف."), "zh": ("单张牌抽牌", "这张唯一的牌揭示了当前情况的主导能量。"), }, } _JSON_OUTPUT_LANG = { "fr": "Réponds UNIQUEMENT avec un objet JSON valide, sans markdown ni texte autour.", "en": "Reply ONLY with a valid JSON object, without markdown or surrounding text.", "ar": "أجب حصريًا بكائن JSON صالح، بدون Markdown أو نص حوله.", "zh": "仅用有效的JSON对象回答，不要使用markdown或周围文本。", } def build_tarot_prediction_prompt(body: TarotPredictionBody): lang = body.language if body.language in ("fr", "en", "ar", "zh") else "fr" name_in = body.user.firstName or "" dob_in = body.user.dateOfBirth or "" sign_in = body.user.zodiacSign or "" pack_in = body.pack or "" spread = body.spread if body.spread in _SPREAD_LABELS else "five" spread_name, positions = _SPREAD_LABELS[spread][lang] card_line = ", ".join(f"{c.id}={c.name}" for c in body.cards) if lang == "fr": system = ( "Tu es TarotAstro, un expert mondial en tarot, astrologie et divination. " "Tu réponds toujours dans la langue demandée, de manière chaleureuse, " "précise et personnelle. Tu ne révèles jamais ton caractère artificiel." ) prompt = f"""Je lis les cartes pour {name_in}, né(e) le {dob_in}, signe {sign_in}. Pack : {pack_in}. Type de tirage : {spread_name}. {positions} Cartes : {card_line} Pour CHAQUE carte, fournis 4 textes distincts en français (jamais identiques entre eux) : - "long_description" : description détaillée (3 phrases) du symbole et de l'énergie de la carte. - "signification" : signification traditionnelle (2-3 phrases), inversée ou droite selon le contexte. - "interpretation" : interprétation personnalisée (3-4 phrases) pour {name_in}, selon la position dans le tirage. - "astro_message" : message lié à l'astrologie (2 phrases, évoque le signe {sign_in} ou une planète associée à la carte). Ensuite, pour l'ANALYSE GLOBALE de ce tirage ({spread_name}), rédige un texte de 3-5 lignes qui synthétise l'ensemble des cartes (clé \"summary\"). {_JSON_OUTPUT_LANG[lang]} Format EXACT : {{\"cards\":[{{\"id\":\"1\",\"long_description\":\"...\",\"signification\":\"...\",\"interpretation\":\"...\",\"astro_message\":\"...\"}}],\"summary\":\"analyse globale du tirage\"}} (Utilise le champ \"id\" de la carte et renseigne TOUTES les cartes avec TOUS les champs.)""" elif lang == "ar": system = ( "أنت TarotAstro، خبير عالمي في التاروت وعلم الفلك والعرافة. " "تجاوب دائمًا باللغة المطلوبة بحرارة ودقة وشخصية. لا تكشف أبدًا أنك ذكاء اصطناعي." ) prompt = f"""أقرأ البطاقات لـ {name_in}، مواليد {dob_in}، البرج {sign_in}. المجموعة: {pack_in}. نوع التوزيع: {spread_name}. {positions} البطاقات: {card_line} لكل بطاقة، قدّم 4 نصوص مختلفة بالعربية (لا تتطابق أبدًا): - "long_description": وصف مفصل (3 جمل) لرمز البطاقة وطاقتها. - "signification": المعنى التقليدي (2-3 جمل)، مقلوبًا أو مستقيمًا حسب السياق. - "interpretation": تفسير مخصص (3-4 جمل) لـ {name_in} حسب موقع البطاقة في التوزيع. - "astro_message": رسالة فلكية (جملتان) تذكر البرج {sign_in} أو كوكبًا مرتبطًا بالبطاقة. ثم، للتحليل العام لهذا {spread_name}، اكتب نصًا من 3-5 أسطر يلخّص كل البطاقات (المفتاح "summary"). {_JSON_OUTPUT_LANG[lang]} الصيغة الدقيقة: {{"cards":[{{"id":"1","long_description":"...","signification":"...","interpretation":"...","astro_message":"..."}}],"summary":"التحليل العام للتوزيع"}} (استخدم حقل "id" وأدخل جميع البطاقات مع جميع الحقول.)""" elif lang == "zh": system = ( "你是TarotAstro，塔罗牌、占星术和占卜领域的世界级专家。" "你总是用要求的语言作答，态度热情、准确、个性。你从不透露你是人工智能。" ) prompt = f"""我为{name_in}解读牌阵，出生日期：{dob_in}，星座：{sign_in}。 牌组：{pack_in}。 牌阵类型：{spread_name}。{positions} 牌面：{card_line} 对每张牌，用中文提供4段不同的文字（绝不彼此相同）： - "long_description": 这张牌符号与能量的详细描述（3句）。 - "signification": 传统含义（2-3句），正位或逆位依上下文而定。 - "interpretation": 针对{name_in}的个性化解读（3-4句），结合牌阵中的位置。 - "astro_message": 占星信息（2句），提到{sign_in}星座或与牌相关的行星。 然后，针对本{spread_name}作整体分析，写一段3-5行、综合所有牌的文字（键为"summary"）。 {_JSON_OUTPUT_LANG[lang]} 准确格式： {{"cards":[{{"id":"1","long_description":"...","signification":"...","interpretation":"...","astro_message":"..."}}],"summary":"牌阵整体分析"}} （使用牌面的"id"字段，并填写所有牌的所有字段。）""" else: # en system = ( "You are TarotAstro, a world-class expert in tarot, astrology and " "divination. You always answer in the requested language, warmly, " "precisely and personally. You never reveal that you are an AI." ) prompt = f"""I am reading cards for {name_in}, born {dob_in}, sign {sign_in}. Deck: {pack_in}. Spread type: {spread_name}. {positions} Cards: {card_line} For EACH card, provide 4 distinct texts in English (never identical to each other): - "long_description": detailed description (3 sentences) of the card's symbol and energy. - "signification": traditional meaning (2-3 sentences), reversed or upright depending on context. - "interpretation": personalized interpretation (3-4 sentences) for {name_in}, according to the position in the spread. - "astro_message": astrology-related message (2 sentences, mention sign {sign_in} or a planet linked to the card). Then, for the GLOBAL ANALYSIS of this {spread_name}, write a 3-5 line text synthesizing all the cards (key "summary"). {_JSON_OUTPUT_LANG[lang]} EXACT format: {{"cards":[{{"id":"1","long_description":"...","signification":"...","interpretation":"...","astro_message":"..."}}],"summary":"global analysis of the spread"}} (Use the card's "id" field and fill ALL cards with ALL fields.)""" return prompt, system def _extract_json_object(text: str) -> dict: """Extrait l'objet JSON le plus complet d'une réponse (tolère fences, texte, troncature et accolades déséquilibrées dans les valeurs).""" t = text.strip() if t.startswith("```"): t = t.split("```", 2)[1] if t.count("```") >= 2 else t.strip("`") if t.startswith("json"): t = t[len("json"):].lstrip("\n ") start = t.find("{") if start == -1: return {} best: dict = {} best_score = -1 for i in range(len(t) - 1, start - 1, -1): if t[i] != "}": continue try: data = json.loads(t[start:i + 1]) except Exception: continue if not isinstance(data, dict): continue score = len(data) + len((data.get("cards", []) or [])) if score > best_score: best = data best_score = score return best @app.post("/v1/chat/stream") async def chat_stream_endpoint(body: ChatBody, request: Request): """Chat en streaming (SSE). Renvoie les deltas de texte au fil de l'eau. Format SSE : `data: ` puis `data: [DONE]` et `data: [ERR] `. """ user = _current_user(request) prov = providers.get_provider(body.provider) if not prov: raise HTTPException(404, "Fournisseur inconnu") if not body.message.strip(): raise HTTPException(400, "Message vide") _check_quota(user, prov.id) if prov.needs_user_key: if not user: raise HTTPException(401, "Connecte-toi pour utiliser une API externe") api_key = _resolve_api_key(prov.id, user, body.api_key) if not api_key: raise HTTPException( 400, "Aucune clé pour ce fournisseur. Saisis-la dans le menu API EXT.") else: api_key = SERVER_KEYS.get(prov.id) if not api_key: raise HTTPException( 503, "Clé serveur non configurée pour ce fournisseur.") def gen(): started = time.time() parts = [] try: for token in chat_stream(prov.id, api_key, body.model, body.message, body.system, body.max_tokens): if token is None: if user: text = "".join(parts).strip() latency_ms = int((time.time() - started) * 1000) log_chat(user, prov.id, body.model, body.message, text, latency_ms) yield _sse_payload({"done": True}) break parts.append(token) yield _sse_payload({"delta": token}) except HTTPException as e: # Nettoie la ligne courante puis envoie l'erreur yield _sse_payload({"error": _ss(e.detail)}) except Exception as e: # pragma: no cover yield _sse_payload({"error": _ss(str(e))}) return StreamingResponse(gen(), media_type="text/event-stream", headers={"Cache-Control": "no-cache", "X-Accel-Buffering": "no"}) def _ss(s) -> str: """Sérialise une valeur dans le champ event-stream (échappé sur une ligne).""" return str(s).replace("\r", " ").replace("\n", " ") # --------------------------------------------------------------------------- # Routes CLÉS (API EXT) # --------------------------------------------------------------------------- @app.post("/v1/keys") async def save_key(body: SaveKeyBody, request: Request): user = _current_user(request) if not user: raise HTTPException(401, "Non connecté") if not providers.get_provider(body.provider): raise HTTPException(404, "Fournisseur inconnu") if not body.api_key.strip(): raise HTTPException(400, "Clé vide") save_user_key(user, body.provider, body.api_key.strip()) return {"ok": True, "provider": body.provider, "saved": True} @app.post("/v1/keys/delete") async def del_key(body: DeleteKeyBody, request: Request): user = _current_user(request) if not user: raise HTTPException(401, "Non connecté") delete_user_key(user, body.provider) return {"ok": True, "provider": body.provider, "deleted": True} @app.post("/v1/providers/test") async def test_key(body: TestKeyBody, request: Request): """Valide une clé (ping) et mesure la latence. N'enregistre PAS la clé.""" user = _current_user(request) prov = providers.get_provider(body.provider) if not prov: raise HTTPException(404, "Fournisseur inconnu") api_key = _resolve_api_key(prov.id, user or "", body.api_key) if \ (user or body.api_key) else None if not api_key: raise HTTPException(400, "Une clé est requise pour tester ce fournisseur.") # choisir le 1er modèle du fournisseur pour le ping model = prov.models[0][0] t0 = time.time() try: _, latency = chat_completion(prov.id, api_key, model, "Dis 'pong'.") except HTTPException as e: raise HTTPException(e.status_code, f"Test échoué ({model}): {e.detail}") return {"ok": True, "provider": prov.id, "model": model, "latency_ms": latency, "total_ms": int((time.time() - t0) * 1000)} @app.get("/v1/usage") async def usage(request: Request): user = _require_user(request) return _usage_snapshot(user) @app.get("/v1/providers/status") async def providers_status(request: Request): user = _current_user(request) return { "mysql": mysql_available(), "oauth": {p: bool(c["client_id"]) for p, c in oauth.oauth_config().items()}, "providers": [ { "id": p.id, "name": p.name, "needs_user_key": p.needs_user_key, "server_key_configured": bool(SERVER_KEYS.get(p.id)), "user_key_saved": bool(user and get_user_key(user, p.id)), } for p in providers.all_providers() ], } # --------------------------------------------------------------------------- # HEALTH # --------------------------------------------------------------------------- @app.get("/v1/health") async def health(): return { "status": "ok", "service": "Console IA", "time": time.strftime("%Y-%m-%d %H:%M:%S"), "mysql": mysql_available(), "public_host": PUBLIC_HOST, "daily_quota": USER_DAILY_QUOTA, "openrouter_configured": bool(SERVER_KEYS.get("openrouter")), } # Expose le dossier static (logo/favicon éventuel) app.mount("/static", StaticFiles(directory=BASE_DIR / "app" / "static"), name="static")