262 lines
11 KiB
Python
262 lines
11 KiB
Python
#!/usr/bin/env python3
|
|
"""Normalize CSV/TSV/XLSX ad reports into deterministic Ad-level evidence JSON."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import csv
|
|
import datetime as dt
|
|
import json
|
|
import math
|
|
import re
|
|
import sys
|
|
import zipfile
|
|
from collections import defaultdict
|
|
from pathlib import Path
|
|
from xml.etree import ElementTree as ET
|
|
|
|
NS = {"m": "http://schemas.openxmlformats.org/spreadsheetml/2006/main",
|
|
"r": "http://schemas.openxmlformats.org/officeDocument/2006/relationships"}
|
|
|
|
ALIASES = {
|
|
"date": ["date", "日期", "day", "report date"],
|
|
"campaign": ["campaign", "campaign name", "平台 campaign name", "external_campaign_name"],
|
|
"ad_group": ["ad group", "ad group name", "平台 ad group name", "external_ad_group_name", "adgroup"],
|
|
"ad": ["ad", "ad name", "平台 ad name", "external_ad_name", "广告名称", "素材名称"],
|
|
"spend": ["spend", "cost", "消耗", "花费", "广告花费"],
|
|
"installs": ["installs", "install", "归因安装量", "安装量", "转化数"],
|
|
"retained": ["d1 retained", "d1 retention users", "d1 留存人数", "次留人数", "次日留存人数"],
|
|
"impressions": ["impressions", "展示量", "曝光量"],
|
|
"clicks": ["clicks", "点击量", "归因点击量"],
|
|
}
|
|
|
|
|
|
def norm(value: object) -> str:
|
|
return re.sub(r"[\s_\-/]+", " ", str(value or "").strip().lower())
|
|
|
|
|
|
def col_index(reference: str) -> int:
|
|
letters = re.match(r"[A-Z]+", reference or "A").group(0)
|
|
n = 0
|
|
for char in letters:
|
|
n = n * 26 + ord(char) - 64
|
|
return n - 1
|
|
|
|
|
|
def xlsx_rows(path: Path, sheet_name: str | None = None):
|
|
with zipfile.ZipFile(path) as archive:
|
|
shared = []
|
|
if "xl/sharedStrings.xml" in archive.namelist():
|
|
root = ET.fromstring(archive.read("xl/sharedStrings.xml"))
|
|
for item in root.findall("m:si", NS):
|
|
shared.append("".join(node.text or "" for node in item.iter(f"{{{NS['m']}}}t")))
|
|
workbook = ET.fromstring(archive.read("xl/workbook.xml"))
|
|
rels_root = ET.fromstring(archive.read("xl/_rels/workbook.xml.rels"))
|
|
rels = {item.attrib["Id"]: item.attrib["Target"] for item in rels_root}
|
|
sheets = []
|
|
for sheet in workbook.find("m:sheets", NS):
|
|
target = rels[sheet.attrib[f"{{{NS['r']}}}id"]].lstrip("/")
|
|
if not target.startswith("xl/"):
|
|
target = "xl/" + target
|
|
sheets.append((sheet.attrib["name"], target))
|
|
if sheet_name:
|
|
matches = [entry for entry in sheets if entry[0] == sheet_name]
|
|
if not matches:
|
|
raise ValueError(f"Sheet not found: {sheet_name}; available: {', '.join(x[0] for x in sheets)}")
|
|
sheets = matches
|
|
name, target = sheets[0]
|
|
root = ET.fromstring(archive.read(target))
|
|
for row in root.findall(".//m:sheetData/m:row", NS):
|
|
values = {}
|
|
for cell in row.findall("m:c", NS):
|
|
idx = col_index(cell.attrib.get("r", "A1"))
|
|
kind = cell.attrib.get("t")
|
|
value_node = cell.find("m:v", NS)
|
|
inline = cell.find("m:is", NS)
|
|
raw = value_node.text if value_node is not None else ""
|
|
if kind == "s" and raw:
|
|
value = shared[int(raw)]
|
|
elif kind == "inlineStr" and inline is not None:
|
|
value = "".join(n.text or "" for n in inline.iter(f"{{{NS['m']}}}t"))
|
|
elif kind == "b":
|
|
value = raw == "1"
|
|
else:
|
|
value = raw
|
|
values[idx] = value
|
|
if values:
|
|
yield name, [values.get(i, "") for i in range(max(values) + 1)]
|
|
|
|
|
|
def tabular_rows(path: Path, sheet: str | None = None):
|
|
suffix = path.suffix.lower()
|
|
if suffix == ".xlsx":
|
|
iterator = xlsx_rows(path, sheet)
|
|
try:
|
|
sheet_name, headers = next(iterator)
|
|
except StopIteration:
|
|
raise ValueError("Workbook contains no rows")
|
|
for _, row in iterator:
|
|
yield sheet_name, headers, row
|
|
return
|
|
delimiter = "\t" if suffix == ".tsv" else ","
|
|
with path.open("r", encoding="utf-8-sig", newline="") as handle:
|
|
reader = csv.reader(handle, delimiter=delimiter)
|
|
try:
|
|
headers = next(reader)
|
|
except StopIteration:
|
|
raise ValueError("File contains no rows")
|
|
for row in reader:
|
|
yield path.name, headers, row
|
|
|
|
|
|
def resolve_columns(headers: list[str], mapping: dict[str, str] | None = None) -> dict[str, int]:
|
|
normalized = {norm(header): i for i, header in enumerate(headers)}
|
|
result = {}
|
|
for field, aliases in ALIASES.items():
|
|
requested = (mapping or {}).get(field)
|
|
candidates = [requested] if requested else aliases
|
|
for candidate in candidates:
|
|
if norm(candidate) in normalized:
|
|
result[field] = normalized[norm(candidate)]
|
|
break
|
|
missing = [field for field in ("date", "ad", "spend", "installs") if field not in result]
|
|
if missing:
|
|
raise ValueError(f"Missing required columns: {', '.join(missing)}. Headers: {headers}")
|
|
return result
|
|
|
|
|
|
def number(value: object) -> float:
|
|
text = str(value or "").strip().replace(",", "").replace("$", "").replace("¥", "")
|
|
if not text or text.lower() in {"nan", "none", "null", "-"}:
|
|
return 0.0
|
|
if text.endswith("%"):
|
|
return float(text[:-1]) / 100
|
|
return float(text)
|
|
|
|
|
|
def date_value(value: object) -> dt.date:
|
|
text = str(value or "").strip()
|
|
if re.fullmatch(r"\d+(\.\d+)?", text):
|
|
serial = int(float(text))
|
|
if serial > 1000:
|
|
return dt.date(1899, 12, 30) + dt.timedelta(days=serial)
|
|
for fmt in ("%Y-%m-%d", "%Y/%m/%d", "%Y.%m.%d", "%m/%d/%Y", "%d/%m/%Y"):
|
|
try:
|
|
return dt.datetime.strptime(text[:10], fmt).date()
|
|
except ValueError:
|
|
pass
|
|
try:
|
|
return dt.datetime.fromisoformat(text.replace("Z", "+00:00")).date()
|
|
except ValueError as exc:
|
|
raise ValueError(f"Unrecognized date: {text!r}") from exc
|
|
|
|
|
|
def ratio(numerator: float, denominator: float):
|
|
return None if denominator == 0 else numerator / denominator
|
|
|
|
|
|
def rounded(value):
|
|
if value is None:
|
|
return None
|
|
return round(value, 6)
|
|
|
|
|
|
def metrics(rows):
|
|
spend = sum(r["spend"] for r in rows)
|
|
installs = sum(r["installs"] for r in rows)
|
|
retained = sum(r["retained"] for r in rows)
|
|
return {
|
|
"spend": rounded(spend), "installs": rounded(installs), "retained": rounded(retained),
|
|
"cpi": rounded(ratio(spend, installs)),
|
|
"retention_rate": rounded(ratio(retained, installs)),
|
|
"retention_cost": rounded(ratio(spend, retained)),
|
|
"delivery_days": len({r["date"] for r in rows if r["spend"] > 0}),
|
|
}
|
|
|
|
|
|
def in_window(rows, end: dt.date, days: int):
|
|
start = end - dt.timedelta(days=days - 1)
|
|
return [row for row in rows if start <= row["date"] <= end]
|
|
|
|
|
|
def main(argv=None):
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
parser.add_argument("input", type=Path)
|
|
parser.add_argument("--analysis-date", help="YYYY-MM-DD; defaults to latest source date")
|
|
parser.add_argument("--retention-lag-days", type=int, default=1)
|
|
parser.add_argument("--sheet")
|
|
parser.add_argument("--mapping", type=Path, help="JSON mapping canonical fields to source headers")
|
|
parser.add_argument("--output", type=Path, required=True)
|
|
args = parser.parse_args(argv)
|
|
if not args.input.exists():
|
|
parser.error(f"Input not found: {args.input}")
|
|
mapping = json.loads(args.mapping.read_text(encoding="utf-8")) if args.mapping else None
|
|
records, source_headers, source_sheet = [], None, None
|
|
for source_sheet, source_headers, row in tabular_rows(args.input, args.sheet):
|
|
columns = resolve_columns(source_headers, mapping)
|
|
def get(field):
|
|
idx = columns.get(field)
|
|
return row[idx] if idx is not None and idx < len(row) else ""
|
|
if not any(str(v).strip() for v in row):
|
|
continue
|
|
ad = str(get("ad")).strip()
|
|
if not ad:
|
|
continue
|
|
records.append({
|
|
"date": date_value(get("date")), "campaign": str(get("campaign")).strip(),
|
|
"ad_group": str(get("ad_group")).strip(), "ad": ad,
|
|
"spend": number(get("spend")), "installs": number(get("installs")),
|
|
"retained": number(get("retained")), "impressions": number(get("impressions")),
|
|
"clicks": number(get("clicks")),
|
|
})
|
|
if not records:
|
|
raise ValueError("No usable Ad-level records found")
|
|
latest = max(row["date"] for row in records)
|
|
analysis_date = dt.date.fromisoformat(args.analysis_date) if args.analysis_date else latest
|
|
mature_date = analysis_date - dt.timedelta(days=args.retention_lag_days)
|
|
by_ad = defaultdict(list)
|
|
for row in records:
|
|
by_ad[(row["campaign"], row["ad_group"], row["ad"])].append(row)
|
|
ads = []
|
|
for (campaign, ad_group, ad), rows in by_ad.items():
|
|
all_dates = [r["date"] for r in rows]
|
|
acquisition_7 = in_window(rows, analysis_date, 7)
|
|
retention_7 = in_window(rows, mature_date, 7)
|
|
previous_7 = in_window(rows, mature_date - dt.timedelta(days=7), 7)
|
|
ads.append({
|
|
"ad": ad, "ad_group_context": ad_group or None, "campaign_context": campaign or None,
|
|
"first_date": min(all_dates).isoformat(), "last_date": max(all_dates).isoformat(),
|
|
"age_days_at_analysis": (analysis_date - min(all_dates)).days + 1,
|
|
"analysis_day": metrics([r for r in rows if r["date"] == analysis_date]),
|
|
"recent_7_acquisition": metrics(acquisition_7),
|
|
"recent_7_mature_retention": metrics(retention_7),
|
|
"previous_7_mature_retention": metrics(previous_7),
|
|
"recent_14_mature_retention": metrics(in_window(rows, mature_date, 14)),
|
|
})
|
|
ads.sort(key=lambda x: (-x["recent_7_acquisition"]["spend"], x["ad"]))
|
|
result = {
|
|
"contract_version": "1.0", "source": {"file": args.input.name, "sheet": source_sheet,
|
|
"row_count": len(records), "columns": source_headers},
|
|
"dates": {"analysis_date": analysis_date.isoformat(), "latest_source_date": latest.isoformat(),
|
|
"retention_mature_date": mature_date.isoformat(), "retention_lag_days": args.retention_lag_days},
|
|
"portfolio": {"analysis_day": metrics([r for r in records if r["date"] == analysis_date]),
|
|
"recent_7_acquisition": metrics(in_window(records, analysis_date, 7)),
|
|
"recent_7_mature_retention": metrics(in_window(records, mature_date, 7))},
|
|
"ads": ads,
|
|
"notes": ["Ratios are recomputed from summed numerators and denominators.",
|
|
"Campaign and Ad Group are identity context only; no parent-level ranking is produced.",
|
|
"This file contains evidence only. KPI verdicts and actions require the Skill decision gates."],
|
|
}
|
|
args.output.parent.mkdir(parents=True, exist_ok=True)
|
|
args.output.write_text(json.dumps(result, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
print(f"Wrote {args.output} ({len(ads)} Ads; analysis date {analysis_date})")
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
raise SystemExit(main())
|
|
except (ValueError, zipfile.BadZipFile, json.JSONDecodeError) as exc:
|
|
print(f"error: {exc}", file=sys.stderr)
|
|
raise SystemExit(2)
|