{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())