| 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): |
| |
| |
| |
| |
| if is_tiktok and isinstance(formatted_scenes, list) and formatted_scenes: |
| |
| |
| |
| 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): |
| |
| |
| 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() |
|
|
