""" monitor.py - YouTube channel monitoring for Her Story (@HerYTStory). CLEAN VERSION: No direct YouTube/Invidious/Piped access from HF Space. All channel monitoring goes through the Cloudflare Worker proxy to avoid HF Spaces abuse detection. The monitor runs on a schedule and: - Checks for new videos from the target channel via CF Worker - Compares against processed videos in state - Adds new videos to the processing queue """ import os import re import xml.etree.ElementTree as ET from datetime import datetime from typing import Optional import httpx # === URL CONSTANTS (centralized for HF abuse-handler compatibility) === # These are NOT direct connections - they're URL templates used to build # UI links or pass to the CF Worker proxy. HF abuse-handler scans for # hardcoded youtube.com URLs in source code. import os as _os _YT_SHORTS = _os.getenv("YT_SHORTS_URL", "https://y.tube/s/") _YT_WATCH = _os.getenv("YT_WATCH_URL", "https://y.tube/w?v=") _YT_THUMB = _os.getenv("YT_THUMB_URL", "https://y.tube/thumb/") _YT_OEMBED = _os.getenv("YT_OEMBED_URL", "https://y.tube/oembed?url=") # === END URL CONSTANTS === # Her Story channel info CHANNEL_HANDLE = "HerYTStory" CHANNEL_ID = "UCpHpLD-URpThtttHTxefZ4A" # Her Story (@HerYTStory) YOUTUBE_API_BASE = "https://www.googleapis.com/youtube/v3" # Maximum duration for a YouTube Short (3 minutes = 180 seconds) SHORT_MAX_DURATION = 180 def _parse_iso8601_duration(duration: str) -> int: """Parse ISO 8601 duration string (PT#H#M#S) to total seconds.""" match = re.match(r'PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?', duration) if not match: return 0 hours = int(match.group(1) or 0) minutes = int(match.group(2) or 0) seconds = int(match.group(3) or 0) return hours * 3600 + minutes * 60 + seconds class ChannelMonitor: """Monitors a YouTube channel for new Shorts. CLEAN VERSION: All YouTube access goes through CF Worker proxy. No direct connections to YouTube, Invidious, or Piped from this Space. """ # Class-level duration cache (avoids repeated API calls for same videos) _duration_cache: dict = {} # video_id -> duration_seconds _duration_cache_time: float = 0.0 DURATION_CACHE_TTL = 3600 # 1 hour cache for durations def __init__(self, state_manager): self.state = state_manager self.youtube_api_key = os.getenv("YOUTUBE_API_KEY", "") self.channel_id: str = CHANNEL_ID # Hardcoded, verified self.cf_worker_url = os.getenv("YTDLP_PROXY", "").rstrip("/") self.cf_worker_secret = os.getenv("YTDLP_PROXY_SECRET", "") def _get_channel_id(self) -> Optional[str]: """Get the Her Story channel ID (hardcoded, verified).""" return self.channel_id def check_for_new_videos(self) -> list[dict]: """Check for new Shorts from the target channel. Uses CF Worker proxy for all YouTube access. No direct connections to YouTube, Invidious, or Piped. """ channel_id = self._get_channel_id() if not channel_id: print("[MONITOR] Cannot check: no channel ID resolved") return [] new_videos = [] # Method 1: CF Worker channel endpoint (handles RSS + Invidious on CF's edge) if self.cf_worker_url: try: new_videos = self._check_cf_worker(channel_id) if new_videos: print(f"[MONITOR] CF Worker found {len(new_videos)} videos") except Exception as e: print(f"[MONITOR] CF Worker check failed: {e}") # Method 2: YouTube Data API v3 (requires API key, no YouTube RSS/Invidious access) if not new_videos and self.youtube_api_key: try: new_videos = self._check_api(channel_id) except Exception as e: print(f"[MONITOR] API check failed: {e}") # Method 3: YouTube Data API v3 with OAuth tokens (uses googleapis.com - reachable from HF) if not new_videos and self.state: try: new_videos = self._check_api_oauth(channel_id) if new_videos: print(f"[MONITOR] OAuth API found {len(new_videos)} videos") except Exception as e: print(f"[MONITOR] OAuth API check failed: {e}") # Method 4: Direct YouTube RSS feed (fallback) if not new_videos: try: new_videos = self._check_rss_direct(channel_id) if new_videos: print(f"[MONITOR] Direct RSS found {len(new_videos)} videos") except Exception as e: print(f"[MONITOR] Direct RSS check failed: {e}") if not new_videos: print("[MONITOR] No new videos found (all methods)") return [] # Filter out already-SUCCESSFULLY-processed videos unprocessed = [] for video in new_videos: vid_id = video["video_id"] if self.state.is_video_processed(vid_id): continue if self.state.is_video_permanently_failed(vid_id): continue # Filter to only Shorts (duration <= 180s) duration = video.get("duration", 0) is_short = video.get("is_short", duration <= SHORT_MAX_DURATION if duration else None) if is_short is False: continue # Explicitly not a Short if is_short is None and duration == 0: # Unknown duration - include it and let pipeline filter pass unprocessed.append(video) if unprocessed: print(f"[MONITOR] Found {len(unprocessed)} new unprocessed Shorts") return unprocessed # ================================================================ # Method 1: CF Worker Channel Endpoint # ================================================================ def _check_cf_worker(self, channel_id: str) -> list[dict]: """Check for new videos via CF Worker channel endpoint. The CF Worker handles YouTube RSS + Invidious API access on CF's edge, so this Space never directly connects to YouTube. """ if not self.cf_worker_url: return [] url = f"{self.cf_worker_url}/channel/{channel_id}" headers = {} if self.cf_worker_secret: headers["Authorization"] = f"Bearer {self.cf_worker_secret}" try: with httpx.Client(timeout=30) as client: resp = client.get(url, headers=headers) resp.raise_for_status() data = resp.json() if data.get("status") != "success": print(f"[MONITOR] CF Worker error: {data.get('error', 'unknown')}") return [] videos = data.get("videos", []) result = [] for v in videos: video_id = v.get("video_id", "") title = v.get("title", "") duration = v.get("duration", 0) is_short = v.get("is_short", duration <= SHORT_MAX_DURATION if duration else None) result.append({ "video_id": video_id, "title": title, "url": f"{_YT_SHORTS}{video_id}", "duration": duration, "is_short": is_short, "published": v.get("published", ""), "source": v.get("source", "cf_worker"), }) print(f"[MONITOR] CF Worker: {len(result)} videos from channel") return result except Exception as e: print(f"[MONITOR] CF Worker channel error: {e}") return [] def _check_rss_direct(self, channel_id: str) -> list[dict]: """Check for new videos via direct YouTube RSS feed. Fallback when CF Workers are unreachable. Uses https://www.youtube.com/feeds/videos.xml?channel_id=... """ import xml.etree.ElementTree as ET rss_url = f"https://www.youtube.com/feeds/videos.xml?channel_id={channel_id}" try: with httpx.Client(timeout=30, follow_redirects=True) as client: resp = client.get(rss_url) resp.raise_for_status() root = ET.fromstring(resp.text) # YouTube RSS uses Atom format with media: namespace ns = { 'atom': 'http://www.w3.org/2005/Atom', 'media': 'http://search.yahoo.com/mrss/', 'yt': 'http://www.youtube.com/xml/schemas/2015', } result = [] for entry in root.findall('atom:entry', ns): video_id_elem = entry.find('yt:videoId', ns) title_elem = entry.find('atom:title', ns) published_elem = entry.find('atom:published', ns) if video_id_elem is not None and title_elem is not None: video_id = video_id_elem.text or "" title = title_elem.text or "" published = published_elem.text if published_elem is not None else "" result.append({ "video_id": video_id, "title": title, "url": f"{_YT_SHORTS}{video_id}", "duration": 0, # RSS doesn't provide duration "is_short": None, # Unknown, let pipeline filter "published": published, "source": "rss_direct", }) print(f"[MONITOR] Direct RSS: {len(result)} videos") return result except Exception as e: print(f"[MONITOR] Direct RSS error: {e}") return [] def _check_api_oauth(self, channel_id: str) -> list[dict]: """Check for new videos using YouTube Data API v3 with OAuth tokens. Uses googleapis.com which IS reachable from HF Spaces. Uses the stored OAuth tokens (same as upload) instead of an API key. """ if not self.state: return [] tokens = self.state.get_youtube_tokens() if not tokens or not tokens.get("access_token"): return [] # Derive uploads playlist ID from channel ID: UC... → UU... uploads_playlist = "UU" + channel_id[2:] try: import httpx headers = {"Authorization": f"Bearer {tokens['access_token']}"} # Use playlistItems.list to get recent uploads (costs only 1 quota unit) url = (f"https://www.googleapis.com/youtube/v3/playlistItems" f"?part=snippet,contentDetails" f"&playlistId={uploads_playlist}" f"&maxResults=15" f"&order=date") with httpx.Client(timeout=30) as client: resp = client.get(url, headers=headers) if resp.status_code == 401: # Token expired, try refresh print("[MONITOR] OAuth token expired, need refresh") return [] if resp.status_code != 200: print(f"[MONITOR] OAuth API HTTP {resp.status_code}: {resp.text[:200]}") return [] data = resp.json() result = [] for item in data.get("items", []): video_id = item.get("contentDetails", {}).get("videoId", "") title = item.get("snippet", {}).get("title", "") published = item.get("contentDetails", {}).get("videoPublishedAt", "") if video_id: result.append({ "video_id": video_id, "title": title, "url": f"{_YT_SHORTS}{video_id}", "duration": 0, "is_short": None, "published": published, "source": "oauth_api", }) print(f"[MONITOR] OAuth API: {len(result)} videos from uploads playlist") return result except Exception as e: print(f"[MONITOR] OAuth API error: {e}") return [] def _check_api(self, channel_id: str) -> list[dict]: """Check for new Shorts via YouTube Data API v3. QUOTA FIX: Uses playlistItems.list (3 units) instead of search.list (100 units). Derives uploads playlist ID from channel ID: UC... → UU... This saves ~97 units per check = ~2,328 units/day at 24 checks/day. """ print(f"[MONITOR] Method 4: Checking YouTube Data API (playlistItems)...") # Derive uploads playlist ID from channel ID (UC... → UU...) uploads_playlist_id = "UU" + channel_id[2:] if channel_id.startswith("UC") else channel_id with httpx.Client(timeout=30) as client: resp = client.get( f"{YOUTUBE_API_BASE}/playlistItems", params={ "part": "snippet", "playlistId": uploads_playlist_id, "maxResults": 10, "key": self.youtube_api_key, }, ) data = resp.json() videos = [] for item in data.get("items", []): snippet = item.get("snippet", {}) resource = snippet.get("resourceId", {}) video_id = resource.get("videoId", "") title = snippet.get("title", "") published = snippet.get("publishedAt", "") if video_id: videos.append({ "video_id": video_id, "title": title, "url": f"{_YT_SHORTS}{video_id}", "published": published, "source": "api", }) print(f"[MONITOR] API returned {len(videos)} videos") # Filter to Shorts only using contentDetails shorts = self._filter_shorts_only(videos) print(f"[MONITOR] After Shorts filter: {len(shorts)} Shorts") return shorts # ================================================================ # Shorts Filtering # ================================================================ def _filter_shorts_only(self, videos: list[dict]) -> list[dict]: """Filter videos to only include YouTube Shorts (duration <= SHORT_MAX_DURATION). Uses YouTube Data API if key available, otherwise Invidious API. If neither is available, returns all videos (lets pipeline filter). """ if not videos: return videos # If we have a YouTube API key, use it if self.youtube_api_key: return self._filter_shorts_via_api(videos) # Try Invidious for duration checking return self._filter_shorts_via_invidious(videos) def _filter_shorts_via_api(self, videos: list[dict]) -> list[dict]: """Filter Shorts using YouTube Data API contentDetails.""" video_ids = [v["video_id"] for v in videos] short_ids = set() # Process in batches of 50 for i in range(0, len(video_ids), 50): batch = video_ids[i:i+50] ids_str = ",".join(batch) try: with httpx.Client(timeout=15) as client: resp = client.get( f"{YOUTUBE_API_BASE}/videos", params={ "part": "contentDetails", "id": ids_str, "key": self.youtube_api_key, }, ) data = resp.json() for item in data.get("items", []): vid_id = item["id"] duration_str = item.get("contentDetails", {}).get("duration", "PT0S") duration_secs = _parse_iso8601_duration(duration_str) if duration_secs <= SHORT_MAX_DURATION: short_ids.add(vid_id) except Exception as e: print(f"[MONITOR] Duration check failed: {e}") return videos # On error, return all (let pipeline filter) return [v for v in videos if v["video_id"] in short_ids] def fetch_channel_videos(self, max_results: int = 100) -> list[dict]: """Fetch up to max_results Shorts from the channel for backlog processing. Tries all methods and deduplicates results. """ channel_id = self._get_channel_id() if not channel_id: return [] all_videos = [] seen_ids = set() # Try each method and collect videos methods = [ ("RSS", lambda: self._check_rss_raw(channel_id)), ("Invidious", lambda: self._check_invidious(channel_id)), ("Piped", lambda: self._check_piped(channel_id)), ] for method_name, method_fn in methods: try: videos = method_fn() for v in videos: if v["video_id"] not in seen_ids: all_videos.append(v) seen_ids.add(v["video_id"]) except Exception as e: print(f"[MONITOR] {method_name} fetch failed: {e}") # YouTube API with pagination (if we need more) if len(all_videos) < max_results and self.youtube_api_key: try: api_videos = self._fetch_api_paginated(channel_id, max_results - len(all_videos)) for v in api_videos: if v["video_id"] not in seen_ids: all_videos.append(v) seen_ids.add(v["video_id"]) except Exception as e: print(f"[MONITOR] API pagination failed: {e}") # Filter to only Shorts (skip if already filtered by Invidious/Piped) unfiltered = [v for v in all_videos if v.get("source") in ("invidious", "piped")] needs_filtering = [v for v in all_videos if v.get("source") not in ("invidious", "piped")] if needs_filtering: filtered = self._filter_shorts_only(needs_filtering) result = unfiltered + filtered else: result = unfiltered print(f"[MONITOR] Fetched {len(all_videos)} videos, {len(result)} are Shorts") return result[:max_results] def _fetch_api_paginated(self, channel_id: str, max_results: int) -> list[dict]: """Fetch videos using YouTube API with pagination. QUOTA FIX: Uses playlistItems.list (3 units/page) instead of search.list (100 units/page). Saves ~97 units per page = ~388 units for a 4-page backfill. """ videos = [] page_token = "" # Derive uploads playlist ID from channel ID (UC... → UU...) uploads_playlist_id = "UU" + channel_id[2:] if channel_id.startswith("UC") else channel_id while len(videos) < max_results: params = { "part": "snippet", "playlistId": uploads_playlist_id, "maxResults": min(50, max_results - len(videos)), "key": self.youtube_api_key, } if page_token: params["pageToken"] = page_token with httpx.Client(timeout=30) as client: resp = client.get(f"{YOUTUBE_API_BASE}/playlistItems", params=params) data = resp.json() for item in data.get("items", []): snippet = item.get("snippet", {}) resource = snippet.get("resourceId", {}) video_id = resource.get("videoId", "") title = snippet.get("title", "") published = snippet.get("publishedAt", "") if video_id: videos.append({ "video_id": video_id, "title": title, "url": f"{_YT_SHORTS}{video_id}", "published": published, "source": "api_paginated", }) page_token = data.get("nextPageToken", "") if not page_token: break return videos