#!/usr/bin/env python3
"""
YouTube name-search fallback utility for SMA Radar
Searches YouTube by company name when channel_id is not available
"""
import sys
import re
import time
import json
import urllib.parse
import urllib.request
import xml.etree.ElementTree as ET
from datetime import datetime, timezone
from pathlib import Path
sys.path.insert(0, str(Path(__file__).parent.parent))
try:
import yaml
except ImportError:
yaml = None
CONFIG_PATH = Path(__file__).parent.parent / "watch_entities.yaml"
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124 Safari/537.36",
"Accept-Language": "de-DE,de;q=0.9,es;q=0.8,en;q=0.7",
}
def youtube_search(query, max_results=5):
"""Search YouTube and return video results"""
q = urllib.parse.quote(query)
url = f"https://www.youtube.com/results?search_query={q}"
req = urllib.request.Request(url, headers=HEADERS)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
raw = resp.read()
html = raw.decode('utf-8', errors='ignore')
# Look for video IDs
video_matches = re.findall(r'"videoId":"([^"]{11})"', html)
channel_matches = re.findall(r'"channelId":"([^"]+)"', html)
return list(set(video_matches))[:max_results], list(set(channel_matches))[:max_results]
except Exception as e:
print(f"[YT-SEARCH] Error searching for '{query}': {e}")
return [], []
def youtube_channel_rss(channel_id):
"""Get videos from YouTube channel RSS"""
url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}"
req = urllib.request.Request(url, headers=HEADERS)
try:
with urllib.request.urlopen(req, timeout=10) as resp:
raw = resp.read()
root = ET.fromstring(raw)
ns = {"a": "http://www.w3.org/2005/Atom", "yt": "http://www.youtube.com/xml/schemas/2015"}
entries = root.findall("a:entry", ns)
results = []
for e in entries[:5]: # Limit to 5 most recent
vid = e.findtext("yt:videoId", "", ns)
title = (e.findtext("a:title", "", ns) or "").strip()
published = e.findtext("a:published", "", ns)
if vid and title:
results.append({
"video_id": vid,
"title": title,
"published": published,
"url": f"https://www.youtube.com/watch?v={vid}"
})
return results
except Exception as e:
print(f"[YT-SEARCH] Error getting RSS for channel '{channel_id}': {e}")
return []
def is_relevant_channel(channel_info, company_name):
"""Check if a YouTube channel is relevant to the company"""
# Simple heuristic: check if company name appears in channel title or description
# In a real implementation, we might check the channel's 'about' page or video titles
company_lower = company_name.lower()
# For now, we'll return True for channels we find via name search
# A more sophisticated implementation would validate the channel
return True
def find_youtube_channel_for_entity(entity):
"""Find YouTube channel for an entity using name search"""
name = entity["canonical_name"]
entity_id = entity["entity_id"]
# Skip if already has a channel ID
if entity.get("youtube_channel_id"):
print(f"[YT-SEARCH] {name} already has channel ID: {entity['youtube_channel_id']}")
return entity["youtube_channel_id"]
# Search queries to try
search_queries = [
f'"{name}" Mallorca',
f'{name} Mallorca',
f'{name} property management Mallorca',
f'{name} house care Mallorca',
f'{name} home management Mallorca',
]
print(f"[YT-SEARCH] Searching for YouTube channel for: {name}")
for query in search_queries:
print(f" Trying query: {query}")
videos, channels = youtube_search(query, max_results=10)
if channels:
# Check each channel for relevance
for channel_id in channels:
if is_relevant_channel({"id": channel_id}, name):
# Get channel info to verify
rss_results = youtube_channel_rss(channel_id)
if rss_results:
print(f"[YT-SEARCH] Found relevant channel: {channel_id}")
print(f" Recent videos: {len(rss_results)}")
for video in rss_results[:2]:
print(f" - {video['title'][:50]}...")
return channel_id
else:
print(f"[YT-SEARCH] Channel {channel_id} found but no RSS content")
else:
print(f" No channels found for query: {query}")
print(f"[YT-SEARCH] No suitable YouTube channel found for {name}")
return None
def update_entity_youtube_id(entity_id, youtube_channel_id):
"""Update an entity's YouTube channel ID in the YAML file"""
try:
with open(CONFIG_PATH, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
updated = False
for entity in data.get("watch_entities", []):
if entity.get("entity_id") == entity_id:
if entity.get("youtube_channel_id") != youtube_channel_id:
entity["youtube_channel_id"] = youtube_channel_id
updated = True
print(f"[YT-SEARCH] Updated {entity_id} with YouTube ID: {youtube_channel_id}")
else:
print(f"[YT-SEARCH] {entity_id} already has YouTube ID: {youtube_channel_id}")
break
if updated:
with open(CONFIG_PATH, "w", encoding="utf-8") as fh:
yaml.dump(data, fh, default_flow_style=False, allow_unicode=True)
print(f"[YT-SEARCH] Successfully updated watch_entities.yaml")
return True
else:
print(f"[YT-SEARCH] Entity {entity_id} not found in watch_entities.yaml")
return False
except Exception as e:
print(f"[YT-SEARCH] Error updating watch_entities.yaml: {e}")
return False
def main():
"""Main function to find and update YouTube channel IDs for entities"""
if yaml is None:
print("[YT-SEARCH] PyYAML not available")
return 1
try:
with open(CONFIG_PATH, encoding="utf-8") as fh:
data = yaml.safe_load(fh)
except Exception as e:
print(f"[YT-SEARCH] Error loading watch_entities.yaml: {e}")
return 1
entities = [e for e in data.get("watch_entities", [])
if e.get("active") and e.get("verification", {}).get("status") == "verified"]
print(f"[YT-SEARCH] Checking {len(entities)} verified active entities...")
updated_count = 0
checked_count = 0
for entity in entities:
entity_id = entity["entity_id"]
name = entity["canonical_name"]
# Skip entities that already have a YouTube channel ID
if entity.get("youtube_channel_id"):
print(f"[YT-SEARCH] Skipping {name} (already has channel ID)")
continue
checked_count += 1
print(f"\\n[YT-SEARCH] Processing: {name} ({entity_id})")
# Find YouTube channel
channel_id = find_youtube_channel_for_entity(entity)
if channel_id:
# Update the entity
if update_entity_youtube_id(entity_id, channel_id):
updated_count += 1
# Be respectful with rate limiting
time.sleep(2)
else:
print(f"[YT-SEARCH] No YouTube channel found for {name}")
print(f"\\n[YT-SEARCH] Summary:")
print(f" Entities checked: {checked_count}")
print(f" Entities updated: {updated_count}")
return 0 if updated_count >= 0 else 1
if __name__ == "__main__":
sys.exit(main())