ChemaVXandClaude Fable 5
960990327a
Build & Deploy ResearchOwl / build-and-push (push) Successful in 10s
Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
223 lines
9.1 KiB
Python
223 lines
9.1 KiB
Python
#!/usr/bin/env python3
|
|
"""Saca el refresh token de YouTube. Se ejecuta UNA vez, en tu máquina.
|
|
|
|
python scripts/youtube_oauth.py --client-id XXX --client-secret YYY
|
|
|
|
Abre el navegador, te pide permiso para subir vídeos a tu canal, y escupe el
|
|
refresh token para meterlo en Infisical como `youtube-refresh-token`.
|
|
|
|
Con `--paste` no levanta servidor local: da permiso desde CUALQUIER dispositivo
|
|
(un iPad vale), deja que la pestaña de redirección falle al cargar — no hay
|
|
nadie escuchando, es lo esperado — y pega aquí la URL completa de la barra de
|
|
direcciones, que lleva el `code=` dentro. También se puede canalizar por stdin:
|
|
|
|
echo 'http://localhost:8765/?code=4/0Ab...' | \\
|
|
python scripts/youtube_oauth.py --paste --client-id XXX --client-secret YYY
|
|
|
|
Sólo stdlib: esto corre fuera del contenedor, en el portátil de quien lo lance,
|
|
y no debería exigir instalar nada.
|
|
|
|
--------------------------------------------------------------------------
|
|
Antes de ejecutarlo, en https://console.cloud.google.com:
|
|
|
|
1. Crea un proyecto (o usa uno) y habilita **YouTube Data API v3**.
|
|
2. Pantalla de consentimiento OAuth → External.
|
|
3. **Pásala a «In production»**, no la dejes en «Testing». En Testing, Google
|
|
revoca los refresh tokens a los SIETE DÍAS y el bot se cae solo el martes que
|
|
viene. Al publicarla verás un aviso de "app no verificada" al dar permiso;
|
|
con «Advanced → continuar» basta, porque el único usuario eres tú.
|
|
4. Credenciales → Crear → ID de cliente OAuth → tipo **Aplicación de escritorio**.
|
|
Ese tipo acepta redirecciones a localhost en cualquier puerto, que es lo que
|
|
usa este script.
|
|
|
|
Y lo que conviene saber antes de invertir la tarde: los vídeos subidos por API
|
|
desde un proyecto sin auditar quedan **restringidos a privado**, y el candado es
|
|
del proyecto, no del vídeo. Se levanta pasando la auditoría de cumplimiento de
|
|
Google, no desde Studio. Con esto el bot te deja el vídeo en el canal con los
|
|
metadatos puestos; publicarlo sigue siendo un clic tuyo.
|
|
"""
|
|
import argparse
|
|
import http.server
|
|
import json
|
|
import os
|
|
import socket
|
|
import sys
|
|
import threading
|
|
import urllib.error
|
|
import urllib.parse
|
|
import urllib.request
|
|
import webbrowser
|
|
|
|
AUTH_URL = "https://accounts.google.com/o/oauth2/v2/auth"
|
|
TOKEN_URL = "https://oauth2.googleapis.com/token"
|
|
SCOPE = "https://www.googleapis.com/auth/youtube.upload"
|
|
|
|
# En modo --paste nadie escucha en el puerto, pero el redirect_uri del
|
|
# intercambio tiene que ser IDÉNTICO al de la autorización, así que va fijo.
|
|
PASTE_PORT = 8765
|
|
|
|
_PAGE = """<!doctype html><meta charset="utf-8">
|
|
<title>ResearchOwl</title>
|
|
<body style="font-family:system-ui;max-width:32rem;margin:6rem auto;line-height:1.6">
|
|
<h1>{heading}</h1><p>{body}</p></body>"""
|
|
|
|
|
|
class _Catcher(http.server.BaseHTTPRequestHandler):
|
|
"""Recoge el ?code= de la redirección y para."""
|
|
|
|
result: dict = {}
|
|
|
|
def do_GET(self):
|
|
query = urllib.parse.parse_qs(urllib.parse.urlparse(self.path).query)
|
|
_Catcher.result = {k: v[0] for k, v in query.items()}
|
|
ok = "code" in _Catcher.result
|
|
page = _PAGE.format(
|
|
heading="✅ Listo" if ok else "❌ Algo falló",
|
|
body=("Ya puedes cerrar esta pestaña y volver a la terminal."
|
|
if ok else
|
|
f"Google devolvió: {_Catcher.result.get('error', 'sin código')}"))
|
|
body = page.encode("utf-8")
|
|
self.send_response(200)
|
|
self.send_header("Content-Type", "text/html; charset=utf-8")
|
|
self.send_header("Content-Length", str(len(body)))
|
|
self.end_headers()
|
|
self.wfile.write(body)
|
|
|
|
def log_message(self, *args):
|
|
pass # sin ruido de servidor en la terminal
|
|
|
|
|
|
def _free_port() -> int:
|
|
with socket.socket() as s:
|
|
s.bind(("127.0.0.1", 0))
|
|
return s.getsockname()[1]
|
|
|
|
|
|
def _code_from_paste(text: str) -> str:
|
|
"""Saca el `code` de lo que pegue el usuario: la URL de la redirección
|
|
entera, o el código a secas si fue más listo que el script."""
|
|
text = text.strip().strip('"').strip("'")
|
|
if not text:
|
|
return ""
|
|
if "?" in text or text.startswith(("http", "localhost")):
|
|
query = urllib.parse.urlparse(text).query
|
|
return urllib.parse.parse_qs(query).get("code", [""])[0]
|
|
return text
|
|
|
|
|
|
def _post_form(url: str, data: dict) -> dict:
|
|
payload = urllib.parse.urlencode(data).encode()
|
|
request = urllib.request.Request(
|
|
url, data=payload,
|
|
headers={"Content-Type": "application/x-www-form-urlencoded"})
|
|
try:
|
|
with urllib.request.urlopen(request, timeout=30) as resp:
|
|
return json.loads(resp.read())
|
|
except urllib.error.HTTPError as e:
|
|
detail = e.read().decode("utf-8", "replace")
|
|
raise SystemExit(f"\n❌ Google rechazó el intercambio ({e.code}):\n{detail}")
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(
|
|
description="Obtiene el refresh token de YouTube para ResearchOwl.")
|
|
parser.add_argument("--client-id", default=os.environ.get("YOUTUBE_CLIENT_ID"))
|
|
parser.add_argument("--client-secret",
|
|
default=os.environ.get("YOUTUBE_CLIENT_SECRET"))
|
|
parser.add_argument("--no-browser", action="store_true",
|
|
help="no abrir el navegador, sólo imprimir la URL")
|
|
parser.add_argument("--paste", action="store_true",
|
|
help="sin servidor local: autoriza desde cualquier "
|
|
"dispositivo y pega aquí la URL de la redirección "
|
|
"(la pestaña dará error de conexión; es normal)")
|
|
args = parser.parse_args()
|
|
|
|
if not args.client_id or not args.client_secret:
|
|
parser.error("hacen falta --client-id y --client-secret "
|
|
"(o YOUTUBE_CLIENT_ID / YOUTUBE_CLIENT_SECRET en el entorno)")
|
|
|
|
port = PASTE_PORT if args.paste else _free_port()
|
|
redirect_uri = f"http://localhost:{port}"
|
|
auth_url = f"{AUTH_URL}?" + urllib.parse.urlencode({
|
|
"client_id": args.client_id,
|
|
"redirect_uri": redirect_uri,
|
|
"response_type": "code",
|
|
"scope": SCOPE,
|
|
# Los dos juntos, y no por gusto: sin access_type=offline no hay refresh
|
|
# token, y sin prompt=consent Google deja de mandarlo en cuanto ya diste
|
|
# permiso una vez — el fallo clásico de "me sale null la segunda vez".
|
|
"access_type": "offline",
|
|
"prompt": "consent",
|
|
})
|
|
|
|
if args.paste:
|
|
print(f"\nAbre esto EN CUALQUIER DISPOSITIVO y da permiso (verás un "
|
|
f"aviso de app no verificada; «Advanced» → continuar):\n\n"
|
|
f" {auth_url}\n\n"
|
|
f"La pestaña final fallará al cargar (localhost:{port} no existe "
|
|
f"ahí) — es lo esperado. Copia la URL COMPLETA de la barra de "
|
|
f"direcciones. El código caduca en unos 10 minutos.")
|
|
try:
|
|
pasted = input("\nPega aquí la URL de la redirección: ")
|
|
except EOFError:
|
|
pasted = ""
|
|
code = _code_from_paste(pasted)
|
|
if not code:
|
|
print("\n❌ Ahí no venía ningún `code`. Vuelve a lanzar y pega la "
|
|
"URL entera de la barra de direcciones (empieza por "
|
|
"http://localhost:…).", file=sys.stderr)
|
|
return 1
|
|
else:
|
|
server = http.server.HTTPServer(("127.0.0.1", port), _Catcher)
|
|
waiter = threading.Thread(target=server.handle_request, daemon=True)
|
|
waiter.start()
|
|
|
|
print(f"\nAbre esto y da permiso (verás un aviso de app no verificada; "
|
|
f"«Advanced» → continuar):\n\n {auth_url}\n")
|
|
if not args.no_browser:
|
|
webbrowser.open(auth_url)
|
|
print("Esperando la redirección… (Ctrl-C para abortar)")
|
|
|
|
waiter.join(timeout=300)
|
|
server.server_close()
|
|
if waiter.is_alive():
|
|
print("\n❌ Nadie llegó a la redirección en 5 minutos.",
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
code = _Catcher.result.get("code")
|
|
if not code:
|
|
print(f"\n❌ Sin código. Google dijo: "
|
|
f"{_Catcher.result.get('error', '(nada)')}", file=sys.stderr)
|
|
return 1
|
|
|
|
tokens = _post_form(TOKEN_URL, {
|
|
"code": code,
|
|
"client_id": args.client_id,
|
|
"client_secret": args.client_secret,
|
|
"redirect_uri": redirect_uri,
|
|
"grant_type": "authorization_code",
|
|
})
|
|
|
|
refresh = tokens.get("refresh_token")
|
|
if not refresh:
|
|
print("\n❌ Google no devolvió refresh_token. Suele pasar cuando ya "
|
|
"habías dado permiso antes: revoca el acceso en "
|
|
"https://myaccount.google.com/permissions y repite.",
|
|
file=sys.stderr)
|
|
return 1
|
|
|
|
print("\n✅ Refresh token:\n")
|
|
print(f" {refresh}\n")
|
|
print("Mételo en Infisical (proyecto researchowl) como:\n")
|
|
print(" youtube-client-id =", args.client_id)
|
|
print(" youtube-client-secret =", args.client_secret)
|
|
print(" youtube-refresh-token =", refresh)
|
|
print("\nRecuerda: si la pantalla de consentimiento sigue en «Testing», "
|
|
"este token caduca en 7 días.")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|