434 lines
17 KiB
Python
434 lines
17 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]:
|
|
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()
|