Add aggregation helper script

scripts/aggregate_video_products.py

@@ -0,0 +1,45 @@
+#!/usr/bin/env python3
+"""Aggregate SW Ads creative_breakdown JSON into product-level video metrics."""
+import argparse, json
+from pathlib import Path
+
+def ratio(n, d): return n / d if d else None
+
+def main():
+    p = argparse.ArgumentParser()
+    p.add_argument("input", type=Path)
+    p.add_argument("--product-id", action="append", dest="product_ids")
+    p.add_argument("--top", type=int, default=5)
+    a = p.parse_args()
+    data = json.loads(a.input.read_text(encoding="utf-8"))
+    cols = [c["name"] for c in data["columns"]]
+    idx = {name: i for i, name in enumerate(cols)}
+    required = {"external_creative_id","creative_identity_kind","creative_product_id","spend","gross_revenue","orders","product_impressions","product_clicks"}
+    missing = sorted(required - idx.keys())
+    if missing: raise SystemExit("missing required columns: " + ", ".join(missing))
+    selected, groups = set(a.product_ids or []), {}
+    for row in data["rows"]:
+        cid = str(row[idx["external_creative_id"]] or "")
+        pid = str(row[idx["creative_product_id"]] or "unknown")
+        if row[idx["creative_identity_kind"]] == "product_card" or cid.startswith("product_card:"): continue
+        if selected and pid not in selected: continue
+        g = groups.setdefault(pid,{"product_id":pid,"video_count":0,"spend":0.0,"gross_revenue":0.0,"orders":0.0,"product_impressions":0.0,"product_clicks":0.0,"videos":[]})
+        g["video_count"] += 1
+        for f in ("spend","gross_revenue","orders","product_impressions","product_clicks"): g[f] += row[idx[f]] or 0
+        v = {"external_creative_id":cid}
+        for f in ("external_creative_name","creative_creator_name"):
+            if f in idx: v[f] = row[idx[f]]
+        for f in ("spend","gross_revenue","orders","product_impressions","product_clicks"): v[f] = row[idx[f]] or 0
+        g["videos"].append(v)
+    out=[]
+    for g in groups.values():
+        g["video_ctr"] = ratio(g["product_clicks"],g["product_impressions"])
+        g["order_conversion_rate"] = ratio(g["orders"],g["product_clicks"])
+        g["roas"] = ratio(g["gross_revenue"],g["spend"])
+        g["cost_per_order"] = ratio(g["spend"],g["orders"])
+        g["videos"].sort(key=lambda v:(v["orders"],v["gross_revenue"],-v["spend"]),reverse=True)
+        g["top_videos"] = g.pop("videos")[:max(a.top,0)]; out.append(g)
+    out.sort(key=lambda g:(g["orders"],g["gross_revenue"]),reverse=True)
+    print(json.dumps({"products":out},ensure_ascii=False,indent=2))
+
+if __name__ == "__main__": main()