#!/usr/bin/env python3 """Build a privacy-minimized AgentStage dataset from the TaurusX order workbook.""" from __future__ import annotations import argparse import json import re from collections import Counter, defaultdict from datetime import datetime from pathlib import Path from typing import Any import openpyxl DEFAULT_SOURCE = Path( "/Users/wangyuan/Documents/SHAREit/运营/TaurusX搬单/武汉自投投放下单_AM&Media Buy.xlsx" ) TARGET_PLATFORM = "TaurusxDSP" REQUIRED_COLUMNS = {"状态", "PromotionAppName", "pkg", "投放平台", "投放国家"} PLATFORM_ALIASES = { "mtg": "MTG", "mintegral": "MTG", "bigo": "Bigo", "kwai": "Kwai", "kwa": "Kwai", "lacuna": "Lacuna", "p_api_adsterra": "Adsterra", "adsterra": "Adsterra", "m_api_self_propellerads": "PropellerAds", "properllerads": "PropellerAds", "propellerads": "PropellerAds", "rollerads": "RollerAds", "mediago": "MediaGo", "media go": "MediaGo", "taurusxdsp": TARGET_PLATFORM, "taurusx dsp": TARGET_PLATFORM, "taurusx": TARGET_PLATFORM, "webeye/taurusx": TARGET_PLATFORM, "unity": "Unity Ads", "unity ads": "Unity Ads", "loopme": "LoopMe", "smartyads": "SmartyAds", "tiktok/pangel": "TikTok/Pangle", "tiktok/pangle": "TikTok/Pangle", "pi": "PI", } def clean_text(value: Any) -> str: if value is None: return "" return re.sub(r"\s+", " ", str(value)).strip() def normalize_platform(value: Any) -> str: raw = clean_text(value) lowered = raw.casefold() # Account annotations occur in several Kwai spellings. if lowered.startswith("kwai") or lowered.startswith("kwa "): return "Kwai" return PLATFORM_ALIASES.get(lowered, raw) def normalize_status(value: Any) -> str: raw = clean_text(value) if raw == "投放中": return "投放中" if raw == "暂停": return "暂停" if not raw or raw.casefold() == "none": return "未标注" return "其他/不可投" def parse_workbook(source: Path) -> tuple[list[dict[str, str]], dict[str, int]]: if not source.is_file(): raise FileNotFoundError(f"Workbook not found: {source}") # Do not use read_only=True: DingTalk exports have an incorrect A1 dimension. workbook = openpyxl.load_workbook(source, read_only=False, data_only=True) sheet = workbook.active rows = sheet.iter_rows(values_only=True) headers = [clean_text(value) for value in next(rows)] missing = REQUIRED_COLUMNS.difference(headers) if missing: raise ValueError(f"Missing required columns: {', '.join(sorted(missing))}") indexes = {name: headers.index(name) for name in REQUIRED_COLUMNS} records: list[dict[str, str]] = [] counters = Counter( source_rows=0, valid_rows=0, excluded_examples=0, excluded_incomplete=0, excluded_invalid_pkg=0, ) for row in rows: counters["source_rows"] += 1 status_raw = clean_text(row[indexes["状态"]]) if status_raw == "示例": counters["excluded_examples"] += 1 continue product = clean_text(row[indexes["PromotionAppName"]]) pkg = clean_text(row[indexes["pkg"]]) country = clean_text(row[indexes["投放国家"]]).upper() platform = normalize_platform(row[indexes["投放平台"]]) if not pkg or not country or not platform: counters["excluded_incomplete"] += 1 continue if re.match(r"^https?://", pkg, flags=re.IGNORECASE): counters["excluded_invalid_pkg"] += 1 continue records.append( { "product": product or "未命名产品", "pkg": pkg, "country": country, "platform": platform, "status": normalize_status(status_raw), } ) counters["valid_rows"] += 1 workbook.close() return records, dict(counters) def aggregate(records: list[dict[str, str]], counters: dict[str, int], source: Path) -> dict[str, Any]: groups: dict[tuple[str, str], dict[str, Any]] = {} for row in records: key = (row["pkg"].casefold(), row["country"]) group = groups.setdefault( key, { "pkg": row["pkg"], "country": row["country"], "products": set(), "platforms": defaultdict(lambda: {"statuses": Counter(), "order_count": 0}), "order_count": 0, }, ) group["products"].add(row["product"]) detail = group["platforms"][row["platform"]] detail["statuses"][row["status"]] += 1 detail["order_count"] += 1 group["order_count"] += 1 output_records: list[dict[str, Any]] = [] for group in groups.values(): platform_names = sorted(group["platforms"], key=lambda name: (name == TARGET_PLATFORM, name.casefold())) target_present = TARGET_PLATFORM in group["platforms"] other_platforms = [name for name in platform_names if name != TARGET_PLATFORM] platform_details = [] for name in platform_names: detail = group["platforms"][name] platform_details.append( { "platform": name, "statuses": dict(sorted(detail["statuses"].items())), "order_count": detail["order_count"], } ) output_records.append( { "product": " / ".join(sorted(group["products"])), "pkg": group["pkg"], "country": group["country"], "platforms": platform_names, "other_platforms": other_platforms, "platform_details": platform_details, "target_present": target_present, "missing_target": bool(other_platforms) and not target_present, "order_count": group["order_count"], } ) output_records.sort( key=lambda row: ( not row["missing_target"], row["country"], row["pkg"].casefold(), ) ) comparison_records = [row for row in output_records if row["other_platforms"]] covered_records = [row for row in comparison_records if row["target_present"]] missing_records = [row for row in comparison_records if row["missing_target"]] all_platforms = sorted({row["platform"] for row in records}, key=lambda name: (name == TARGET_PLATFORM, name.casefold())) platform_summary = [] for platform in all_platforms: platform_rows = [row for row in output_records if platform in row["platforms"]] platform_summary.append( { "platform": platform, "key_count": len(platform_rows), "order_count": sum( next(detail["order_count"] for detail in row["platform_details"] if detail["platform"] == platform) for row in platform_rows ), "target_overlap_count": sum(row["target_present"] for row in platform_rows), } ) platform_summary.sort(key=lambda row: (-row["key_count"], row["platform"].casefold())) country_groups: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in comparison_records: country_groups[row["country"]].append(row) country_summary = [ { "country": country, "comparison_count": len(rows), "covered_count": sum(row["target_present"] for row in rows), "missing_count": sum(row["missing_target"] for row in rows), "coverage_rate": round(sum(row["target_present"] for row in rows) / len(rows), 4), } for country, rows in country_groups.items() ] country_summary.sort(key=lambda row: (-row["missing_count"], row["country"])) generated_at = datetime.now().astimezone().isoformat(timespec="seconds") comparison_count = len(comparison_records) return { "meta": { "title": "TaurusX订单", "source": "文稿 / SHAREit / 运营 / TaurusX搬单 / 武汉自投投放下单_AM&Media Buy.xlsx", "generated_at": generated_at, "target_platform": TARGET_PLATFORM, **counters, "pkg_country_count": len(output_records), "comparison_count": comparison_count, "covered_count": len(covered_records), "missing_count": len(missing_records), "target_only_count": sum(not row["other_platforms"] and row["target_present"] for row in output_records), "coverage_rate": round(len(covered_records) / comparison_count, 4) if comparison_count else 0, }, "platform_summary": platform_summary, "country_summary": country_summary, "records": output_records, } def main() -> None: parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--source", type=Path, default=DEFAULT_SOURCE, help="Source XLSX workbook") parser.add_argument("--output", type=Path, default=Path(__file__).with_name("data.json"), help="Output JSON") args = parser.parse_args() records, counters = parse_workbook(args.source) payload = aggregate(records, counters, args.source) args.output.parent.mkdir(parents=True, exist_ok=True) args.output.write_text(json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8") print( f"Wrote {args.output}: {payload['meta']['pkg_country_count']} pkg-country keys, " f"{payload['meta']['missing_count']} missing {TARGET_PLATFORM}" ) if __name__ == "__main__": main()