init: voypace ad report

2026-09-10 15:26:54 +08:00
commit 3c525a9672
9 changed files with 1833 additions and 0 deletions
@@ -0,0 +1,129 @@
+#!/usr/bin/env python3
+# 在 ads-center 上每天运行：取前一天3渠道数据 -> 合并 mtg_data.json -> 生成矩阵 -> git push
+# 缺表渠道(Adsterra/TrafficStars/Eagllwin)由 excel_fetch.py 从运营 Excel 补数（可选，找不到就跳过）
+# 由 launchd 定时触发。Python: /Users/ads-center/py314env/bin/python
+import sqlite3, json, subprocess, sys, os
+from collections import defaultdict
+from datetime import date, timedelta
+
+DB = "/Users/ads-center/Documents/database/report_api.db"
+REPO = "/Users/ads-center/voypace-report-sync/voypace-ad-report"
+
+def log(m): print(f"[sync] {m}", flush=True)
+
+# 目标日期：前一天（可用参数覆盖，便于补数）
+TARGET = sys.argv[1] if len(sys.argv) > 1 else (date.today() - timedelta(days=1)).isoformat()
+
+os.chdir(REPO)
+# 先 pull 最新，避免冲突
+subprocess.run(["git", "pull", "--rebase", "-q"], cwd=REPO)
+
+camps = json.load(open("campaigns.json"))
+CAMPS = {r["campid"]: {"product": r["product"], "pkg": r["pkg"]} for r in camps}
+IDS = list(CAMPS.keys())
+
+conn = sqlite3.connect(DB); conn.row_factory = sqlite3.Row
+agg = defaultdict(lambda: {"impression":0,"clicks":0,"conversions":0,"spend":0.0})
+meta = {}
+
+def cid_of(t):
+    t = str(t or "")
+    for c in IDS:
+        if c in t: return c
+    return None
+
+# Mintegral（广告单元名可能落在 Offer_Name 或 Campaign_Id，两字段都匹配）
+like = " OR ".join([f"Offer_Name LIKE '%{c}%' OR Campaign_Id LIKE '%{c}%'" for c in IDS])
+for r in conn.execute(f"SELECT Date d,Offer_Name nm,Campaign_Id cmp,Location loc,Impression imp,Click clk,Conversion cv,Spend sp FROM mintegral WHERE Date=? AND ({like})", (TARGET,)):
+    c = cid_of(r["nm"]) or cid_of(r["cmp"])
+    if not c: continue
+    k = (r["d"],"Mintegral",c,(r["loc"] or "").upper()); a=agg[k]
+    a["impression"]+=int(r["imp"] or 0); a["clicks"]+=int(r["clk"] or 0)
+    a["conversions"]+=int(r["cv"] or 0); a["spend"]+=float(r["sp"] or 0)
+    meta[k]=(r["nm"] or r["cmp"],CAMPS[c])
+
+# Bigo
+like = " OR ".join([f"campaignName LIKE '%{c}%'" for c in IDS])
+for r in conn.execute(f"SELECT aggregateTime d,campaignName nm,country loc,impression imp,click clk,real_result_conversion cv,totalCost cost FROM bigo WHERE aggregateTime=? AND ({like})", (TARGET,)):
+    c = cid_of(r["nm"])
+    if not c: continue
+    k = (r["d"],"Bigo",c,(r["loc"] or "").upper()); a=agg[k]
+    a["impression"]+=int(r["imp"] or 0); a["clicks"]+=int(r["clk"] or 0)
+    a["conversions"]+=int(r["cv"] or 0); a["spend"]+=float(r["cost"] or 0)/100.0
+    meta[k]=(r["nm"],CAMPS[c])
+
+# MediaGo (spend = cpc*click)
+like = " OR ".join([f"campaign_name LIKE '%{c}%'" for c in IDS])
+for r in conn.execute(f"SELECT date d,campaign_name nm,adset_country loc,vimp imp,click clk,cv cvn,cpc FROM mediago WHERE date=? AND ({like})", (TARGET,)):
+    c = cid_of(r["nm"])
+    if not c: continue
+    k = (r["d"],"MediaGo",c,(r["loc"] or "").upper()); a=agg[k]
+    clk=int(r["clk"] or 0)
+    a["impression"]+=int(r["imp"] or 0); a["clicks"]+=clk
+    a["conversions"]+=int(r["cvn"] or 0); a["spend"]+=float(r["cpc"] or 0)*clk
+    meta[k]=(r["nm"],CAMPS[c])
+conn.close()
+
+new_rows=[]
+for k,a in agg.items():
+    dt,ch,c,loc=k; nm,info=meta[k]
+    imp,clk,cv=a["impression"],a["clicks"],a["conversions"]
+    new_rows.append({"date":dt,"channel":ch,"campid":c,"offer_name":nm,"product":info["product"],"pkg":info["pkg"],
+        "location":loc,"bid_type":"","impression":imp,"clicks":clk,"conversions":cv,
+        "ctr":round(clk/imp,5) if imp else 0.0,"cvr":round(cv/clk,5) if clk else 0.0,"spend":round(a["spend"],2)})
+log(f"{TARGET} DB三渠道新增匹配 {len(new_rows)} 条")
+
+# 可选：从运营 Excel 补齐缺表渠道（Adsterra/TrafficStars/Eagllwin 等）
+excel_rows = []
+try:
+    ef = os.path.join(REPO, "excel_fetch.py")
+    if os.path.exists(ef):
+        p = subprocess.run([sys.executable, ef], cwd=REPO, capture_output=True, text=True)
+        if p.returncode == 0 and p.stdout.strip():
+            excel_rows = json.loads(p.stdout)
+            # 只保留目标日期，保持“每天增量”语义（补数全量跑时可去掉此过滤）
+            excel_rows = [r for r in excel_rows if r.get("date") == TARGET]
+        else:
+            log(f"Excel 补数无输出/失败：{p.stderr[:120]}")
+except Exception as e:
+    log(f"Excel 补数跳过：{e}")
+if excel_rows:
+    log(f"Excel 补数新增 {len(excel_rows)} 条")
+    new_rows += excel_rows
+
+# 合并（按 date+channel+campid+location 去重）
+existing = json.load(open("mtg_data.json")) if os.path.exists("mtg_data.json") else []
+merged = {(r["date"],r.get("channel",""),r["campid"],r["location"]): r for r in existing}
+for r in new_rows:
+    merged[(r["date"],r["channel"],r["campid"],r["location"])] = r
+data = sorted(merged.values(), key=lambda x:(x["date"],x.get("channel",""),-x["impression"]))
+json.dump(data, open("mtg_data.json","w"), ensure_ascii=False, indent=2)
+log(f"合并后共 {len(data)} 条")
+
+# 重新生成矩阵状态（近3天单日最高展示>=100 且产品订单为Active -> Active）
+orders = json.load(open("orders.json"))
+pkg_status={}
+for o in orders:
+    for p in str(o["pkg"]).replace(" ","").split("/"):
+        if p: pkg_status[p]=o.get("status","Pause")
+g=defaultdict(lambda: defaultdict(int))
+for r in data: g[(r["channel"],r["pkg"])][r["date"]]+=r["impression"]
+alldates=sorted(set(r["date"] for r in data)); last3=alldates[-3:]
+ov={}
+for (ch,pkg),days in g.items():
+    ps=pkg_status.get(pkg,"Pause")
+    if ps!="Active": ov[f"{ch}|{pkg}"]="Pause"; continue
+    total=sum(days.values()); rmax=max([days.get(dt,0) for dt in last3]) if last3 else 0
+    ov[f"{ch}|{pkg}"]= "none" if total==0 else ("Active" if rmax>=100 else "Pause")
+json.dump(ov, open("matrix_overrides.json","w"), ensure_ascii=False, indent=2)
+log("矩阵状态已刷新")
+
+# git push
+subprocess.run(["git","add","mtg_data.json","matrix_overrides.json"], cwd=REPO)
+r=subprocess.run(["git","commit","-m",f"data: daily sync {TARGET}"], cwd=REPO, capture_output=True, text=True)
+if "nothing to commit" in (r.stdout+r.stderr):
+    log("无变化，跳过 push")
+else:
+    p=subprocess.run(["git","push","-q","origin","main"], cwd=REPO, capture_output=True, text=True)
+    log("已 push" if p.returncode==0 else f"push失败: {p.stderr[:200]}")
+log("完成")