"""YouTube Data API v3 client: quota guard, TTL cache, typed faults. Every number the server returns originates from a live API response made here at request time, or from the TTL cache of a previous live response. Nothing is fabricated; failures surface as structured errors. """ import os import time from collections import OrderedDict from datetime import datetime, timedelta from zoneinfo import ZoneInfo import httpx import config from src import analysis API_BASE = "https://www.googleapis.com/youtube/v3" _PACIFIC = ZoneInfo("America/Los_Angeles") _QUOTA_UPSTREAM_REASONS = {"quotaExceeded", "dailyLimitExceeded"} class ToolFault(Exception): """Structured tool failure, surfaced to the client as {"ok": false, "error": {...}}.""" def __init__(self, code, message, retryable): super().__init__(message) self.code = code self.message = message self.retryable = retryable def to_response(self): return { "ok": False, "error": { "code": self.code, "message": self.message, "retryable": self.retryable, }, } class RequestMeter: """Counts quota units spent by a single tool invocation (reported in meta).""" def __init__(self): self.units = 0 class QuotaGuard: """Tracks units spent per Google quota day (resets at midnight Pacific).""" def __init__(self, budget): self.budget = budget self._day = None self._spent = 0 def _roll(self): today = datetime.now(_PACIFIC).date() if today != self._day: self._day = today self._spent = 0 @property def remaining(self): self._roll() return max(0, self.budget - self._spent) def next_reset_iso(self): now = datetime.now(_PACIFIC) reset = (now + timedelta(days=1)).replace( hour=0, minute=0, second=0, microsecond=0 ) return reset.isoformat() def precheck(self, worst_case_units): self._roll() if worst_case_units > self.remaining: raise ToolFault( "QUOTA_EXHAUSTED", f"Daily YouTube quota budget is spent ({self._spent}/{self.budget} units " f"used today; this call needs up to {worst_case_units}). The budget " f"resets at midnight Pacific: {self.next_reset_iso()}. Retry after that.", True, ) def spend(self, units, meter): self._roll() self._spent += units meter.units += units quota = QuotaGuard(config.DAILY_UNIT_BUDGET) class TTLCache: """In-memory TTL cache keyed on normalized inputs. Stores only real fetched responses (caching is not seeding). Capped; evicts oldest entries first.""" def __init__(self, max_entries): self.max_entries = max_entries self._entries = OrderedDict() # key -> (expires_at, payload, fetched_at_iso) def get(self, key): entry = self._entries.get(key) if entry is None: return None expires_at, payload, fetched_at = entry if time.monotonic() >= expires_at: del self._entries[key] return None return payload, fetched_at def put(self, key, payload, ttl_s, fetched_at): self._entries[key] = (time.monotonic() + ttl_s, payload, fetched_at) self._entries.move_to_end(key) while len(self._entries) > self.max_entries: self._entries.popitem(last=False) cache = TTLCache(config.CACHE_MAX_ENTRIES) _client = httpx.AsyncClient(timeout=config.UPSTREAM_TIMEOUT_S) def api_key_present(): return bool(os.environ.get(config.YT_API_KEY_ENV, "").strip()) def _api_key(): key = os.environ.get(config.YT_API_KEY_ENV, "").strip() if not key: raise ToolFault( "MISSING_API_KEY", f"{config.YT_API_KEY_ENV} is not configured on the server, so no live " f"YouTube data can be fetched. The operator must set it; retrying will " f"not help until then.", False, ) return key def _sanitize(text, key): return text.replace(key, "***") if key else text def _error_details(response): try: err = response.json().get("error", {}) errors = err.get("errors") or [{}] return errors[0].get("reason", "unknown"), err.get("message", response.text[:200]) except Exception: return "unknown", response.text[:200] async def _get(resource, params, cost, meter): key = _api_key() quota.spend(cost, meter) try: response = await _client.get( f"{API_BASE}/{resource}", params={**params, "key": key} ) except httpx.TimeoutException: raise ToolFault( "UPSTREAM_TIMEOUT", f"YouTube API '{resource}' call exceeded {config.UPSTREAM_TIMEOUT_S}s. " f"Retry shortly.", True, ) except httpx.HTTPError as exc: raise ToolFault( "YT_API_ERROR", _sanitize(f"YouTube API '{resource}' request failed: {exc}. Retry may help.", key), True, ) if response.status_code == 200: return response.json() reason, message = _error_details(response) if reason in _QUOTA_UPSTREAM_REASONS: raise ToolFault( "QUOTA_EXHAUSTED", f"YouTube reports the API key's daily quota is exhausted. Quota resets " f"at midnight Pacific: {quota.next_reset_iso()}. Retry after that.", True, ) raise ToolFault( "YT_API_ERROR", _sanitize( f"YouTube API '{resource}' returned HTTP {response.status_code} " f"({reason}): {message}. Retry may help.", key, ), True, ) async def search_videos(query, region_code, published_after_iso, max_results, meter): return await _get( "search", { "part": "snippet", "type": "video", "q": query, "order": "relevance", "maxResults": max_results, "regionCode": region_code, "publishedAfter": published_after_iso, }, config.COST_SEARCH, meter, ) async def list_videos(video_ids, meter): return await _get( "videos", { "part": "snippet,statistics,contentDetails,liveStreamingDetails", "id": ",".join(video_ids), }, config.COST_LIST, meter, ) async def list_channels(channel_ids, meter): return await _get( "channels", {"part": "snippet,statistics,contentDetails", "id": ",".join(channel_ids)}, config.COST_LIST, meter, ) async def channel_by_handle(handle, meter): return await _get( "channels", {"part": "snippet,statistics,contentDetails", "forHandle": handle}, config.COST_LIST, meter, ) async def playlist_page(playlist_id, max_results, page_token, meter): params = { "part": "contentDetails", "playlistId": playlist_id, "maxResults": max_results, } if page_token: params["pageToken"] = page_token return await _get("playlistItems", params, config.COST_LIST, meter) def normalize_video(item): """Flatten a videos.list item. views is None when the count is hidden. stream is True for live, upcoming, and past live broadcasts.""" snippet = item.get("snippet", {}) stats = item.get("statistics", {}) seconds = analysis.duration_seconds(item.get("contentDetails", {}).get("duration")) published = snippet.get("publishedAt") return { "id": item.get("id", ""), "title": snippet.get("title", "").strip(), "channel_id": snippet.get("channelId", ""), "channel_title": snippet.get("channelTitle", "").strip(), "published_at": analysis.parse_timestamp(published) if published else None, "views": int(stats["viewCount"]) if "viewCount" in stats else None, "seconds": seconds, "stream": snippet.get("liveBroadcastContent") in ("live", "upcoming") or "liveStreamingDetails" in item, } def normalize_channel(item): """Flatten a channels.list item. subs is None when the count is hidden.""" snippet = item.get("snippet", {}) stats = item.get("statistics", {}) hidden = stats.get("hiddenSubscriberCount", False) return { "id": item.get("id", ""), "title": snippet.get("title", "").strip(), "subs": None if hidden else int(stats.get("subscriberCount", 0)), "view_count": int(stats.get("viewCount", 0)), "video_count": int(stats.get("videoCount", 0)), "uploads": item.get("contentDetails", {}) .get("relatedPlaylists", {}) .get("uploads"), }