from sqlalchemy.pool import NullPool import os import time import json import hashlib import threading import re import subprocess import shutil import logging import tempfile import uuid import asyncio import base64 import io import logging logger = logging.getLogger("app") from datetime import datetime, timezone from collections import deque from typing import Optional, Dict, Any, List from fastapi import ( FastAPI, Request, Body, Query, Header, BackgroundTasks, File, UploadFile, Form, HTTPException, status ) from fastapi.responses import JSONResponse, StreamingResponse, HTMLResponse, FileResponse from sqlalchemy import create_engine, text as sql_text # Optional external helpers import requests # Optional ML libs try: import torch except Exception: torch = None try: from sentence_transformers import SentenceTransformer except Exception: SentenceTransformer = None try: from transformers import AutoTokenizer, AutoModelForSeq2SeqLM, pipeline as hf_pipeline except Exception: AutoTokenizer = None AutoModelForSeq2SeqLM = None hf_pipeline = None # Optional TTS (Coqui) try: from TTS.api import TTS TTS_AVAILABLE = True except Exception: TTS_AVAILABLE = False # Optional language module try: import language as language_module # type: ignore LANGUAGE_MODULE_AVAILABLE = True except Exception: language_module = None LANGUAGE_MODULE_AVAILABLE = False # Optional emojis helper try: from emojis import get_emoji, get_category_for_mood # type: ignore EMOJIS_AVAILABLE = True except Exception: EMOJIS_AVAILABLE = False def get_category_for_mood(m): return "neutral" def get_emoji(cat, intensity=0.5): return "πŸ€–" # Import custom modules try: from voicecloner import synthesize_speech, is_available as tts_is_available, cache_speaker_sample VOICECLONER_AVAILABLE = True logger.info("voicecloner module loaded successfully") except Exception as e: VOICECLONER_AVAILABLE = False logger.warning(f"voicecloner module not available: {e}") try: from coder import Coder CODER_AVAILABLE = True logger.info("coder module loaded successfully") except Exception as e: CODER_AVAILABLE = False logger.warning(f"coder module not available: {e}") import traceback logger.error(f"Coder import traceback: {traceback.format_exc()}") try: from videogenerator import VideoGenerator VIDEOGEN_AVAILABLE = True except Exception: VIDEOGEN_AVAILABLE = False logger.warning("videogenerator module not available") try: from image_editor import ImageEditor IMAGE_EDITOR_AVAILABLE = True logger.info("image_editor module loaded successfully") except Exception as e: IMAGE_EDITOR_AVAILABLE = False logger.warning(f"image_editor module not available: {e}") # Optional langdetect try: from langdetect import detect as detect_lang except Exception: detect_lang = None # Optional fuzzy matching for spell tolerance try: from difflib import SequenceMatcher FUZZY_AVAILABLE = True except Exception: FUZZY_AVAILABLE = False # Moderator pipeline (optional) moderator = None try: if hf_pipeline is not None: moderator = hf_pipeline("text-classification", model="unitary/toxic-bert", device=-1) except Exception: moderator = None # Detect whether python-multipart is available (package name: multipart) try: import multipart # type: ignore HAVE_MULTIPART = True except Exception: HAVE_MULTIPART = False # Pillow for image editing try: from PIL import Image, ImageOps, ImageFilter, ImageDraw, ImageFont PIL_AVAILABLE = True except Exception: PIL_AVAILABLE = False # Config via environment ADMIN_KEY = os.environ.get("ADMIN_KEY") DATABASE_URL = os.environ.get("DATABASE_URL", "sqlite:///justice_user.db") KNOWLEDGEDATABASE_URL = os.environ.get("KNOWLEDGEDATABASE_URL", DATABASE_URL) EMBED_MODEL_NAME = os.environ.get("EMBED_MODEL_NAME", "paraphrase-multilingual-MiniLM-L12-v2") TRANSLATION_CACHE_DIR = os.environ.get("TRANSLATION_CACHE_DIR", "./translation_models") LLM_MODEL_PATH = os.environ.get("LLM_MODEL_PATH", "") SAVE_MEMORY_CONFIDENCE = float(os.environ.get("SAVE_MEMORY_CONFIDENCE", "0.45")) MAX_INPUT_SIZE = int(os.environ.get("MAX_INPUT_SIZE", "1000000")) OLLAMA_MODEL = os.environ.get("OLLAMA_MODEL", "llama3") OLLAMA_HTTP_URL = os.environ.get("OLLAMA_HTTP_URL", "http://localhost:11434") OLLAMA_AUTO_PULL = os.environ.get("OLLAMA_AUTO_PULL", "0") in ("1", "true", "yes") MODEL_TIMEOUT = float(os.environ.get("MODEL_TIMEOUT", "10")) # TTS settings TTS_MODEL_NAME = os.environ.get("TTS_MODEL_NAME", "tts_models/multilingual/multi-dataset/xtts_v2") TTS_DEVICE = os.environ.get("TTS_DEVICE", "cuda" if (torch is not None and torch.cuda.is_available()) else "cpu") TTS_USE_HALF = os.environ.get("TTS_USE_HALF", "1") in ("1", "true", "yes") # Logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger("justicebrain") # heartbeat & start timestamp last_heartbeat = {"time": datetime.utcnow().replace(tzinfo=timezone.utc).isoformat(), "ok": True} app_start_time = time.time() # DB engines engine_user = create_engine( DATABASE_URL, poolclass=NullPool, connect_args={"check_same_thread": False} if DATABASE_URL.startswith("sqlite") else {} ) engine_knowledge = create_engine( KNOWLEDGEDATABASE_URL, poolclass=NullPool, connect_args={"check_same_thread": False} if KNOWLEDGEDATABASE_URL.startswith("sqlite") else {} ) app = FastAPI(title="Justice Brain β€” Backend") # βœ… Serve generated videos from /tmp/video_sandbox from fastapi.staticfiles import StaticFiles video_dir = os.getenv("VIDEO_SANDBOX_DIR", "/tmp/video_sandbox") # βœ… Create the folder if it doesn’t exist yet (prevents runtime error) os.makedirs(video_dir, exist_ok=True) # βœ… Mount the directory for frontend access app.mount("/static/video_sandbox", StaticFiles(directory=video_dir), name="videos") # Initialize custom modules coder_instance = None video_generator = None image_editor = None try: if CODER_AVAILABLE: coder_instance = Coder() logger.info("Coder instance initialized successfully") except Exception as e: logger.error(f"Failed to initialize Coder: {e}") import traceback logger.error(f"Coder init traceback: {traceback.format_exc()}") CODER_AVAILABLE = False try: if VIDEOGEN_AVAILABLE: video_generator = VideoGenerator() logger.info("VideoGenerator instance initialized successfully") except Exception as e: logger.error(f"Failed to initialize VideoGenerator: {e}") VIDEOGEN_AVAILABLE = False try: if IMAGE_EDITOR_AVAILABLE: image_editor = ImageEditor() logger.info("ImageEditor instance initialized successfully") except Exception as e: logger.error(f"Failed to initialize ImageEditor: {e}") IMAGE_EDITOR_AVAILABLE = False # ------------------------- # Database schema creation # ------------------------- def ensure_tables(): dialect_k = engine_knowledge.dialect.name with engine_knowledge.begin() as conn: if dialect_k == "sqlite": conn.execute(sql_text(""" CREATE TABLE IF NOT EXISTS knowledge ( id INTEGER PRIMARY KEY AUTOINCREMENT, text TEXT, reply TEXT, language TEXT DEFAULT 'und', embedding BLOB, category TEXT DEFAULT 'general', topic TEXT DEFAULT 'general', confidence FLOAT DEFAULT 0, source TEXT, meta TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) else: conn.execute(sql_text(""" CREATE TABLE IF NOT EXISTS knowledge ( id SERIAL PRIMARY KEY, text TEXT, reply TEXT, language TEXT DEFAULT 'und', embedding BYTEA, category TEXT DEFAULT 'general', topic TEXT DEFAULT 'general', confidence FLOAT DEFAULT 0, source TEXT, meta JSONB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) dialect_u = engine_user.dialect.name with engine_user.begin() as conn: if dialect_u == "sqlite": conn.execute(sql_text(""" CREATE TABLE IF NOT EXISTS user_memory ( id INTEGER PRIMARY KEY AUTOINCREMENT, user_id TEXT, username TEXT, ip TEXT, text TEXT, reply TEXT, language TEXT DEFAULT 'und', mood TEXT, confidence FLOAT DEFAULT 0, topic TEXT DEFAULT 'general', source TEXT, meta TEXT, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) else: conn.execute(sql_text(""" CREATE TABLE IF NOT EXISTS user_memory ( id SERIAL PRIMARY KEY, user_id TEXT, username TEXT, ip TEXT, text TEXT, reply TEXT, language TEXT DEFAULT 'und', mood TEXT, confidence FLOAT DEFAULT 0, topic TEXT DEFAULT 'general', source TEXT, meta JSONB, created_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP ); """)) ensure_tables() def ensure_column_exists(table: str, column: str, col_def_sql: str): dialect = engine_user.dialect.name try: with engine_user.begin() as conn: if dialect == "sqlite": try: rows = conn.execute(sql_text(f"PRAGMA table_info({table})")).fetchall() existing = [r[1] for r in rows] if column not in existing: conn.execute(sql_text(f"ALTER TABLE {table} ADD COLUMN {col_def_sql}")) except Exception: pass else: try: conn.execute(sql_text(f"ALTER TABLE {table} ADD COLUMN IF NOT EXISTS {col_def_sql}")) except Exception: pass except Exception: pass ensure_column_exists("knowledge", "reply", "reply TEXT") ensure_column_exists("user_memory", "reply", "reply TEXT") # ------------------------- # Utility helpers # ------------------------- def sanitize_knowledge_text(t: Any) -> str: if not isinstance(t, str): return str(t or "").strip() s = t.strip() try: parsed = json.loads(s) if isinstance(parsed, dict) and "text" in parsed: return str(parsed["text"]).strip() except Exception: pass if (s.startswith('"') and s.endswith('"')) or (s.startswith("'") and s.endswith("'")): s = s[1:-1].strip() return " ".join(s.split()) def dedupe_sentences(text: str) -> str: if not text: return text sentences = [] seen = set() for chunk in re.split(r'\n+', text): parts = re.split(r'(?<=[.?!])\s+', chunk) for sent in parts: s = sent.strip() if not s: continue if s in seen: continue seen.add(s) sentences.append(s) return "\n".join(sentences) _EMOJI_PATTERN = re.compile( "[" "\U0001F600-\U0001F64F" "\U0001F300-\U0001F5FF" "\U0001F680-\U0001F6FF" "\U0001F1E0-\U0001F1FF" "\u2600-\u26FF" "\u2700-\u27BF" "]+", flags=re.UNICODE ) def extract_emojis(text: str) -> List[str]: if not text: return [] return _EMOJI_PATTERN.findall(text) def emoji_sentiment_score(emojis: List[str]) -> float: if not emojis: return 0.0 score = 0.0 for e in "".join(emojis): ord_val = ord(e) if 0x1F600 <= ord_val <= 0x1F64F: score += 0.5 elif 0x2600 <= ord_val <= 0x26FF: score += 0.1 return max(-1.0, min(1.0, score / max(1, len(emojis)))) # ------------------------- # Language detection & translation # ------------------------- _translation_model_cache: Dict[str, Any] = {} def detect_language_safe(text: str) -> str: text = (text or "").strip() if not text: return "en" if LANGUAGE_MODULE_AVAILABLE: try: if hasattr(language_module, "detect"): out = language_module.detect(text) if out: return out if hasattr(language_module, "detect_language"): out = language_module.detect_language(text) if out: return out except Exception: pass lower = text.lower() greetings = {"hola":"es","bonjour":"fr","hallo":"de","ciao":"it","こんにけは":"ja","δ½ ε₯½":"zh","μ•ˆλ…•ν•˜μ„Έμš”":"ko"} for k, v in greetings.items(): if k in lower: return v if re.search(r'[\u4e00-\u9fff]', text): return "zh" if re.search(r'[\u3040-\u30ff]', text): return "ja" letters = re.findall(r'[A-Za-z]', text) if len(letters) >= max(1, 0.6 * len(text)): return "en" if detect_lang is not None: try: out = detect_lang(text) if out: return out except Exception: pass return "und" def translate_text(text: str, src: str, tgt: str) -> str: if not text: return text if LANGUAGE_MODULE_AVAILABLE: try: if hasattr(language_module, "translate"): out = language_module.translate(text, src, tgt) if out: return out if src in ("en", "eng") and hasattr(language_module, "translate_from_en"): out = language_module.translate_from_en(text, tgt) if out: return out if tgt in ("en", "eng") and hasattr(language_module, "translate_to_en"): out = language_module.translate_to_en(text, src) if out: return out except Exception: pass src_code = (src or "und").split("-")[0].lower() tgt_code = (tgt or "und").split("-")[0].lower() if not re.fullmatch(r"[a-z]{2,3}", src_code) or not re.fullmatch(r"[a-z]{2,3}", tgt_code): return text key = f"{src_code}-{tgt_code}" try: if key in _translation_model_cache: tokenizer, model = _translation_model_cache[key] inputs = tokenizer([text], return_tensors="pt", truncation=True) outputs = model.generate(**inputs, max_length=1024) return tokenizer.batch_decode(outputs, skip_special_tokens=True)[0] except Exception: pass try: if AutoTokenizer is not None and AutoModelForSeq2SeqLM is not None: model_name = f"Helsinki-NLP/opus-mt-{src_code}-{tgt_code}" tokenizer = AutoTokenizer.from_pretrained(model_name, cache_dir=TRANSLATION_CACHE_DIR) model = AutoModelForSeq2SeqLM.from_pretrained(model_name, cache_dir=TRANSLATION_CACHE_DIR) _translation_model_cache[key] = (tokenizer, model) inputs = tokenizer([text], return_tensors="pt", truncation=True) outputs = model.generate(**inputs, max_length=1024) return tokenizer.batch_decode(outputs, skip_special_tokens=True)[0] except Exception: pass return text def translate_to_english(text: str, src_lang: str) -> str: src = (src_lang or "und").split("-")[0].lower() if src in ("en", "eng", "", "und"): return text return translate_text(text, src, "en") def translate_from_english(text: str, tgt_lang: str) -> str: tgt = (tgt_lang or "und").split("-")[0].lower() if tgt in ("en", "eng", "", "und"): return text return translate_text(text, "en", tgt) # ------------------------- # Embeddings helpers # ------------------------- embed_model = None def try_load_embed(): global embed_model if SentenceTransformer is None: logger.info("[JusticeAI] SentenceTransformer not available") return try: embed_model = SentenceTransformer(EMBED_MODEL_NAME, device="cpu") logger.info(f"[JusticeAI] Loaded embed model: {EMBED_MODEL_NAME}") except Exception as e: embed_model = None logger.warning(f"[JusticeAI] failed to load embed model: {e}") def embed_to_bytes(text: str) -> Optional[bytes]: if embed_model is None: return None try: emb = embed_model.encode([text], convert_to_tensor=True)[0] return emb.cpu().numpy().tobytes() except Exception: return None def bytes_to_tensor(b: bytes): """ Convert embedding bytes (as stored in DB) back to a torch tensor if possible. Returns None if conversion not possible. """ if b is None: return None if torch is None: return None try: import numpy as _np arr = _np.frombuffer(b, dtype=_np.float32) # If embed_model is available, try to infer dimension from it if embed_model is not None: # some sentence-transformers return float32 vectors return torch.from_numpy(arr) return torch.from_numpy(arr) except Exception as e: logger.debug(f"bytes_to_tensor conversion failed: {e}") return None # ------------------------- # Blocking with timeout helper (for non-TTS blocking ops) # ------------------------- async def run_blocking_with_timeout(func, *args, timeout: float = MODEL_TIMEOUT): loop = asyncio.get_running_loop() fut = loop.run_in_executor(None, lambda: func(*args)) return await asyncio.wait_for(fut, timeout=timeout) # ------------------------- # Ollama helpers # ------------------------- def ollama_cli_available() -> bool: return shutil.which("ollama") is not None def ollama_http_available() -> bool: try: resp = requests.get(f"{OLLAMA_HTTP_URL}/health", timeout=1.0) return resp.status_code == 200 except Exception: return False def call_ollama_http(prompt: str, model: str = OLLAMA_MODEL, timeout_s: int = MODEL_TIMEOUT) -> Optional[str]: try: url = f"{OLLAMA_HTTP_URL}/api/generate" payload = {"model": model, "prompt": prompt, "max_tokens": 256} headers = {"Content-Type": "application/json"} r = requests.post(url, json=payload, headers=headers, timeout=min(timeout_s, MODEL_TIMEOUT)) if r.status_code == 200: try: obj = r.json() for key in ("output", "text", "result", "generations"): if key in obj: return obj[key] if isinstance(obj[key], str) else json.dumps(obj[key]) return r.text except Exception: return r.text else: logger.debug(f"ollama HTTP status {r.status_code}") return None except Exception as e: logger.debug(f"ollama HTTP call failed: {e}") return None def call_ollama_cli(prompt: str, model: str = OLLAMA_MODEL, timeout_s: int = MODEL_TIMEOUT) -> Optional[str]: if not ollama_cli_available(): return None try: proc = subprocess.run(["ollama", "run", model, "--prompt", prompt], capture_output=True, text=True, timeout=min(timeout_s, MODEL_TIMEOUT)) if proc.returncode == 0: return proc.stdout.strip() or proc.stderr.strip() else: logger.debug(f"ollama CLI rc={proc.returncode}") return None except Exception as e: logger.debug(f"ollama CLI call exception: {e}") return None def infer_topic_with_ollama(msg: str, topics: List[str], model: str = OLLAMA_MODEL, timeout_s: int = MODEL_TIMEOUT) -> Optional[str]: if not msg or not topics: return None topics_escaped = [t.replace('"','\\"') for t in topics] topics_list = ", ".join(f'"{t}"' for t in topics_escaped) escaped_msg = msg.replace('"', '\\"') prompt = ( "You are a strict topic classifier. Given a user message, choose the single best topic from this list: " f"[{topics_list}]. If none match, return topic \"none\". Return ONLY a JSON object with a single key \"topic\" and the chosen topic string.\n\n" f"Message: \"{escaped_msg}\"\n\n" "Respond with JSON only. Example: {\"topic\": \"security\"}" ) out = call_ollama_http(prompt, model=model, timeout_s=timeout_s) if out: try: j = json.loads(out) if isinstance(j, dict) and "topic" in j: t = j["topic"] if t in topics: return t if t == "none": return None except Exception: try: idx = out.find("{") if idx >= 0: j = json.loads(out[idx:]) t = j.get("topic") if t in topics: return t except Exception: pass out = call_ollama_cli(prompt, model=model, timeout_s=timeout_s) if out: try: j = json.loads(out) if isinstance(j, dict) and "topic" in j: t = j["topic"] if t in topics: return t if t == "none": return None except Exception: try: idx = out.find("{") if idx >= 0: j = json.loads(out[idx:]) t = j.get("topic") if t in topics: return t except Exception: pass return None # ------------------------- # Simple fallback topic inference (NEW) # ------------------------- def fuzzy_match_score(s1: str, s2: str) -> float: """ Calculate fuzzy match score between two strings (0.0 to 1.0). Handles spell errors and variations. """ if not FUZZY_AVAILABLE: return 1.0 if s1.lower() == s2.lower() else 0.0 return SequenceMatcher(None, s1.lower(), s2.lower()).ratio() def infer_topic_from_message(msg: str, topics: List[str]) -> Optional[str]: """ Fallback topic inference: tries keyword matching against topic names and common words. Returns the first matching topic or None. """ if not msg or not topics: return None low = msg.lower() # Try exact topic token matches first for t in topics: if not t: continue token = str(t).lower() if token and token in low: return t # split topic into words and check for w in re.split(r'[\s\-_]+', token): if w and re.search(r'\b' + re.escape(w) + r'\b', low): return t # Try fuzzy matching for spell tolerance if FUZZY_AVAILABLE: best_match = None best_score = 0.0 for t in topics: if not t: continue token = str(t).lower() # Check fuzzy match against whole message score = fuzzy_match_score(token, low) if score > 0.7 and score > best_score: best_score = score best_match = t # Check fuzzy match against individual words for word in low.split(): if len(word) > 3: # Only check meaningful words score = fuzzy_match_score(token, word) if score > 0.75 and score > best_score: best_score = score best_match = t if best_match: return best_match # If no direct match, try heuristics: map some keywords to topics heuristics = { "security": ["security", "vulnerability", "exploit", "attack", "auth", "password", "login"], "billing": ["bill", "invoice", "payment", "charge", "price", "cost"], "installation": ["install", "setup", "deploy", "deployment", "configure"], "general": ["help", "question", "how", "what", "why", "issue", "problem"] } for topic, kws in heuristics.items(): for kw in kws: if kw in low: # if topic exists in known topics return it, else skip if topic in topics: return topic return None def infer_topic_with_embeddings(msg: str, topics: List[str], knowledge_rows: List[dict]) -> Optional[str]: """ Use cosine similarity on embeddings to infer the best matching topic. This provides semantic understanding instead of just keyword matching. """ if not embed_model or not topics or not knowledge_rows: return None try: # Compute query embedding q_emb = embed_model.encode([msg], convert_to_tensor=True, show_progress_bar=False)[0] # Group knowledge by topic and compute average embedding per topic topic_embeddings = {} topic_counts = {} for kr in knowledge_rows: t = kr.get("topic", "general") if t not in topics: continue emb_bytes = kr.get("embedding") if emb_bytes is None: continue emb_tensor = bytes_to_tensor(emb_bytes) if emb_tensor is None: continue if t not in topic_embeddings: topic_embeddings[t] = emb_tensor topic_counts[t] = 1 else: topic_embeddings[t] = topic_embeddings[t] + emb_tensor topic_counts[t] += 1 # Average the embeddings for t in topic_embeddings: topic_embeddings[t] = topic_embeddings[t] / topic_counts[t] if not topic_embeddings: return None # Compute cosine similarity with each topic best_topic = None best_score = 0.0 for t, t_emb in topic_embeddings.items(): try: score = float(torch.nn.functional.cosine_similarity(q_emb.unsqueeze(0), t_emb.unsqueeze(0), dim=1)[0]) if score > best_score: best_score = score best_topic = t except Exception: continue # Only return if confidence is high enough if best_score > 0.4: logger.info(f"[topic inference] embedding-based: {best_topic} (score={best_score:.2f})") return best_topic except Exception as e: logger.debug(f"[topic inference] embedding error: {e}") return None # ------------------------- # Boilerplate detection & reply helpers # ------------------------- def is_boilerplate_candidate(s: str) -> bool: s_low = (s or "").strip().lower() generic = ["i don't know", "not sure", "maybe", "perhaps", "justiceai is a unified intelligence dashboard"] if len(s_low) < 8: return True return any(g in s_low for g in generic) def generate_creative_reply(candidates: List[str]) -> str: all_sent = [] seen = set() for c in candidates: for s in re.split(r'(?<=[.?!])\s+', c): st = s.strip() if not st or st in seen or is_boilerplate_candidate(st): continue seen.add(st) all_sent.append(st) if not all_sent: return "I don't have enough context yet β€” can you give more details?" return "\n".join(all_sent[:5]) def detect_mood(text: str) -> str: lower = (text or "").lower() positive = ["great", "thanks", "awesome", "happy", "love", "excellent", "cool", "yes", "good"] negative = ["sad", "bad", "problem", "angry", "hate", "fail", "no", "error", "issue"] if any(w in lower for w in positive): return "positive" if any(w in lower for w in negative): return "negative" return "neutral" def should_append_emoji(user_text: str, reply_text: str, mood: str, flags: Dict) -> str: if flags.get("toxic"): return "" if EMOJIS_AVAILABLE: try: cat = get_category_for_mood(mood) return get_emoji(cat, 0.6) except Exception: return "" return "" # ------------------------- # TTS: optimized loader and endpoints # ------------------------- _tts_model = None _tts_lock = threading.Lock() _speaker_hash_cache: Dict[str, str] = {} _tts_loaded_event = threading.Event() def compute_file_sha256(path: str) -> str: h = hashlib.sha256() with open(path, "rb") as f: while True: b = f.read(8192) if not b: break h.update(b) return h.hexdigest() def get_tts_model_blocking(): global _tts_model if not TTS_AVAILABLE: raise RuntimeError("TTS.api not available on server") with _tts_lock: if _tts_model is None: model_name = os.environ.get("TTS_MODEL_NAME", TTS_MODEL_NAME) device = os.environ.get("TTS_DEVICE", TTS_DEVICE) logger.info(f"[TTS] Loading model {model_name} on device {device}") _tts_model = TTS(model_name) try: if device and torch is not None: if device.startswith("cuda") and torch.cuda.is_available(): try: _tts_model.to(device) except Exception: pass try: torch.backends.cudnn.benchmark = True except Exception: pass if TTS_USE_HALF: try: if hasattr(_tts_model, "model") and hasattr(_tts_model.model, "half"): _tts_model.model.half() except Exception: pass try: torch.set_num_threads(int(os.environ.get("TORCH_NUM_THREADS", "4"))) except Exception: pass else: try: torch.set_num_threads(int(os.environ.get("TORCH_NUM_THREADS", "4"))) except Exception: pass except Exception as e: logger.debug(f"[TTS] model device tuning warning: {e}") logger.info("[TTS] model loaded") _tts_loaded_event.set() return _tts_model def _save_upload_file_tmp(upload_file: UploadFile) -> str: suffix = os.path.splitext(upload_file.filename)[1] or ".wav" fd, tmp_path = tempfile.mkstemp(suffix=suffix, prefix="tts_speaker_") os.close(fd) with open(tmp_path, "wb") as f: content = upload_file.file.read() f.write(content) return tmp_path # Preload TTS in background (best-effort) if TTS_AVAILABLE: threading.Thread(target=lambda: (get_tts_model_blocking()), daemon=True).start() # /speak_json and /speak endpoints @app.post("/speak_json") async def speak_json(background_tasks: BackgroundTasks, payload: dict = Body(...)): text = payload.get("text", "") if not text or not text.strip(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Field 'text' is required") voice_b64 = payload.get("voice_wav_b64") language = payload.get("language") speaker_path = None if voice_b64: try: data = base64.b64decode(voice_b64) fd, speaker_path = tempfile.mkstemp(suffix=".wav", prefix="tts_speaker_json_") os.close(fd) with open(speaker_path, "wb") as f: f.write(data) speaker_hash = compute_file_sha256(speaker_path) cached = _speaker_hash_cache.get(speaker_hash) if cached and os.path.exists(cached): try: os.remove(speaker_path) except Exception: pass speaker_path = cached else: _speaker_hash_cache[speaker_hash] = speaker_path background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), speaker_path) except Exception: raise HTTPException(status_code=400, detail="Invalid base64 in 'voice_wav_b64'") out_fd, out_path = tempfile.mkstemp(suffix=".wav", prefix="tts_out_json_") os.close(out_fd) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), out_path) try: tts = get_tts_model_blocking() except Exception: try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=500, detail="TTS model not available") def synth(): kwargs = {} if speaker_path: kwargs["speaker_wav"] = speaker_path if language: kwargs["language"] = language tts.tts_to_file(text=text, file_path=out_path, **kwargs) return out_path loop = asyncio.get_running_loop() try: await loop.run_in_executor(None, synth) except Exception: try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=500, detail="TTS synthesis failed") return FileResponse(path=out_path, filename=f"speech-{uuid.uuid4().hex}.wav", media_type="audio/wav", background=background_tasks) if HAVE_MULTIPART: @app.post("/speak") async def speak( background_tasks: BackgroundTasks, text: str = Form(...), voice_wav: Optional[UploadFile] = File(None), language: Optional[str] = Form(None), ): if not text or not text.strip(): raise HTTPException(status_code=status.HTTP_400_BAD_REQUEST, detail="Field 'text' is required") if not TTS_AVAILABLE: raise HTTPException(status_code=status.HTTP_503_SERVICE_UNAVAILABLE, detail="TTS engine not available on server. Please install TTS library.") speaker_path = None if voice_wav is not None: try: speaker_path = _save_upload_file_tmp(voice_wav) speaker_hash = compute_file_sha256(speaker_path) cached = _speaker_hash_cache.get(speaker_hash) if cached and os.path.exists(cached): try: os.remove(speaker_path) except Exception: pass speaker_path = cached else: _speaker_hash_cache[speaker_hash] = speaker_path except Exception as e: logger.error(f"Voice sample processing failed: {e}") raise HTTPException(status_code=500, detail=f"Failed to process uploaded voice sample: {str(e)}") out_fd, out_path = tempfile.mkstemp(suffix=".wav", prefix="tts_out_") os.close(out_fd) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), out_path) try: tts = get_tts_model_blocking() except Exception as e: logger.error(f"TTS model loading failed: {e}") try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=503, detail=f"TTS model not available: {str(e)}") kwargs = {} if speaker_path: kwargs["speaker_wav"] = speaker_path if language: kwargs["language"] = language try: if torch is not None and torch.cuda.is_available() and TTS_USE_HALF: try: with torch.inference_mode(): with torch.cuda.amp.autocast(): tts.tts_to_file(text=text, file_path=out_path, **kwargs) except Exception as e: logger.warning(f"GPU synthesis failed, trying CPU: {e}") with torch.inference_mode(): tts.tts_to_file(text=text, file_path=out_path, **kwargs) else: if torch is not None: with torch.inference_mode(): tts.tts_to_file(text=text, file_path=out_path, **kwargs) else: tts.tts_to_file(text=text, file_path=out_path, **kwargs) except Exception as e: logger.error(f"TTS synthesis failed: {e}") try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=500, detail=f"TTS synthesis failed: {str(e)}") filename = f"speech-{uuid.uuid4().hex}.wav" return FileResponse(path=out_path, filename=filename, media_type="audio/wav", background=background_tasks) else: @app.post("/speak") async def speak_unavailable(): raise HTTPException( status_code=501, detail="Multipart support not available. Install python-multipart (pip install python-multipart) to enable /speak with file uploads. Use /speak_json with base64-encoded speaker sample instead." ) # ------------------------- # Image Editor: endpoints using the new image_editor module # ------------------------- @app.post("/image_edit_json") async def image_edit_json(background_tasks: BackgroundTasks, payload: dict = Body(...)): """ JSON endpoint for advanced image editing with AI capabilities. Body: { "image_b64": "" OR "image_url": "http://...", "operations": [ {op definitions} ], "prompt": "natural language edit request (e.g., 'add text: Hello', 'blur background')", "format": "png" # optional } Returns: edited image file response. """ if not IMAGE_EDITOR_AVAILABLE or image_editor is None: raise HTTPException(status_code=503, detail="Image editing requires Pillow. Install with pip install pillow") image_b64 = payload.get("image_b64") image_url = payload.get("image_url") operations = payload.get("operations", []) prompt = payload.get("prompt", "") out_format = (payload.get("format") or "png").lower() # Parse natural language prompt into operations using image_editor if prompt and not operations: operations = image_editor.parse_edit_prompt(prompt) if not image_b64 and not image_url: raise HTTPException(status_code=400, detail="Provide either image_b64 or image_url") in_fd, in_path = tempfile.mkstemp(suffix=".input") os.close(in_fd) try: if image_b64: try: data = base64.b64decode(image_b64) except Exception: raise HTTPException(status_code=400, detail="Invalid base64 for image_b64") with open(in_path, "wb") as f: f.write(data) else: try: resp = requests.get(image_url, timeout=10) if resp.status_code != 200: raise HTTPException(status_code=400, detail="Failed to download image_url") with open(in_path, "wb") as f: f.write(resp.content) except Exception: raise HTTPException(status_code=400, detail="Failed to download image_url") except HTTPException: try: if os.path.exists(in_path): os.remove(in_path) except Exception: pass raise except Exception: try: if os.path.exists(in_path): os.remove(in_path) except Exception: pass raise HTTPException(status_code=500, detail="Failed to save input image") ext = "." + out_format if not out_format.startswith(".") else out_format out_fd, out_path = tempfile.mkstemp(suffix=ext, prefix="img_edit_out_") os.close(out_fd) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), out_path) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), in_path) try: loop = asyncio.get_running_loop() await loop.run_in_executor(None, lambda: image_editor.perform_operations(in_path, operations, out_path)) except Exception as e: logger.exception("Image edit failed") try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=500, detail=f"Image edit failed: {e}") return FileResponse(path=out_path, filename=f"image-{uuid.uuid4().hex}{ext}", media_type="image/png", background=background_tasks) if HAVE_MULTIPART: @app.post("/image_edit") async def image_edit( background_tasks: BackgroundTasks, operations: str = Form(...), # JSON string describing ops image: Optional[UploadFile] = File(None), image_url: Optional[str] = Form(None), format: Optional[str] = Form("png"), ): if not IMAGE_EDITOR_AVAILABLE or image_editor is None: raise HTTPException(status_code=503, detail="Image editing requires Pillow. Install with pip install pillow") try: ops = json.loads(operations) if operations else [] except Exception: raise HTTPException(status_code=400, detail="Invalid JSON in operations") if image is None and not image_url: raise HTTPException(status_code=400, detail="Provide uploaded image file or image_url") in_fd, in_path = tempfile.mkstemp(suffix=".input") os.close(in_fd) try: if image is not None: content = await image.read() with open(in_path, "wb") as f: f.write(content) else: try: resp = requests.get(image_url, timeout=10) if resp.status_code != 200: raise HTTPException(status_code=400, detail="Failed to download image_url") with open(in_path, "wb") as f: f.write(resp.content) except Exception: raise HTTPException(status_code=400, detail="Failed to download image_url") except HTTPException: try: if os.path.exists(in_path): os.remove(in_path) except Exception: pass raise except Exception: try: if os.path.exists(in_path): os.remove(in_path) except Exception: pass raise HTTPException(status_code=500, detail="Failed to save uploaded image") out_ext = "." + (format or "png").lstrip(".") out_fd, out_path = tempfile.mkstemp(suffix=out_ext, prefix="img_edit_out_") os.close(out_fd) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), out_path) background_tasks.add_task(lambda p: os.path.exists(p) and os.remove(p), in_path) try: loop = asyncio.get_running_loop() await loop.run_in_executor(None, lambda: image_editor.perform_operations(in_path, ops, out_path)) except Exception as e: logger.exception("Image edit failed (multipart)") try: if os.path.exists(out_path): os.remove(out_path) except Exception: pass raise HTTPException(status_code=500, detail=f"Image edit failed: {e}") return FileResponse(path=out_path, filename=f"image-{uuid.uuid4().hex}{out_ext}", media_type="image/png", background=background_tasks) else: @app.post("/image_edit") async def image_edit_unavailable(): raise HTTPException( status_code=501, detail="Multipart support not available. Install python-multipart (pip install python-multipart) to enable /image_edit with uploads. Use /image_edit_json instead." ) # ------------------------- # Metrics, language.bin, and small helpers # ------------------------- recent_request_times = deque() recent_learning_timestamps = deque() response_time_ema: Optional[float] = None EMA_ALPHA = 0.2 def record_request(duration_s: float): global response_time_ema ts = time.time() recent_request_times.append((ts, duration_s)) while recent_request_times and recent_request_times[0][0] < ts - 3600: recent_request_times.popleft() if response_time_ema is None: response_time_ema = duration_s else: response_time_ema = EMA_ALPHA * duration_s + (1 - EMA_ALPHA) * response_time_ema def record_learn_event(): ts = time.time() recent_learning_timestamps.append(ts) while recent_learning_timestamps and recent_learning_timestamps[0] < ts - 3600: recent_learning_timestamps.popleft() @app.get("/metrics") async def metrics(): try: with engine_knowledge.connect() as c: k = c.execute(sql_text("SELECT COUNT(*) FROM knowledge")).scalar() or 0 except Exception: k = -1 try: with engine_user.connect() as c: u = c.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0 except Exception: u = -1 reqs_last_hour = sum(1 for ts, _ in recent_request_times if ts >= time.time() - 3600) if 'recent_request_times' in globals() else 0 return { "ok": True, "uptime_s": round(time.time() - app_start_time, 2) if 'app_start_time' in globals() else None, "knowledge_count": int(k), "user_memory_count": int(u), "requests_last_hour": int(reqs_last_hour) } @app.get("/language.bin") async def language_bin(): path = "language.bin" if os.path.exists(path): return FileResponse(path, media_type="application/octet-stream") return JSONResponse(status_code=404, content={"error": "language.bin not found", "hint": "Place file at ./language.bin or upload it"}) # ------------------------- # Startup warmups # ------------------------- @app.on_event("startup") async def startup_event(): logger.info("[JusticeAI] startup: warming optional components") if SentenceTransformer is not None: def warm_embed(): try: try_load_embed() except Exception as e: logger.debug(f"[startup] embed warmup error: {e}") threading.Thread(target=warm_embed, daemon=True).start() if OLLAMA_AUTO_PULL and ollama_cli_available(): try: subprocess.run(["ollama", "pull", OLLAMA_MODEL], timeout=300) logger.info("[startup] attempted ollama pull") except Exception as e: logger.debug(f"[startup] ollama pull failed: {e}") logger.info("[JusticeAI] startup complete") # ------------------------- # Knowledge endpoints (add/add-bulk/leaderboard/reembed/model-status/health) # ------------------------- def _require_admin(x_admin_key: Optional[str]): if ADMIN_KEY is None: raise HTTPException(status_code=403, detail="Server not configured for admin operations.") if not x_admin_key or x_admin_key != ADMIN_KEY: raise HTTPException(status_code=403, detail="Invalid admin key.") @app.post("/add") async def add_knowledge(data: dict = Body(...), x_admin_key: Optional[str] = Header(None, alias="X-Admin-Key")): """ Add a single knowledge entry. Requires X-Admin-Key header matching ADMIN_KEY. Body fields: - text: required - reply: optional - topic: required """ # enforce admin try: _require_admin(x_admin_key) except HTTPException: # keep previous behavior of returning JSONResponse for auth failure return JSONResponse(status_code=403, content={"error": "Invalid or missing admin key."}) if not isinstance(data, dict): return JSONResponse(status_code=400, content={"error": "Invalid body"}) text_data = sanitize_knowledge_text(data.get("text", "") or "") reply = sanitize_knowledge_text(data.get("reply", "") or "") topic = str(data.get("topic", "") or "").strip() if not topic: return JSONResponse(status_code=400, content={"error": "Topic is required"}) if not text_data: return JSONResponse(status_code=400, content={"error": "Text is required"}) detected = detect_language_safe(text_data) or "und" if detected not in ("en", "eng", "und"): try: text_data = translate_to_english(text_data, detected) detected = "en" except Exception: return JSONResponse(status_code=400, content={"error": "translation failed"}) emb_bytes = None if embed_model is not None: try: emb_bytes = await run_blocking_with_timeout(lambda: embed_to_bytes(text_data), timeout=MODEL_TIMEOUT) except Exception: emb_bytes = None # Use proper parameter binding. For SQLite, bytes are accepted. try: with engine_knowledge.begin() as conn: if emb_bytes: conn.execute(sql_text( "INSERT INTO knowledge (text, reply, language, embedding, category, topic, confidence, meta, source) " "VALUES (:t, :r, :lang, :e, 'manual', :topic, :conf, :meta, :source)" ), {"t": text_data, "r": reply, "lang": detected, "e": emb_bytes, "topic": topic, "conf": 0.9, "meta": json.dumps({"manual": True}), "source": "admin"}) else: conn.execute(sql_text( "INSERT INTO knowledge (text, reply, language, category, topic, confidence, meta, source) " "VALUES (:t, :r, :lang, 'manual', :topic, :conf, :meta, :source)" ), {"t": text_data, "r": reply, "lang": detected, "topic": topic, "conf": 0.9, "meta": json.dumps({"manual": True}), "source": "admin"}) record_learn_event() return {"status": "βœ… Knowledge added", "text": text_data, "topic": topic, "language": detected} except Exception as e: logger.exception("add failed") return JSONResponse(status_code=500, content={"error": "failed to store knowledge", "details": str(e)}) @app.post("/add-bulk") async def add_bulk(data: List[dict] = Body(...), x_admin_key: Optional[str] = Header(None, alias="X-Admin-Key")): """ Add many knowledge entries. Requires admin key. """ try: _require_admin(x_admin_key) except HTTPException: return JSONResponse(status_code=403, content={"error": "Invalid or missing admin key."}) if not isinstance(data, list): return JSONResponse(status_code=400, content={"error": "Expected an array"}) added = 0 errors = [] for i, it in enumerate(data): try: if not isinstance(it, dict): errors.append({"index": i, "error": "not object"}); continue text_data = sanitize_knowledge_text(it.get("text", "") or "") topic = str(it.get("topic", "") or "").strip() reply = sanitize_knowledge_text(it.get("reply", "") or "") if not text_data or not topic: errors.append({"index": i, "error": "missing text or topic"}); continue detected = detect_language_safe(text_data) or "und" if detected not in ("en", "eng", "und"): errors.append({"index": i, "error": "non-english; skip"}); continue emb_bytes = None if embed_model is not None: try: emb_bytes = await run_blocking_with_timeout(lambda: embed_to_bytes(text_data), timeout=MODEL_TIMEOUT) except Exception: emb_bytes = None with engine_knowledge.begin() as conn: if emb_bytes: conn.execute(sql_text( "INSERT INTO knowledge (text, reply, language, embedding, category, topic, source) VALUES (:t, :r, :lang, :e, 'manual', :topic, :source)" ), {"t": text_data, "r": reply, "lang": "en", "e": emb_bytes, "topic": topic, "source": "admin"}) else: conn.execute(sql_text( "INSERT INTO knowledge (text, reply, language, category, topic, source) VALUES (:t, :r, :lang, 'manual', :topic, :source)" ), {"t": text_data, "r": reply, "lang": "en", "topic": topic, "source": "admin"}) added += 1 except Exception as e: logger.exception("add-bulk item error") errors.append({"index": i, "error": str(e)}) if added: record_learn_event() return {"added": added, "errors": errors} @app.get("/leaderboard") async def leaderboard(topic: str = Query("general")): t = str(topic or "general").strip() or "general" try: with engine_knowledge.begin() as conn: rows = conn.execute(sql_text(""" SELECT id, text, reply, language, category, confidence, created_at FROM knowledge WHERE topic = :topic ORDER BY confidence DESC, created_at DESC LIMIT 20 """), {"topic": t}).fetchall() out = [] for r in rows: text_en = r[1] or "" lang = r[3] or "und" display_text = text_en if lang and lang not in ("en", "eng", "", "und"): try: display_text = translate_to_english(text_en, lang) except Exception: display_text = text_en created_at = r[6] out.append({ "id": r[0], "text": display_text, "reply": r[2], "language": lang, "category": r[4], "confidence": round(r[5] or 0.0, 2), "created_at": created_at.isoformat() if hasattr(created_at, "isoformat") else str(created_at) }) return {"topic": t, "top_20": out} except Exception as e: logger.exception("leaderboard failed") return JSONResponse(status_code=500, content={"error": "failed to fetch leaderboard", "details": str(e)}) @app.post("/reembed") async def reembed_all(data: dict = Body(...), x_admin_key: str = Header(None, alias="X-Admin-Key")): if ADMIN_KEY is None: return JSONResponse(status_code=403, content={"error": "Server not configured for admin operations."}) if x_admin_key != ADMIN_KEY: return JSONResponse(status_code=403, content={"error": "Invalid admin key."}) if embed_model is None: return JSONResponse(status_code=503, content={"error": "Embedding model not ready."}) confirm = str(data.get("confirm", "") or "").strip() if confirm != "REEMBED": return JSONResponse(status_code=400, content={"error": "confirm token required."}) batch_size = int(data.get("batch_size", 100)) try: with engine_knowledge.begin() as conn: rows = conn.execute(sql_text("SELECT id, text FROM knowledge ORDER BY id")).fetchall() ids_texts = [(r[0], r[1]) for r in rows] total = len(ids_texts) updated = 0 for i in range(0, total, batch_size): batch = ids_texts[i:i+batch_size] texts = [t for _, t in batch] try: embs = await run_blocking_with_timeout(lambda: embed_model.encode(texts, convert_to_tensor=True), timeout=MODEL_TIMEOUT) except Exception: embs = None if embs is None: continue for j, (kid, _) in enumerate(batch): emb_bytes = embs[j].cpu().numpy().tobytes() with engine_knowledge.begin() as conn: conn.execute(sql_text("UPDATE knowledge SET embedding = :e, updated_at = CURRENT_TIMESTAMP WHERE id = :id"), {"e": emb_bytes, "id": kid}) updated += 1 return {"status": "βœ… Re-embed complete", "total_rows": total, "updated": updated} except Exception as e: logger.exception("reembed failed") return JSONResponse(status_code=500, content={"error": "reembed failed", "details": str(e)}) @app.get("/model-status") async def model_status(): return { "embed_loaded": embed_model is not None, "ollama_cli": ollama_cli_available(), "ollama_http": ollama_http_available(), "moderator": moderator is not None, "language_module": LANGUAGE_MODULE_AVAILABLE, "tts_available": TTS_AVAILABLE, "multipart_available": HAVE_MULTIPART, "pillow_available": PIL_AVAILABLE, "voicecloner_available": VOICECLONER_AVAILABLE, "coder_available": CODER_AVAILABLE, "videogen_available": VIDEOGEN_AVAILABLE, "image_editor_available": IMAGE_EDITOR_AVAILABLE } @app.get("/health") async def health(): try: with engine_knowledge.connect() as c: k = c.execute(sql_text("SELECT COUNT(*) FROM knowledge")).scalar() or 0 except Exception: k = -1 try: with engine_user.connect() as c: u = c.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0 except Exception: u = -1 return {"ok": True, "knowledge_count": int(k), "user_memory_count": int(u), "uptime_s": round(time.time() - app_start_time, 2), "heartbeat": last_heartbeat} # ------------------------- # Chat endpoint (topic-scoped, user-memory isolated) # ------------------------- @app.post("/chat") async def chat(request: Request, data: dict = Body(...)): t0 = time.time() # Performance optimization: Use caching cache_key = None if isinstance(data, dict): msg = str(data.get("message", "") or data.get("text", "") or "").strip() if msg: cache_key = hashlib.md5(msg.encode()).hexdigest() # Accept both "message" and "text" if isinstance(data, dict): raw_msg = str(data.get("message", "") or data.get("text", "") or "").strip() else: raw_msg = str(data or "").strip() if not raw_msg: record_request(time.time() - t0) return JSONResponse(status_code=400, content={"error": "Empty message"}) username = data.get("username", "anonymous") if isinstance(data, dict) else "anonymous" user_ip = request.client.host if request.client else "0.0.0.0" user_id = hashlib.sha256(f"{user_ip}-{username}".encode()).hexdigest() topic_hint = str(data.get("topic", "") or "").strip() if isinstance(data, dict) else "" include_steps = bool(data.get("include_steps", False) if isinstance(data, dict) else False) detected_lang = detect_language_safe(raw_msg) reply_lang = detected_lang if detected_lang and detected_lang != "und" else "en" # Translate incoming to English for retrieval if needed en_msg = raw_msg if detected_lang not in ("en", "eng", "", "und"): try: en_msg = translate_to_english(raw_msg, detected_lang) except Exception: en_msg = raw_msg # Load ALL knowledge entries first (needed for embedding-based topic inference) try: with engine_knowledge.begin() as conn: all_rows = conn.execute(sql_text("SELECT id, text, reply, language, embedding, topic FROM knowledge ORDER BY created_at DESC")).fetchall() except Exception as e: record_request(time.time() - t0) return JSONResponse(status_code=500, content={"error": "failed to read knowledge", "details": str(e)}) all_knowledge_rows = [{"id": r[0], "text": r[1] or "", "reply": r[2] or "", "lang": r[3] or "und", "embedding": r[4], "topic": r[5] or "general"} for r in all_rows] # Get list of known topics known_topics = list(set([kr.get("topic", "general") for kr in all_knowledge_rows if kr.get("topic")])) # Determine topic: Embeddings first (best), then Ollama, then keyword matching topic = "general" try: if not topic_hint: chosen = None # 1. Try embedding-based topic inference (BEST - semantic understanding) if embed_model is not None and all_knowledge_rows: try: chosen = infer_topic_with_embeddings(en_msg, known_topics, all_knowledge_rows) if chosen: logger.info(f"[topic] Selected via embeddings: {chosen}") except Exception as e: logger.debug(f"[topic] embedding inference failed: {e}") # 2. Fallback to Ollama if embeddings didn't work if not chosen: try: if (ollama_http_available() or ollama_cli_available()) and known_topics: possible = infer_topic_with_ollama(en_msg, known_topics) if possible: chosen = possible logger.info(f"[topic] Selected via Ollama: {chosen}") except Exception as e: logger.debug(f"[topic] ollama inference failed: {e}") # 3. Final fallback to keyword/fuzzy matching if not chosen: chosen = infer_topic_from_message(en_msg, known_topics) if chosen: logger.info(f"[topic] Selected via keyword/fuzzy: {chosen}") topic = chosen or "general" else: topic = topic_hint or "general" except Exception as e: logger.warning(f"[topic] inference error: {e}") topic = topic_hint or "general" logger.info(f"[chat] Final topic: {topic}") # Moderation flags = {} try: if moderator is not None: mod_res = moderator(raw_msg[:1024]) if isinstance(mod_res, list) and mod_res: lbl = mod_res[0].get('label', '').lower() sc = float(mod_res[0].get('score', 0.0)) if 'toxic' in lbl or sc > 0.85: flags['toxic'] = True except Exception: pass # Filter knowledge entries for this topic only knowledge_rows = [kr for kr in all_knowledge_rows if kr.get("topic") == topic] # Retrieval using cosine similarity with spell tolerance matches: List[str] = [] confidence = 0.0 match_lang = "en" try: # If we have an embed model, use semantic similarity (BEST approach) if embed_model is not None and knowledge_rows: stored_embs = [] stored_indices = [] # Collect stored embeddings for i, kr in enumerate(knowledge_rows): if kr.get("embedding") is not None: t = bytes_to_tensor(kr["embedding"]) if t is not None: stored_embs.append(t) stored_indices.append(i) # Use stored embeddings if available if torch is not None and stored_embs: try: # Stack stored embeddings embs_tensor = torch.stack(stored_embs) # Compute query embedding q_emb = await run_blocking_with_timeout( lambda: embed_model.encode([en_msg], convert_to_tensor=True, show_progress_bar=False)[0], timeout=MODEL_TIMEOUT ) if not isinstance(q_emb, torch.Tensor): q_emb = torch.from_numpy(q_emb.cpu().numpy()) # Compute cosine similarity try: scores = torch.nn.functional.cosine_similarity(q_emb.unsqueeze(0), embs_tensor, dim=1) except Exception: scores = torch.nn.functional.cosine_similarity(embs_tensor, q_emb.unsqueeze(0), dim=1) # Collect candidates with scores cand = [] for idx, s in enumerate(scores): i_orig = stored_indices[idx] kr = knowledge_rows[i_orig] candidate_text = (kr["reply"] or kr["text"]).strip() if is_boilerplate_candidate(candidate_text): continue s_float = float(s) # Lower threshold for better recall if s_float >= 0.25: cand.append({ "text": candidate_text, "lang": kr["lang"], "score": s_float }) # Sort by score cand = sorted(cand, key=lambda x: -x["score"]) matches = [c["text"] for c in cand[:5]] # Top 5 matches confidence = float(cand[0]["score"]) if cand else 0.0 match_lang = cand[0]["lang"] if cand else "en" logger.info(f"[retrieval] Found {len(matches)} matches via embeddings, best score: {confidence:.2f}") except asyncio.TimeoutError: logger.warning("[retrieval] embedding encode timed out") except Exception as e: logger.warning(f"[retrieval] embedding error: {e}") # Fallback: compute embeddings on the fly if no stored embeddings if not matches and knowledge_rows: try: texts = [kr["text"] for kr in knowledge_rows] embs = await run_blocking_with_timeout( lambda: embed_model.encode(texts, convert_to_tensor=True, show_progress_bar=False), timeout=MODEL_TIMEOUT ) q_emb = await run_blocking_with_timeout( lambda: embed_model.encode([en_msg], convert_to_tensor=True, show_progress_bar=False)[0], timeout=MODEL_TIMEOUT ) try: scores = torch.nn.functional.cosine_similarity(q_emb.unsqueeze(0), embs, dim=1) except Exception: scores = torch.nn.functional.cosine_similarity(embs, q_emb.unsqueeze(0), dim=1) cand = [] for i in range(scores.shape[0]): s = float(scores[i]) kr = knowledge_rows[i] candidate_text = (kr["reply"] or kr["text"]).strip() if is_boilerplate_candidate(candidate_text): continue if s >= 0.25: cand.append({ "text": candidate_text, "lang": kr["lang"], "score": s }) cand = sorted(cand, key=lambda x: -x["score"]) matches = [c["text"] for c in cand[:5]] confidence = float(cand[0]["score"]) if cand else 0.0 match_lang = cand[0]["lang"] if cand else "en" logger.info(f"[retrieval] Found {len(matches)} matches via on-the-fly embeddings, best score: {confidence:.2f}") except asyncio.TimeoutError: logger.warning("[retrieval] embedding encode timed out") except Exception as e: logger.warning(f"[retrieval] embedding error: {e}") # Final fallback: fuzzy keyword matching with spell tolerance if not matches and knowledge_rows: logger.info("[retrieval] Using fuzzy keyword matching fallback") cand = [] for kr in knowledge_rows: txt = (kr["reply"] or kr["text"]) or "" txt_lower = txt.lower() msg_lower = en_msg.lower() # Exact substring match if msg_lower in txt_lower: if not is_boilerplate_candidate(txt): cand.append({"text": txt, "lang": kr["lang"], "score": 0.8}) continue # Fuzzy matching for spell tolerance if FUZZY_AVAILABLE and len(en_msg) > 3: # Check fuzzy match against text fuzzy_score = fuzzy_match_score(en_msg, txt) if fuzzy_score > 0.6: if not is_boilerplate_candidate(txt): cand.append({"text": txt, "lang": kr["lang"], "score": fuzzy_score * 0.7}) continue # Check fuzzy match against individual words msg_words = [w for w in msg_lower.split() if len(w) > 3] txt_words = [w for w in txt_lower.split() if len(w) > 3] for msg_word in msg_words: for txt_word in txt_words: word_score = fuzzy_match_score(msg_word, txt_word) if word_score > 0.75: if not is_boilerplate_candidate(txt): cand.append({"text": txt, "lang": kr["lang"], "score": word_score * 0.5}) break # Remove duplicates and sort seen = set() unique_cand = [] for c in cand: if c["text"] not in seen: seen.add(c["text"]) unique_cand.append(c) cand = sorted(unique_cand, key=lambda x: -x["score"]) matches = [c["text"] for c in cand[:5]] confidence = float(cand[0]["score"]) if cand else 0.0 match_lang = cand[0]["lang"] if cand else "en" logger.info(f"[retrieval] Found {len(matches)} matches via fuzzy matching, best score: {confidence:.2f}") except Exception as e: logger.warning(f"[retrieval] error: {e}") matches = [] # Compose reply strictly from topic matches if matches and confidence >= 0.6: reply_en = matches[0] elif matches: reply_en = generate_creative_reply(matches[:5]) else: base = "This is outside the project, I can only help with problems related to the project." if reply_lang and reply_lang not in ("en", "eng", "und"): try: base = translate_from_english(base, reply_lang) except Exception: pass reply_final = base # Persist user memory (even on low confidence), skipping toxic content try: if not flags.get('toxic', False): with engine_user.begin() as conn: conn.execute(sql_text( "INSERT INTO user_memory (user_id, username, ip, text, reply, language, mood, confidence, topic, source) " "VALUES (:uid, :uname, :ip, :text, :reply, :lang, :mood, :conf, :topic, :source)" ), {"uid": user_id, "uname": username, "ip": user_ip, "text": raw_msg, "reply": reply_final, "lang": detected_lang, "mood": detect_mood(raw_msg + " " + reply_final), "conf": float(confidence), "topic": topic, "source": "chat"}) conn.execute(sql_text( "DELETE FROM user_memory WHERE id NOT IN (SELECT id FROM user_memory WHERE user_id = :uid ORDER BY created_at DESC LIMIT 10) AND user_id = :uid" ), {"uid": user_id}) except Exception as e: logger.debug(f"user_memory store error: {e}") record_request(time.time() - t0) return {"reply": reply_final, "topic": topic, "language": reply_lang, "emoji": "", "confidence": round(confidence,2), "flags": flags} # Post-process and translate back to user's language reply_en = dedupe_sentences(reply_en) reply_final = reply_en # Determine target language for translation target_lang = reply_lang if reply_lang and reply_lang not in ("en", "eng", "und", "") else None # If match was in a different language, try to use that if match_lang and match_lang not in ("en", "eng", "und", ""): # If user's language matches the match language, use it if target_lang and target_lang.split("-")[0].lower() == match_lang.split("-")[0].lower(): target_lang = match_lang # Translate to user's language if target_lang: lang_code = target_lang.split("-")[0].lower() try: logger.info(f"[translation] Translating reply from en to {lang_code}") reply_final = translate_from_english(reply_en, lang_code) reply_final = dedupe_sentences(reply_final) logger.info(f"[translation] Translation successful") except Exception as exc: logger.warning(f"[translation] failed to translate reply_en -> {lang_code}: {exc}") reply_final = reply_en else: logger.info("[translation] No translation needed, using English") # Mood & emoji append emoji = "" try: mood = detect_mood(raw_msg + " " + reply_final) if EMOJIS_AVAILABLE: try: cand = get_emoji(get_category_for_mood(mood), 0.6) if cand and cand not in reply_final and len(reply_final) + len(cand) < 1200: reply_final = f"{reply_final} {cand}" emoji = cand except Exception: emoji = "" except Exception: emoji = "" # Persist user memory (only in user DB) and prune to last 10 try: if not flags.get('toxic', False): with engine_user.begin() as conn: conn.execute(sql_text( "INSERT INTO user_memory (user_id, username, ip, text, reply, language, mood, confidence, topic, source) " "VALUES (:uid, :uname, :ip, :text, :reply, :lang, :mood, :conf, :topic, :source)" ), {"uid": user_id, "uname": username, "ip": user_ip, "text": raw_msg, "reply": reply_final, "lang": detected_lang, "mood": detect_mood(raw_msg + " " + reply_final), "conf": float(confidence), "topic": topic, "source": "chat"}) conn.execute(sql_text( "DELETE FROM user_memory WHERE id NOT IN (SELECT id FROM user_memory WHERE user_id = :uid ORDER BY created_at DESC LIMIT 10) AND user_id = :uid" ), {"uid": user_id}) except Exception as e: logger.debug(f"user_memory persist error: {e}") duration = time.time() - t0 record_request(duration) if include_steps: reply_final = f"{reply_final}\n\n[Debug: topic={topic} confidence={round(confidence,2)}]" return {"reply": reply_final, "topic": topic, "language": reply_lang, "emoji": emoji, "confidence": round(confidence,2), "flags": flags} @app.post("/response") async def response_wrapper(request: Request, data: dict = Body(...)): return await chat(request, data) @app.post("/verify-admin") async def verify_admin(x_admin_key: str = Header(None, alias="X-Admin-Key")): if ADMIN_KEY is None: return JSONResponse(status_code=403, content={"error": "Server not configured for admin operations."}) if not x_admin_key or x_admin_key != ADMIN_KEY: return JSONResponse(status_code=403, content={"valid": False, "error": "Invalid or missing admin key."}) return {"valid": True} @app.post("/cleardatabase") async def clear_database(data: dict = Body(...), x_admin_key: str = Header(None, alias="X-Admin-Key")): if ADMIN_KEY is None: return JSONResponse(status_code=403, content={"error": "Server not configured for admin operations."}) if x_admin_key != ADMIN_KEY: return JSONResponse(status_code=403, content={"error": "Invalid admin key."}) confirm = str(data.get("confirm", "") or "").strip() if confirm != "CLEAR_DATABASE": return JSONResponse(status_code=400, content={"error": "confirm token required."}) try: with engine_knowledge.begin() as conn: k_count = conn.execute(sql_text("SELECT COUNT(*) FROM knowledge")).scalar() or 0 conn.execute(sql_text("DELETE FROM knowledge")) with engine_user.begin() as conn: u_count = conn.execute(sql_text("SELECT COUNT(*) FROM user_memory")).scalar() or 0 conn.execute(sql_text("DELETE FROM user_memory")) return {"status": "βœ… Cleared database", "deleted_knowledge": int(k_count), "deleted_user_memory": int(u_count)} except Exception as e: logger.exception("clear failed") return JSONResponse(status_code=500, content={"error": "failed to clear database", "details": str(e)}) # ------------------------- # Coder endpoints # ------------------------- @app.post("/coder/run") async def coder_run_code(data: dict = Body(...)): """Execute code in sandbox""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") code = data.get("code", "") lang = data.get("language", "python") timeout = int(data.get("timeout", 15)) if not code: raise HTTPException(status_code=400, detail="Code is required") try: result = coder_instance.run_code(code, lang, timeout) return result except Exception as e: logger.exception("Coder run failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/debug") async def coder_debug_code(data: dict = Body(...)): """Debug code in sandbox""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") code = data.get("code", "") lang = data.get("language", "python") if not code: raise HTTPException(status_code=400, detail="Code is required") try: result = coder_instance.debug_code(code, lang) return result except Exception as e: logger.exception("Coder debug failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/fix") async def coder_fix_code(data: dict = Body(...)): """Automatically fix code issues""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") code = data.get("code", "") lang = data.get("language", "python") if not code: raise HTTPException(status_code=400, detail="Code is required") try: result = coder_instance.fix_code(code, lang) return result except Exception as e: logger.exception("Coder fix failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/generate") async def generate_code(data: dict = Body(...)): """Generate code from natural language request""" if not CODER_AVAILABLE: raise HTTPException( status_code=503, detail="Coder module not available. Please check server logs and ensure all dependencies are installed." ) if coder_instance is None: raise HTTPException( status_code=503, detail="Coder instance not initialized. Please restart the server." ) request = data.get("request", "") lang = data.get("language", "python") if not request: raise HTTPException(status_code=400, detail="Request is required") try: result = coder_instance.generate_code(request, lang) return result except AttributeError as e: logger.exception("Code generation failed - method not found") return JSONResponse( status_code=500, content={"error": f"Code generation method not available: {str(e)}"} ) except Exception as e: logger.exception("Code generation failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/preview/start") async def coder_start_preview(data: dict = Body(...)): """Start preview server""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") lang = data.get("language", "html") port = int(data.get("port", 8000)) html_content = data.get("html") # optional HTML body try: result = coder_instance.start_preview(lang=lang, port=port, html_content=html_content) return result except Exception as e: logger.exception("Preview start failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/preview/stop") async def coder_stop_preview(): """Stop preview server""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") try: result = coder_instance.stop_preview() return result except Exception as e: logger.exception("Preview stop failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/coder/preview/info") async def coder_preview_info(): """Get preview server info""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") try: result = coder_instance.get_preview_info() return result except Exception as e: logger.exception("Preview info failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/file/write") async def coder_write_file(data: dict = Body(...)): """Write file to sandbox""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") filename = data.get("filename", "") content = data.get("content", "") if not filename: raise HTTPException(status_code=400, detail="Filename is required") try: result = coder_instance.write_file(filename, content) return result except Exception as e: logger.exception("File write failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.post("/coder/file/read") async def coder_read_file(data: dict = Body(...)): """Read file from sandbox""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") filename = data.get("filename", "") if not filename: raise HTTPException(status_code=400, detail="Filename is required") try: result = coder_instance.read_file(filename) return result except Exception as e: logger.exception("File read failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/coder/files") async def coder_list_files(): """List files in sandbox""" if not CODER_AVAILABLE or coder_instance is None: raise HTTPException(status_code=503, detail="Coder module not available") try: result = coder_instance.list_files() return result except Exception as e: logger.exception("File list failed") return JSONResponse(status_code=500, content={"error": str(e)}) # ------------------------- # Video Generator endpoints # ------------------------- @app.post("/video/generate") async def video_generate(background_tasks: BackgroundTasks, data: dict = Body(...)): """Generate video from prompt""" if not VIDEOGEN_AVAILABLE or video_generator is None: raise HTTPException(status_code=503, detail="Video generator not available") prompt = data.get("prompt", "") num_frames = int(data.get("num_frames", 16)) fps = int(data.get("fps", 8)) enhance = bool(data.get("enhance", False)) if not prompt: raise HTTPException(status_code=400, detail="Prompt is required") try: loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, lambda: video_generator.generate( prompt=prompt, num_frames=num_frames, fps=fps, enhance=enhance ) ) return result except Exception as e: logger.exception("Video generation failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/video/history") async def video_history(limit: int = Query(20)): """Get video generation history""" if not VIDEOGEN_AVAILABLE or video_generator is None: raise HTTPException(status_code=503, detail="Video generator not available") try: history = video_generator.get_history(limit) return {"history": history} except Exception as e: logger.exception("Video history failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/video/status") async def video_status(): """Get video generator status""" if not VIDEOGEN_AVAILABLE or video_generator is None: raise HTTPException(status_code=503, detail="Video generator not available") try: status = video_generator.get_status() return status except Exception as e: logger.exception("Video status failed") return JSONResponse(status_code=500, content={"error": str(e)}) @app.get("/", response_class=HTMLResponse) async def frontend_dashboard(): try: health = requests.get("http://localhost:7860/health", timeout=1).json() except Exception: health = {"status": "starting", "db_status": "unknown", "stars": 0, "db_metrics": {}} db_metrics = health.get("db_metrics") or {} knowledge_count = db_metrics.get("knowledge_count", "?") user_memory_count = db_metrics.get("user_memory_count", "?") stars = health.get("stars", 0) hb = last_heartbeat try: hb_display = f'{hb.get("time")} (ok={hb.get("ok")})' if isinstance(hb, dict) else str(hb) except Exception: hb_display = str(hb) startup_time_local = round(time.time() - app_start_time, 2) try: with open("frontend.html", "r") as f: html = f.read() except Exception: html = "

Frontend file not found

" html = html.replace("%%HEALTH_STATUS%%", str(health.get("status", "starting"))) html = html.replace("%%KNOWLEDGE_COUNT%%", str(knowledge_count)) html = html.replace("%%USER_MEMORY_COUNT%%", str(user_memory_count)) html = html.replace("%%STARS%%", "⭐" * int(stars) if isinstance(stars, int) else str(stars)) html = html.replace("%%HB_DISPLAY%%", hb_display) html = html.replace("%%FOOTER_TIME%%", datetime.utcnow().isoformat()) html = html.replace("%%STARTUP_TIME%%", str(startup_time_local)) return HTMLResponse(html) # ------------------------- # Run server # ------------------------- if __name__ == "__main__": # Preload TTS and embeddings in background to reduce first-request latency if TTS_AVAILABLE: try: threading.Thread(target=lambda: get_tts_model_blocking(), daemon=True).start() except Exception: pass if SentenceTransformer is not None: try: threading.Thread(target=try_load_embed, daemon=True).start() except Exception: pass app_start_time = time.time() import uvicorn port = int(os.environ.get("PORT", 7860)) uvicorn.run("app:app", host="0.0.0.0", port=port, log_level="info")