Manta-tracking/update.sh

#!/bin/sh
set -eu

ROOT_DIR=$(CDPATH= cd -- "$(dirname -- "$0")" && pwd)
HOUR_CN=$(TZ=Asia/Shanghai date +%H)

# AgentStage invokes this hourly. Only the 12:00 China-time run refreshes data.
if [ "${FORCE_UPDATE:-0}" != "1" ] && [ "$HOUR_CN" != "12" ]; then
  exit 0
fi

TMP_DIR=$(mktemp -d)
trap 'rm -rf "$TMP_DIR"' EXIT HUP INT TERM
RAW_CSV="$TMP_DIR/tableau.csv"

if [ -n "${SOURCE_CSV:-}" ]; then
  cp "$SOURCE_CSV" "$RAW_CSV"
else
  TABLEAU_URL=${TABLEAU_CSV_URL:-'https://mars.ushareit.org/views/DSPSpend/DSPDailyData.csv?:showVizHome=no&%E7%BB%B4%E5%BA%A61=pkg&%E7%BB%B4%E5%BA%A62=adping_id'}
  if [ -n "${TABLEAU_COOKIE:-}" ]; then
    curl --fail --silent --show-error --location \
      --cookie "$TABLEAU_COOKIE" "$TABLEAU_URL" --output "$RAW_CSV"
  else
    curl --fail --silent --show-error --location "$TABLEAU_URL" --output "$RAW_CSV"
  fi
fi

python3 - "$RAW_CSV" "$ROOT_DIR/campaigns.csv" "$ROOT_DIR/data.json" <<'PY'
import csv
import datetime as dt
import io
import json
import pathlib
import sys

source_path = pathlib.Path(sys.argv[1])
campaign_path = pathlib.Path(sys.argv[2])
output_path = pathlib.Path(sys.argv[3])
raw = source_path.read_bytes()
head = raw[:500].lower()
if b"<html" in head or b"tableau server" in head:
    raise SystemExit("Tableau returned a login page; configure TABLEAU_COOKIE or an authenticated TABLEAU_CSV_URL")

if raw.startswith((b"\xff\xfe", b"\xfe\xff")) or b"\x00" in raw[:200]:
    text = raw.decode("utf-16").lstrip("\ufeff")
else:
    text = raw.decode("utf-8-sig")

sample = text[:4096]
delimiter = "\t" if sample.count("\t") > sample.count(",") else ","
reader = csv.DictReader(io.StringIO(text), delimiter=delimiter)
with campaign_path.open(encoding="utf-8-sig", newline="") as f:
    campaigns = list(csv.DictReader(f))
campaign_by_key = {(r["pkg"], r["adping_id"]): r for r in campaigns if r["status"] == "Active"}
active_pkgs = sorted({r["pkg"] for r in campaigns if r["status"] == "Active"})

def pick(row, *names):
    for name in names:
        if name in row:
            return row[name]
    return ""

def number(value):
    value = (value or "").strip().replace(",", "").replace("$", "")
    if value.endswith("%"):
        value = value[:-1]
    try:
        return float(value)
    except ValueError:
        return 0.0

campaign_daily_map = {}
for row in reader:
    pkg = pick(row, "维度1", "pkg", "Pkg").strip()
    adping_id = pick(row, "维度2", "adping_id", "adpingid").strip()
    campaign = campaign_by_key.get((pkg, adping_id))
    if not campaign:
        continue
    date = pick(row, "dt1", "dt", "date").strip()
    if not date:
        continue
    key = (date, pkg, adping_id)
    item = campaign_daily_map.setdefault(key, {
        "date": date, "pkg": pkg, "adping_id": adping_id,
        "group": campaign["group"], "geo": campaign["geo"],
        "platform": campaign["platform"], "owner": campaign["owner"],
        "impression": 0, "click": 0, "conversion": 0, "spend": 0.0,
    })
    item["impression"] += round(number(pick(row, "impression", "Impression")))
    item["click"] += round(number(pick(row, "click", "Click")))
    item["conversion"] += round(number(pick(row, "conversions", "conversion", "Conversion")))
    item["spend"] += number(pick(row, "spend", "Spend"))

def metrics(item):
    imp = item["impression"]
    click = item["click"]
    conv = item["conversion"]
    spend = item["spend"]
    item["spend"] = round(spend, 2)
    item["ctr"] = click / imp if imp else None
    item["cvr"] = conv / click if click else None
    item["ivr"] = conv / imp if imp else None
    item["cac"] = spend / conv if conv else None
    item["ecpi"] = spend / conv if conv else None
    return item

campaign_daily = [metrics(v) for _, v in sorted(campaign_daily_map.items())]
daily_map = {}
for row in campaign_daily:
    key = (row["date"], row["pkg"])
    item = daily_map.setdefault(key, {"date": row["date"], "pkg": row["pkg"], "impression": 0, "click": 0, "conversion": 0, "spend": 0.0})
    for field in ("impression", "click", "conversion", "spend"):
        item[field] += row[field]
daily = [metrics(v) for _, v in sorted(daily_map.items())]
summary = []
for pkg in active_pkgs:
    rows = [r for r in daily if r["pkg"] == pkg]
    total = {"pkg": pkg, "impression": sum(r["impression"] for r in rows), "click": sum(r["click"] for r in rows), "conversion": sum(r["conversion"] for r in rows), "spend": sum(r["spend"] for r in rows)}
    total["has_data"] = bool(rows)
    total["first_date"] = rows[0]["date"] if rows else None
    total["last_date"] = rows[-1]["date"] if rows else None
    summary.append(metrics(total))

dates = sorted({r["date"] for r in daily})
payload = {
    "meta": {
        "generated_at": dt.datetime.now(dt.timezone(dt.timedelta(hours=8))).isoformat(timespec="seconds"),
        "timezone": "Asia/Shanghai",
        "source": "Tableau / DSP Spend / DSP Daily Data",
        "source_url": "https://mars.ushareit.org/#/views/DSPSpend/DSPDailyData?:iid=1",
        "active_pkg_source": "DingTalk / Manta人群标签外投 / Campaign info",
        "date_start": dates[0] if dates else None,
        "date_end": dates[-1] if dates else None,
        "active_pkg_count": len(active_pkgs),
        "active_campaign_count": len(campaigns),
        "campaign_with_data_count": len({(r["pkg"], r["adping_id"]) for r in campaign_daily}),
        "pkg_with_data_count": sum(1 for x in summary if x["has_data"]),
        "definitions": {
            "CTR": "click / impression",
            "CVR": "conversion / click",
            "CAC": "spend / conversion",
            "IVR": "conversion / impression",
            "eCPI": "spend / conversion",
        },
    },
    "summary": summary,
    "daily": daily,
    "campaigns": campaigns,
    "campaign_daily": campaign_daily,
}

tmp = output_path.with_suffix(".json.tmp")
tmp.write_text(json.dumps(payload, ensure_ascii=False, separators=(",", ":")), encoding="utf-8")
tmp.replace(output_path)
PY