""" Competitive Topic Research & Viral Explainer Intelligence Engine. Researches the highest-viewed YouTube videos and viral narrative structures for given topics, and autonomously discovers high-velocity popular topics. """ import re import json import urllib.request import urllib.parse import logging from typing import Dict, Any, List, Optional from .remote import RemotePC logger = logging.getLogger("YTFactory.Research") # High-retention explainer niches with proven multi-million view formulas POPULAR_EXPLAINER_NICHES = [ "Lost Civilizations & Ancient Archaeology", "Astrophysics & Space Paradoxes", "Origins of Human Civilization & Anthropology", "History of Computing & Revolutionary Technology", "Deep Ocean Mysteries & Earth Sciences", "Unsolved Historical Enigmas & Cold Cases", "Economic Scandals & Corporate History", "Neuroscience & Psychology Secrets" ] class ViralTopicResearcher: def __init__(self, llm_pc: Optional[RemotePC] = None, llm_port: int = 8002): self.llm_pc = llm_pc self.llm_port = llm_port self.user_agent = "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/120.0.0.0 Safari/537.36" def _search_web_insights(self, topic: str) -> str: """Fetch search snippets and YouTube search results for the topic.""" query = f"{topic} documentary explained" clean_q = urllib.parse.quote(query) url = f"https://html.duckduckgo.com/html/?q={clean_q}" req = urllib.request.Request(url, headers={"User-Agent": self.user_agent}) snippets = [] try: with urllib.request.urlopen(req, timeout=8) as resp: html = resp.read().decode("utf-8", errors="ignore") # Extract search result snippets matches = re.findall(r' Dict[str, Any]: """ Research multi-million view YouTube explainer videos on this topic to extract: - The most viral psychological hook & curiosity gap - Key narrative revelations and counter-intuitive insights - Step-by-step explainer pacing (Kurzgesagt / Veritasium / ColdFusion style) - Core visual motifs for artistic consistency """ logger.info(f"[RESEARCH] Researching top-performing viral explainer structures for: '{topic}'...") web_context = self._search_web_insights(topic) prompt = f"""You are a master YouTube retention strategist and documentary showrunner (studying 10M+ view explainer channels like Veritasium, Kurzgesagt, ColdFusion, Wendover, and Johnny Harris). TOPIC: "{topic}" WEB SEARCH CONTEXT: {web_context[:1500]} OBJECTIVE: Analyze why the most viral videos on this topic succeed and formulate a comprehensive high-retention blueprint for our original 7-12 minute explainer documentary. Provide: 1. "viral_hook_angle": The irresistible curiosity gap / central question that hooks viewers in the first 10 seconds. 2. "common_misconceptions": 2-3 widely believed myths that we will dramatically debunk. 3. "key_revelations": 3-4 mind-blowing historical/scientific milestones or paradoxes to structure the video around. 4. "narrative_arc": 5-act explainer progression: - Act 1 (The Hook & Mystery) - Act 2 (The Prevailing Myth) - Act 3 (The Breakthrough Discovery) - Act 4 (The Mechanism / Step-by-Step Reality) - Act 5 (The Paradigm Shift & Modern Epiphany) 5. "visual_style_anchor": A cohesive 10-15 word visual art direction description to ensure ALL scene clips look visually unified. STRICT JSON OUTPUT: {{ "viral_hook_angle": "...", "common_misconceptions": ["...", "..."], "key_revelations": ["...", "...", "..."], "narrative_arc": ["Act 1: ...", "Act 2: ...", "Act 3: ...", "Act 4: ...", "Act 5: ..."], "visual_style_anchor": "cinematic documentary style, warm archival tones, 35mm film texture, photorealistic 8k" }}""" if self.llm_pc and self.llm_pc.check(self.llm_port): try: payload = { "model": "llama", "messages": [{"role": "user", "content": prompt}], "temperature": 0.7, "max_tokens": 2048 } r = self.llm_pc.post_json(self.llm_port, "/v1/chat/completions", payload, timeout=60) if r and "choices" in r: content = r["choices"][0]["message"]["content"] # Extract JSON s = content.find('{') e = content.rfind('}') if s != -1 and e > s: dossier = json.loads(content[s:e+1]) logger.info(f"[RESEARCH] Extracted viral hook: '{dossier.get('viral_hook_angle', '')[:60]}...'") return dossier except Exception as e: logger.warning(f"[RESEARCH] LLM topic analysis note: {e}") # Fallback structured research dossier return { "viral_hook_angle": f"Everything we thought we understood about {topic} was challenged by a single discovery.", "common_misconceptions": [f"Standard historical accounts of {topic} overlook the key evidence."], "key_revelations": [f"The untold origin and catalyst behind {topic}."], "narrative_arc": ["The Opening Mystery", "The Historical Consensus", "The Revolutionary Discovery", "The Hidden Mechanism", "The Modern Impact"], "visual_style_anchor": "cinematic documentary style, warm archival color grade, 35mm film texture, photorealistic 8k" } def discover_popular_trending_topics(self, niche: str = "", count: int = 3) -> List[Dict[str, Any]]: """ Autonomously research and generate high-velocity popular YouTube explainer topics. """ logger.info(f"[AUTO-RESEARCH] Discovering {count} high-velocity YouTube explainer topics...") target_niche = niche if niche else POPULAR_EXPLAINER_NICHES[0] prompt = f"""You are the Head of Programming for top-tier YouTube explainer channels (like Kurzgesagt, ColdFusion, Lemmino, Wendover). Generate {count} distinct, multi-million-view potential documentary explainer video topics for the niche: "{target_niche}". Requirements for EACH topic: - Must have massive viral appeal, strong curiosity gaps, and deep explainer substance. - Target duration: 7 to 12 minutes. - Titles must be high CTR, intriguing, and professional. STRICT JSON OUTPUT: {{ "topics": [ {{ "topic": "The Full Intriguing Video Title", "curiosity_angle": "Why viewers can't resist clicking", "target_niche": "{target_niche}" }} ] }}""" if self.llm_pc and self.llm_pc.check(self.llm_port): try: payload = { "model": "llama", "messages": [{"role": "user", "content": prompt}], "temperature": 0.8, "max_tokens": 2048 } r = self.llm_pc.post_json(self.llm_port, "/v1/chat/completions", payload, timeout=60) if r and "choices" in r: content = r["choices"][0]["message"]["content"] s = content.find('{') e = content.rfind('}') if s != -1 and e > s: data = json.loads(content[s:e+1]) topics = data.get("topics", []) if topics: logger.info(f"[AUTO-RESEARCH] Discovered {len(topics)} viral topics ready for production.") return topics except Exception as e: logger.warning(f"[AUTO-RESEARCH] LLM discovery note: {e}") # Fallback curated high-velocity explainer topics fallback_catalog = [ {"topic": "The Mystery of Göbekli Tepe and the Civilization Before History", "curiosity_angle": "How hunter gatherers built stone monoliths 7,000 years before Stonehenge."}, {"topic": "How the First Microprocessor Silently Changed the Human Race", "curiosity_angle": "The hidden engineering race in 1971 that birthed modern personal computing."}, {"topic": "The Black Hole Information Paradox: How the Universe Destroys Reality", "curiosity_angle": "The terrifying theoretical paradox that baffled Stephen Hawking for decades."}, {"topic": "The True Story of the Antikythera Mechanism: The 2,000-Year-Old Analog Computer", "curiosity_angle": "Ancient Greek gears that shouldn't have existed for another millennium."}, {"topic": "How Beer Built Human Civilization: The Agriculture Hypothesis", "curiosity_angle": "Why humanity abandoned nomadic life to brew the world's first fermented grains."} ] return fallback_catalog[:count]