SidneyZhang/StudioLift
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

feat(scripts): 添加 PEP 723 脚本元数据并修复 HTTP/2 伪头过滤问题

Browse Source 
- 为 build_studio_urls.py 和 lookup_groups.py 添加 PEP 723 依赖声明
- 修复 HTTP/2 伪头导致 requests 抛 InvalidHeader 的问题
- 添加 CDP 端口连通性预检查及启动指引
This commit is contained in:
Sidney Zhang
2026-08-27 13:33:34 +08:00
parent b8bd749f43
commit 9020752eed
23 changed files with 885 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files

245
scripts/youtube_export_interceptor.py
View File

@@ -1,60 +1,49 @@
# -*- coding: utf-8 -*-
# /// script
# requires-python = ">=3.10"
# dependencies = [
# "playwright>=1.40",
# ]
# ///
r"""
YouTube Studio 内容管理器「导出当前视图 → 逗号分隔值 (.csv)」下载流程 —— 拦截/解码/自动保存/重名去重 参考实现。
YouTube Studio 内容管理器「导出当前视图 → 逗号分隔值 (.csv)」下载脚本
（拦截响应 + 解码 zip + 完整性校验 + 自动保存到下载目录 + 重名去重）。
## 已实证的下载生成机制（2026-08-21 抓包确认）
机制（已实证）：前端点击导出后向后端
POST https://studio.youtube.com/youtubei/v1/yta_web/csv_export?alt=json
后端把打好的 zip 以 base64 内联在响应 `zippedData` 字段里（开头 `UEsDBBQ` 即 ZIP 文件头 `PK`）。
注意两个字段细节（均为实证结论）：
- `zippedData` 是 **URL-safe base64**（-/_ 代替 +//，且省略 = 填充）；
- 当前界面导出入口是右上角**下载图标按钮**：只有 `aria-label="导出当前视图"`，没有可见文本。
本脚本拦截该响应，base64 解码并校验 zip 完整性后按 `<维度标签> <起始日>_<结束日> <账号名>.zip`
写盘，重名自动加 ` (n)` 后缀，n 从 1 起；未捕获到有效响应时自动重试一次触发导出。
1. 前端点击「导出当前视图 → 逗号分隔值 (.csv)」后，向后端发起：
POST https://studio.youtube.com/youtubei/v1/yta_web/csv_export?alt=json
请求体 `exportQuery` 内含 joinRequest 各节点（表格数据/图表数据/总计），以及
日期范围 `dateIdRange.inclusiveStart` / `dateIdRange.exclusiveEnd`（都是 YYYYMMDD）。
登录态：脚本不负责登录，必须复用已登录 YouTube Studio 的浏览器会话，二选一：
- --user-data-dir + --channel chrome|msedge ：用已登录的用户数据目录启动（需先关闭该浏览器）
- --connect http://localhost:9222 ：附加到已在调试端口运行的浏览器
2. 后端**不在服务器上生成一个可下载的 URL**，而是直接把打好的 zip 以
**base64 字符串**内联在响应里：
{ "responseContext": {...}, "zippedData": "<base64>" }
实测 `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 中产物一致）
<维度标签> <inclusiveStart>_<exclusiveEnd> <账号名>.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 "<explore URL>" --channel chrome ^
--user-data-dir "%LOCALAPPDATA%\Google\Chrome\User Data"
方式 B（附加到已在调试端口运行的浏览器，无需关闭）：
python youtube_export_interceptor.py --url "<explore URL>" --connect http://localhost:9222
# 先启动： chrome.exe --remote-debugging-port=9222 或 msedge.exe --remote-debugging-port=9222
用法：
python youtube_export_download.py --selftest
python youtube_export_download.py --url "<explore URL>" --channel chrome --user-data-dir "C:/Users/<you>/AppData/Local/Google/Chrome/User Data"
python youtube_export_download.py --url "<explore URL>" --connect http://localhost:9222
"""
import argparse
import base64
import io
import json
import os
import re
import sys
import time
import zipfile
# 本机 Windows 下载目录。可改为 os.path.expanduser("~") / "Downloads"。
DOWNLOAD_DIR = r"D:\Downloads"
DOWNLOAD_DIR = r"D:\Downloads" # 默认下载目录，可用 --download-dir 覆盖
# 维度类型 -> 文件名前缀标签（生产环境请从页面「维度」按钮文本读取，这里兜底映射）。
EXPORT_RESPONSE_TIMEOUT = 30 # 每次触发导出后等待 csv_export 响应的秒数
MAX_EXPORT_ATTEMPTS = 2 # 未捕获到有效响应时的最大触发次数（1 次重试）
# 维度类型 -> 文件名前缀标签（生产环境建议从页面「维度」按钮文本读取，这里兜底映射）。
DIMENSION_LABEL = {
"VIDEO": "内容",
"USER": "频道",
@@ -64,15 +53,8 @@ DIMENSION_LABEL = {
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）。
"""
"""返回不冲突的落盘路径；重名按 `名称 (n).后缀` 递增，n 从 1 起。"""
directory = os.path.abspath(directory)
base, ext = os.path.splitext(filename)
candidate = os.path.join(directory, filename)
@@ -84,20 +66,16 @@ def dedup_path(directory, filename):
def build_export_filename(export_query, account_name, dimension_label=None):
"""从 csv_export 请求体 `exportQuery` 反推导出文件名。
与实测产物命名一致：`<维度标签> <inclusiveStart>_<exclusiveEnd> <账号名>.zip`
"""
"""从 csv_export 请求体 `exportQuery` 反推文件名。"""
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 [])
nodes = export_query.get("joinRequest", {}).get("nodes") or []
for node in nodes:
q = (node.get("value", {}).get("query") or {})
q = node.get("value", {}).get("query") or {}
if not date_range:
tr = q.get("timeRange", {}).get("dateIdRange")
if tr and tr.get("inclusiveStart"):
@@ -117,17 +95,46 @@ def build_export_filename(export_query, account_name, dimension_label=None):
def decode_zipped_data(payload):
"""把 csv_export 响应 payload 里的 zippedData 解码为 zip 字节流。"""
"""把 csv_export 响应 payload 里的 zippedData 解码为 zip 字节流。
该字段是 URL-safe base64（-/_ 代替 +//，且省略 = 填充）：标准 b64decode 遇
-/_ 或非 4 对齐长度会抛 Incorrect padding，或静默解出损坏字节流。先补齐填充，
再用 altchars 兼容 URL-safe 与标准两种字符表；解码后校验 PK 文件头。
"""
zipped = payload.get("zippedData")
if not zipped:
raise ValueError("响应中缺少 zippedData 字段")
return base64.b64decode(zipped)
z = str(zipped).strip()
data = base64.b64decode(z + "=" * (-len(z) % 4), altchars=b"-_")
if not data.startswith(b"PK"):
raise ValueError(
"zippedData 解码结果不是 zip 字节流（缺 PK 文件头），"
"响应可能被网关改写或导出接口已变动")
return data
# --------------------------------------------------------------------------- #
# Playwright 主流程（拦截响应 -> 解码 -> 去重 -> 落盘） #
# --------------------------------------------------------------------------- #
def intercept_and_save(page, account_name):
def verify_zip(data):
"""校验内存 zip 完整性：结构可读、成员 CRC 全过；否则抛 ValueError。"""
try:
with zipfile.ZipFile(io.BytesIO(data)) as zf:
bad = zf.testzip()
except zipfile.BadZipFile as e:
raise ValueError(f"zip 结构损坏: {e}") from e
if bad is not None:
raise ValueError(f"zip 成员 CRC 校验失败: {bad}")
def trigger_export(page):
"""点「导出当前视图」→「逗号分隔值 (.csv)」。
导出入口是右上角的下载图标按钮：只有 aria-label、无可见文本，
优先按 label 定位（get_by_text 兜底旧版有可见文本的界面）。
"""
page.get_by_label("导出当前视图").or_(page.get_by_text("导出当前视图")).first.click()
page.get_by_text("逗号分隔值 (.csv)").click()
def intercept_and_save(page, account_name, download_dir):
"""给 page 绑定 response 拦截器：命中 csv_export 就把 zip 保存到下载目录。"""
import pathlib
@@ -139,18 +146,17 @@ def intercept_and_save(page, account_name):
try:
payload = response.json()
data = decode_zipped_data(payload)
verify_zip(data)
# 反推文件名：请求体在 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 = "export.zip" # 反推失败兜底，避免丢内容
filename = re.sub(r"[\\/:*?\"<>|]", "_", filename) # Windows 非法字符
path = dedup_path(DOWNLOAD_DIR, filename)
path = dedup_path(download_dir, filename)
pathlib.Path(path).write_bytes(data)
saved.append((filename, len(data), path))
except Exception as e: # noqa: BLE001
@@ -168,29 +174,25 @@ def _default_user_data_dir(channel):
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：用已登录的用户数据目录启动持久化上下文（需先关闭该浏览器）
- 两者都不传： 新建空白会话（大概率未登录，仅作占位/调试）
"""
def run(url, user_data_dir=None, channel=None, cdp_url=None,
download_dir=None, account_name=None):
"""启动/连接浏览器并触发导出。必须复用已登录会话，否则跳 Google 登录页。"""
from playwright.sync_api import sync_playwright
download_dir = download_dir or DOWNLOAD_DIR
os.makedirs(download_dir, exist_ok=True)
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,
@@ -206,27 +208,34 @@ def run(url, user_data_dir=None, channel=None, cdp_url=None):
page = context.new_page()
page.goto(url, wait_until="domcontentloaded")
if "studio.youtube.com" not in page.url and "accounts.google" in page.url:
if "accounts.google" in page.url:
print("[!] 当前会话未登录，已跳转到 Google 登录页。", file=sys.stderr)
print(" 请用 --user-data-dir 复用已登录浏览器，或先手动登录后再试。",
print(" 请用 --user-data-dir 或 --connect 复用已登录浏览器后重试。",
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
if not account_name:
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)
saved = intercept_and_save(page, account_name, download_dir)
# 触发导出：点「导出当前视图」→「逗号分隔值 (.csv)」
page.get_by_text("导出当前视图").click()
page.get_by_text("逗号分隔值 (.csv)").click()
# 触发导出并等待响应；未捕获到有效 zip（未触发/响应无效）自动重试一次
for attempt in range(1, MAX_EXPORT_ATTEMPTS + 1):
trigger_export(page)
deadline = time.time() + EXPORT_RESPONSE_TIMEOUT
while not saved and time.time() < deadline:
page.wait_for_timeout(500)
if saved:
break
if attempt < MAX_EXPORT_ATTEMPTS:
print(f"[重试] 第 {attempt} 次未捕获到有效导出响应，自动重试……",
file=sys.stderr)
page.wait_for_timeout(3000)
if not saved:
print("未捕获到 csv_export 响应，请确认已点击导出且登录态有效。")
else:
@@ -239,15 +248,11 @@ def run(url, user_data_dir=None, channel=None, cdp_url=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")
@@ -260,14 +265,9 @@ def selftest():
assert p2.name == "需求文件 (2).zip", p2.name
p2.write_bytes(b"c")
# 2) 若 (1) 已存在，也应跳到 (2)
assert Path(dedup_path(td, "需求文件.zip")).name == "需求文件 (3).zip"
assert Path(dedup_path(td, "其他.zip")).name == "其他.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"}],
@@ -278,7 +278,25 @@ def selftest():
name = build_export_filename(export_query, "WL Media")
assert name == "内容 2026-07-23_2026-08-20 WL Media.zip", name
print("selftest OK：去重与文件名反推逻辑全部通过")
# zip 解码：标准 base64 与 URL-safe 无填充变体都要解出同一字节流
buf = io.BytesIO()
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
zf.writestr("a.csv", "x,y\n1,2")
raw = buf.getvalue()
std = base64.b64encode(raw).decode("ascii")
urlsafe_nopad = std.replace("+", "-").replace("/", "_").rstrip("=")
assert decode_zipped_data({"zippedData": std}) == raw
assert decode_zipped_data({"zippedData": urlsafe_nopad}) == raw
verify_zip(raw)
# 非 zip 字节流要报 PK 文件头错误，而不是静默落盘损坏文件
try:
decode_zipped_data({"zippedData": base64.b64encode(b"not a zip").decode()})
raise AssertionError("非 zip 数据应抛 ValueError")
except ValueError as e:
assert "PK" in str(e), e
print("selftest OK：去重、文件名反推、zip 解码与完整性校验全部通过")
if __name__ == "__main__":
@@ -287,20 +305,23 @@ if __name__ == "__main__":
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")
help="浏览器品牌，复用登录态时必填其一")
ap.add_argument("--connect", help="通过 CDP 附加到已打开浏览器，如 http://localhost:9222")
ap.add_argument("--download-dir", help=f"下载目录，默认 {DOWNLOAD_DIR}")
ap.add_argument("--account-name", help="账号名（用于文件名），默认从页面读取")
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)
run(args.url, user_data_dir=args.user_data_dir, channel=args.channel,
cdp_url=args.connect, download_dir=args.download_dir,
account_name=args.account_name)
else:
selftest()
print("\n实际运行请先: pip install playwright\n"
"复用登录态（必选其一，详见脚本顶部 docstring）：\n"
" 方式 A: python youtube_export_interceptor.py --url \"<explore URL>\" "
"--channel chrome --user-data-dir \"%LOCALAPPDATA%\\Google\\Chrome\\User Data\"\n"
" 方式 B: python youtube_export_interceptor.py --url \"<explore URL>\" "
print("\n实际运行需先准备依赖环境（uv sync 或直接 uv run，见 uv-env-setup 技能），"
"再复用登录态（二选一）：\n"
" 方式 A: python youtube_export_download.py --url \"<explore URL>\" "
"--channel chrome --user-data-dir <你的 Chrome User Data 目录>\n"
" 方式 B: python youtube_export_download.py --url \"<explore URL>\" "
"--connect http://localhost:9222")