ChemaVXandClaude Opus 5
77029fa894
Build & Deploy ResearchOwl / build-and-push (push) Successful in 10s
`/upload_short` contaba lo que la respuesta de la subida decía del vídeo. Eso es la palabra de la API sobre sí misma, y todo el flujo de revisión descansa en ella: el informe de fundamento se lee ANTES de publicar sólo si subir no publica. El 2026-08-12, mirando un vídeo recién subido, la respuesta decía `privacyStatus: private` y el vídeo se veía sin sesión — resultó ser un clic humano en Studio y no un fallo, pero el episodio dejó claro que no había forma de distinguir un caso del otro. Ahora se contrasta: oEmbed contesta 200 a un vídeo que se ve sin sesión y 404 a uno que no. Sin credenciales, sin tocar el scope — `youtube.upload` no puede preguntar por el estado de un vídeo, y ampliarlo a uno que sí pueda significa darle a un token de subida permiso para vaciar el canal. Dos decisiones que van con esto: - **Sólo el 200 es una prueba.** Un 404 no demuestra que el vídeo sea privado: también lo devuelve uno que YouTube aún no ha indexado. Por eso el negativo se mira dos veces y, si sigue negativo, se cuenta como "no se ve desde fuera", no como "es privado". - **No haber podido comprobar no es haber comprobado que no.** Un fallo de red deja `reachable=None` y el parte lo dice, en vez de heredar la garantía que no tiene. Cuando la API dice privado y el vídeo se ve, el aviso va en la PRIMERA línea del mensaje de Telegram: enterarse tiene que costar cero atención. Verificado contra la realidad — el mismo vídeo daba True antes de ocultarlo y False después; un vídeo borrado y uno privado dan False, y uno público True. Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
521 lines
21 KiB
Python
521 lines
21 KiB
Python
"""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"
|
||
|
||
#: Con `youtube.upload` no se le puede PREGUNTAR a la API por el estado de un
|
||
#: vídeo, así que la visibilidad se comprueba desde fuera y sin credenciales:
|
||
#: oEmbed contesta 200 a un vídeo que se ve sin sesión y 401/404 a uno que no.
|
||
#: Es la única forma de contrastar lo que dice la respuesta de la subida sin
|
||
#: cambiar un token que sólo sabe subir por uno que puede vaciar el canal.
|
||
OEMBED_URL = "https://www.youtube.com/oembed"
|
||
#: Segunda pasada por si YouTube aún no había indexado el vídeo recién subido.
|
||
_VISIBILITY_RECHECK_DELAY = 3.0
|
||
|
||
#: 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 = ""
|
||
#: Si el vídeo se ve sin iniciar sesión, comprobado desde fuera en vez de
|
||
#: creerle a la respuesta de la subida. None = no se pudo comprobar.
|
||
reachable: Optional[bool] = None
|
||
|
||
@property
|
||
def visibility_contradiction(self) -> bool:
|
||
"""La API dice privado y el vídeo se ve. Es el caso que hay que gritar:
|
||
todo el flujo de revisión — informe de fundamento primero, publicar
|
||
después — descansa en que subir NO publica."""
|
||
return self.reachable is True and self.privacy_status == "private"
|
||
|
||
@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 ""),
|
||
)
|
||
await _report(on_progress, "🔎 Comprobando la visibilidad…")
|
||
result.reachable = await self.reachable(result.video_id)
|
||
|
||
logger.info("Short subido a YouTube", video_id=result.video_id,
|
||
privacy=result.privacy_status,
|
||
forced_private=result.forced_private,
|
||
reachable=result.reachable)
|
||
if result.visibility_contradiction:
|
||
logger.error("El vídeo se ve sin sesión y la API lo dio por privado",
|
||
video_id=result.video_id)
|
||
return result
|
||
|
||
async def reachable(self, video_id: str) -> Optional[bool]:
|
||
"""¿Se ve este vídeo sin haber iniciado sesión?
|
||
|
||
True = cualquiera con el enlace lo ve. False = no. None = no se pudo
|
||
averiguar.
|
||
|
||
**La asimetría es deliberada.** Un 200 PRUEBA que el vídeo es accesible;
|
||
un 404 no prueba que sea privado, porque también lo devuelve un vídeo
|
||
que YouTube todavía no ha terminado de indexar segundos después de
|
||
subirlo. Por eso sólo el 200 dispara un aviso, y por eso el negativo se
|
||
reintenta una vez antes de darlo por bueno.
|
||
|
||
Nunca levanta: esto contrasta un dato, no lo produce. Si la red falla, el
|
||
vídeo ya está subido y lo que toca es decir que no se pudo comprobar —
|
||
no convertir una comprobación en el motivo de que la subida parezca
|
||
haber fallado.
|
||
"""
|
||
if not video_id:
|
||
return None
|
||
params = {"url": f"https://www.youtube.com/watch?v={video_id}",
|
||
"format": "json"}
|
||
seen: Optional[bool] = None
|
||
for attempt in (1, 2):
|
||
try:
|
||
async with self._session(20) as sess:
|
||
async with sess.get(OEMBED_URL, params=params) as resp:
|
||
status = resp.status
|
||
except (aiohttp.ClientError, OSError) as e:
|
||
logger.warning("No se pudo comprobar la visibilidad",
|
||
video_id=video_id, error=str(e))
|
||
return seen
|
||
if status == 200:
|
||
return True
|
||
if status in (401, 403, 404):
|
||
seen = False
|
||
else:
|
||
logger.warning("oEmbed contestó algo inesperado",
|
||
video_id=video_id, status=status)
|
||
return seen
|
||
if attempt == 1:
|
||
import asyncio
|
||
await asyncio.sleep(_VISIBILITY_RECHECK_DELAY)
|
||
return seen
|
||
|
||
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))
|