"""Subida de un Short a YouTube vía Data API v3. **Lo primero que hay que saber, porque cambia lo que esta pieza puede prometer:** los vídeos subidos con `videos.insert` desde un proyecto de API sin auditar (creados después del 28-jul-2020) quedan *restringidos a privado*, y el candado es del PROYECTO, no del vídeo — no se abre desde Studio, se abre pasando la auditoría de cumplimiento de Google. Así que esto no publica: deja el vídeo en el canal con los metadatos puestos y devuelve el enlace de Studio para que una persona lo revise y le dé a publicar. Ese paso humano no es una limitación que estemos aceptando a regañadientes; es el mismo que defiende `/publish` con los borradores de Ghost. Sin `google-api-python-client` a propósito: es síncrono (bloquearía el loop del bot), arrastra httplib2 y protobuf, y lo que necesitamos son dos peticiones HTTP. El repo ya firma los JWT de Ghost a mano por la misma razón. El token de refresco NO se guarda aquí ni en la DB: llega por entorno desde Infisical. El de acceso vive en memoria y dura una hora. """ from __future__ import annotations import json import re import time from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable, Optional import aiohttp import structlog from src.config import SAFE_ACCEPT_ENCODING, settings logger = structlog.get_logger() __all__ = [ "YouTubeUploader", "UploadedVideo", "build_metadata", "YouTubeError", "YouTubeNotConfigured", "YouTubeAuthError", "YouTubeQuotaExceeded", "YouTubeRejected", "YouTubeDisabled", ] TOKEN_URL = "https://oauth2.googleapis.com/token" UPLOAD_URL = "https://www.googleapis.com/upload/youtube/v3/videos" #: El único scope que hace falta. `youtube.upload` no puede leer ni borrar nada #: del canal: si el token se filtra, lo peor que se puede hacer con él es subir. SCOPE = "https://www.googleapis.com/auth/youtube.upload" #: Márgen antes de que caduque el token de acceso (dura 3600 s). _TOKEN_MARGIN = 120.0 #: Tokens de acceso en memoria por client_id. El bot crea un uploader nuevo en #: cada comando; sin esto, cada subida pagaría un refresco. _token_cache: dict[str, tuple[str, float]] = {} # Límites de la API. Pasarse no da un error bonito: da un 400 genérico. MAX_TITLE = 100 MAX_DESCRIPTION = 5000 MAX_TAGS_CHARS = 460 # el tope real es 500; dejamos aire para las comas MAX_TAG = 60 # una etiqueta suelta más larga que esto no la busca nadie class YouTubeError(Exception): """Cualquier fallo hablando con YouTube.""" class YouTubeDisabled(YouTubeError): """YOUTUBE_ENABLED=false. El interruptor, igual que SHORTSMITH_ENABLED.""" class YouTubeNotConfigured(YouTubeError): """Faltan client id / secret / refresh token.""" class YouTubeAuthError(YouTubeError): """El refresh token no sirve: caducado, revocado o de otro cliente.""" class YouTubeQuotaExceeded(YouTubeError): """Cuota diaria agotada. Se reinicia a medianoche hora del Pacífico.""" class YouTubeRejected(YouTubeError): """YouTube rechazó los metadatos o el fichero.""" def __init__(self, message: str, reason: str = ""): super().__init__(message) self.reason = reason @dataclass class UploadedVideo: video_id: str title: str privacy_status: str #: True si YouTube ignoró el privacy_status pedido y lo dejó en privado. #: Es la firma del candado del proyecto sin auditar. forced_private: bool = False upload_status: str = "" #: Por qué YouTube marcó el vídeo como no reproducible, si lo hizo. rejection_reason: str = "" @property def watch_url(self) -> str: return f"https://youtube.com/shorts/{self.video_id}" @property def studio_url(self) -> str: return f"https://studio.youtube.com/video/{self.video_id}/edit" # -------------------------------------------------------------------------- # Metadatos # -------------------------------------------------------------------------- #: Etiquetas de partida del canal. Las del tema se añaden detrás. BASE_TAGS = ["UAP", "UFO", "declassified", "documentary", "shorts"] #: Palabras que no aportan nada como etiqueta. _STOPWORDS = { "the", "a", "an", "of", "in", "on", "at", "to", "for", "and", "or", "el", "la", "los", "las", "de", "del", "en", "y", "o", "un", "una", } #: De dónde sale una cita de fuente dentro de los props de un shot. Son los #: campos que el spec usa para atribuir, no para rotular. _CITATION_KEYS = ("source", "attribution") def _clean(text: str) -> str: return re.sub(r"\s+", " ", str(text)).strip() def _tags_from(topic: str, spec_id: str = "") -> list[str]: """Etiquetas del tema, sin repetir las de base y sin pasarse de los 500 caracteres que YouTube cuenta sumando toda la lista. El tema entero va primero como una sola etiqueta: partido en palabras deja cosas como "New" y "Mexico" sueltas, que no buscan igual que "Socorro New Mexico 1964". Las palabras sueltas van detrás igualmente, que cuestan poco. """ seen = {t.casefold() for t in BASE_TAGS} tags = list(BASE_TAGS) phrase = _clean(topic)[:MAX_TAG] if phrase and phrase.casefold() not in seen: seen.add(phrase.casefold()) tags.append(phrase) words = re.findall(r"[\w'-]+", f"{topic} {spec_id.replace('_', ' ')}") for word in words: low = word.casefold() if low in seen or low in _STOPWORDS or len(word) < 3: continue seen.add(low) tags.append(word) kept, size = [], 0 for tag in tags: if size + len(tag) + 1 > MAX_TAGS_CHARS: break kept.append(tag) size += len(tag) + 1 return kept def _citations(spec: dict) -> list[str]: """Las atribuciones que el propio Short enseña en pantalla. Verbatim, sin tocar mayúsculas: vienen en caja alta del spec y cualquier intento de suavizarlas convierte FAA en Faa. Van a una descripción que un humano revisa antes de publicar; que las edite él si quiere. """ out: list[str] = [] for shot in spec.get("shots") or []: props = shot.get("props") or {} if not isinstance(props, dict): continue for key in _CITATION_KEYS: value = props.get(key) if isinstance(value, str) and _clean(value): text = _clean(value).lstrip("—-– ") if text and text not in out: out.append(text) return out def build_metadata(spec: dict, topic: str, article_url: Optional[str] = None, privacy_status: Optional[str] = None, category_id: Optional[str] = None) -> dict: """El cuerpo de `videos.insert`, construido desde el shot spec ya guardado. Deliberadamente corto. La descripción no busca posicionar — en Shorts eso lo hace el título — sino ahorrarle a quien revisa teclear el enlace al artículo y las fuentes. Lo que falte se edita en Studio, que es donde va a estar de todas formas. """ meta = spec.get("meta") or {} title = _clean(meta.get("title") or topic)[:MAX_TITLE] parts: list[str] = [_clean(topic)] if article_url: parts.append(f"Full investigation → {article_url}") cited = _citations(spec) if cited: parts.append("Sources cited in this short:\n" + "\n".join(f"— {c}" for c in cited)) # #Shorts no es obligatorio (YouTube clasifica solo por formato vertical y # duración) pero tampoco estorba, y quita la duda cuando el render cambia. parts.append("#Shorts #UAP #UFO") description = "\n\n".join(p for p in parts if p)[:MAX_DESCRIPTION] return { "snippet": { "title": title, "description": description, "tags": _tags_from(topic, str(meta.get("id") or "")), "categoryId": str(category_id or settings.youtube_category_id), "defaultLanguage": "en", "defaultAudioLanguage": "en", }, "status": { "privacyStatus": privacy_status or settings.youtube_privacy, # Obligatorio declararlo. Sin esto la subida puede quedar en un # limbo de "falta información" que no se ve desde la API. "selfDeclaredMadeForKids": False, "embeddable": True, }, } # -------------------------------------------------------------------------- # Cliente # -------------------------------------------------------------------------- class YouTubeUploader: def __init__(self, client_id: Optional[str] = None, client_secret: Optional[str] = None, refresh_token: Optional[str] = None, timeout: Optional[float] = None): self.client_id = client_id or settings.youtube_client_id or "" self.client_secret = client_secret or settings.youtube_client_secret or "" self.refresh_token = refresh_token or settings.youtube_refresh_token or "" self.timeout = timeout or settings.youtube_timeout def is_configured(self) -> bool: return bool(self.client_id and self.client_secret and self.refresh_token) def _session(self, total: float) -> aiohttp.ClientSession: return aiohttp.ClientSession( timeout=aiohttp.ClientTimeout(total=total), # Nunca heredar el default de aiohttp (KNOWN-ISSUES.md, 2026-07-04). headers={"Accept-Encoding": SAFE_ACCEPT_ENCODING}, ) # --- auth -------------------------------------------------------------- async def access_token(self, force: bool = False) -> str: """Un token de acceso vivo, refrescando sólo cuando hace falta.""" if not self.is_configured(): raise YouTubeNotConfigured( "Faltan YOUTUBE_CLIENT_ID / YOUTUBE_CLIENT_SECRET / " "YOUTUBE_REFRESH_TOKEN. Sácalos con scripts/youtube_oauth.py.") cached = _token_cache.get(self.client_id) if cached and not force and cached[1] > time.time() + _TOKEN_MARGIN: return cached[0] payload = { "client_id": self.client_id, "client_secret": self.client_secret, "refresh_token": self.refresh_token, "grant_type": "refresh_token", } try: async with self._session(30) as sess: async with sess.post(TOKEN_URL, data=payload) as resp: body = await resp.text() if resp.status != 200: raise _auth_error(resp.status, body) data = json.loads(body) except aiohttp.ClientError as e: raise YouTubeError(f"no se pudo hablar con el token endpoint: {e}") from e token = data.get("access_token") if not token: raise YouTubeAuthError(f"respuesta de token sin access_token: {body[:200]}") expiry = time.time() + float(data.get("expires_in", 3600)) _token_cache[self.client_id] = (token, expiry) logger.info("Token de YouTube refrescado", expires_in=data.get("expires_in")) return token # --- subida ------------------------------------------------------------ async def upload(self, video_path: str | Path, metadata: dict, on_progress: Optional[Callable[[str], Any]] = None ) -> UploadedVideo: """Sube el fichero y devuelve el vídeo creado. Resumable en dos pasos aunque un Short quepa de sobra en una petición: es el camino documentado para vídeo, separa el rechazo de los metadatos (falla en el paso 1, barato) del de los bytes, y deja la puerta abierta a reanudar si algún día los ficheros crecen. """ if not settings.youtube_enabled: raise YouTubeDisabled( "YOUTUBE_ENABLED=false — la subida está apagada a propósito") path = Path(video_path) if not path.exists(): raise YouTubeError(f"no existe el vídeo: {path}") size = path.stat().st_size if size == 0: raise YouTubeError(f"el vídeo está vacío: {path}") await _report(on_progress, "🔑 Autenticando…") token = await self.access_token() await _report(on_progress, "📡 Abriendo sesión de subida…") location = await self._start(token, metadata, size) await _report(on_progress, f"⬆️ Subiendo {size / 1_048_576:.1f} MB…") video = await self._put(location, path, size) requested = (metadata.get("status") or {}).get("privacyStatus", "private") status = video.get("status") or {} actual = status.get("privacyStatus", requested) result = UploadedVideo( video_id=video.get("id", ""), title=((video.get("snippet") or {}).get("title") or (metadata.get("snippet") or {}).get("title", "")), privacy_status=actual, forced_private=(requested != "private" and actual == "private"), upload_status=status.get("uploadStatus", ""), rejection_reason=(status.get("rejectionReason") or status.get("failureReason") or ""), ) logger.info("Short subido a YouTube", video_id=result.video_id, privacy=result.privacy_status, forced_private=result.forced_private) return result async def _start(self, token: str, metadata: dict, size: int) -> str: """Paso 1: los metadatos. Devuelve la URL de subida (cabecera Location).""" headers = { "Authorization": f"Bearer {token}", "Content-Type": "application/json; charset=UTF-8", "X-Upload-Content-Length": str(size), "X-Upload-Content-Type": "video/mp4", } params = {"uploadType": "resumable", "part": "snippet,status"} try: async with self._session(60) as sess: async with sess.post(UPLOAD_URL, params=params, headers=headers, json=metadata) as resp: if resp.status not in (200, 201): raise _api_error(resp.status, await resp.text()) location = resp.headers.get("Location") except aiohttp.ClientError as e: raise YouTubeError(f"no se pudo abrir la subida: {e}") from e if not location: raise YouTubeError( "YouTube aceptó los metadatos pero no devolvió Location: " "sin esa URL no hay dónde mandar los bytes") return location async def _put(self, location: str, path: Path, size: int) -> dict: """Paso 2: los bytes, de una vez. Un Short son pocos MB.""" headers = {"Content-Type": "video/mp4", "Content-Length": str(size)} try: with path.open("rb") as handle: async with self._session(self.timeout) as sess: async with sess.put(location, data=handle, headers=headers) as resp: body = await resp.text() if resp.status not in (200, 201): raise _api_error(resp.status, body) except aiohttp.ClientError as e: raise YouTubeError(f"la subida se cortó: {e}") from e try: return json.loads(body) except ValueError as e: raise YouTubeError( f"YouTube aceptó el fichero pero devolvió algo que no es JSON: " f"{body[:200]}") from e # -------------------------------------------------------------------------- def _google_error(body: str) -> tuple[str, str]: """(mensaje, reason) del cuerpo de error de Google, que anida el motivo.""" try: error = (json.loads(body) or {}).get("error") or {} except ValueError: return body[:300], "" if isinstance(error, str): # el token endpoint usa el formato OAuth plano return error, error message = error.get("message") or "" reasons = error.get("errors") or [] reason = reasons[0].get("reason", "") if reasons else "" return (message or body[:300]), reason def _auth_error(status: int, body: str) -> YouTubeError: """El fallo del refresco, traducido a algo accionable. `invalid_grant` es casi siempre lo mismo y casi nunca es obvio: la pantalla de consentimiento se quedó en "Testing", y Google revoca los refresh tokens de apps sin publicar a los 7 días. Decirlo aquí ahorra la tarde de buscarlo. """ message, reason = _google_error(body) if "invalid_grant" in (message + reason + body).lower(): return YouTubeAuthError( "El refresh token ya no vale (invalid_grant). La causa habitual es " "que la pantalla de consentimiento de OAuth siga en «Testing»: " "Google revoca esos tokens a los 7 días. Pásala a «In production» " "en la consola de Google Cloud y vuelve a sacar el token con " "scripts/youtube_oauth.py.") return YouTubeAuthError(f"refresco rechazado ({status}): {message}") def _api_error(status: int, body: str) -> YouTubeError: message, reason = _google_error(body) if status == 401: return YouTubeAuthError(f"token no aceptado (401): {message}") if status == 403 and reason in ("quotaExceeded", "uploadLimitExceeded", "rateLimitExceeded"): return YouTubeQuotaExceeded( f"cuota agotada ({reason}): {message}. Se reinicia a medianoche " f"hora del Pacífico.") if status == 403: return YouTubeRejected( f"YouTube denegó la subida ({reason or 403}): {message}", reason) if status == 400: return YouTubeRejected(f"metadatos rechazados: {message}", reason) return YouTubeError(f"YouTube devolvió {status}: {message}") async def _report(callback: Optional[Callable[[str], Any]], text: str) -> None: if not callback: return try: value = callback(text) if hasattr(value, "__await__"): await value except Exception as e: logger.warning("Progreso de subida no enviado", error=str(e))