#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ 根据需求清单（CSV/Excel）批量拼接 YouTube Studio 内容管理器 explore URL。 用法: python build_studio_urls.py -i 需求清单.csv -o 输出.csv python build_studio_urls.py -i 需求清单.xlsx -o 输出.csv 输入列（支持中英文别名，未填可留空）: 所有者名称 : 所有者名称 / owner_name 所有者ID : 所有者ID / owner_id / o 实体名称 : 实体名称 / 群组名称 / 频道名称 / 节目名称 / entity_name / group_name 实体ID : 实体ID / 群组ID / group_id / entity_id / id 实体类型 : 实体类型 / 类型 / entity_type (群组/所有者/频道/节目，或 GROUP/CONTENT_OWNER/CHANNEL/VIDEO) 数据周期 : 数据周期 / 周期 / period / time_period (yyyy.mm.dd-yyyy.mm.dd 或 yyyy.m.d-yyyy.m.d) 国家 : 国家 / 国家/地区 / country / countries (一个或多个，中文名或 ISO 两位代码) 指标 : 指标 / 主指标 / 数据指标 / metric (可留空，则用 CONFIG 默认；中文名或指标代码) 维度 : 维度 / 细分维度 / dimension (可留空，则用 CONFIG 默认；中文名或维度代码) 说明: - 数据周期起止日期均包含在数据范围内，time_period 结束值取结束日后一天的日界线。 - 国家为多个时，ur_values 以 '%27' 包裹、'%7C' 连接（如 美国,日本 -> %27US%27%7C%27JP%27）。 - 中文国家名 -> ISO 代码的映射放在同目录 countries.json（可自行扩充）；已是两位代码的原样透传。 - 指标每行可选（metric 参数，中文名 -> 代码的映射放在同目录 metrics.json，可自行扩充； 已是已知指标代码的原样透传）。留空则用 CONFIG 默认，且只影响 metric / o_column。 - 维度每行可选（dimension 参数，中文名 -> 代码的映射放在同目录 dimensions.json，可自行扩充； 已是已知维度代码的原样透传）。留空则用 CONFIG 默认。 - 其余固定参数（granularity/t_metrics 等）在本文件 CONFIG 中统一配置。 """ import argparse import json import os import re import sys from datetime import date import pandas as pd from urllib.parse import quote # --------------------------------------------------------------------------- # 固定参数（所有需求共用，按需修改） # --------------------------------------------------------------------------- CONFIG = { "explore_type": "TABLE_AND_CHART", "metric": "SUBSCRIBERS_NET_CHANGE", # 主指标 "granularity": "DAY", # DAY / WEEK / MONTH / YEAR "dimension": "USER", # 细分维度 "t_metrics": [ # 表格列指标（可多个） "SUBSCRIBERS_NET_CHANGE", "VIDEO_COUNT_FIRST_PUBLISHED", "ENGAGED_VIEWS", "EXTERNAL_VIEWS", "EXTERNAL_WATCH_TIME", "AVERAGE_WATCH_TIME", "TOTAL_ESTIMATED_EARNINGS", ], "o_column": "SUBSCRIBERS_NET_CHANGE", # 排序字段 "o_direction": "ANALYTICS_ORDER_DIRECTION_DESC", # DESC / ASC "comparison_type": "NONE", } # 常用指标代码集合（供指标代码透传；metrics.json 提供中文/英文别名映射） KNOWN_METRICS = {CONFIG["metric"], CONFIG["o_column"], *CONFIG["t_metrics"]} # 常见维度代码（供维度代码透传；dimensions.json 提供中文别名映射） KNOWN_DIMENSIONS = {CONFIG["dimension"]} # 实体类型 中文/代码 -> URL 参数值（可扩充） ENTITY_TYPE_MAP = { "群组": "GROUP", "GROUP": "GROUP", "所有者": "CONTENT_OWNER", "账号": "CONTENT_OWNER", "CONTENT_OWNER": "CONTENT_OWNER", "频道": "CHANNEL", "CHANNEL": "CHANNEL", "节目": "VIDEO", "视频": "VIDEO", "VIDEO": "VIDEO", } DEFAULT_ENTITY_TYPE = "GROUP" # 日界线锚点 + 整日偏移（见 docs/adr/0001） ANCHOR_DATE = date(2026, 6, 15) ANCHOR_MS = 1781506800000 MS_PER_DAY = 86400000 # 数据周期正则：支持 2026.07.01-2026.08.01 / 2026.7.1-2026.8.1 / 2026-07-01~2026-08-01 PERIOD_RE = re.compile( r"(\d{4})[.\-/](\d{1,2})[.\-/](\d{1,2})\s*[-~～]\s*(\d{4})[.\-/](\d{1,2})[.\-/](\d{1,2})" ) # 输入列名 -> 规范字段（键为去空白、转小写后的列名） COLUMN_ALIASES = { # 所有者 "所有者名称": "owner_name", "ownername": "owner_name", "owner_name": "owner_name", "所有者": "owner_name", "所有者id": "owner_id", "ownerid": "owner_id", "owner_id": "owner_id", "o": "owner_id", # 实体 "实体名称": "entity_name", "entity_name": "entity_name", "群组名称": "entity_name", "group_name": "entity_name", "groupname": "entity_name", "频道名称": "entity_name", "节目名称": "entity_name", "实体id": "entity_id", "entity_id": "entity_id", "entityid": "entity_id", "群组id": "entity_id", "group_id": "entity_id", "groupid": "entity_id", "群组": "entity_id", "id": "entity_id", "实体类型": "entity_type", "entity_type": "entity_type", "entitytype": "entity_type", "类型": "entity_type", "type": "entity_type", # 数据周期 "数据周期": "period", "周期": "period", "period": "period", "time_period": "period", "timeperiod": "period", "日期范围": "period", # 国家 "国家": "countries", "国家/地区": "countries", "国家地区": "countries", "countries": "countries", "country": "countries", "筛选国家": "countries", "地区": "countries", # 指标（每行可选；缺省用 CONFIG） "指标": "metric", "主指标": "metric", "主要指标": "metric", "数据指标": "metric", "metric": "metric", "metric_name": "metric", "metricname": "metric", "指标名": "metric", # 维度（每行可选；缺省用 CONFIG） "维度": "dimension", "细分维度": "dimension", "dimension": "dimension", "dim": "dimension", "细分": "dimension", "维度名": "dimension", } def ts(y, m, d): """日期 -> 日界线 Unix 毫秒（锚点 + 整日偏移）。""" return ANCHOR_MS + (date(y, m, d) - ANCHOR_DATE).days * MS_PER_DAY def normalize_columns(df): """按别名把输入列映射到规范字段。返回 (field->column_index, 未识别列名列表)。""" mapping = {} unknown = [] for col in df.columns: key = str(col).strip().lower() field = COLUMN_ALIASES.get(key) if field: mapping[field] = col else: unknown.append(str(col)) return mapping, unknown def parse_period(text): """解析数据周期，返回 (start_ms, end_ms)，起止日期均含。""" text = str(text).strip() m = PERIOD_RE.search(text) if not m: raise ValueError("无法解析数据周期: %r（应为 yyyy.mm.dd-yyyy.mm.dd）" % text) y1, m1, d1, y2, m2, d2 = (int(g) for g in m.groups()) start_ms = ts(y1, m1, d1) end_ms = ts(y2, m2, d2) + MS_PER_DAY # 结束日包含 return start_ms, end_ms def load_country_map(path): """加载 中文国家名 -> ISO 代码 映射。文件缺失则仅支持两位代码透传。""" if not os.path.exists(path): print("[提示] 未找到国家映射文件 %s，仅支持直接填写两位 ISO 代码" % path, file=sys.stderr) return {} with open(path, "r", encoding="utf-8-sig") as f: data = json.load(f) return {str(k).strip(): str(v).strip().upper() for k, v in data.items()} def parse_countries(text, country_map): """解析国家列（一个或多个，中文名或 ISO 代码），返回 ISO 代码列表。""" if text is None or (isinstance(text, float) and str(text) == "nan"): return [] raw = str(text) parts = re.split(r"[,，、;；\s|]+", raw) codes = [] for part in parts: p = part.strip().strip("'\"").strip() if not p: continue upper = p.upper() if re.fullmatch(r"[A-Z]{2}", upper): # 已是两位 ISO 代码 codes.append(upper) elif p in country_map: codes.append(country_map[p]) else: raise ValueError("未识别的国家: %r（不在映射文件中，也不是两位 ISO 代码）" % p) return codes def load_metric_map(path): """加载 指标中文/英文名 -> 指标代码 映射。文件缺失则仅支持已知指标代码透传。""" if not os.path.exists(path): print("[提示] 未找到指标映射文件 %s，仅支持直接填写指标代码" % path, file=sys.stderr) return {} with open(path, "r", encoding="utf-8-sig") as f: data = json.load(f) mapped = {str(k).strip(): str(v).strip().upper() for k, v in data.items() if str(k).strip() and str(v).strip()} # 允许映射中出现的所有指标代码原样透传（大小写不敏感） for code in set(mapped.values()) | KNOWN_METRICS: mapped.setdefault(code.lower(), code) return mapped def parse_metric(text, metric_map): """指标列 -> 指标代码。空则返回 None（走 CONFIG 默认）；具体代码透传；未知则报错。""" if text is None or (isinstance(text, float) and str(text) == "nan"): return None raw = str(text).strip().strip("'\"").strip() if not raw: return None if raw.upper() in KNOWN_METRICS: # 已是已知指标代码 return raw.upper() key = raw.lower() if key in metric_map: # 中文/英文别名 return metric_map[key] raise ValueError("未识别的指标: %r（不在映射文件中，也不是已知指标代码）" % raw) def load_dimension_map(path): """加载 细分维度中文名 -> 代码 映射。文件缺失则仅支持已知维度代码透传。""" if not os.path.exists(path): print("[提示] 未找到维度映射文件 %s，仅支持直接填写维度代码" % path, file=sys.stderr) return {} with open(path, "r", encoding="utf-8-sig") as f: data = json.load(f) mapped = {str(k).strip(): str(v).strip().upper() for k, v in data.items() if str(k).strip() and str(v).strip()} # 允许映射中出现的所有维度代码原样透传（大小写不敏感） for code in set(mapped.values()) | KNOWN_DIMENSIONS: mapped.setdefault(code.lower(), code) return mapped def parse_dimension(text, dimension_map): """维度列 -> 维度代码。空则返回 None（走 CONFIG 默认）；具体代码透传；未知则报错。""" if text is None or (isinstance(text, float) and str(text) == "nan"): return None raw = str(text).strip().strip("'\"").strip() if not raw: return None if raw.upper() in KNOWN_DIMENSIONS: # 已是已知维度代码 return raw.upper() key = raw.lower() if key in dimension_map: # 中文别名 return dimension_map[key] raise ValueError("未识别的维度: %r（不在映射文件中，也不是已知维度代码）" % raw) def resolve_entity_type(text): """实体类型 -> URL 参数值。空则用默认群组。""" if text is None or (isinstance(text, float) and str(text) == "nan") or str(text).strip() == "": return DEFAULT_ENTITY_TYPE key = str(text).strip() if key in ENTITY_TYPE_MAP: return ENTITY_TYPE_MAP[key] raise ValueError("未识别的实体类型: %r（应为 群组/所有者/频道/节目 或 GROUP/CONTENT_OWNER/CHANNEL/VIDEO）" % key) def build_url(row, country_map, metric_map=None, dimension_map=None): """根据一行需求生成 URL。返回 (url, 状态, 错误信息)。 metric_map / dimension_map 为 名 -> 代码 映射；对应列缺省时回退到 CONFIG 默认。 """ owner_id = row.get("owner_id", "").strip() entity_type = resolve_entity_type(row.get("entity_type", "")) entity_id = row.get("entity_id", "").strip() if not owner_id: return None, "error", "缺少所有者ID" if not entity_id: if entity_type == "CONTENT_OWNER": entity_id = owner_id # 账号整体场景回退 else: return None, "error", "缺少实体ID" period = row.get("period", "").strip() if not period: return None, "error", "缺少数据周期" start_ms, end_ms = parse_period(period) codes = parse_countries(row.get("countries", ""), country_map) # 指标：每行可选；空 -> 主指标/排序字段用 CONFIG 默认；非空 -> 二者都跟随所选指标 sel_metric = parse_metric(row.get("metric", ""), metric_map or {}) metric = sel_metric or CONFIG["metric"] o_column = sel_metric or CONFIG["o_column"] # 维度：每行可选；空 -> 用 CONFIG 默认细分维度 sel_dimension = parse_dimension(row.get("dimension", ""), dimension_map or {}) dimension = sel_dimension or CONFIG["dimension"] base = "https://studio.youtube.com/owner/%s/analytics/tab-overview/period-default/explore" % owner_id params = [] params.append("o=%s" % owner_id) params.append("entity_type=%s" % entity_type) params.append("entity_id=%s" % entity_id) if codes: ur_values = quote("|".join("'%s'" % c for c in codes), safe="") # 'US'|'JP' -> %27US%27%7C%27JP%27 params.append("ur_dimensions=COUNTRY") params.append("ur_values=%s" % ur_values) params.append("ur_inclusive_starts=") params.append("ur_exclusive_ends=") params.append("time_period=%d%%2C%d" % (start_ms, end_ms)) params.append("explore_type=%s" % CONFIG["explore_type"]) params.append("metric=%s" % metric) params.append("granularity=%s" % CONFIG["granularity"]) for m in CONFIG["t_metrics"]: params.append("t_metrics=%s" % m) params.append("dimension=%s" % dimension) params.append("o_column=%s" % o_column) params.append("o_direction=%s" % CONFIG["o_direction"]) params.append("comparison_type=%s" % CONFIG["comparison_type"]) url = base + "?" + "&".join(params) return url, "ok", "" def main(): parser = argparse.ArgumentParser(description="批量拼接 YouTube Studio explore URL") parser.add_argument("-i", "--input", required=True, help="需求清单文件（.csv / .xlsx / .xls）") parser.add_argument("-o", "--output", default=None, help="输出 CSV 路径（默认：输入同目录 studio_urls_output.csv）") parser.add_argument("--countries", default=None, help="国家映射 JSON 路径（默认：脚本同目录 countries.json）") parser.add_argument("--metrics", default=None, help="指标映射 JSON 路径（默认：脚本同目录 metrics.json）") parser.add_argument("--dimensions", default=None, help="维度映射 JSON 路径（默认：脚本同目录 dimensions.json）") args = parser.parse_args() if not os.path.exists(args.input): print("错误：输入文件不存在: %s" % args.input, file=sys.stderr) sys.exit(1) ext = os.path.splitext(args.input)[1].lower() if ext == ".csv": try: df = pd.read_csv(args.input, encoding="utf-8-sig") except UnicodeDecodeError: df = pd.read_csv(args.input, encoding="gbk") # Excel 另存的 ANSI/GBK elif ext in (".xlsx", ".xls"): df = pd.read_excel(args.input) else: print("错误：不支持的输入格式: %s（支持 .csv/.xlsx/.xls）" % ext, file=sys.stderr) sys.exit(1) df = df.dropna(how="all") # 去掉全空行 mapping, unknown = normalize_columns(df) if unknown: print("[提示] 未识别的列（忽略）: %s" % ", ".join(unknown), file=sys.stderr) missing = [f for f in ("owner_id", "period") if f not in mapping] if missing: print("错误：输入缺少必要列: %s" % ", ".join(missing), file=sys.stderr) sys.exit(1) countries_file = args.countries or os.path.join(os.path.dirname(os.path.abspath(__file__)), "countries.json") country_map = load_country_map(countries_file) metrics_file = args.metrics or os.path.join(os.path.dirname(os.path.abspath(__file__)), "metrics.json") metric_map = load_metric_map(metrics_file) dimensions_file = args.dimensions or os.path.join(os.path.dirname(os.path.abspath(__file__)), "dimensions.json") dimension_map = load_dimension_map(dimensions_file) records = [] errors = [] for idx, raw in df.iterrows(): row = {f: ("" if pd.isna(raw[mapping[f]]) else str(raw[mapping[f]])) for f in mapping} try: url, status, msg = build_url(row, country_map, metric_map, dimension_map) except ValueError as e: url, status, msg = None, "error", str(e) if status == "ok": start_ms, end_ms = parse_period(row.get("period", "")) codes = parse_countries(row.get("countries", ""), country_map) sel_metric = parse_metric(row.get("metric", ""), metric_map) metric_code = sel_metric or CONFIG["metric"] sel_dimension = parse_dimension(row.get("dimension", ""), dimension_map) dimension_code = sel_dimension or CONFIG["dimension"] records.append({ "所有者名称": row.get("owner_name", ""), "所有者ID": row.get("owner_id", ""), "实体类型": row.get("entity_type", ""), "实体名称": row.get("entity_name", ""), "实体ID": row.get("entity_id", ""), "数据周期": row.get("period", ""), "国家": row.get("countries", ""), "国家代码": ",".join(codes), "指标": row.get("metric", ""), "指标代码": metric_code, "维度": row.get("dimension", ""), "维度代码": dimension_code, "开始时间戳": start_ms, "结束时间戳": end_ms, "URL": url, }) else: errors.append((idx + 2, row.get("owner_name", ""), row.get("entity_name", ""), msg)) out_path = args.output or os.path.join( os.path.dirname(os.path.abspath(args.input)), "studio_urls_output.csv") out_df = pd.DataFrame(records) out_df.to_csv(out_path, index=False, encoding="utf-8-sig") print("已生成 %d 条 URL -> %s" % (len(records), out_path)) if errors: print("\n以下 %d 行生成失败：" % len(errors), file=sys.stderr) for r in errors: print(" 第%d行 所有者=%s 实体=%s：%s" % r, file=sys.stderr) sys.exit(2) if __name__ == "__main__": main()