voypace-ad-report/adscenter_daily_sync.py

#!/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("完成")