Daily movie and TV show info aggregator with LLM-generated highlights, multi-source scraping, and smart ranking
triggers
["set up moontv daily feed","create a movie and TV show recommendation system","scrape CMS movie sources","generate daily movie highlights with LLM","build a watchlist tracker","aggregate multiple video sources","rank movies by douban score and popularity","create automated daily media reports"]
This skill enables AI agents to help developers use MoonTV OpenClaw, a Python-based movie and TV show aggregator that scrapes multiple CMS sources, ranks content by Douban scores and popularity, generates LLM-powered highlights, and produces daily Markdown reports with watchlist tracking.
What MoonTV Does
MoonTV OpenClaw:
Multi-source aggregation: Concurrently fetches from 400+ CMS sources via a gateway API
Smart deduplication: Removes duplicates by vod_name, keeping first occurrence
5-category ranking: Movies, TV shows, variety shows, short dramas, and special content (Top 5 each)
Dual-path scoring: Uses Douban ratings when available, otherwise falls back to popularity-based scoring
Watchlist tracking: Monitors configured shows for updates
LLM highlights: Generates content highlights via GPT-4o-mini (with fallback to synopsis truncation)
Scrape all sources concurrently (with 10-second timeout per source)
Deduplicate by vod_name
Classify into 5 categories
Score and rank items
Match watchlist items (if configured)
Generate LLM highlights (or fallback)
Render Markdown report to output/
Clean up old reports (>7 days)
Run Tests
cd scripts
python -m pytest test_moontv.py -v
Key Python Modules
1. Main Scraper (moontv_daily.py)
Fetch Gateway Sources:
import requests
import os
from dotenv import load_dotenv
load_dotenv()
deffetch_gateway():
"""Fetch available CMS sources from gateway"""
gateway_url = os.getenv("MOONTV_GATEWAY")
response = requests.get(f"{gateway_url}/api/resource/sources", timeout=10)
response.raise_for_status()
data = response.json()
if data["code"] != 0ornot data["data"]:
raise Exception("Gateway returned no sources")
return data["data"] # List of dicts: [{"name": "...", "api": "..."}, ...]
Scrape CMS Source:
deffetch_cms_data(api_url):
"""Fetch today's data from a single CMS source"""
response = requests.get(
f"{api_url}?ac=videolist&t=1,2,3,4,5",
timeout=10
)
response.raise_for_status()
return response.json()
Deduplicate:
defdeduplicate(items):
"""Remove duplicates by vod_name, keep first occurrence"""
seen = set()
unique = []
for item in items:
name = item.get("vod_name")
if name and name notin seen:
seen.add(name)
unique.append(item)
return unique
Classify Items:
defclassify(items):
"""Classify into 5 categories with priority"""
categories = {
"电影": [], "剧集": [], "综艺": [], "短剧": [], "福利": []
}
for item in items:
type_name = item.get("type_name", "")
# Priority order: 电影 > 剧集 > 综艺 > 短剧 > 福利if"电影"in type_name:
categories["电影"].append(item)
elifany(x in type_name for x in ["连续", "电视剧", "美剧", "韩剧"]):
categories["剧集"].append(item)
elif"综艺"in type_name:
categories["综艺"].append(item)
elif"短剧"in type_name:
categories["短剧"].append(item)
else:
categories["福利"].append(item)
return categories
import json
defmatch_watchlist(all_items, watchlist_file):
"""Match items against watchlist"""try:
withopen(watchlist_file, 'r', encoding='utf-8') as f:
config = json.load(f)
watchlist = config.get("watchlist", [])
except:
return []
matches = []
for watch in watchlist:
for item in all_items:
if item.get("vod_name") == watch["name"]:
matches.append(item)
breakreturn matches
Common Patterns
1. Concurrent Source Scraping
from concurrent.futures import ThreadPoolExecutor, as_completed
defscrape_all_sources(sources):
"""Scrape all CMS sources concurrently"""
all_items = []
with ThreadPoolExecutor(max_workers=10) as executor:
futures = {
executor.submit(fetch_cms_data, src["api"]): src["name"]
for src in sources
}
for future in as_completed(futures):
source_name = futures[future]
try:
data = future.result()
items = data.get("list", [])
all_items.extend(items)
print(f"✓ {source_name}: {len(items)} items")
except Exception as e:
print(f"✗ {source_name}: {e}")
return all_items
2. Filter Low-Quality Content
deffilter_low_quality(items):
"""Remove items with no name, score, or hits"""return [
item for item in items
if item.get("vod_name")
and (item.get("vod_douban_score", 0) > 0or item.get("vod_hits", 0) > 0)
]
3. Extract Episode Number
import re
defextract_episode(item):
"""Extract episode info from vod_remarks"""
remarks = item.get("vod_remarks", "")
# Match patterns like "更新至12集", "第10集", "EP08"
patterns = [
r'更新至(\d+)集',
r'第(\d+)集',
r'EP(\d+)',
r'(\d+)集全'
]
for pattern in patterns:
match = re.search(pattern, remarks)
ifmatch:
returnf"第{match.group(1)}集"return remarks if remarks else""
4. Clean Old Reports
import os
from datetime import datetime, timedelta
from pathlib import Path
defcleanup_old_reports(output_dir, days=7):
"""Delete reports older than N days"""
cutoff = datetime.now() - timedelta(days=days)
for file in Path(output_dir).glob("moontv_*.md"):
if file.stat().st_mtime < cutoff.timestamp():
file.unlink()
print(f"Deleted old report: {file.name}")
Troubleshooting
Gateway Not Accessible
Symptom: requests.exceptions.RequestException when calling gateway
Solution: Check MOONTV_GATEWAY in .env, verify network access:
This skill provides everything an AI agent needs to help developers deploy, configure, and customize MoonTV OpenClaw for automated movie/TV content aggregation and recommendation.