import ast import logging import re from datetime import datetime, timezone import threading from fastapi import FastAPI from pydantic import BaseModel from engines.scene_engine import SceneEngine from engines.explanation_engine import ExplanationEngine from engines.content_engine import ContentEngine from engines.image_generation_engine import ImageGenerationEngine from engines.prompt_intelligence_engine import PromptIntelligenceEngine from engines.tiktok_content_engine import TikTokContentEngine from engines.onyx_brain import OnyxBrain from engines.memory_store import ( add_to_history, set_style, set_mode, set_ideas, set_selected_idea, get_context ) logger = logging.getLogger("ONYX") logging.basicConfig(level=logging.INFO) app = FastAPI() print("đŸ”„ APP STARTED") scene_engine = SceneEngine() explanation_engine = ExplanationEngine() content_engine = ContentEngine() prompt_intelligence_engine = PromptIntelligenceEngine() tiktok_engine = TikTokContentEngine(scene_engine) onyx_brain = OnyxBrain() REQUEST_LOCK = threading.Lock() LAST_DRAFT_IDEAS = [] LAST_SELECTED_IDEA = { "index": None, "text": None } LAST_ASSISTANT_RESPONSE = "" LAST_CREATIVE_CONTEXT = "" try: image_generation_engine = ImageGenerationEngine() print("đŸ–Œïž ImageGenerationEngine OK") except Exception as e: image_generation_engine = None logger.warning(f"ImageGenerationEngine indisponible: {e}") print("⚠ ImageGenerationEngine indisponible :", e) class RequestModel(BaseModel): prompt: str style: str = "tiktok" def normalize(text: str) -> str: return re.sub(r"\s+", " ", (text or "").strip()) def remember_response(raw_prompt: str, response: dict): global LAST_ASSISTANT_RESPONSE try: scenes = response.get("scenes", []) response_text = "\n\n".join( [scene.get("text", "") for scene in scenes if scene.get("text")] ).strip() if response_text: LAST_ASSISTANT_RESPONSE = response_text if raw_prompt and response_text: add_to_history(raw_prompt, response_text) except Exception as e: logger.warning(f"Memory history skipped: {e}") return response def make_text_response(text: str, mode: str = "draft", style: str = "tiktok"): return { "status": "ok", "mode": mode, "style": style, "scenes": [ { "id": 1, "text": text.strip(), "image": "" } ] } def is_simple_chat_message(prompt: str) -> bool: p = normalize(prompt).lower() simple_messages = [ "salut", "bonjour", "coucou", "hey", "yo", "bonsoir", "merci", "merci !", "merci.", "ok", "okay", "d accord", "d'accord", "oui", "non", "ça va", "ca va", "ça va ?", "ca va ?", "ça va merci", "ca va merci", "ça va merci et toi", "ca va merci et toi", "et toi", "comment tu vas", "comment tu vas ?", "comment ça va", "comment ca va" ] return p in simple_messages def is_continuation_command(prompt: str) -> bool: p = normalize(prompt).lower() continuation_commands = [ "continue", "continu", "suite", "la suite", "ensuite", "aprĂšs", "apres", "poursuis", "reprends", "reprend", "continue la suite", "vas-y continue", "vas y continue", "continue comme ça", "continue comme ca" ] return p in continuation_commands def is_develop_request(prompt: str) -> bool: p = normalize(prompt).lower() keywords = [ "dĂ©veloppe", "developpe", "dĂ©velopper", "developper", "version finale", "finale", "final", "finalise", "finaliser", "continue", "continu", "suite", "ensuite", "poursuis", "reprends", "reprend", "transforme en aperçu", "transforme en apercu", "transforme en final", "fais l'aperçu", "fais l’aperçu", "fais le final", "passe en aperçu", "passe en apercu", "passe en final" ] return any(k in p for k in keywords) def is_conversation_request(prompt: str) -> bool: p = normalize(prompt).lower() if not p: return False if is_continuation_command(p): return False if is_develop_request(p): return False if is_simple_chat_message(p): return True creation_keywords = [ "tiktok", "tik tok", "reel", "reels", "short", "shorts", "vidĂ©o", "video", "scĂšne", "scene", "histoire", "script", "idĂ©e", "idee", "contenu", "brouillon", "aperçu", "apercu", "final", "cinĂ©ma", "cinema", "storyboard", "hook", "twist", "gĂ©nĂšre", "genere", "crĂ©e", "cree", "imagine", "invente", "rĂ©dige", "redige", "dragon ball", "dbz", "anime", "manga" ] if any(k in p for k in creation_keywords): return False conversation_signals = [ "salut", "bonjour", "coucou", "j ai besoin", "j'ai besoin", "j’ai besoin", "j aurais besoin", "j'aurais besoin", "j’aurais besoin", "j ai une question", "j'ai une question", "j’ai une question", "question", "rĂ©ponse", "reponse", "rĂ©ponses", "reponses", "peux tu", "peux-tu", "peut tu", "peut-tu", "tu peux", "aide moi", "aide-moi", "explique", "c est quoi", "c'est quoi", "c’est quoi", "comment", "pourquoi", "est ce que", "est-ce que", "dit moi", "dis moi", "dis-moi", "dit-moi", "ça va", "ca va", "et toi", "quelles fonction", "quelle fonction", "que peut tu", "que peux tu", "que peut-tu", "que peux-tu", "que peut tu faire", "que peux tu faire", "tu sais faire quoi", "tu peux faire quoi", "tu en penses quoi", "tu penses quoi", "ton avis", "avis", "ça peut marcher", "ca peut marcher" ] if any(k in p for k in conversation_signals): return True if len(p.split()) <= 5: return True return False def should_use_memory(prompt: str) -> bool: p = normalize(prompt).lower() if not p: return False if is_simple_chat_message(p): return False if is_continuation_command(p): return True if len(p.split()) <= 2: memory_short_keywords = [ "continue", "suite", "ensuite", "poursuis", "reprends", "reprend", "finalise", "dĂ©veloppe", "developpe", "finale", "final" ] return any(k == p for k in memory_short_keywords) memory_keywords = [ "continue", "suite", "ensuite", "poursuis", "reprends", "reprend", "reprendre", "dĂ©veloppe", "developpe", "dĂ©velopper", "developper", "finalise", "finaliser", "dans ce style", "mĂȘme style", "meme style", "comme avant", "pareil", "mĂȘme univers", "meme univers", "la mĂȘme histoire", "la meme histoire", "cette histoire", "histoire prĂ©cĂ©dente", "histoire precedente", "ce personnage", "cet univers", "cette idĂ©e", "cette idee", "idĂ©e sĂ©lectionnĂ©e", "idee selectionnee" ] return any(k in p for k in memory_keywords) def is_new_creative_request(prompt: str) -> bool: p = normalize(prompt).lower() if not p: return False if should_use_memory(p): return False if is_direct_selection_request(p): return False new_request_keywords = [ "donne moi", "donne-moi", "trouve", "propose", "invente", "imagine", "crĂ©e", "cree", "gĂ©nĂšre", "genere", "Ă©cris", "ecris", "rĂ©dige", "redige", "idĂ©es", "idees", "idĂ©e", "idee", "tiktok", "tik tok", "cinĂ©ma", "cinema", "histoire", "script", "dragon ball", "dbz", "anime", "manga" ] return any(k in p for k in new_request_keywords) def reset_current_selection(): global LAST_SELECTED_IDEA global LAST_CREATIVE_CONTEXT LAST_SELECTED_IDEA["index"] = None LAST_SELECTED_IDEA["text"] = None LAST_CREATIVE_CONTEXT = "" try: set_selected_idea(None) except Exception: pass def build_memory_context_prompt(prompt: str) -> str: global LAST_ASSISTANT_RESPONSE global LAST_CREATIVE_CONTEXT if not should_use_memory(prompt): return prompt try: memory_context = get_context() except Exception as e: logger.warning(f"Memory context unavailable: {e}") memory_context = "" useful_context_parts = [] if memory_context: useful_context_parts.append(f"Historique mĂ©moire :\n{memory_context}") if LAST_SELECTED_IDEA.get("text"): useful_context_parts.append( f"IdĂ©e sĂ©lectionnĂ©e actuellement :\n{LAST_SELECTED_IDEA.get('text')}" ) if LAST_CREATIVE_CONTEXT: useful_context_parts.append( f"Contexte crĂ©atif actuel :\n{LAST_CREATIVE_CONTEXT}" ) if LAST_ASSISTANT_RESPONSE: useful_context_parts.append( f"DerniĂšre rĂ©ponse Onyx :\n{LAST_ASSISTANT_RESPONSE}" ) if not useful_context_parts: return prompt return f""" Contexte Onyx : {chr(10).join(useful_context_parts)} Demande actuelle : {prompt} RĂšgle : Utilise ce contexte uniquement parce que la demande actuelle demande une suite, un dĂ©veloppement, un avis ou une continuitĂ©. Ne redĂ©marre jamais une nouvelle conversation si la demande est : continue, suite, ensuite, dĂ©veloppe, finale. Ne mĂ©lange jamais ce contexte avec une nouvelle demande crĂ©ative qui possĂšde un nouveau sujet. """.strip() def build_chat_prompt(prompt: str) -> str: return f""" Tu es Onyx AI, un assistant intelligent, naturel et polyvalent. RĂ©ponds comme une IA moderne : - naturellement - clairement - utilement - sans inventer une histoire si l'utilisateur discute simplement - sans partir en scĂšne ou en narration sauf si l'utilisateur demande une crĂ©ation - sans rĂ©pondre comme un robot CapacitĂ©s Ă  prĂ©senter si on te demande ce que tu sais faire : - rĂ©pondre Ă  des questions - expliquer des sujets - aider Ă  crĂ©er des idĂ©es, scripts, scĂšnes, contenus TikTok/Shorts/Reels - aider Ă  prĂ©parer des prompts d’images - aider Ă  organiser un projet, du code, du business ou une stratĂ©gie - accompagner l’utilisateur dans son projet Onyx RĂšgles importantes : - Si l'utilisateur dit bonjour ou discute, rĂ©ponds simplement. - Si l'utilisateur demande tes fonctions, prĂ©sente tes capacitĂ©s. - Si l'utilisateur demande si tu peux gĂ©nĂ©rer des images, rĂ©ponds que tu peux aider Ă  crĂ©er/prĂ©parer des prompts et Ă  lancer la gĂ©nĂ©ration selon le moteur image disponible dans Onyx. - Ne dĂ©marre jamais une histoire sans demande claire de crĂ©ation. - Si l'utilisateur demande ton avis sur une idĂ©e prĂ©cĂ©dente, donne un avis utile et concret. - RĂ©ponds en français. Demande utilisateur : {prompt} """.strip() def chat_response(prompt: str, mode: str = "draft", style: str = "tiktok"): try: text = content_engine.chat(build_chat_prompt(prompt)) return make_text_response(text, mode=mode, style=style) except Exception as e: logger.warning(f"Chat engine failed: {e}") return make_text_response( "Je suis lĂ . Dis-moi ce que tu veux faire ou ce que tu veux savoir.", mode=mode, style=style ) def detect_style(raw_prompt: str, req_style: str = "tiktok") -> str: p = normalize(raw_prompt).lower() s = normalize(req_style).lower() if "[onyx_style_cinema]" in p: return "cinema" if "[onyx_style_tiktok]" in p: return "tiktok" if s in ["cinema", "cinĂ©ma", "cine", "cinĂ©", "cinematic"]: return "cinema" return "tiktok" def detect_generation_mode(prompt: str) -> str: p = normalize(prompt).lower() if is_continuation_command(p): return "continue" if "[onyx_studio_final" in p: return "final" if "[onyx_studio_preview" in p: return "preview" if "[onyx_studio_draft" in p: return "draft" if "[final]" in p or "mode final" in p: return "final" if "[preview]" in p or "mode preview" in p: return "preview" if "[draft]" in p or "mode draft" in p: return "draft" if p in ["finale", "final", "version finale", "finalise"]: return "final" if p in ["aperçu", "apercu", "preview", "dĂ©veloppe", "developpe"]: return "preview" return "draft" def sanitize_prompt_for_generation(prompt: str) -> str: clean = (prompt or "").strip() clean = re.sub(r"\[ONYX_STYLE_TIKTOK\]", "", clean, flags=re.IGNORECASE) clean = re.sub(r"\[ONYX_STYLE_CINEMA\]", "", clean, flags=re.IGNORECASE) sujet_matches = re.findall( r"Sujet\s*:\s*(.*)$", clean, flags=re.IGNORECASE | re.DOTALL ) if sujet_matches: extracted = sujet_matches[-1].strip() if extracted: return normalize(extracted) demande_matches = re.findall( r"Demande utilisateur\s*:\s*(.*)$", clean, flags=re.IGNORECASE | re.DOTALL ) if demande_matches: extracted = demande_matches[-1].strip() if extracted: return normalize(extracted) markers = [ "[ONYX_STUDIO_DRAFT]", "[ONYX_STUDIO_PREVIEW]", "[ONYX_STUDIO_FINAL]", "[ONYX_STYLE_TIKTOK]", "[ONYX_STYLE_CINEMA]", "[draft]", "[preview]", "[final]", "mode draft", "mode preview", "mode final" ] result = clean for marker in markers: result = re.sub(re.escape(marker), "", result, flags=re.IGNORECASE) return normalize(result) def is_tiktok_like_request(prompt: str) -> bool: p = normalize(prompt).lower() if is_continuation_command(p): return True keywords = [ "tiktok", "tik tok", "reel", "reels", "short", "shorts", "vidĂ©o courte", "video courte", "contenu viral", "hook", "script vidĂ©o", "script video", "faire une vidĂ©o", "faire une video", "crĂ©e une vidĂ©o", "cree une video", "contenu court", "mini script", "idĂ©e de contenu", "idee de contenu", "studio onyx", "dragon ball", "dbz", "anime", "manga" ] return any(k in p for k in keywords) def detect_requested_scene_count(prompt: str, default_count: int) -> int: p = normalize(prompt).lower() if "6 scĂšnes" in p or "6 scenes" in p: return 6 if "5 scĂšnes" in p or "5 scenes" in p: return 5 if "4 scĂšnes" in p or "4 scenes" in p: return 4 if "3 scĂšnes" in p or "3 scenes" in p: return 3 if "2 scĂšnes" in p or "2 scenes" in p: return 2 if "1 scĂšne" in p or "1 scene" in p: return 1 return default_count def detect_selected_idea(prompt: str): p = normalize(prompt).lower() patterns = [ (1, [r"\bidĂ©e 1\b", r"\bidee 1\b", r"\bchoix 1\b", r"\boption 1\b"]), (2, [r"\bidĂ©e 2\b", r"\bidee 2\b", r"\bchoix 2\b", r"\boption 2\b"]), (3, [r"\bidĂ©e 3\b", r"\bidee 3\b", r"\bchoix 3\b", r"\boption 3\b"]), ] for idx, regex_list in patterns: for regex in regex_list: if re.search(regex, p): return idx return None def is_direct_selection_request(prompt: str) -> bool: p = normalize(prompt).lower() direct_patterns = [ r"^(je\s+choisis\s+)?(l['’]?)?idĂ©e\s*1$", r"^(je\s+choisis\s+)?(l['’]?)?idee\s*1$", r"^choix\s*1$", r"^option\s*1$", r"^(je\s+choisis\s+)?(l['’]?)?idĂ©e\s*2$", r"^(je\s+choisis\s+)?(l['’]?)?idee\s*2$", r"^choix\s*2$", r"^option\s*2$", r"^(je\s+choisis\s+)?(l['’]?)?idĂ©e\s*3$", r"^(je\s+choisis\s+)?(l['’]?)?idee\s*3$", r"^choix\s*3$", r"^option\s*3$", ] return any(re.search(pattern, p) for pattern in direct_patterns) def build_effective_develop_prompt(prompt: str, mode: str, selected_style: str) -> str: global LAST_CREATIVE_CONTEXT selected_text = LAST_SELECTED_IDEA.get("text") if selected_text: base = f""" Sujet principal obligatoire : {selected_text} Demande utilisateur : {prompt} Style demandĂ© : {selected_style} RĂšgle absolue : Tu dois rester uniquement sur le sujet principal obligatoire. Tu ne dois pas introduire un autre sujet, objet, personnage ou univers sans lien. Tu ne dois jamais recommencer l’histoire depuis le dĂ©but. Si la demande est une suite : - Continue directement l’histoire prĂ©cĂ©dente - Ajoute de nouveaux Ă©vĂ©nements cohĂ©rents - Augmente la tension, la rĂ©vĂ©lation ou la progression - Ne reformule pas les scĂšnes prĂ©cĂ©dentes - Ne rĂ©pĂšte pas les mĂȘmes idĂ©es """.strip() LAST_CREATIVE_CONTEXT = base return base memory_prompt = build_memory_context_prompt(prompt) LAST_CREATIVE_CONTEXT = memory_prompt return memory_prompt def parse_possible_dict_string(value): if not isinstance(value, str): return value cleaned = value.strip() if not ( (cleaned.startswith("{") and cleaned.endswith("}")) or (cleaned.startswith("[") and cleaned.endswith("]")) ): return value try: return ast.literal_eval(cleaned) except Exception: return value def clean_raw_scene_text(text: str) -> str: cleaned = normalize(text) parsed = parse_possible_dict_string(cleaned) if isinstance(parsed, dict): return normalize(parsed.get("text", "")) cleaned = re.sub( r"^\{['\"]id['\"]\s*:\s*\d+\s*,\s*['\"]text['\"]\s*:\s*['\"]", "", cleaned, flags=re.IGNORECASE ) cleaned = re.sub(r"['\"]\s*,\s*['\"]image['\"]\s*:\s*['\"].*?['\"]\s*\}$", "", cleaned) cleaned = re.sub(r"['\"]\s*\}$", "", cleaned) return normalize(cleaned) def normalize_scene_item(scene, fallback_id: int): scene = parse_possible_dict_string(scene) if isinstance(scene, dict): text = normalize(scene.get("text", "")) image = scene.get("image", "") or "" else: text = normalize(str(scene)) image = "" text = clean_raw_scene_text(text) if not text: return None return { "id": fallback_id, "text": text, "image": image } def dedupe_scenes(formatted_scenes): cleaned = [] seen = set() for item in formatted_scenes: raw_text = normalize(item.get("text", "")).lower() text_key = raw_text[:80] if not text_key: continue if text_key in seen: continue seen.add(text_key) cleaned.append(item) for i, item in enumerate(cleaned): item["id"] = i + 1 return cleaned def enforce_mode_behavior(formatted_scenes, mode: str, is_tiktok: bool = False): # ONYX_TIKTOK_NATIVE_RICH_BLOCKS_V8 # ONYX_TIKTOK_QUALITY_GATE_V9 # Reconstruit des cartes TikTok propres et applique un contrĂŽle qualitĂ© dĂ©terministe # pour rendre Visuel / Voix off / Texte Ă©cran / Rythme rĂ©ellement distincts et exploitables. if is_tiktok and isinstance(formatted_scenes, list) and formatted_scenes: # ONYX_TIKTOK_INTENT_ROUTER_V10 # Les demandes de sujets/idĂ©es TikTok doivent rester des cartes SUJET. # Le quality gate V9 ne doit pas les convertir en scĂšnes narratives. def _is_topic_card(value): if isinstance(value, dict): raw = str(value.get('text') or value.get('content') or value.get('description') or '') else: raw = str(value or '') return bool(re.search(r'(?im)^\s*SUJET\s*\d+\s*:', raw)) if mode == 'draft' and all(_is_topic_card(item) for item in formatted_scenes): cleaned_topics = [] for idx, item in enumerate(formatted_scenes[:3], 1): raw = str(item.get('text') or item.get('content') or item.get('description') or '') if isinstance(item, dict) else str(item or '') raw = re.sub(r'(?im)^\s*SUJET\s*\d+\s*:', 'SUJET ' + str(idx) + ' :', raw.strip(), count=1) raw = re.sub(r'\n{3,}', '\n\n', raw).strip() cleaned_topics.append({'id': idx, 'text': raw, 'title': 'SUJET ' + str(idx), 'image': item.get('image', '') if isinstance(item, dict) else ''}) return cleaned_topics def _field(text, label): labels = ['Visuel', 'Voix off', 'Texte Ă©cran', 'Rythme'] idx = labels.index(label) next_labels = labels[idx + 1:] if next_labels: nxt = '|'.join([re.escape(x) + r'\s*:' for x in next_labels]) pat = r'(?is)' + re.escape(label) + r'\s*:\s*(.*?)(?=\s*(?:' + nxt + r')|$)' else: pat = r'(?is)' + re.escape(label) + r'\s*:\s*(.*)$' m = re.search(pat, text or '') return re.sub(r'\s+', ' ', m.group(1)).strip(' -:;,.') if m else '' def _strip_old_labels(text): value = re.sub(r'(?i)\b(Hook|Titre|Beat\s*\d+|Twist|Chute|Concept|Suite\s*\d+|Chute suivante)\s*:', ' ', text or '') value = re.sub(r'(?i)^\s*SC[ÈE]NE\s*\d+\s*:?\s*', '', value) value = re.sub(r'\s+', ' ', value).strip(' -:;,.') return value def _clean_quotes(text): value = re.sub(r'["“”«»]', '', text or '') value = re.sub(r"^['’]+|['’]+$", '', value.strip()) return re.sub(r'\s+', ' ', value).strip(' -:;,.') def _shorten(text, limit=96): text = re.sub(r'\s+', ' ', (text or '').strip(' -:;,.')) return text if len(text) <= limit else text[:limit - 3].rstrip() + '...' def _keyword_from(text): stop = set('avec dans pour les des une un aux sur qui que quoi dont mais donc puis tres trĂšs plus tout tous comme sans sous entre vers chez par est sont etre ĂȘtre avoir fait font son ses leur leurs votre notre titre beat concept idee idĂ©e scene scĂšne hook twist chute voix off visuel texte ecran Ă©cran rythme aujourd hui alors quand cette cet ces celui celle cela dans vient faire fait regarde jamais toujours pourquoi comment debut dĂ©but vraiment presque maintenant personne juste'.split()) words = [] for w in re.sub(r"[-:;,!?\.()\[\]{}\"'’]+", ' ', text or '').split(): w2 = ''.join(ch for ch in w.lower() if ch.isalnum()) if len(w2) >= 4 and w2 not in stop and w2 not in words: words.append(w2) return words[0].upper() if words else 'MOMENT' def _contains_action(text): action_roots = [ 'avance', 'recule', 'tombe', 'court', 'saute', 'attrape', 'lance', 'ouvre', 'ferme', 'tourne', 'montre', 'filme', 'zoome', 'tape', 'pousse', 'glisse', 'sourit', 'fixe', 'regarde', 'rĂ©agit', 'reagit', 'explose', 'apparaĂźt', 'apparait', 'disparaĂźt', 'disparait', 'entre', 'sort', 'dĂ©voile', 'devoile', 'renverse', 'bloque', 'cache', 'dĂ©place', 'deplace', 'tremble', 'panique' ] low = (text or '').lower() return any(root in low for root in action_roots) def _visual_from(seed, idx): key = _keyword_from(seed).lower() if idx == 1: base = 'Gros plan vertical : le sujet se fige, regarde camĂ©ra, dĂ©tail ' + key + ' visible.' elif idx == 5: base = 'Dernier plan serrĂ© : rĂ©action muette, accessoire au premier plan, rĂ©sultat impossible Ă  rater.' elif idx == 4: base = 'Plan Ă©paule rapide : le sujet recule brusquement, dĂ©cor qui bouge, chaos lisible.' elif idx == 3: base = 'Plan rapprochĂ© : main qui hĂ©site, objet qui dĂ©rape, expression de panique nette.' else: base = 'Plan court camĂ©ra main : action simple, regard surpris, transition vers le problĂšme.' return _shorten(base, 128) def _upgrade_visual(visual, seed, idx): visual = _clean_quotes(_strip_old_labels(visual)) low = visual.lower() forbidden_narration = any(x in low for x in [' raconte ', ' explique ', ' pense ', ' se souvient ', ' dĂ©couvre que ', ' comprend que ', ' va devoir ']) missing_camera = not any(x in low for x in ['plan', 'gros plan', 'camĂ©ra', 'camera', 'travelling', 'zoom', 'vertical', 'champ', 'contrechamp']) if not visual or len(visual.split()) > 20 or not _contains_action(visual) or forbidden_narration: return _visual_from(seed or visual, idx) if missing_camera: visual = 'Plan vertical : ' + visual[0].lower() + visual[1:] if visual else visual return _shorten(visual, 128) def _screen_from(text, idx): raw = _clean_quotes(text).upper() raw = re.sub(r'[^A-ZÀÂÄÇÉÈÊËÎÏÔÖÙÛÜƾƒÆ0-9\s-]', ' ', raw) raw = re.sub(r'\s+', ' ', raw).strip(' -') words = [w for w in raw.split() if len(w) > 1] bland = {'AIDE', 'PRET', 'PRÊT', 'OUPS', 'OK', 'GO', 'FIN', 'VOIR', 'À', 'A'} if 2 <= len(words) <= 4 and not (len(words) == 1 and words[0] in bland): return ' '.join(words[:4]) key = _keyword_from(text) templates = { 1: ['STOP', key], 2: [key, 'DÉRAPE'], 3: ['ÇA', 'TOURNE', 'MAL'], 4: ['TWIST', 'TOTAL'], 5: ['FIN', 'IMPOSSIBLE'], } return ' '.join(templates.get(idx, [key, 'DÉRAPE'])) def _voice_from(text, seed, idx): voice = _clean_quotes(_strip_old_labels(text or seed)) generic = ['aujourd hui je vais', 'dans cette vidĂ©o', 'nous allons voir', 'voici comment', 'salut tout le monde'] if not voice or any(g in voice.lower() for g in generic): key = _keyword_from(seed).lower() presets = { 1: 'LĂ , on sait dĂ©jĂ  que ça va mal finir.', 2: 'Le pire, c’est qu’il croit encore contrĂŽler la situation.', 3: 'À ce moment-lĂ , tout part beaucoup trop vite.', 4: 'Personne n’était prĂȘt pour ce retournement.', 5: 'Et voilĂ  pourquoi cette scĂšne reste en tĂȘte.' } voice = presets.get(idx, 'LĂ , le ' + key + ' change tout.') voice = _shorten(voice, 118) if voice and voice[-1] not in '.!?': voice += '.' return voice def _rhythm_from(text, idx): rhythm = _clean_quotes(_strip_old_labels(text)) low = rhythm.lower() too_vague = (not rhythm) or len(rhythm.split()) < 3 or any(x in low for x in ['dynamique et entraĂźnant', 'dynamique et entrainant', 'curieux et rapide', 'tempo rapide', 'rythme rapide']) if not too_vague: return _shorten(rhythm, 76) presets = { 1: 'Cut immĂ©diat, zoom rĂ©action.', 2: 'Plan bref, jump cut net.', 3: 'AccĂ©lĂ©ration, cut sur dĂ©tail.', 4: 'Rupture sĂšche, zoom panique.', 5: 'Micro-pause, chute en cut.' } return presets.get(idx, 'Cut rapide, rĂ©action lisible.') rich = [] max_count = 3 if mode == 'draft' else (4 if mode == 'continue' else 5) for idx, item in enumerate(formatted_scenes[:max_count], 1): if isinstance(item, dict): text = str(item.get('text') or item.get('content') or item.get('description') or '').strip() else: text = str(item).strip() if not text: continue has_rich = all(label in text for label in ['Visuel', 'Voix off', 'Texte Ă©cran', 'Rythme']) if has_rich: visual = _field(text, 'Visuel') voice = _field(text, 'Voix off') screen = _field(text, 'Texte Ă©cran') rhythm = _field(text, 'Rythme') seed = voice or visual or text else: seed = _strip_old_labels(text) visual = '' voice = seed screen = '' rhythm = '' seed = _strip_old_labels(seed or text) voice = _voice_from(voice, seed, idx) visual = _upgrade_visual(visual, seed or voice, idx) screen = _screen_from(screen or voice or seed, idx) rhythm = _rhythm_from(rhythm, idx) block = 'SCÈNE ' + str(idx) + chr(10) + 'Visuel : ' + visual + chr(10) + 'Voix off : ' + voice + chr(10) + 'Texte Ă©cran : ' + screen + chr(10) + 'Rythme : ' + rhythm rich.append({'id': idx, 'text': block, 'title': 'SCÈNE ' + str(idx), 'image': item.get('image', '') if isinstance(item, dict) else ''}) if rich: return rich if not formatted_scenes: return formatted_scenes if mode == "continue": return formatted_scenes[:4] if is_tiktok: if mode == "draft": return formatted_scenes[:3] if mode == "preview": return formatted_scenes[:5] return formatted_scenes[:5] if mode == "draft": return formatted_scenes[:3] if mode == "preview": return formatted_scenes[:5] return formatted_scenes[:6] def split_ideas_from_text(raw_text: str): # ONYX_TIKTOK_PRESERVE_RICH_BLOCKS_V5 # PrĂ©serve les sorties TikTok enrichies au lieu de les convertir en cartes gĂ©nĂ©riques. if isinstance(raw_text, str) and "Visuel" in raw_text and "Voix off" in raw_text and "Rythme" in raw_text: import re as _onyx_re _chunks = [c.strip() for c in _onyx_re.split(r'(?im)(?=^\s*SC[ÈE]NE\s*\d+)', raw_text) if c.strip()] if not _chunks and raw_text.strip(): _chunks = [raw_text.strip()] _preserved = [] for _idx, _chunk in enumerate(_chunks[:5], 1): if not _onyx_re.match(r'(?im)^\s*SC[ÈE]NE\s*\d+', _chunk): _chunk = 'SCÈNE ' + str(_idx) + chr(10) + _chunk _preserved.append({'text': _chunk.strip(), 'title': 'SCÈNE ' + str(_idx)}) if _preserved: return _preserved if not raw_text: return [] text = str(raw_text).strip() dict_matches = re.findall( r"\{['\"]id['\"]\s*:\s*\d+\s*,\s*['\"]text['\"]\s*:\s*([\"'])(.*?)\1(?:\s*,\s*['\"]image['\"]\s*:\s*([\"']).*?\3)?\s*\}", text, flags=re.DOTALL ) if dict_matches: ideas = [] for match in dict_matches[:3]: idea_text = clean_raw_scene_text(match[1]) if idea_text: ideas.append({ "id": len(ideas) + 1, "text": idea_text, "image": "" }) return ideas parsed = parse_possible_dict_string(text) if isinstance(parsed, list): ideas = [] for i, item in enumerate(parsed[:3]): normalized = normalize_scene_item(item, i + 1) if normalized: ideas.append(normalized) return ideas if isinstance(parsed, dict): one_item = normalize_scene_item(parsed, 1) return [one_item] if one_item else [] text = clean_raw_scene_text(text) parts = re.split( r"(?:^|\n)\s*(?:IdĂ©e|Idee)\s*([123])\s*:\s*", text, flags=re.IGNORECASE ) ideas = [] if len(parts) > 1: i = 1 while i < len(parts) - 1: idea_text = clean_raw_scene_text(parts[i + 1]) if idea_text: ideas.append({ "id": len(ideas) + 1, "text": idea_text, "image": "" }) i += 2 return ideas[:3] def attach_images_if_possible( formatted_scenes, intent: str, mode: str, continuity_profile=None, shot_plan=None ): if not formatted_scenes: return formatted_scenes if intent not in ["anime", "tiktok", "story", "emotional"]: for item in formatted_scenes: item["image"] = "" return formatted_scenes if image_generation_engine is None: logger.info("đŸ–Œïž Moteur image indisponible -> rĂ©ponse texte seule") for item in formatted_scenes: item["image"] = "" return formatted_scenes scene_texts = [item.get("text", "").strip() for item in formatted_scenes] if mode == "draft": for item in formatted_scenes: item["image"] = "" return formatted_scenes if mode == "preview": preview_prompt = scene_texts[0] if scene_texts else "" image_url = image_generation_engine.generate_safe(preview_prompt) for i, item in enumerate(formatted_scenes): item["image"] = image_url if i == 0 and image_url else "" return formatted_scenes images = image_generation_engine.generate_batch_v3( scenes=scene_texts, visual_bible=continuity_profile or {}, shot_plan=shot_plan or [], creative_mode=intent ) for i, item in enumerate(formatted_scenes): item["image"] = images[i] if i < len(images) else "" return formatted_scenes @app.get("/") def root(): return {"status": "ONYX API RUNNING 🚀 ONYX BRAIN ROUTER CONTINUE FIX"} @app.get("/health") def health(): return { "status": "ok", "service": "onyx-ai-backend", "version": "v3-reference-based-pipeline", "generate_content": "available", "timestamp": datetime.now(timezone.utc).isoformat() } @app.post("/generate_content") def generate_content(req: RequestModel): global LAST_DRAFT_IDEAS global LAST_SELECTED_IDEA global LAST_CREATIVE_CONTEXT acquired = REQUEST_LOCK.acquire(blocking=False) if not acquired: logger.warning("⛔ Une gĂ©nĂ©ration est dĂ©jĂ  en cours") return make_text_response( "Une gĂ©nĂ©ration est dĂ©jĂ  en cours. Attends la fin complĂšte avant d’envoyer une nouvelle demande.", mode="draft", style=req.style ) try: raw_prompt = normalize(req.prompt) selected_style = detect_style(raw_prompt, req.style) print("🎬 Prompt:", raw_prompt) print("🎹 Style:", selected_style) if not raw_prompt: return {"status": "error", "message": "Prompt vide"} prompt = sanitize_prompt_for_generation(raw_prompt) mode = detect_generation_mode(raw_prompt) if is_continuation_command(prompt): mode = "continue" brain_decision = onyx_brain.detect_mode(prompt) brain_type = brain_decision.get("type") brain_mode = brain_decision.get("mode") brain_creative_mode = brain_decision.get("creative_mode") if brain_mode and mode == "draft" and not is_continuation_command(prompt): mode = brain_mode set_style(selected_style) set_mode(mode) logger.info(f"đŸŽ›ïž Mode detected: {mode}") logger.info(f"🎹 Style detected: {selected_style}") logger.info(f"đŸ§Œ Prompt clean: {prompt}") logger.info(f"🧠 OnyxBrain decision: {brain_decision}") is_studio = "[onyx_studio_" in raw_prompt.lower() is_tiktok = is_tiktok_like_request(prompt) or brain_type == "tiktok" is_followup = should_use_memory(prompt) is_new_request = is_new_creative_request(prompt) or ( brain_type == "tiktok" and not is_followup ) if is_new_request and not is_followup: logger.info("đŸ§Œ Nouvelle demande crĂ©ative dĂ©tectĂ©e -> reset contexte sĂ©lectionnĂ©") reset_current_selection() selected_index = detect_selected_idea(prompt) if is_direct_selection_request(prompt): if not LAST_DRAFT_IDEAS: response = make_text_response( "Aucune idĂ©e rĂ©cente Ă  sĂ©lectionner. Lance d’abord un Brouillon.", mode="draft", style=selected_style ) return remember_response(prompt, response) if selected_index is None or selected_index < 1 or selected_index > len(LAST_DRAFT_IDEAS): response = make_text_response( "Choix invalide. Utilise idĂ©e 1, idĂ©e 2 ou idĂ©e 3.", mode="draft", style=selected_style ) return remember_response(prompt, response) selected_text = LAST_DRAFT_IDEAS[selected_index - 1] LAST_SELECTED_IDEA["index"] = selected_index LAST_SELECTED_IDEA["text"] = selected_text LAST_CREATIVE_CONTEXT = selected_text set_selected_idea(selected_text) response = { "status": "ok", "mode": "draft", "style": selected_style, "scenes": [ { "id": 1, "text": f"IdĂ©e {selected_index} sĂ©lectionnĂ©e : {selected_text}", "image": "" } ] } return remember_response(prompt, response) if not is_studio and not is_tiktok and is_conversation_request(prompt): logger.info("💬 Mode conversation Onyx prioritaire") response = chat_response( prompt=build_memory_context_prompt(prompt), mode=mode, style=selected_style ) return remember_response(prompt, response) if is_develop_request(prompt): prompt = build_effective_develop_prompt(prompt, mode, selected_style) prompt_with_memory = build_memory_context_prompt(prompt) if selected_style == "tiktok" and ( is_studio or is_tiktok or is_develop_request(prompt) or (LAST_SELECTED_IDEA.get("text") and is_followup) ): logger.info("🎬 TikTok Engine activĂ©") if mode == "continue": engine_mode = "continue" elif mode == "preview": engine_mode = "preview" elif mode == "final": engine_mode = "final" else: engine_mode = "draft" scenes = tiktok_engine.generate(prompt_with_memory, engine_mode) formatted = [] for i, scene in enumerate(scenes): item = normalize_scene_item(scene, i + 1) if item: formatted.append(item) formatted = dedupe_scenes(formatted) formatted = enforce_mode_behavior(formatted, engine_mode, is_tiktok=True) formatted = dedupe_scenes(formatted) if engine_mode == "draft" and not is_develop_request(prompt): LAST_DRAFT_IDEAS = [item["text"] for item in (formatted[:5] if locals().get('is_tiktok') else formatted[:3])] LAST_SELECTED_IDEA["index"] = None LAST_SELECTED_IDEA["text"] = None LAST_CREATIVE_CONTEXT = "\n\n".join(LAST_DRAFT_IDEAS) set_ideas(LAST_DRAFT_IDEAS) set_selected_idea(None) logger.info(f"đŸ’Ÿ Draft ideas mĂ©morisĂ©es: {LAST_DRAFT_IDEAS}") else: LAST_CREATIVE_CONTEXT = "\n\n".join( [item.get("text", "") for item in formatted if item.get("text")] ).strip() if not formatted: response = make_text_response( "Je n’ai pas pu gĂ©nĂ©rer un contenu correct.", mode=mode, style=selected_style ) return remember_response(prompt, response) response = { "status": "ok", "mode": mode, "style": selected_style, "scenes": formatted } return remember_response(prompt, response) intent = prompt_intelligence_engine.detect_intent(prompt) if brain_type == "scene": intent = brain_creative_mode or intent or "story" logger.info(f"🧠 Intent detected: {intent}") if not is_studio and not is_tiktok and brain_type != "scene": logger.info("💬 SĂ©curitĂ© conversation IA") response = chat_response( prompt=prompt_with_memory, mode=mode, style=selected_style ) return remember_response(prompt, response) if selected_style == "cinema" and ( is_studio or is_develop_request(prompt) or is_tiktok or brain_type == "scene" or (LAST_SELECTED_IDEA.get("text") and is_followup) ): intent = "story" logger.info("đŸŽ„ Mode CinĂ©ma forcĂ© -> intent story") if intent == "latest_info": response = make_text_response( content_engine.latest_info_guard(prompt_with_memory), mode=mode, style=selected_style ) return remember_response(prompt, response) if intent == "explanation": response = make_text_response( explanation_engine.explain(prompt_with_memory), mode=mode, style=selected_style ) return remember_response(prompt, response) if intent == "business": response = make_text_response( content_engine.business(prompt_with_memory), mode=mode, style=selected_style ) return remember_response(prompt, response) if intent == "general": response = chat_response( prompt=prompt_with_memory, mode=mode, style=selected_style ) return remember_response(prompt, response) if mode == "draft": default_scene_count = 3 elif mode == "preview": default_scene_count = 4 elif mode == "continue": default_scene_count = 4 else: default_scene_count = 5 if intent in ['tiktok', 'anime', 'story', 'emotional'] else 3 scene_count = detect_requested_scene_count( prompt=prompt, default_count=default_scene_count ) if mode == "preview": scene_count = min(scene_count, 5) if mode == "draft": scene_count = 3 if mode == "continue": scene_count = 4 if is_tiktok: scene_count = 5 selected_style = "tiktok" creative_prompt = f""" Tu es Onyx, un directeur crĂ©atif spĂ©cialisĂ© dans les scripts TikTok courts, viraux et faciles Ă  tourner. Mission : CrĂ©er exactement {scene_count} scĂšnes pour une vidĂ©o TikTok de 25 Ă  35 secondes Ă  partir de la demande utilisateur. Structure obligatoire : - ScĂšne 1 : HOOK immĂ©diat en moins de 2 secondes, avec une phrase qui donne envie de regarder la suite. - ScĂšnes 2 Ă  {scene_count - 1} : progression claire, visuelle et rythmĂ©e, avec une nouvelle information ou action Ă  chaque scĂšne. - DerniĂšre scĂšne : payoff fort, twist, rĂ©vĂ©lation, bĂ©nĂ©fice clair ou appel Ă  l’action naturel. Format obligatoire pour chaque scĂšne : SCÈNE X Visuel : dĂ©cris ce que l’on voit Ă  l’écran de façon concrĂšte. Voix off : Ă©cris une phrase courte et parlĂ©e, naturelle pour TikTok. Texte Ă©cran : Ă©cris un texte trĂšs court Ă  afficher Ă  l’écran. Rythme : indique le tempo de la scĂšne en quelques mots. RĂšgles de qualitĂ© : - RĂ©ponds uniquement en français. - Ne donne pas d’introduction avant les scĂšnes et pas de conclusion aprĂšs les scĂšnes. - Écris comme un crĂ©ateur TikTok professionnel, pas comme un article. - Chaque scĂšne doit ĂȘtre concrĂšte, filmable et comprĂ©hensible immĂ©diatement. - Garde le mĂȘme sujet principal du dĂ©but Ă  la fin. - Utilise des phrases courtes, dynamiques et naturelles. - Évite les rĂ©pĂ©titions, les gĂ©nĂ©ralitĂ©s et le blabla. - Si le sujet est une idĂ©e, transforme-la en histoire visuelle. - Si le sujet est un produit, montre le problĂšme, la transformation puis le bĂ©nĂ©fice. - Si le sujet est Ă©motionnel, commence par une tension humaine forte. Demande utilisateur : {prompt} """ if selected_style == "cinema": if mode == "draft": creative_prompt = f""" Tu es Onyx, un rĂ©alisateur crĂ©atif spĂ©cialisĂ© en concepts cinĂ©matographiques. Mode : Brouillon Mission : CrĂ©er exactement 3 idĂ©es cinĂ©matographiques diffĂ©rentes Ă  partir de la demande utilisateur. Format obligatoire : IdĂ©e 1 : ... IdĂ©e 2 : ... IdĂ©e 3 : ... RĂšgles absolues : - Les 3 idĂ©es doivent ĂȘtre indĂ©pendantes. - IdĂ©e 1, IdĂ©e 2 et IdĂ©e 3 ne doivent PAS former une suite. - Chaque idĂ©e doit raconter un concept diffĂ©rent. - Ne fais pas une histoire dĂ©coupĂ©e en 3 blocs. - Ne continue jamais l'idĂ©e prĂ©cĂ©dente. - Chaque idĂ©e doit pouvoir devenir son propre court-mĂ©trage. - Chaque idĂ©e contient 2 Ă  4 phrases maximum. - Style cinĂ©ma, visuel, immersif, mystĂ©rieux ou Ă©motionnel. - Pas de blabla. - Pas d'explication de mĂ©thode. - Pas de rĂ©pĂ©tition. Demande : {prompt_with_memory} """ elif mode == "continue": creative_prompt = f""" Tu es Onyx, un rĂ©alisateur expert en suites cinĂ©matographiques. Mode : Suite Mission : Continuer directement l’histoire prĂ©cĂ©dente en exactement {scene_count} scĂšnes nouvelles. RĂšgles : - Ne recommence jamais depuis le dĂ©but. - Ne reformule jamais les scĂšnes prĂ©cĂ©dentes. - Continue le mĂȘme sujet, le mĂȘme ton et le mĂȘme univers. - Ajoute de nouveaux Ă©vĂ©nements cohĂ©rents. - Augmente la tension, le mystĂšre ou la rĂ©vĂ©lation. - Chaque scĂšne doit faire avancer l’histoire. - Finir sur une ouverture forte ou une rĂ©vĂ©lation. - Pas de rĂ©pĂ©tition. - Pas de texte gĂ©nĂ©rique. Contexte et demande : {prompt_with_memory} """ else: creative_prompt = f""" Tu es Onyx, un rĂ©alisateur expert en narration cinĂ©matographique. Mode : {mode} Mission : CrĂ©er exactement {scene_count} scĂšnes cinĂ©matographiques immersives, puissantes et cohĂ©rentes. RĂšgles : - Style film / court-mĂ©trage. - Narration visuelle, Ă©motionnelle et immersive. - Chaque scĂšne doit faire progresser l’histoire. - Garde le mĂȘme sujet principal du dĂ©but Ă  la fin. - Pas de rĂ©pĂ©tition. - Pas de contenu vide. IMPORTANT : - Le mode Final doit ĂȘtre PLUS INTENSE que l’aperçu. - Ajouter une montĂ©e dramatique claire. - Ajouter un climax fort. - Ajouter une fin marquante ou inattendue. - Ne jamais simplement reformuler l’aperçu. Demande : {prompt_with_memory} """ else: creative_prompt = f""" Tu es Onyx, un assistant crĂ©atif spĂ©cialisĂ© en storytelling immersif. Mode : {mode} Mission : CrĂ©er exactement {scene_count} scĂšnes cohĂ©rentes, fluides et captivantes. RĂšgles : - RĂ©ponds uniquement en français. - Chaque scĂšne contient 2 Ă  4 phrases maximum. - Chaque scĂšne doit faire avancer l’histoire. - Le ton doit ĂȘtre visuel et immersif. - Garde une logique claire du dĂ©but Ă  la fin. - Garde le mĂȘme sujet principal du dĂ©but Ă  la fin. - Pas de rĂ©pĂ©tition. - Pas de contenu vide. Demande : {prompt_with_memory} """ scenes = scene_engine.generate( user_prompt=creative_prompt, scene_count=scene_count, creative_mode=intent ) if selected_style == "cinema" and mode == "draft": raw_text = "\n".join([str(scene) for scene in scenes]) formatted = split_ideas_from_text(raw_text) if not formatted: formatted = [] for i, scene in enumerate(scenes): item = normalize_scene_item(scene, i + 1) if item: formatted.append(item) formatted = dedupe_scenes(formatted) formatted = formatted[:5] if locals().get('is_tiktok') else formatted[:3] if not formatted: response = make_text_response( "Je n’ai pas pu gĂ©nĂ©rer une rĂ©ponse correcte.", mode=mode, style=selected_style ) return remember_response(prompt, response) LAST_DRAFT_IDEAS = [item["text"] for item in (formatted[:5] if locals().get('is_tiktok') else formatted[:3])] LAST_SELECTED_IDEA["index"] = None LAST_SELECTED_IDEA["text"] = None LAST_CREATIVE_CONTEXT = "\n\n".join(LAST_DRAFT_IDEAS) set_ideas(LAST_DRAFT_IDEAS) set_selected_idea(None) logger.info(f"đŸ’Ÿ Draft ideas cinĂ©ma mĂ©morisĂ©es: {LAST_DRAFT_IDEAS}") response = { "status": "ok", "mode": mode, "style": selected_style, "scenes": (formatted[:5] if locals().get("is_tiktok") else formatted[:3]) } return remember_response(prompt, response) formatted = [] for i, scene in enumerate(scenes): item = normalize_scene_item(scene, i + 1) if item: formatted.append(item) if not formatted: response = make_text_response( "Je n’ai pas pu gĂ©nĂ©rer une rĂ©ponse correcte.", mode=mode, style=selected_style ) return remember_response(prompt, response) formatted = dedupe_scenes(formatted) formatted = enforce_mode_behavior(formatted, mode) formatted = dedupe_scenes(formatted) if mode == "draft": LAST_DRAFT_IDEAS = [item["text"] for item in (formatted[:5] if locals().get('is_tiktok') else formatted[:3])] LAST_SELECTED_IDEA["index"] = None LAST_SELECTED_IDEA["text"] = None LAST_CREATIVE_CONTEXT = "\n\n".join(LAST_DRAFT_IDEAS) set_ideas(LAST_DRAFT_IDEAS) set_selected_idea(None) logger.info(f"đŸ’Ÿ Draft ideas mĂ©morisĂ©es: {LAST_DRAFT_IDEAS}") else: LAST_CREATIVE_CONTEXT = "\n\n".join( [item.get("text", "") for item in formatted if item.get("text")] ).strip() continuity_profile = scene_engine.build_visual_bible( user_prompt=prompt, scenes=formatted, creative_mode=intent ) shot_plan = scene_engine.build_shot_plan( user_prompt=prompt, scenes=formatted, continuity_profile=continuity_profile, creative_mode=intent ) formatted = attach_images_if_possible( formatted_scenes=formatted, intent=intent, mode=mode, continuity_profile=continuity_profile, shot_plan=shot_plan ) response = { "status": "ok", "mode": mode, "style": selected_style, "scenes": formatted } return remember_response(prompt, response) except Exception as e: logger.error(f"GLOBAL ERROR: {e}") print("❌ GLOBAL ERROR:", e) return { "status": "error", "message": str(e) } finally: if acquired: REQUEST_LOCK.release()