282 lines
10 KiB
Python
282 lines
10 KiB
Python
#!/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]:
|
|
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(),
|
|
"platforms": defaultdict(
|
|
lambda: {"statuses": Counter(), "adpingids": set(), "order_count": 0}
|
|
),
|
|
"order_count": 0,
|
|
},
|
|
)
|
|
group["products"].add(row["product"])
|
|
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]
|
|
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"],
|
|
"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()
|