184 lines
8.6 KiB
Python
184 lines
8.6 KiB
Python
"""
|
|
Trending topics discovery engine for 24/7 video generation.
|
|
Fetches real-time search trends from Google Trends RSS and leverages AMDLLM LLM to
|
|
brainstorm viral, high-retention faceless documentary topics across niches.
|
|
"""
|
|
|
|
import urllib.request
|
|
import xml.etree.ElementTree as ET
|
|
import json
|
|
import logging
|
|
import random
|
|
import re
|
|
from typing import List, Dict, Optional, Any
|
|
from .remote import RemotePC
|
|
|
|
logger = logging.getLogger("YTFactory.Trending")
|
|
|
|
NICHES = {
|
|
"history": "Ancient civilizations, forgotten empires, historical conspiracies, lost technologies, archaeological discoveries",
|
|
"space": "Cosmic mysteries, black holes, NASA discoveries, alien worlds, quantum astrophysics, future space missions",
|
|
"science": "Mind-bending paradoxes, deep ocean biology, cutting-edge genetics, unexplained natural phenomena, human longevity",
|
|
"tech": "Artificial Intelligence breakthroughs, robotics revolution, future civilization, cyber warfare, quantum computing",
|
|
"mysteries": "Unsolved cold cases, bizarre disappearances, ancient maps, oceanic anomalies, secret underground vaults",
|
|
"business": "Mega-corporation empires, greatest marketing hacks, financial crashes, economic warfare, billionaire secrets"
|
|
}
|
|
|
|
class TrendingFetcher:
|
|
def __init__(self, amdllm_pc: Optional[RemotePC] = None, llm_port: int = 8002):
|
|
self.pc = amdllm_pc
|
|
self.llm_port = llm_port
|
|
|
|
def get_google_trends(self, geo: str = "US", max_items: int = 15) -> List[str]:
|
|
"""Fetch raw trending search queries from Google Trends RSS."""
|
|
url = f"https://trends.google.com/trending/rss?geo={geo}"
|
|
trends = []
|
|
try:
|
|
req = urllib.request.Request(
|
|
url,
|
|
headers={"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=10) as resp:
|
|
root = ET.fromstring(resp.read())
|
|
for item in root.findall(".//item"):
|
|
title = item.find("title")
|
|
if title is not None and title.text:
|
|
trends.append(title.text.strip())
|
|
if len(trends) >= max_items:
|
|
break
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch Google Trends RSS: {e}")
|
|
return trends
|
|
|
|
def brainstorm_niche_topics(self, niche: str, count: int = 5) -> List[str]:
|
|
"""Use AMDLLM LLM to brainstorm viral documentary topics for a given niche."""
|
|
if not self.pc or not self.pc.check(self.llm_port):
|
|
logger.warning("AMDLLM not available for brainstorming, using curated fallback topics.")
|
|
return self._fallback_topics(niche, count)
|
|
|
|
niche_desc = NICHES.get(niche.lower(), niche)
|
|
prompt = f"""You are a master YouTube content strategist specializing in viral faceless documentaries.
|
|
Generate {count} engaging, viral, high-CTR documentary video topics for the niche: "{niche}" ({niche_desc}).
|
|
|
|
Requirements:
|
|
- Strong curiosity hooks, intriguing mysteries, or surprising revelations
|
|
- Optimized for 5-10 minute faceless video documentaries
|
|
- Output STRICT JSON only: {{"topics": ["Topic 1", "Topic 2", ...]}}
|
|
- Do NOT output any markdown, explanations, or code blocks."""
|
|
|
|
payload = {
|
|
"model": "llama",
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": 0.8,
|
|
"max_tokens": 1024
|
|
}
|
|
r = self.pc.post_json(self.llm_port, "/v1/chat/completions", payload, timeout=60)
|
|
if r and "choices" in r:
|
|
content = r["choices"][0]["message"]["content"].strip()
|
|
content = re.sub(r'```(?:json)?', '', content).strip('` \n')
|
|
try:
|
|
data = json.loads(content)
|
|
topics = data.get("topics", [])
|
|
if topics:
|
|
return topics[:count]
|
|
except Exception as e:
|
|
logger.warning(f"Error parsing LLM niche topics: {e}")
|
|
|
|
return self._fallback_topics(niche, count)
|
|
|
|
def convert_trends_to_video_topics(self, raw_trends: List[str], count: int = 5) -> List[str]:
|
|
"""Transform raw search queries into compelling documentary video topics."""
|
|
if not raw_trends:
|
|
return self.brainstorm_niche_topics(random.choice(list(NICHES.keys())), count)
|
|
|
|
if not self.pc or not self.pc.check(self.llm_port):
|
|
return [f"The Shocking Truth About {t}" for t in raw_trends[:count]]
|
|
|
|
prompt = f"""You are a viral YouTube documentary producer.
|
|
Transform the following real-time search trends into {count} high-retention faceless documentary topics:
|
|
{json.dumps(raw_trends[:10])}
|
|
|
|
Requirements:
|
|
- Turn simple keywords into deep, narrative-driven documentary angles
|
|
- Output STRICT JSON only: {{"topics": ["Topic 1", "Topic 2", ...]}}
|
|
- No markdown formatting or extra text."""
|
|
|
|
payload = {
|
|
"model": "llama",
|
|
"messages": [{"role": "user", "content": prompt}],
|
|
"temperature": 0.7,
|
|
"max_tokens": 1024
|
|
}
|
|
r = self.pc.post_json(self.llm_port, "/v1/chat/completions", payload, timeout=60)
|
|
if r and "choices" in r:
|
|
content = r["choices"][0]["message"]["content"].strip()
|
|
content = re.sub(r'```(?:json)?', '', content).strip('` \n')
|
|
try:
|
|
data = json.loads(content)
|
|
topics = data.get("topics", [])
|
|
if topics:
|
|
return topics[:count]
|
|
except Exception as e:
|
|
logger.warning(f"Error parsing trend conversion JSON: {e}")
|
|
|
|
return [f"The Untold Story of {t}" for t in raw_trends[:count]]
|
|
|
|
def get_trending_topics(self, niche: Optional[str] = None, count: int = 3) -> List[str]:
|
|
"""Main method to get trending topics, either niche-based or real-time web trends."""
|
|
if niche:
|
|
return self.brainstorm_niche_topics(niche, count)
|
|
|
|
raw = self.get_google_trends(max_items=10)
|
|
if raw:
|
|
return self.convert_trends_to_video_topics(raw, count)
|
|
else:
|
|
return self.brainstorm_niche_topics(random.choice(list(NICHES.keys())), count)
|
|
|
|
def _fallback_topics(self, niche: str, count: int) -> List[str]:
|
|
curated = {
|
|
"history": [
|
|
"The Lost Library of Alexandria and Its Forgotten Secrets",
|
|
"Why the Roman Empire Actually Collapsed",
|
|
"The Ancient Underground City of Derinkuyu",
|
|
"The Real Reason the Pyramids Were Built",
|
|
"The Dark Secrets of the Medieval Black Plague"
|
|
],
|
|
"space": [
|
|
"What If the James Webb Telescope Found an Artificial Structure?",
|
|
"The Great Attractor: The Cosmic Monster Pulling Our Galaxy",
|
|
"What Actually Happens Inside a Supermassive Black Hole?",
|
|
"The Terrifying Theory of the Dead Universe",
|
|
"Europa: The Hidden Ocean That Could Harbor Alien Life"
|
|
],
|
|
"science": [
|
|
"The Deepest Point on Earth: Secrets of the Mariana Trench",
|
|
"The Quantum Paradox That Breaks Modern Physics",
|
|
"Can We Stop Aging? The Biological Immortality Breakthrough",
|
|
"The Mysterious Sounds Recorded in the Deepest Mines",
|
|
"The Hive Mind: How Ants Outsmart Supercomputers"
|
|
],
|
|
"tech": [
|
|
"The Rise of Autonomous AI Agents: What Comes Next?",
|
|
"How Quantum Computers Will Break Global Encryption",
|
|
"The Silicon Shield: How One Island Controls All Global Tech",
|
|
"Humanoid Robots in Everyday Life: The Next 10 Years",
|
|
"The Hidden Global Fiber-Optic Cables Running Under the Ocean"
|
|
],
|
|
"mysteries": [
|
|
"The Voynich Manuscript: The Unbreakable Ancient Code",
|
|
"The Mystery of the Mary Celeste Ship",
|
|
"The Lost Amber Room: History's Greatest Stolen Treasure",
|
|
"The Strange Case of the Dyatlov Pass Incident",
|
|
"The Bermuda Triangle of Space: The South Atlantic Anomaly"
|
|
],
|
|
"business": [
|
|
"How One Company Quietly Controls 90% of the World's Microchips",
|
|
"The Greatest Wall Street Heist in History",
|
|
"The Collapse of the South Sea Bubble: The First Stock Mania",
|
|
"How Dubai Built a Trillion-Dollar Metropolis Out of the Desert",
|
|
"The Dark Economics Behind Fast Fashion Giants"
|
|
]
|
|
}
|
|
pool = curated.get(niche.lower(), curated["history"])
|
|
random.shuffle(pool)
|
|
return pool[:count]
|