269 lines
10 KiB
Python
269 lines
10 KiB
Python
"""生成 Ozon 报告数据。
|
||
|
||
取数范围：当月 1 号至昨天。
|
||
指标：Install、Revenue、CAC from client report。
|
||
Revenue 取源表的 Daily Spend 列（该列即收入口径）。
|
||
月度指标：Current Month Budget、FFR，按 Total/KZ/UZ/KG/AM 读取源表。
|
||
|
||
列位置通过表头识别，不依赖固定列号。脚本不包含任何账号、Token 或私钥。
|
||
"""
|
||
|
||
import csv
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.request
|
||
from datetime import datetime, timedelta
|
||
from zoneinfo import ZoneInfo
|
||
|
||
CSV_URL = (
|
||
"https://docs.google.com/spreadsheets/d/"
|
||
"1KaNOwaZ3DqjTXRK8XhF1OGF9qUmOh0NLXOgxVrh14h0/"
|
||
"export?format=csv&gid=1445753035"
|
||
)
|
||
TIMEZONE = ZoneInfo("Asia/Shanghai")
|
||
COUNTRIES = ["KZ", "UZ", "KG", "AM"]
|
||
# 平台只支持“每隔几小时”，因此按小时调度，仅在此时刻真正更新（Asia/Shanghai）
|
||
UPDATE_HOUR = 11
|
||
|
||
DATE_HEADERS = {"date"}
|
||
GEO_HEADERS = {"geo", "country"}
|
||
INSTALL_HEADERS = {"install", "installs"}
|
||
"""Revenue 使用源表 Daily Spend 列；Revenue 表头若存在则优先使用。"""
|
||
REVENUE_HEADERS = {"revenue", "income from client", "income from client($)"}
|
||
REVENUE_FALLBACK_HEADERS = {"daily spend", "daily spend($)", "spend"}
|
||
CAC_HEADERS = {"cac from client report", "cac from client report($)", "cac"}
|
||
BUDGET_LABELS = {"current month budget", "month budget", "budget"}
|
||
FFR_LABELS = {"ffr%", "ffr", "ffr(%)"}
|
||
|
||
|
||
def clean(value):
|
||
return " ".join(str(value or "").replace("\xa0", " ").split())
|
||
|
||
|
||
def key(value):
|
||
return clean(value).lower()
|
||
|
||
|
||
def cell(row, index):
|
||
if index is None or index < 0 or index >= len(row):
|
||
return ""
|
||
return row[index]
|
||
|
||
|
||
def parse_date(value):
|
||
text = clean(value)
|
||
for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%Y%m%d", "%m/%d/%Y", "%d/%m/%Y"):
|
||
try:
|
||
return datetime.strptime(text, fmt).date()
|
||
except ValueError:
|
||
pass
|
||
match = re.fullmatch(r"(\d{4})\D{1,3}(\d{1,2})\D{1,3}(\d{1,2})", text)
|
||
if not match:
|
||
return None
|
||
try:
|
||
return datetime(*map(int, match.groups())).date()
|
||
except ValueError:
|
||
return None
|
||
|
||
|
||
def parse_number(value):
|
||
text = clean(value).replace(",", "").replace("%", "")
|
||
if not text or text in {"-", "—", "#DIV/0!", "#N/A", "#REF!", "#VALUE!"}:
|
||
return None
|
||
negative = text.startswith("(") and text.endswith(")")
|
||
cleaned = re.sub(r"[^0-9.\-]", "", text)
|
||
if cleaned in {"", "-", ".", "-."}:
|
||
return None
|
||
try:
|
||
number = float(cleaned)
|
||
except ValueError:
|
||
return None
|
||
return -number if negative else number
|
||
|
||
|
||
def find_blocks(rows):
|
||
"""按表头识别每个国家的横向数据区块。"""
|
||
blocks = []
|
||
for row_index, row in enumerate(rows):
|
||
date_columns = [i for i, value in enumerate(row) if key(value) in DATE_HEADERS]
|
||
if not date_columns:
|
||
continue
|
||
for position, start in enumerate(date_columns):
|
||
end = date_columns[position + 1] if position + 1 < len(date_columns) else len(row)
|
||
block = {"header_row": row_index, "date": start}
|
||
for column in range(start, end):
|
||
name = key(cell(row, column))
|
||
if name in GEO_HEADERS:
|
||
block.setdefault("geo", column)
|
||
elif name in INSTALL_HEADERS:
|
||
block.setdefault("installs", column)
|
||
elif name in REVENUE_HEADERS:
|
||
block["revenue"] = column
|
||
elif name in REVENUE_FALLBACK_HEADERS:
|
||
block.setdefault("revenue_fallback", column)
|
||
elif name in CAC_HEADERS:
|
||
block.setdefault("cac", column)
|
||
if "revenue" not in block and "revenue_fallback" in block:
|
||
block["revenue"] = block["revenue_fallback"]
|
||
if "geo" in block and "installs" in block:
|
||
blocks.append(block)
|
||
return blocks
|
||
|
||
|
||
def find_monthly_metrics(rows):
|
||
"""读取顶部 Current Month Budget 与 FFR 的 Total/各国数值。"""
|
||
metrics = {"budget": {}, "ffr": {}}
|
||
for row_index, row in enumerate(rows[:30]):
|
||
labels = {key(value) for value in row}
|
||
if "total" not in labels:
|
||
continue
|
||
columns = {}
|
||
for column, value in enumerate(row):
|
||
name = clean(value).upper()
|
||
if name == "TOTAL":
|
||
columns["TOTAL"] = column
|
||
elif name in COUNTRIES:
|
||
columns[name] = column
|
||
if len(columns) < 2:
|
||
continue
|
||
window = rows[row_index: min(row_index + 12, len(rows))]
|
||
for line in window:
|
||
labels_in_line = [key(value) for value in line]
|
||
target = None
|
||
if any(name in BUDGET_LABELS for name in labels_in_line):
|
||
target = "budget"
|
||
elif any(name in FFR_LABELS for name in labels_in_line):
|
||
target = "ffr"
|
||
if target is None:
|
||
continue
|
||
for name, column in columns.items():
|
||
number = parse_number(cell(line, column))
|
||
if number is not None and name not in metrics[target]:
|
||
metrics[target][name] = number
|
||
if metrics["budget"] or metrics["ffr"]:
|
||
break
|
||
return metrics
|
||
|
||
|
||
def build_payload(rows, now):
|
||
end_date = now.date() - timedelta(days=1)
|
||
start_date = end_date.replace(day=1)
|
||
days = (end_date - start_date).days + 1
|
||
|
||
blocks = find_blocks(rows)
|
||
if not blocks:
|
||
raise SystemExit("未能通过表头识别国家数据区块，已中止，不覆盖旧数据。")
|
||
|
||
daily = {}
|
||
for row in rows:
|
||
for block in blocks:
|
||
day = parse_date(cell(row, block["date"]))
|
||
if day is None or not (start_date <= day <= end_date):
|
||
continue
|
||
country = clean(cell(row, block["geo"])).upper()
|
||
if country not in COUNTRIES:
|
||
continue
|
||
item = daily.setdefault(
|
||
(country, day.isoformat()),
|
||
{"installs": 0.0, "revenue": 0.0, "cac": None},
|
||
)
|
||
item["installs"] += parse_number(cell(row, block.get("installs"))) or 0.0
|
||
item["revenue"] += parse_number(cell(row, block.get("revenue"))) or 0.0
|
||
cac = parse_number(cell(row, block.get("cac")))
|
||
if cac is not None and item["cac"] is None:
|
||
item["cac"] = cac
|
||
|
||
records = []
|
||
for country in COUNTRIES:
|
||
for offset in range(days):
|
||
day = (start_date + timedelta(days=offset)).isoformat()
|
||
values = daily.get((country, day))
|
||
records.append({
|
||
"date": day,
|
||
"country": country,
|
||
"installs": round(values["installs"], 2) if values else 0,
|
||
"revenue": round(values["revenue"], 4) if values else 0,
|
||
"cac": values["cac"] if values else None,
|
||
})
|
||
|
||
monthly = find_monthly_metrics(rows)
|
||
summary = []
|
||
for country in COUNTRIES:
|
||
country_rows = [row for row in records if row["country"] == country]
|
||
cac_values = [row["cac"] for row in country_rows if row["cac"] is not None]
|
||
summary.append({
|
||
"country": country,
|
||
"installs": round(sum(row["installs"] for row in country_rows), 2),
|
||
"revenue": round(sum(row["revenue"] for row in country_rows), 4),
|
||
"avg_cac": round(sum(cac_values) / len(cac_values), 4) if cac_values else None,
|
||
"cac_days": len(cac_values),
|
||
"budget_usd": monthly["budget"].get(country),
|
||
"ffr_percent": monthly["ffr"].get(country),
|
||
})
|
||
|
||
total_budget = monthly["budget"].get("TOTAL")
|
||
if total_budget is None:
|
||
country_budgets = [row["budget_usd"] for row in summary if row["budget_usd"] is not None]
|
||
total_budget = round(sum(country_budgets), 4) if country_budgets else None
|
||
|
||
kpis = {
|
||
"TOTAL": {
|
||
"installs": round(sum(row["installs"] for row in records), 2),
|
||
"revenue": round(sum(row["revenue"] for row in records), 4),
|
||
"budget_usd": total_budget,
|
||
"ffr_percent": monthly["ffr"].get("TOTAL"),
|
||
}
|
||
}
|
||
for row in summary:
|
||
kpis[row["country"]] = {
|
||
"installs": row["installs"],
|
||
"revenue": row["revenue"],
|
||
"budget_usd": row["budget_usd"],
|
||
"ffr_percent": row["ffr_percent"],
|
||
}
|
||
|
||
return {
|
||
"title": "Ozon · 当月投放表现",
|
||
"generated_at": now.isoformat(timespec="seconds"),
|
||
"timezone": "Asia/Shanghai",
|
||
"period": {"start": start_date.isoformat(), "end": end_date.isoformat(), "days": days},
|
||
"countries": COUNTRIES,
|
||
"kpis": kpis,
|
||
"summary": summary,
|
||
"records": records,
|
||
"definitions": {
|
||
"period": "当月 1 号至昨天",
|
||
"installs": "各国家区块的 Install",
|
||
"revenue": "各国家区块的 Revenue（USD），取源表 Daily Spend 列",
|
||
"cac": "各国家区块的 CAC from client report 原始值；不改算、不标记为 USD",
|
||
"budget_usd": "源表 Current Month Budget，按 Total 与各国读取",
|
||
"ffr_percent": "源表 FFR，按 Total 与各国读取",
|
||
},
|
||
}
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
raise SystemExit("用法: update_report_data.py <输出文件> [--force]")
|
||
|
||
now = datetime.now(TIMEZONE)
|
||
forced = "--force" in sys.argv[2:]
|
||
if not forced and now.hour != UPDATE_HOUR:
|
||
print(f"SKIP {now:%Y-%m-%d %H:%M} 非 {UPDATE_HOUR}:00（Asia/Shanghai），不更新")
|
||
return
|
||
|
||
request = urllib.request.Request(CSV_URL, headers={"User-Agent": "AgentStage-Ozon-Report/2.0"})
|
||
with urllib.request.urlopen(request, timeout=90) as response:
|
||
text = response.read().decode("utf-8-sig")
|
||
rows = list(csv.reader(text.splitlines()))
|
||
payload = build_payload(rows, now)
|
||
with open(sys.argv[1], "w", encoding="utf-8") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
period = payload["period"]
|
||
print(f"OK {period['start']}..{period['end']} records={len(payload['records'])}")
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|