#!/usr/bin/env python3 """Connect your own Chrome login to the download site (Python 3.10+). Install: python -m pip install playwright Run: python youtube_connect.py Only youtube.com cookies are sent, after you finish signing in yourself. The site's single-use connection code identifies the receiving user. """ import argparse import getpass import json import os from pathlib import Path import re import shutil import socket import subprocess import sys import tempfile import time from urllib.error import HTTPError, URLError from urllib.parse import urlsplit from urllib.request import HTTPRedirectHandler, Request, build_opener DEFAULT_SITE = "https://amusement.info/video-downloader" def find_chrome(): candidates = [shutil.which(name) for name in ("google-chrome", "google-chrome-stable", "chromium", "chrome")] candidates += ["/Applications/Google Chrome.app/Contents/MacOS/Google Chrome"] for name in ("LOCALAPPDATA", "PROGRAMFILES", "PROGRAMFILES(X86)"): if os.environ.get(name): candidates.append(str(Path(os.environ[name]) / "Google/Chrome/Application/chrome.exe")) for candidate in candidates: if candidate and Path(candidate).is_file(): return candidate raise RuntimeError("未找到 Chrome，请安装 Google Chrome，或使用 --chrome 指定其可执行文件。") def browser_cookies(rows): lines = ["# Netscape HTTP Cookie File"] for cookie in rows: domain = cookie["domain"] host = domain.lstrip(".").lower() if host != "youtube.com" and not host.endswith(".youtube.com"): continue fields = [("#HttpOnly_" if cookie.get("httpOnly") else "") + domain, "TRUE" if domain.startswith(".") else "FALSE", cookie.get("path", "/"), "TRUE" if cookie.get("secure") else "FALSE", str(max(0, int(cookie.get("expires", 0)))), cookie["name"], cookie["value"]] if any(re.search(r"[\x00-\x1f\x7f]", value) for value in fields): raise RuntimeError("浏览器 Cookie 格式异常，请使用网页导入入口重试。") lines.append("\t".join(fields)) return "\n".join(lines) + "\n" def validate_site(value): url = urlsplit(value) if url.scheme != "https" or not url.hostname or url.username or url.password or url.query or url.fragment: raise RuntimeError("连接地址必须是 HTTPS 网站地址，不能包含密码、参数或片段。") return value.rstrip("/") class NoRedirect(HTTPRedirectHandler): def redirect_request(self, *args, **kwargs): return None def upload(site, code, cookies): request = Request(site + "/api/youtube/session/pair/complete", data=json.dumps({"code": code, "cookies": cookies}).encode(), headers={"Content-Type": "application/json"}, method="POST") # Never forward credentials across a redirect to another endpoint. try: with build_opener(NoRedirect()).open(request, timeout=30) as response: result = json.loads(response.read(4096)) return result.get("resumed_count", 0) except HTTPError as error: if error.code == 400: raise RuntimeError("会话或连接码未被接受，请检查是否已登录，并在网页生成新的连接码。") from None raise RuntimeError("网站暂时无法接收会话，请稍后重试。") from None except (URLError, ValueError, OSError): raise RuntimeError("连接网站失败，请检查网络后重试；网页也支持导入 Cookie。") from None def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--site", default=DEFAULT_SITE, help="接收会话的 HTTPS 站点") parser.add_argument("--chrome", help="本机 Chrome 的可执行文件") args = parser.parse_args() site = validate_site(args.site) try: from playwright.sync_api import sync_playwright except ImportError: raise RuntimeError("请先运行 python -m pip install playwright，再运行本工具。") from None chrome = args.chrome or find_chrome() print("会话将保存到：" + site) print("请在该网站的「YouTube 登录会话 → 连接浏览器」中生成连接码（20 分钟有效）。") code = getpass.getpass("粘贴连接码后按回车（不显示输入）：").strip() if not re.fullmatch(r"[A-Za-z0-9_-]{43}", code): raise RuntimeError("连接码格式不正确，请完整复制网页生成的连接码。") with socket.socket() as sock: sock.bind(("127.0.0.1", 0)) port = sock.getsockname()[1] cookies = "" with tempfile.TemporaryDirectory(prefix="youtube-connect-") as profile: process = subprocess.Popen([chrome, "--user-data-dir=" + profile, "--remote-debugging-address=127.0.0.1", "--remote-debugging-port=" + str(port), "--no-first-run", "--no-default-browser-check", "--new-window", "https://www.youtube.com/"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL) try: print("请在新打开的 Chrome 中手动登录 YouTube，完成 Google 要求的验证，并返回 YouTube 首页。") print("登录窗口使用单独的临时资料，不读取你日常浏览器的账号。不要关闭此窗口。") input("看到自己的账号头像后，回到这里按回车：") if process.poll() is not None: raise RuntimeError("登录窗口已关闭或未能启动，请重新运行工具。") # Attach only after the user finishes normal, interactive login. with sync_playwright() as playwright: browser = playwright.chromium.connect_over_cdp(f"http://127.0.0.1:{port}", timeout=15000) context = browser.contexts[0] pages = [page for page in context.pages if urlsplit(page.url).hostname == "www.youtube.com"] logged_in = any(page.evaluate("() => window.ytcfg?.get('LOGGED_IN') === true") for page in pages) if not logged_in: raise RuntimeError("尚未确认 YouTube 登录。请完成登录后重试；Google 拦截登录时可改用网页导入 Cookie。") # Stop active YouTube pages before exporting, then close this # temporary profile to avoid continued browser cookie rotation. for page in pages: page.goto("https://www.youtube.com/robots.txt", wait_until="domcontentloaded", timeout=30000) cookies = browser_cookies(context.cookies(["https://www.youtube.com/"])) browser.new_browser_cdp_session().send("Browser.close") finally: try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.terminate() try: process.wait(timeout=5) except subprocess.TimeoutExpired: process.kill() process.wait(timeout=5) resumed = upload(site, code, cookies) cookies = code = "" print(f"会话已保存；已恢复 {resumed} 条等待验证的任务。请回到网页查看状态。") print("保存会话不代表所有视频均可下载；如站点再次要求验证，请重新连接。") if __name__ == "__main__": try: main() except (KeyboardInterrupt, EOFError): print("\n已取消连接。") sys.exit(1) except RuntimeError as error: print(str(error), file=sys.stderr) sys.exit(1) except Exception: # Browser/network exceptions can contain credentials or temporary URLs. print("连接未完成，请检查 Chrome 和网络后重试，或使用网页的 Cookie 导入入口。", file=sys.stderr) sys.exit(1)