343 lines
17 KiB
Python
343 lines
17 KiB
Python
"""
|
|
Script Generation Module using AMDLLM (llama.cpp Qwen3.5-7B).
|
|
Generates bilingual documentary scripts, A/B test titles, SEO metadata,
|
|
dynamic chapter markers, visual generation prompts, and stock search queries.
|
|
"""
|
|
|
|
import json
|
|
import logging
|
|
import re
|
|
from typing import Dict, Any, List, Optional
|
|
from .remote import RemotePC
|
|
|
|
logger = logging.getLogger("YTFactory.Script")
|
|
|
|
class ScriptGen:
|
|
def __init__(self, pcs: Dict[str, RemotePC], llm_port: int = 8002):
|
|
self.pcs = pcs
|
|
self.llm_port = llm_port
|
|
|
|
def _build_master_prompt(self, topic: str, duration: int) -> str:
|
|
# Pacing: ~140 words per minute for professional documentary delivery.
|
|
num_segments = max(4, duration // 20)
|
|
target_words = max(70, int(duration * 140 / 60))
|
|
words_per_seg = max(25, target_words // num_segments)
|
|
|
|
return f"""You are an award-winning documentary writer and YouTube strategist.
|
|
Generate a complete, high-retention faceless documentary script for the topic: "{topic}".
|
|
|
|
SPECIFICATIONS:
|
|
- Target Duration: Exactly ~{duration} seconds of continuous narration
|
|
- Total Word Count: ~{target_words} spoken words across all segments (~{words_per_seg} words per segment)
|
|
- Number of Segments: Exactly {num_segments} chronological segments
|
|
- Tone: Mysterious, engaging, cinematic, informative, highly gripping
|
|
- Spoken Depth: Each segment MUST contain a full, rich spoken paragraph ({words_per_seg - 5} to {words_per_seg + 10} words) so the narration fills the entire video duration without silence or rushing.
|
|
|
|
For EACH segment provide:
|
|
1. "chapter_title": Short 2-4 word chapter title (e.g., "The Ancient Mystery", "The Critical Discovery")
|
|
2. "text": English spoken narration (rich 2-4 sentence narrative paragraph, natural pacing, compelling storytelling)
|
|
3. "visual_prompt": Highly detailed visual scene description for an AI video/image generator (cinematic lighting, camera angle, 8k documentary style)
|
|
4. "stock_query": 2-4 word search term to find relevant stock b-roll footage online (e.g., "ancient library books", "desert ruins aerial")
|
|
5. "on_screen_text": Optional 2-5 word lower-third label, key statistic, date, or scene title to display on screen (e.g., "Depth: 11,034m", "Year: 1960", "Pacific Ocean Abyss", or "" if none)
|
|
6. "is_historical": boolean (true if this segment depicts past events, ancient civilizations, historical figures, archaeological discoveries, or vintage eras; false if modern/contemporary)
|
|
7. "duration_estimate": Estimated duration in seconds ({max(10, duration // num_segments)}s)
|
|
|
|
METADATA REQUIREMENTS:
|
|
- "titles": Exactly 3 distinct viral, high-CTR YouTube titles for A/B testing (max 85 chars each)
|
|
- "hook": Compelling 5-second opening question or statement
|
|
- "description": Comprehensive 3-paragraph YouTube description with engaging summary, key takeaways, and relevant hashtags (#Documentary #History #Science)
|
|
- "tags": 12-18 highly relevant YouTube search tags
|
|
|
|
STRICT OUTPUT FORMAT:
|
|
Output MUST be a single, valid JSON object starting with {{ and ending with }}. Do NOT include markdown code blocks (```), comments, or text before/after.
|
|
|
|
JSON Schema:
|
|
{{
|
|
"titles": ["Title A", "Title B", "Title C"],
|
|
"hook": "Opening hook line...",
|
|
"description": "SEO description...",
|
|
"tags": ["tag1", "tag2", "tag3"],
|
|
"script_segments": [
|
|
{{
|
|
"chapter_title": "Chapter Name",
|
|
"text": "English spoken narration for segment 1...",
|
|
"visual_prompt": "Cinematic visual description...",
|
|
"stock_query": "relevant broll keyword",
|
|
"on_screen_text": "Key Metric or Title",
|
|
"is_historical": true,
|
|
"duration_estimate": 18
|
|
}}
|
|
]
|
|
}}"""
|
|
|
|
def _extract_json(self, text: str) -> Dict[str, Any]:
|
|
"""Sanitize and parse JSON from LLM response with multiple fallback passes."""
|
|
text = text.strip()
|
|
text = re.sub(r'^\s*```(?:json)?\s*', '', text, flags=re.IGNORECASE)
|
|
text = re.sub(r'\s*```\s*$', '', text)
|
|
text = text.strip()
|
|
|
|
start = text.find('{')
|
|
end = text.rfind('}')
|
|
if start == -1 or end <= start:
|
|
raise ValueError("No JSON object found in response")
|
|
|
|
json_str = text[start:end + 1]
|
|
# Clean trailing commas
|
|
json_str = re.sub(r',\s*}', '}', json_str)
|
|
json_str = re.sub(r',\s*]', ']', json_str)
|
|
|
|
try:
|
|
return json.loads(json_str)
|
|
except json.JSONDecodeError:
|
|
pass
|
|
|
|
# Try json5 or partial bracket matching if available
|
|
try:
|
|
import json5
|
|
return json5.loads(json_str)
|
|
except Exception:
|
|
pass
|
|
|
|
for end_idx in range(end, start, -1):
|
|
if text[end_idx] == '}':
|
|
candidate = text[start:end_idx + 1]
|
|
candidate = re.sub(r',\s*}', '}', candidate)
|
|
candidate = re.sub(r',\s*]', ']', candidate)
|
|
try:
|
|
return json.loads(candidate)
|
|
except json.JSONDecodeError:
|
|
continue
|
|
|
|
raise ValueError("Unable to parse JSON from LLM response after all sanitization passes.")
|
|
|
|
def _chat(self, pc: RemotePC, prompt: str, max_tokens: int = 8192) -> Optional[str]:
|
|
payload = {
|
|
"model": "llama",
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": 0.7,
|
|
"max_tokens": max_tokens
|
|
}
|
|
r = pc.post_json(self.llm_port, "/v1/chat/completions", payload, timeout=600)
|
|
if not r or "choices" not in r:
|
|
return None
|
|
return r["choices"][0]["message"]["content"]
|
|
|
|
def _translate_to_spanish(self, pc: RemotePC, script_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Translate narration, titles, chapters, description, and on-screen text into Latin American Spanish."""
|
|
logger.info("[SCRIPT] Translating script to Latin-American Spanish...")
|
|
|
|
segments = script_data.get("script_segments", [])
|
|
titles = script_data.get("titles", [])
|
|
desc = script_data.get("description", "")
|
|
|
|
translation_payload = {
|
|
"titles": titles,
|
|
"narrations": [s.get("text", "") for s in segments],
|
|
"chapters": [s.get("chapter_title", f"Part {i+1}") for i, s in enumerate(segments)],
|
|
"on_screen_texts": [s.get("on_screen_text", "") for s in segments],
|
|
"description": desc
|
|
}
|
|
|
|
prompt = f"""You are a professional documentary translator.
|
|
Translate the following English YouTube documentary package into fluent, natural Latin-American Spanish for voiceover and YouTube upload.
|
|
|
|
English Data:
|
|
{json.dumps(translation_payload, ensure_ascii=False, indent=2)}
|
|
|
|
Requirements:
|
|
- "titles_es": 3 catchy Spanish titles matching the English titles
|
|
- "narrations_es": array of exact same length containing the Spanish voiceover text
|
|
- "chapters_es": array of exact same length containing short 2-4 word Spanish chapter titles
|
|
- "on_screen_texts_es": array of exact same length containing Spanish translations of the on-screen text labels (keep empty if original is empty)
|
|
- "description_es": full Spanish translation of the YouTube description with Spanish hashtags
|
|
- Output STRICT JSON only with keys: {{"titles_es": [...], "narrations_es": [...], "chapters_es": [...], "on_screen_texts_es": [...], "description_es": "..."}}
|
|
- No markdown formatting or extra text."""
|
|
|
|
for attempt in (1, 2):
|
|
text = self._chat(pc, prompt, max_tokens=4096)
|
|
if not text:
|
|
continue
|
|
try:
|
|
es_data = self._extract_json(text)
|
|
narr_es = es_data.get("narrations_es", [])
|
|
titles_es = es_data.get("titles_es", [])
|
|
chapters_es = es_data.get("chapters_es", [])
|
|
ost_es = es_data.get("on_screen_texts_es", [])
|
|
desc_es = es_data.get("description_es", "")
|
|
|
|
if len(narr_es) == len(segments):
|
|
for i, seg in enumerate(segments):
|
|
seg["text_es"] = narr_es[i]
|
|
seg["chapter_title_es"] = chapters_es[i] if i < len(chapters_es) else seg.get("chapter_title", "")
|
|
seg["on_screen_text_es"] = ost_es[i] if i < len(ost_es) else seg.get("on_screen_text", "")
|
|
|
|
script_data["titles_es"] = titles_es if len(titles_es) >= len(titles) else titles
|
|
script_data["description_es"] = desc_es or desc
|
|
logger.info("[SCRIPT] Spanish translation completed successfully.")
|
|
return script_data
|
|
else:
|
|
logger.warning(f"[SCRIPT] Segment length mismatch on ES translation ({len(narr_es)} vs {len(segments)})")
|
|
except Exception as e:
|
|
logger.warning(f"[SCRIPT] ES translation parse attempt {attempt} failed: {e}")
|
|
|
|
logger.warning("[SCRIPT] Spanish translation failed after retries. Using English fallbacks.")
|
|
for seg in segments:
|
|
seg.setdefault("text_es", seg.get("text", ""))
|
|
seg.setdefault("chapter_title_es", seg.get("chapter_title", ""))
|
|
seg.setdefault("on_screen_text_es", seg.get("on_screen_text", ""))
|
|
script_data["titles_es"] = script_data.get("titles", [])
|
|
script_data["description_es"] = script_data.get("description", "")
|
|
return script_data
|
|
|
|
def _proofread_and_refine_script(self, pc: RemotePC, script_data: Dict[str, Any], topic: str) -> Dict[str, Any]:
|
|
"""
|
|
Secondary LLM pass to proofread, fact-check, and fix proper noun/phonetic errors
|
|
(e.g., 'white tecapel' -> 'Whitechapel', 'tutankamen' -> 'Tutankhamun').
|
|
"""
|
|
logger.info("[SCRIPT:AUDIT] Running secondary LLM proofreading & fact-checking pass...")
|
|
|
|
prompt = f"""You are a master documentary script editor, fact-checker, and phonetics auditor.
|
|
Your job is to rigorously review and correct this draft documentary script for the topic: "{topic}".
|
|
|
|
AUDITING OBJECTIVES:
|
|
1. Proper Nouns & Historical Places:
|
|
- Identify and correct ANY misspellings, phonetic corruptions, or garbled proper nouns (for example, if you see "white tecapel" or "white te capel", change it to "Whitechapel"; fix any names of people, cities, landmarks, vessels, historical dates, or scientific terms).
|
|
2. Phonetic Cleanliness for Text-to-Speech:
|
|
- Ensure words are written cleanly so speech synthesizers pronounce them naturally without awkward syllable breaks.
|
|
3. Flow & Grammar:
|
|
- Enhance sentence rhythm, natural documentary pacing, and dramatic intrigue.
|
|
4. Structure:
|
|
- Preserve the exact same JSON schema, segment count, keys, visual prompts, and metadata.
|
|
|
|
DRAFT SCRIPT TO AUDIT:
|
|
{json.dumps(script_data, ensure_ascii=False, indent=2)}
|
|
|
|
STRICT OUTPUT FORMAT:
|
|
Return ONLY the corrected script as a single valid JSON object starting with {{ and ending with }}. No markdown, no commentary."""
|
|
|
|
try:
|
|
raw_text = self._chat(pc, prompt, max_tokens=8192)
|
|
if raw_text:
|
|
refined = self._extract_json(raw_text)
|
|
if "script_segments" in refined and len(refined["script_segments"]) == len(script_data.get("script_segments", [])):
|
|
logger.info("[SCRIPT:AUDIT] Secondary LLM proofreading pass passed successfully.")
|
|
return refined
|
|
else:
|
|
logger.warning("[SCRIPT:AUDIT] Segment count mismatch in audit pass. Keeping original.")
|
|
except Exception as e:
|
|
logger.warning(f"[SCRIPT:AUDIT] Secondary LLM audit pass encountered error: {e}. Keeping original.")
|
|
|
|
return script_data
|
|
|
|
def _proofread_spanish_translation(self, pc: RemotePC, script_data: Dict[str, Any], topic: str) -> Dict[str, Any]:
|
|
"""
|
|
Secondary LLM pass to proofread Spanish translation for natural Latin-American idioms,
|
|
proper noun spelling (e.g. 'Whitechapel', 'Alejandría'), accent marks, and pronunciation.
|
|
"""
|
|
logger.info("[SCRIPT:AUDIT] Running secondary LLM Spanish audit pass...")
|
|
|
|
segments = script_data.get("script_segments", [])
|
|
payload = {
|
|
"titles_es": script_data.get("titles_es", []),
|
|
"description_es": script_data.get("description_es", ""),
|
|
"segments_es": [
|
|
{
|
|
"chapter_title_es": s.get("chapter_title_es", ""),
|
|
"text_es": s.get("text_es", ""),
|
|
"on_screen_text_es": s.get("on_screen_text_es", "")
|
|
}
|
|
for s in segments
|
|
]
|
|
}
|
|
|
|
prompt = f"""You are a master Spanish documentary editor and translator.
|
|
Review and polish this Spanish translation for topic "{topic}".
|
|
|
|
OBJECTIVES:
|
|
1. Fix proper nouns (keep standard names like "Whitechapel", "Jack el Destripador", "Londres", "Alejandría" accurate).
|
|
2. Correct any phonetic spelling mistakes or awkward phrasing.
|
|
3. Ensure natural Latin-American documentary narration tone and correct accent marks (tildes).
|
|
4. Preserve the exact number of segments.
|
|
|
|
Spanish Data:
|
|
{json.dumps(payload, ensure_ascii=False, indent=2)}
|
|
|
|
STRICT OUTPUT FORMAT:
|
|
Return STRICT JSON only with keys: {{"titles_es": [...], "description_es": "...", "segments_es": [{{"chapter_title_es": "...", "text_es": "...", "on_screen_text_es": "..."}}, ...]}}"""
|
|
|
|
try:
|
|
raw_text = self._chat(pc, prompt, max_tokens=8192)
|
|
if raw_text:
|
|
refined = self._extract_json(raw_text)
|
|
segs_es = refined.get("segments_es", [])
|
|
if len(segs_es) == len(segments):
|
|
for i, s in enumerate(segments):
|
|
s["text_es"] = segs_es[i].get("text_es", s.get("text_es", ""))
|
|
s["chapter_title_es"] = segs_es[i].get("chapter_title_es", s.get("chapter_title_es", ""))
|
|
s["on_screen_text_es"] = segs_es[i].get("on_screen_text_es", s.get("on_screen_text_es", ""))
|
|
if "titles_es" in refined and refined["titles_es"]:
|
|
script_data["titles_es"] = refined["titles_es"]
|
|
if "description_es" in refined and refined["description_es"]:
|
|
script_data["description_es"] = refined["description_es"]
|
|
logger.info("[SCRIPT:AUDIT] Spanish audit pass completed successfully.")
|
|
except Exception as e:
|
|
logger.warning(f"[SCRIPT:AUDIT] Spanish audit pass error: {e}")
|
|
|
|
return script_data
|
|
|
|
def generate(self, topic: str, duration: int) -> Dict[str, Any]:
|
|
"""Generate master bilingual script with secondary LLM verification & proofreading passes."""
|
|
logger.info(f"[SCRIPT] Generating script for topic: '{topic}' (~{duration}s)...")
|
|
pc = self.pcs.get("amdllm")
|
|
if not pc or not pc.check(self.llm_port):
|
|
raise RuntimeError(f"AMDLLM llama.cpp not reachable on port {self.llm_port}")
|
|
|
|
prompt = self._build_master_prompt(topic, duration)
|
|
|
|
for attempt in range(1, 4):
|
|
logger.info(f"[SCRIPT] Requesting script generation from AMDLLM (attempt {attempt}/3)...")
|
|
raw_text = self._chat(pc, prompt, max_tokens=8192)
|
|
if not raw_text:
|
|
logger.warning(f"[SCRIPT] AMDLLM returned empty response on attempt {attempt}")
|
|
continue
|
|
|
|
try:
|
|
data = self._extract_json(raw_text)
|
|
|
|
# Validation & Normalization
|
|
titles = data.get("titles", [])
|
|
if isinstance(titles, str):
|
|
titles = [titles]
|
|
while len(titles) < 3:
|
|
titles.append(f"{topic} - Part {len(titles)+1}")
|
|
data["titles"] = titles[:3]
|
|
data["title"] = titles[0]
|
|
|
|
segments = data.get("script_segments", [])
|
|
if not segments:
|
|
raise ValueError("No script_segments generated")
|
|
|
|
for i, s in enumerate(segments):
|
|
s.setdefault("chapter_title", f"Part {i+1}")
|
|
s.setdefault("duration_estimate", max(15, duration // len(segments)))
|
|
s.setdefault("stock_query", topic)
|
|
if not s.get("visual_prompt"):
|
|
s["visual_prompt"] = f"Cinematic documentary footage illustrating: {s.get('text', topic)}"
|
|
|
|
# ── Step 1.5: Secondary LLM Proofreading & Fact-Checking Pass (English) ──
|
|
data = self._proofread_and_refine_script(pc, data, topic)
|
|
|
|
# ── Step 1.6: Spanish Translation ──
|
|
data = self._translate_to_spanish(pc, data)
|
|
|
|
# ── Step 1.7: Secondary LLM Spanish Translation Audit ──
|
|
data = self._proofread_spanish_translation(pc, data, topic)
|
|
|
|
logger.info(f"[SCRIPT] Master script successfully created & audited: '{data['title']}' ({len(data.get('script_segments', []))} segments)")
|
|
return data
|
|
|
|
except Exception as e:
|
|
logger.warning(f"[SCRIPT] Parse error on attempt {attempt}: {e}")
|
|
|
|
raise RuntimeError("Failed to generate and parse script after 3 attempts.")
|