Sidney Zhang
b8bd749f43
为 build_studio_urls.py 添加指标和维度的可配置支持，包括： - 新增 metrics.json 和 dimensions.json 映射文件加载 - 支持中文/英文别名映射及指标代码透传 - 列别名扩展以识别"指标"和"维度"列 - 空值时自动回退到 CONFIG 默认值
387 lines
17 KiB
Python
387 lines
17 KiB
Python
# -*- 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 TestMetricColumn:
|
||
def test_metric_selected_and_default(self, tmp_path):
|
||
"""指标列：一行选"观看时长"，一行留空走 CONFIG 默认。"""
|
||
src = tmp_path / "需求.csv"
|
||
pd.DataFrame([
|
||
{"所有者ID": "MC123", "实体ID": "G001", "数据周期": "2026.07.01-2026.08.01",
|
||
"指标": "观看时长"},
|
||
{"所有者ID": "MC123", "实体ID": "G002", "数据周期": "2026.07.01-2026.08.01",
|
||
"指标": ""},
|
||
]).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 list(df.columns) == OUTPUT_COLUMNS
|
||
assert df.loc[0, "指标"] == "观看时长"
|
||
assert df.loc[0, "指标代码"] == "EXTERNAL_WATCH_TIME"
|
||
assert "metric=EXTERNAL_WATCH_TIME" in df.loc[0, "URL"]
|
||
assert "o_column=EXTERNAL_WATCH_TIME" in df.loc[0, "URL"]
|
||
# 空白行：回退 CONFIG 默认
|
||
assert df.loc[1, "指标代码"] == "SUBSCRIBERS_NET_CHANGE"
|
||
assert "metric=SUBSCRIBERS_NET_CHANGE" in df.loc[1, "URL"]
|
||
|
||
def test_unknown_metric_exits_2(self, tmp_path):
|
||
src = tmp_path / "需求.csv"
|
||
src.write_text(
|
||
"所有者ID,实体ID,数据周期,指标\nMC123,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 == 2
|
||
assert "未识别的指标" in r.stderr
|
||
assert "第2行" in r.stderr
|
||
|
||
def test_metric_code_passthrough(self, tmp_path):
|
||
src = tmp_path / "需求.csv"
|
||
src.write_text(
|
||
"所有者ID,实体ID,数据周期,指标\nMC123,G001,2026.07.01-2026.08.01,external_views\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
|
||
df = read_output(out)
|
||
assert df.loc[0, "指标代码"] == "EXTERNAL_VIEWS"
|
||
assert "metric=EXTERNAL_VIEWS" in df.loc[0, "URL"]
|
||
|
||
def test_metric_alias_tax_revenue(self, tmp_path):
|
||
"""「税前收益/税前收入」别名 -> TOTAL_ESTIMATED_EARNINGS。"""
|
||
src = tmp_path / "需求.csv"
|
||
src.write_text(
|
||
"所有者ID,实体ID,数据周期,指标\nMC123,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
|
||
df = read_output(out)
|
||
assert df.loc[0, "指标代码"] == "TOTAL_ESTIMATED_EARNINGS"
|
||
assert "metric=TOTAL_ESTIMATED_EARNINGS" in df.loc[0, "URL"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 维度（每行可选）
|
||
# ---------------------------------------------------------------------------
|
||
class TestDimensionColumn:
|
||
def test_dimension_selected_and_default(self, tmp_path):
|
||
"""维度列：一行选"地理位置"，一行留空走 CONFIG 默认。"""
|
||
src = tmp_path / "需求.csv"
|
||
pd.DataFrame([
|
||
{"所有者ID": "MC123", "实体ID": "G001", "数据周期": "2026.07.01-2026.08.01",
|
||
"维度": "地理位置"},
|
||
{"所有者ID": "MC123", "实体ID": "G002", "数据周期": "2026.07.01-2026.08.01",
|
||
"维度": ""},
|
||
]).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 list(df.columns) == OUTPUT_COLUMNS
|
||
assert df.loc[0, "维度"] == "地理位置"
|
||
assert df.loc[0, "维度代码"] == "COUNTRY"
|
||
assert "dimension=COUNTRY" in df.loc[0, "URL"]
|
||
# 空白行：回退 CONFIG 默认
|
||
assert df.loc[1, "维度代码"] == "USER"
|
||
assert "dimension=USER" in df.loc[1, "URL"]
|
||
|
||
def test_unknown_dimension_exits_2(self, tmp_path):
|
||
src = tmp_path / "需求.csv"
|
||
src.write_text(
|
||
"所有者ID,实体ID,数据周期,维度\nMC123,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 == 2
|
||
assert "未识别的维度" in r.stderr
|
||
assert "第2行" in r.stderr
|
||
|
||
def test_dimension_code_passthrough(self, tmp_path):
|
||
src = tmp_path / "需求.csv"
|
||
src.write_text(
|
||
"所有者ID,实体ID,数据周期,维度\nMC123,G001,2026.07.01-2026.08.01,video\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
|
||
df = read_output(out)
|
||
assert df.loc[0, "维度代码"] == "VIDEO"
|
||
assert "dimension=VIDEO" in df.loc[0, "URL"]
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 失败行与退出码
|
||
# ---------------------------------------------------------------------------
|
||
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
|