#!/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 = {"状态", "Adpingid", "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 sort_adpingids(values: set[str]) -> list[str]: def sort_key(value: str) -> tuple[int, int | str]: return (0, int(value)) if value.isdigit() else (1, value.casefold()) return sorted(values, key=sort_key) 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 adpingid = clean_text(row[indexes["Adpingid"]]) 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( { "adpingid": adpingid, "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]: status_priority = {"投放中": 0, "暂停": 1, "未标注": 2, "其他/不可投": 3} 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(), "adpingids": set(), "order_statuses": Counter(), "platforms": defaultdict( lambda: {"statuses": Counter(), "adpingids": set(), "order_count": 0} ), "order_count": 0, }, ) group["products"].add(row["product"]) group["order_statuses"][row["status"]] += 1 if row["adpingid"]: group["adpingids"].add(row["adpingid"]) detail = group["platforms"][row["platform"]] detail["statuses"][row["status"]] += 1 if row["adpingid"]: detail["adpingids"].add(row["adpingid"]) 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] order_statuses = sorted( group["order_statuses"], key=lambda status: (status_priority.get(status, 99), status) ) platform_details = [] for name in platform_names: detail = group["platforms"][name] platform_details.append( { "platform": name, "adpingids": sort_adpingids(detail["adpingids"]), "statuses": dict(sorted(detail["statuses"].items())), "order_count": detail["order_count"], } ) output_records.append( { "adpingids": sort_adpingids(group["adpingids"]), "product": " / ".join(sorted(group["products"])), "pkg": group["pkg"], "country": group["country"], "order_statuses": order_statuses, "order_status_counts": {status: group["order_statuses"][status] for status in order_statuses}, "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 parse_mtg1013(source: Path, payload: dict[str, Any]) -> dict[str, Any]: """Parse MTG performance rows and relate them to the order coverage keys.""" import csv from decimal import Decimal, InvalidOperation if not source.is_file(): raise FileNotFoundError(f"MTG CSV not found: {source}") required = { "Adping ID", "Pkg Name", "Target Geo", "Valid Clicks", "Postback Installs", "Revenue", "Cost", } coverage_by_key = { (row["pkg"].casefold(), row["country"]): row for row in payload["records"] } def number(value: Any, field: str, line: int) -> Decimal: try: return Decimal(clean_text(value) or "0") except InvalidOperation as exc: raise ValueError(f"Invalid {field} on CSV line {line}: {value}") from exc def decimal_value(value: Decimal, places: str = "0.0001") -> float: return float(value.quantize(Decimal(places))) records: list[dict[str, Any]] = [] with source.open("r", encoding="utf-8-sig", newline="") as handle: reader = csv.DictReader(handle) missing = required.difference(reader.fieldnames or []) if missing: raise ValueError(f"Missing MTG CSV columns: {', '.join(sorted(missing))}") for line, row in enumerate(reader, start=2): adpingid = clean_text(row["Adping ID"]) pkg = clean_text(row["Pkg Name"]) country = clean_text(row["Target Geo"]).upper() if not adpingid or not pkg or not country: raise ValueError(f"Incomplete MTG CSV key on line {line}") if re.match(r"^https?://", pkg, flags=re.IGNORECASE): raise ValueError(f"Invalid MTG CSV pkg URL on line {line}") clicks = int(number(row["Valid Clicks"], "Valid Clicks", line)) installs = int(number(row["Postback Installs"], "Postback Installs", line)) revenue = number(row["Revenue"], "Revenue", line) cost = number(row["Cost"], "Cost", line) key = (pkg.casefold(), country) coverage = coverage_by_key.get(key) records.append( { "adpingid": adpingid, "pkg": pkg, "country": country, "valid_clicks": clicks, "installs": installs, "revenue": decimal_value(revenue), "cost": decimal_value(cost), "profit": decimal_value(revenue - cost), "roas": round(float(revenue / cost), 4) if cost else None, "install_rate": round(installs / clicks, 4) if clicks else None, "order_match": coverage is not None, "adpingid_match": bool(coverage and adpingid in coverage["adpingids"]), "target_present": bool(coverage and coverage["target_present"]), "missing_target": bool(coverage and coverage["missing_target"]), } ) # Rows without a valid click are intentionally excluded from the MTG module, # including its summaries, linked filters, pagination, and CSV export. records = [row for row in records if row["valid_clicks"] > 0] records.sort(key=lambda row: (-row["cost"], -row["valid_clicks"], row["adpingid"])) total_clicks = sum(row["valid_clicks"] for row in records) total_installs = sum(row["installs"] for row in records) total_revenue = sum(Decimal(str(row["revenue"])) for row in records) total_cost = sum(Decimal(str(row["cost"])) for row in records) by_key: dict[tuple[str, str], list[dict[str, Any]]] = defaultdict(list) by_country: dict[str, list[dict[str, Any]]] = defaultdict(list) for row in records: by_key[(row["pkg"].casefold(), row["country"])].append(row) by_country[row["country"]].append(row) for key, rows in by_key.items(): coverage = coverage_by_key.get(key) if not coverage: continue coverage["mtg1013"] = { "adpingids": sort_adpingids({row["adpingid"] for row in rows}), "valid_clicks": sum(row["valid_clicks"] for row in rows), "installs": sum(row["installs"] for row in rows), "revenue": round(sum(row["revenue"] for row in rows), 4), "cost": round(sum(row["cost"] for row in rows), 4), } country_summary = [] for country, rows in by_country.items(): revenue = sum(row["revenue"] for row in rows) cost = sum(row["cost"] for row in rows) country_summary.append( { "country": country, "rows": len(rows), "valid_clicks": sum(row["valid_clicks"] for row in rows), "installs": sum(row["installs"] for row in rows), "revenue": round(revenue, 4), "cost": round(cost, 4), "profit": round(revenue - cost, 4), "roas": round(revenue / cost, 4) if cost else None, } ) country_summary.sort(key=lambda row: (-row["cost"], row["country"])) return { "meta": { "source": source.name, "rows": len(records), "unique_keys": len(by_key), "unique_adpingids": len({row["adpingid"] for row in records}), "matched_order_rows": sum(row["order_match"] for row in records), "matched_adpingid_rows": sum(row["adpingid_match"] for row in records), "unmatched_order_rows": sum(not row["order_match"] for row in records), "valid_clicks": total_clicks, "installs": total_installs, "revenue": decimal_value(total_revenue), "cost": decimal_value(total_cost), "profit": decimal_value(total_revenue - total_cost), "roas": round(float(total_revenue / total_cost), 4) if total_cost else None, "install_rate": round(total_installs / total_clicks, 4) if total_clicks else None, }, "country_summary": country_summary, "records": 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( "--mtg-source", type=Path, default=DEFAULT_SOURCE.with_name("mtg1013.csv"), help="MTG 10/13 performance CSV", ) 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) payload["mtg1013"] = parse_mtg1013(args.mtg_source, payload) 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}, " f"{payload['mtg1013']['meta']['rows']} MTG rows" ) if __name__ == "__main__": main()