feat(scripts): 添加批量生成 YouTube Studio 内容管理器 URL 的脚本
This commit is contained in:
268
tests/test_build_studio_urls_cli.py
Normal file
268
tests/test_build_studio_urls_cli.py
Normal file
@@ -0,0 +1,268 @@
|
||||
# -*- coding: utf-8 -*-
|
||||
"""build_studio_urls.py CLI 端到端测试。
|
||||
|
||||
以子进程运行脚本（与真实使用方式一致：uv run python build_studio_urls.py -i ... -o ...），
|
||||
覆盖输入格式（CSV UTF-8-BOM / GBK / XLSX）、成功输出、失败行与退出码约定：
|
||||
|
||||
退出码 0 ：全部行生成成功
|
||||
退出码 1 ：输入文件不存在 / 格式不支持 / 缺必要列
|
||||
退出码 2 ：存在失败行（成功行仍写入输出，失败行逐行打印在 stderr）
|
||||
"""
|
||||
|
||||
import pandas as pd
|
||||
import pytest
|
||||
|
||||
from conftest import ROOT, URL_BUILDER_SCRIPT, run_script
|
||||
|
||||
ASSET_XLSX = ROOT / "assets" / "需求输入示例.xlsx"
|
||||
|
||||
HEADERS = ["所有者名称", "所有者ID", "实体类型", "实体名称", "实体ID", "数据周期", "国家"]
|
||||
|
||||
OUTPUT_COLUMNS = ["所有者名称", "所有者ID", "实体类型", "实体名称", "实体ID",
|
||||
"数据周期", "国家", "国家代码", "开始时间戳", "结束时间戳", "URL"]
|
||||
|
||||
|
||||
def run_builder(args):
|
||||
return run_script(URL_BUILDER_SCRIPT, args)
|
||||
|
||||
|
||||
def read_output(path):
|
||||
return pd.read_csv(path, encoding="utf-8-sig")
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 输入格式
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestInputFormats:
|
||||
def test_csv_utf8_bom(self, tmp_path):
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
",".join(HEADERS) + "\n"
|
||||
"示例所有者,MC123,群组,示例群组,G001,2026.07.01-2026.08.01,美国\n",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert "已生成 1 条 URL" in r.stdout
|
||||
df = read_output(out)
|
||||
assert list(df.columns) == OUTPUT_COLUMNS
|
||||
assert len(df) == 1
|
||||
assert df.loc[0, "国家代码"] == "US"
|
||||
assert df.loc[0, "URL"].startswith("https://studio.youtube.com/owner/MC123/")
|
||||
|
||||
def test_csv_gbk_fallback(self, tmp_path):
|
||||
"""Excel 另存 ANSI/GBK 编码 CSV 也能读。"""
|
||||
src = tmp_path / "需求_gbk.csv"
|
||||
src.write_text(
|
||||
",".join(HEADERS) + "\n"
|
||||
"示例所有者,MC123,群组,示例群组,G001,2026.07.01-2026.08.01,美国\n",
|
||||
encoding="gbk",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
df = read_output(out)
|
||||
assert len(df) == 1
|
||||
assert df.loc[0, "国家代码"] == "US"
|
||||
|
||||
def test_xlsx_input(self, tmp_path):
|
||||
src = tmp_path / "需求.xlsx"
|
||||
pd.DataFrame([
|
||||
{"所有者名称": "示例所有者", "所有者ID": "MC123", "实体类型": "群组",
|
||||
"实体名称": "示例群组", "实体ID": "G001",
|
||||
"数据周期": "2026.07.01-2026.08.01", "国家": "US"},
|
||||
]).to_excel(src, index=False)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
df = read_output(out)
|
||||
assert len(df) == 1
|
||||
assert "ur_values=%27US%27" in df.loc[0, "URL"]
|
||||
|
||||
def test_unsupported_extension_exits_1(self, tmp_path):
|
||||
src = tmp_path / "需求.txt"
|
||||
src.write_text("所有者ID,数据周期\nMC123,2026.07.01-2026.08.01", encoding="utf-8")
|
||||
r = run_builder(["-i", str(src), "-o", str(tmp_path / "out.csv")])
|
||||
assert r.returncode == 1
|
||||
assert "不支持的输入格式" in r.stderr
|
||||
|
||||
def test_missing_file_exits_1(self, tmp_path):
|
||||
r = run_builder(["-i", str(tmp_path / "不存在.csv"), "-o", str(tmp_path / "out.csv")])
|
||||
assert r.returncode == 1
|
||||
assert "输入文件不存在" in r.stderr
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 成功路径与输出内容
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestHappyPath:
|
||||
def test_output_content(self, tmp_path):
|
||||
src = tmp_path / "需求.csv"
|
||||
pd.DataFrame([
|
||||
{"所有者名称": "所有者A", "所有者ID": "MC123", "实体类型": "群组",
|
||||
"实体名称": "群组1", "实体ID": "G001",
|
||||
"数据周期": "2026.07.01-2026.08.01", "国家": "美国,日本"},
|
||||
{"所有者名称": "所有者B", "所有者ID": "MC456", "实体类型": "频道",
|
||||
"实体名称": "频道1", "实体ID": "UCxyz",
|
||||
"数据周期": "2026.8.1-2026.8.31", "国家": ""},
|
||||
]).to_csv(src, index=False, encoding="utf-8-sig")
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
df = read_output(out)
|
||||
assert len(df) == 2
|
||||
|
||||
row0, row1 = df.iloc[0], df.iloc[1]
|
||||
assert row0["国家代码"] == "US,JP"
|
||||
assert "ur_values=%27US%27%7C%27JP%27" in row0["URL"]
|
||||
# 时间戳 = 日界线毫秒（独立于模块重算：锚点 2026-06-15 + 整日偏移）
|
||||
anchor = 1781506800000
|
||||
assert row0["开始时间戳"] == anchor + 16 * 86400000 # 2026-07-01
|
||||
assert row0["结束时间戳"] == anchor + 48 * 86400000 # 2026-08-01 次日
|
||||
# 无国家行：不含国家筛选参数，实体类型为频道
|
||||
assert "ur_values" not in row1["URL"]
|
||||
assert "entity_type=CHANNEL" in row1["URL"]
|
||||
assert "entity_id=UCxyz" in row1["URL"]
|
||||
assert row1["国家代码"] == "" or pd.isna(row1["国家代码"])
|
||||
|
||||
def test_default_output_next_to_input(self, tmp_path):
|
||||
"""不传 -o 时输出到输入同目录 studio_urls_output.csv。"""
|
||||
src = tmp_path / "sub"
|
||||
src.mkdir()
|
||||
inp = src / "需求.csv"
|
||||
inp.write_text(
|
||||
"所有者ID,实体ID,数据周期\nMC123,G001,2026.07.01-2026.08.01",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
r = run_builder(["-i", str(inp)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert (src / "studio_urls_output.csv").exists()
|
||||
assert "已生成 1 条 URL" in r.stdout
|
||||
|
||||
def test_blank_rows_dropped(self, tmp_path):
|
||||
"""全空行不产出 URL、不算失败。"""
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
"所有者ID,实体ID,数据周期\n"
|
||||
"MC123,G001,2026.07.01-2026.08.01\n"
|
||||
",,\n"
|
||||
"MC456,G002,2026.07.01-2026.08.01\n",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert "已生成 2 条 URL" in r.stdout
|
||||
|
||||
def test_custom_countries_json(self, tmp_path):
|
||||
cj = tmp_path / "my_countries.json"
|
||||
cj.write_text('{"梦幻国": "ZZ"}', encoding="utf-8")
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
"所有者ID,实体ID,数据周期,国家\nMC123,G001,2026.07.01-2026.08.01,梦幻国",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out), "--countries", str(cj)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
df = read_output(out)
|
||||
assert df.loc[0, "国家代码"] == "ZZ"
|
||||
assert "ur_values=%27ZZ%27" in df.loc[0, "URL"]
|
||||
|
||||
def test_unknown_columns_warned_but_ok(self, tmp_path):
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
"所有者ID,实体ID,数据周期,备注\nMC123,G001,2026.07.01-2026.08.01,随便写",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
assert "未识别的列" in r.stderr and "备注" in r.stderr
|
||||
assert len(read_output(out)) == 1
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 失败行与退出码
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestErrorRows:
|
||||
def test_error_row_exits_2_and_reports_line(self, tmp_path):
|
||||
"""失败行：退出码 2，stderr 报「第N行」（N 按文件行号，含表头）；成功行照常输出。"""
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
"所有者ID,实体ID,数据周期\n"
|
||||
"MC123,G001,2026.07.01-2026.08.01\n" # 第2行 成功
|
||||
"MC123,,2026.07.01-2026.08.01\n" # 第3行 缺实体ID -> 失败
|
||||
"MC456,G002,2026.07.01-2026.08.01\n", # 第4行 成功
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 2
|
||||
assert "已生成 2 条 URL" in r.stdout
|
||||
assert "以下 1 行生成失败" in r.stderr
|
||||
assert "第3行" in r.stderr
|
||||
assert "缺少实体ID" in r.stderr
|
||||
assert len(read_output(out)) == 2
|
||||
|
||||
def test_missing_required_column_exits_1(self, tmp_path):
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text("所有者名称,实体名称\nA,B\n", encoding="utf-8-sig")
|
||||
r = run_builder(["-i", str(src), "-o", str(tmp_path / "out.csv")])
|
||||
assert r.returncode == 1
|
||||
# 提示缺失的规范字段名（owner_id 与 period 同时缺失时都列出）
|
||||
assert "缺少必要列" in r.stderr
|
||||
assert "owner_id" in r.stderr
|
||||
assert "period" in r.stderr
|
||||
|
||||
def test_all_rows_fail_still_writes_empty_output(self, tmp_path):
|
||||
src = tmp_path / "需求.csv"
|
||||
src.write_text(
|
||||
"所有者ID,实体ID,数据周期\nMC123,,2026.07.01-2026.08.01\n",
|
||||
encoding="utf-8-sig",
|
||||
)
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(src), "-o", str(out)])
|
||||
assert r.returncode == 2
|
||||
assert "已生成 0 条 URL" in r.stdout
|
||||
assert out.exists() # 空结果也落盘（仅表头）
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# 集成：真实资产文件
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestRealAsset:
|
||||
def test_example_asset_generates_all_urls(self, tmp_path):
|
||||
"""assets/需求输入示例.xlsx：6 条示例（群组/所有者/频道/节目）全部成功。"""
|
||||
out = tmp_path / "out.csv"
|
||||
r = run_builder(["-i", str(ASSET_XLSX), "-o", str(out)])
|
||||
assert r.returncode == 0, r.stderr
|
||||
df = read_output(out)
|
||||
assert len(df) == 6
|
||||
# 所有者整体行：实体ID 为空时回退用所有者ID
|
||||
owner_row = df[df["实体类型"] == "所有者"].iloc[0]
|
||||
assert "entity_type=CONTENT_OWNER" in owner_row["URL"]
|
||||
assert "entity_id=bqSUnNpU67xJ51TxH4PKpQ" in owner_row["URL"]
|
||||
# 实体类型枚举全覆盖
|
||||
assert set(df["实体类型"]) == {"群组", "所有者", "频道", "节目"}
|
||||
# 每行 URL 均含 owner 路径与 time_period
|
||||
for url in df["URL"]:
|
||||
assert url.startswith(
|
||||
"https://studio.youtube.com/owner/bqSUnNpU67xJ51TxH4PKpQ/analytics/")
|
||||
assert "time_period=" in url
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# CLI 参数
|
||||
# ---------------------------------------------------------------------------
|
||||
class TestCliArgs:
|
||||
def test_help(self):
|
||||
r = run_builder(["--help"])
|
||||
assert r.returncode == 0
|
||||
assert "批量拼接 YouTube Studio explore URL" in r.stdout
|
||||
|
||||
def test_input_required(self):
|
||||
r = run_builder([])
|
||||
assert r.returncode != 0 # argparse 缺 -i 报错退出码 2
|
||||
assert "required" in r.stderr or "-i" in r.stderr
|
||||
Reference in New Issue
Block a user