# -*- coding: utf-8 -*- r""" YouTube Studio 内容管理器「导出当前视图 → 逗号分隔值 (.csv)」下载流程 —— 拦截/解码/自动保存/重名去重 参考实现。 ## 已实证的下载生成机制（2026-08-21 抓包确认） 1. 前端点击「导出当前视图 → 逗号分隔值 (.csv)」后，向后端发起： POST https://studio.youtube.com/youtubei/v1/yta_web/csv_export?alt=json 请求体 `exportQuery` 内含 joinRequest 各节点（表格数据/图表数据/总计），以及 日期范围 `dateIdRange.inclusiveStart` / `dateIdRange.exclusiveEnd`（都是 YYYYMMDD）。 2. 后端**不在服务器上生成一个可下载的 URL**，而是直接把打好的 zip 以 **base64 字符串**内联在响应里： { "responseContext": {...}, "zippedData": "" } 实测 `zippedData` 以 `UEsDBBQ...` 开头（即 `PK\\x03\\x04`，ZIP 魔数）， base64 解码后得到 zip，内含 `表格数据.csv`、`图表数据.csv`、`总计.csv`。 3. 前端把 `zippedData` base64 解码 → Blob → 触发浏览器下载。 由于没有出现指向下载文件的 GET/跳转，判定为「客户端 Blob 下载」而非服务端重定向。 4. 浏览器自身已自动保存到本机默认下载目录（本机为 `D:\\Downloads`），无需"另存为"确认； 且 Chromium 对重名文件会自动追加 ` (1)`、` (2)` …后缀。 ## 结论：可以跨越下载流程 - 在响应层拦截 `csv_export`，拿到 `zippedData`，自行 base64 解码并写盘， 即可完全掌控「保存目录 + 文件名 + 重名去重」，不依赖浏览器的下载管理器和弹窗。 - 或者只用 Chromium 的下载偏好（auto-download + 内置去重）让它自动落盘。 ## 文件名约定（与实测 D:\\Downloads 中产物一致） <维度标签> _ <账号名>.zip 例：内容 2026-07-23_2026-08-20 WL Media.zip - 维度标签：VIDEO -> 内容；USER -> 频道（生产中建议从页面"维度"按钮文本读取） - 日期格式 YYYY-MM-DD：inclusiveStart 与 exclusiveEnd 各取 YYYYMMDD 转 YYYY-MM-DD - 账号名：右上角账号按钮文本 自测（无需 Playwright）： python youtube_export_interceptor.py --selftest 实际运行（需 Playwright；必须复用已登录 YouTube Studio 的浏览器会话，否则跳登录页）： 方式 A（用已登录的用户数据目录启动，需先关闭 Chrome/Edge）： python youtube_export_interceptor.py --url "" --channel chrome ^ --user-data-dir "%LOCALAPPDATA%\Google\Chrome\User Data" 方式 B（附加到已在调试端口运行的浏览器，无需关闭）： python youtube_export_interceptor.py --url "" --connect http://localhost:9222 # 先启动： chrome.exe --remote-debugging-port=9222 或 msedge.exe --remote-debugging-port=9222 """ import argparse import base64 import json import os import re import sys # 本机 Windows 下载目录。可改为 os.path.expanduser("~") / "Downloads"。 DOWNLOAD_DIR = r"D:\Downloads" # 维度类型 -> 文件名前缀标签（生产环境请从页面「维度」按钮文本读取，这里兜底映射）。 DIMENSION_LABEL = { "VIDEO": "内容", "USER": "频道", "CONTENT_OWNER": "内容", } CSV_EXPORT_PATH = "/youtubei/v1/yta_web/csv_export" # --------------------------------------------------------------------------- # # 重名去重：需求文件.zip -> 需求文件 (1).zip -> 需求文件 (2).zip ... # # --------------------------------------------------------------------------- # def dedup_path(directory, filename): """返回不冲突的落盘路径。重名时按 `名称 (n).后缀` 递增，n 从 1 开始。 规则与用户要求一致：第二次同名保存 `xx (1).zip`，第三次 `xx (2).zip`，以此类推； 若 `xx (1).zip` 也已存在，则继续找 `xx (2).zip`（即取最小无冲突的 n）。 """ directory = os.path.abspath(directory) base, ext = os.path.splitext(filename) candidate = os.path.join(directory, filename) n = 1 while os.path.exists(candidate): candidate = os.path.join(directory, f"{base} ({n}){ext}") n += 1 return candidate def build_export_filename(export_query, account_name, dimension_label=None): """从 csv_export 请求体 `exportQuery` 反推导出文件名。 与实测产物命名一致：`<维度标签> _ <账号名>.zip` """ def fmt_dateid(yyyymmdd): s = str(yyyymmdd) return f"{s[0:4]}-{s[4:6]}-{s[6:8]}" # 从任意一个 joinRequest 节点取 dateIdRange date_range = None dimension = None nodes = (export_query.get("joinRequest", {}).get("nodes") or []) for node in nodes: q = (node.get("value", {}).get("query") or {}) if not date_range: tr = q.get("timeRange", {}).get("dateIdRange") if tr and tr.get("inclusiveStart"): date_range = (tr["inclusiveStart"], tr.get("exclusiveEnd")) dims = q.get("dimensions") or [] if dimension is None and dims: dimension = dims[0].get("type") if not date_range: raise ValueError("无法从 exportQuery 解析日期范围") if dimension_label is None: dimension_label = DIMENSION_LABEL.get(dimension or "", "") start, end = date_range return f"{dimension_label} {fmt_dateid(start)}_{fmt_dateid(end)} {account_name}.zip" def decode_zipped_data(payload): """把 csv_export 响应 payload 里的 zippedData 解码为 zip 字节流。""" zipped = payload.get("zippedData") if not zipped: raise ValueError("响应中缺少 zippedData 字段") return base64.b64decode(zipped) # --------------------------------------------------------------------------- # # Playwright 主流程（拦截响应 -> 解码 -> 去重 -> 落盘） # # --------------------------------------------------------------------------- # def intercept_and_save(page, account_name): """给 page 绑定 response 拦截器：命中 csv_export 就把 zip 保存到下载目录。""" import pathlib saved = [] def on_response(response): if CSV_EXPORT_PATH not in response.url: return try: payload = response.json() data = decode_zipped_data(payload) # 反推文件名：请求体在 response.request.post_data 里不总是可读， # 这里从已捕获的 body 兜底；找不到就用时间戳命名，保证不误覆盖。 filename = None try: body = json.loads(response.request.post_data or "{}") filename = build_export_filename(body.get("exportQuery", {}), account_name) except Exception: filename = f"export-{response.request.headers.get('date', '')}.zip" filename = re.sub(r"[\\/:*?\"<>|]", "_", filename) # Windows 非法字符 path = dedup_path(DOWNLOAD_DIR, filename) pathlib.Path(path).write_bytes(data) saved.append((filename, len(data), path)) except Exception as e: # noqa: BLE001 print(f"[interceptor] 处理 csv_export 响应失败: {e}", file=sys.stderr) page.on("response", on_response) return saved def _default_user_data_dir(channel): """返回指定浏览器的默认用户数据目录（Windows），用于复用已登录会话。""" _local = os.environ.get("LOCALAPPDATA") or os.path.expanduser(r"~\AppData\Local") if channel == "msedge": return os.path.join(_local, "Microsoft", "Edge", "User Data") return os.path.join(_local, "Google", "Chrome", "User Data") def run(url, user_data_dir=None, channel=None, cdp_url=None): """启动/连接浏览器并触发导出。 关键：必须复用「已登录 YouTube Studio」的浏览器会话，否则会跳 Google 登录页。 - cdp_url： 附加到已在调试端口运行的浏览器（推荐，无需关闭浏览器） - user_data_dir：用已登录的用户数据目录启动持久化上下文（需先关闭该浏览器） - 两者都不传： 新建空白会话（大概率未登录，仅作占位/调试） """ from playwright.sync_api import sync_playwright with sync_playwright() as p: browser = None context = None if cdp_url: # 方式 B：附加到已打开、已登录的浏览器（先以调试端口启动浏览器） browser = p.chromium.connect_over_cdp(cdp_url) context = browser.contexts[0] if browser.contexts else \ browser.new_context(accept_downloads=True) page = context.new_page() page.goto(url, wait_until="domcontentloaded") elif user_data_dir: # 方式 A：用已登录的用户数据目录启动（cookies 复用；必须先关闭同名浏览器） context = p.chromium.launch_persistent_context( user_data_dir=user_data_dir or _default_user_data_dir(channel), channel=channel, headless=False, accept_downloads=True, args=["--disable-blink-features=AutomationControlled"], ) page = context.new_page() page.goto(url, wait_until="domcontentloaded") else: browser = p.chromium.launch(headless=False, channel=channel) context = browser.new_context(accept_downloads=True) page = context.new_page() page.goto(url, wait_until="domcontentloaded") if "studio.youtube.com" not in page.url and "accounts.google" in page.url: print("[!] 当前会话未登录，已跳转到 Google 登录页。", file=sys.stderr) print(" 请用 --user-data-dir 复用已登录浏览器，或先手动登录后再试。", file=sys.stderr) # 账号名：右上角账号按钮（示例选择器，按实际页面微调） account_name = "WL Media" try: account_name = page.locator( "ytcp-account-item button, .account-switcher button" ).first.inner_text(timeout=5000).strip() or account_name except Exception: pass saved = intercept_and_save(page, account_name) # 触发导出：点「导出当前视图」→「逗号分隔值 (.csv)」 page.get_by_text("导出当前视图").click() page.get_by_text("逗号分隔值 (.csv)").click() page.wait_for_timeout(3000) if not saved: print("未捕获到 csv_export 响应，请确认已点击导出且登录态有效。") else: for name, size, path in saved: print(f"已保存: {path} ({size} bytes)") if browser is not None: browser.close() elif context is not None: context.close() # --------------------------------------------------------------------------- # # 自测 # # --------------------------------------------------------------------------- # def selftest(): import tempfile from pathlib import Path with tempfile.TemporaryDirectory() as td: # 1) 用户示例：需求文件.zip 第二次 -> 需求文件 (1).zip，第三次 -> (2).zip p0 = Path(dedup_path(td, "需求文件.zip")) assert p0.name == "需求文件.zip", p0.name p0.write_bytes(b"a") p1 = Path(dedup_path(td, "需求文件.zip")) assert p1.name == "需求文件 (1).zip", p1.name p1.write_bytes(b"b") p2 = Path(dedup_path(td, "需求文件.zip")) assert p2.name == "需求文件 (2).zip", p2.name p2.write_bytes(b"c") # 2) 若 (1) 已存在，也应跳到 (2) assert Path(dedup_path(td, "需求文件.zip")).name == "需求文件 (3).zip" # 3) 无冲突时不加后缀 p3 = Path(dedup_path(td, "其他.zip")) assert p3.name == "其他.zip", p3.name # 4) 文件名反推（对应实测产物「内容 2026-07-23_2026-08-20 WL Media.zip」） export_query = { "joinRequest": {"nodes": [{"value": {"query": { "dimensions": [{"type": "VIDEO"}], "timeRange": {"dateIdRange": { "inclusiveStart": 20260723, "exclusiveEnd": 20260820}}, }}}]}, } name = build_export_filename(export_query, "WL Media") assert name == "内容 2026-07-23_2026-08-20 WL Media.zip", name print("selftest OK：去重与文件名反推逻辑全部通过") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--selftest", action="store_true", help="仅跑去重/命名自测") ap.add_argument("--url", help="explore URL") ap.add_argument("--user-data-dir", help="浏览器用户数据目录（复用登录态，需先关闭该浏览器）") ap.add_argument("--channel", choices=["chrome", "msedge"], help="浏览器品牌：chrome / msedge（复用登录态时必填其一）") ap.add_argument("--connect", help="通过 CDP 附加到已打开的浏览器，如 http://localhost:9222") args = ap.parse_args() if args.selftest: selftest() elif args.url: run(args.url, user_data_dir=args.user_data_dir, channel=args.channel, cdp_url=args.connect) else: selftest() print("\n实际运行请先: pip install playwright\n" "复用登录态（必选其一，详见脚本顶部 docstring）：\n" " 方式 A: python youtube_export_interceptor.py --url \"\" " "--channel chrome --user-data-dir \"%LOCALAPPDATA%\\Google\\Chrome\\User Data\"\n" " 方式 B: python youtube_export_interceptor.py --url \"\" " "--connect http://localhost:9222")