671 lines
26 KiB
Python
671 lines
26 KiB
Python
"""生成 top10-ffr 报表数据（多 KA，按 tab 展示）。
|
||
|
||
取数范围：当月 1 号至昨天。
|
||
口径：
|
||
- Revenue 来自各 KA 日报表的 Daily Spend / Spend / Cost（客户口径消耗）
|
||
- Cost 来自 Sands Report 2.0，按包名汇总（需本地抓取 sands_cost.json）
|
||
- GP = Revenue − Cost；GP rate = GP ÷ Revenue × 100
|
||
- Target 来自 Oversea FY2627 Monthly FFR 表对应 KA 行
|
||
|
||
读取优先公开 CSV 导出，失败回退本地 credentials.json。脚本不含任何凭据。
|
||
"""
|
||
|
||
import csv
|
||
import io
|
||
import json
|
||
import re
|
||
import sys
|
||
import urllib.request
|
||
from collections import defaultdict
|
||
from datetime import datetime, timedelta
|
||
from pathlib import Path
|
||
from zoneinfo import ZoneInfo
|
||
|
||
BASE = Path(__file__).resolve().parent
|
||
CONFIG_FILE = BASE / "accounts.json"
|
||
TIMEZONE = ZoneInfo("Asia/Shanghai")
|
||
SKIP_MARKS = ("总计", "合计", "total")
|
||
SPEND_NAMES = {"cost"}
|
||
SPEND_BLOCKLIST = ("cap", "budget", "rate", "expend")
|
||
INSTALL_BLOCKLIST = ("cap", "rate", "overall", "superset", "rejected", "reject")
|
||
|
||
|
||
def clean(value):
|
||
return " ".join(str(value or "").replace("\xa0", " ").split())
|
||
|
||
|
||
def norm(value):
|
||
return re.sub(r"\s+", "", str(value or "")).lower()
|
||
|
||
|
||
def cell(row, index):
|
||
if index is None or index < 0 or index >= len(row):
|
||
return ""
|
||
return row[index]
|
||
|
||
|
||
def letter(index):
|
||
if index < 26:
|
||
return chr(65 + index)
|
||
return chr(64 + index // 26) + chr(65 + index % 26)
|
||
|
||
|
||
def parse_date(value):
|
||
text = clean(value)
|
||
for fmt in ("%Y/%m/%d", "%Y-%m-%d", "%Y%m%d", "%m/%d/%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("$", "").replace("%", "")
|
||
if not text or text in {"-", "—"} or text.startswith("#"):
|
||
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 is_install_header(text):
|
||
"""Install 列有 install / installs / Installations / Install(new device) 等写法。"""
|
||
name = clean(text).lower()
|
||
if not name.startswith("install"):
|
||
return False
|
||
return not any(word in name for word in INSTALL_BLOCKLIST)
|
||
|
||
|
||
def is_spend_header(text):
|
||
"""Revenue 列写法：spend / Spend / Daily Spend / Spends USD / Cost。"""
|
||
name = clean(text).lower()
|
||
if name in SPEND_NAMES:
|
||
return True
|
||
if not (name.startswith("spend") or name.startswith("daily spend")):
|
||
return False
|
||
return not any(word in name for word in SPEND_BLOCKLIST)
|
||
|
||
|
||
def fetch_public(spreadsheet_id, gid):
|
||
url = (
|
||
f"https://docs.google.com/spreadsheets/d/{spreadsheet_id}"
|
||
f"/export?format=csv&gid={gid}"
|
||
)
|
||
request = urllib.request.Request(url, headers={"User-Agent": "AgentStage-Top10/1.0"})
|
||
with urllib.request.urlopen(request, timeout=120) as response:
|
||
return list(csv.reader(io.StringIO(response.read().decode("utf-8-sig"))))
|
||
|
||
|
||
def fetch_private(spreadsheet_id, gid):
|
||
import gspread
|
||
from google.oauth2.service_account import Credentials
|
||
|
||
candidates = [BASE / "credentials.json"] + [p / "credentials.json" for p in BASE.parents]
|
||
path = next((item for item in candidates if item.exists()), None)
|
||
if path is None:
|
||
raise FileNotFoundError("未找到 credentials.json")
|
||
creds = Credentials.from_service_account_file(
|
||
str(path), scopes=["https://www.googleapis.com/auth/spreadsheets.readonly"]
|
||
)
|
||
spreadsheet = gspread.authorize(creds).open_by_key(spreadsheet_id)
|
||
for worksheet in spreadsheet.worksheets():
|
||
if worksheet.id == gid:
|
||
return worksheet.get_all_values()
|
||
raise LookupError(f"未找到 gid={gid}")
|
||
|
||
|
||
def fetch_rows(spreadsheet_id, gid):
|
||
try:
|
||
rows = fetch_public(spreadsheet_id, gid)
|
||
if rows:
|
||
return rows, "public"
|
||
except Exception:
|
||
pass
|
||
return fetch_private(spreadsheet_id, gid), "credentials"
|
||
|
||
|
||
def find_blocks(rows):
|
||
"""定位表头行与其中所有横向区块（同一行可能有多套 Date/Install/Spend）。"""
|
||
for index, row in enumerate(rows[:40]):
|
||
date_columns = [i for i, v in enumerate(row) if clean(v).lower() == "date"]
|
||
if not date_columns:
|
||
continue
|
||
blocks = []
|
||
for position, start in enumerate(date_columns):
|
||
end = date_columns[position + 1] if position + 1 < len(date_columns) else len(row)
|
||
block = {"date": start}
|
||
for column in range(start, end):
|
||
name = cell(row, column)
|
||
if "install" not in block and is_install_header(name):
|
||
block["install"] = [column]
|
||
elif "spend" not in block and is_spend_header(name):
|
||
block["spend"] = [column]
|
||
if "install" in block or "spend" in block:
|
||
blocks.append(block)
|
||
if blocks:
|
||
return index, blocks
|
||
return None, []
|
||
|
||
|
||
def single_index(value):
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, int):
|
||
return value
|
||
text = clean(value).upper()
|
||
if not text:
|
||
return None
|
||
if text.isdigit():
|
||
return int(text)
|
||
index = 0
|
||
for char in text:
|
||
if not ("A" <= char <= "Z"):
|
||
return None
|
||
index = index * 26 + (ord(char) - 64)
|
||
return index - 1
|
||
|
||
|
||
def column_index(value):
|
||
"""支持单列或多列求和：'D' 或 ['D','Q','AD','AT']。"""
|
||
if value is None:
|
||
return None
|
||
if isinstance(value, (list, tuple)):
|
||
items = [single_index(item) for item in value]
|
||
items = [item for item in items if item is not None]
|
||
return items or None
|
||
index = single_index(value)
|
||
return None if index is None else [index]
|
||
|
||
|
||
def read_campaign_sheet(rows, start_date, end_date, label, overrides=None):
|
||
"""按表头自动识别；overrides 可显式指定表头行与列，避免取到本币列。"""
|
||
overrides = overrides or {}
|
||
forced_row = overrides.get("header_row")
|
||
if forced_row:
|
||
header_row = int(forced_row) - 1
|
||
if not 0 <= header_row < len(rows):
|
||
print(f" {label}: 指定的表头行 {forced_row} 超出范围，跳过")
|
||
return {}
|
||
block = {}
|
||
header = rows[header_row]
|
||
date_index = column_index(overrides.get("date_col"))
|
||
if date_index:
|
||
block["date"] = date_index[0]
|
||
for key, field in (("install", "install_col"), ("spend", "revenue_col")):
|
||
index = column_index(overrides.get(field))
|
||
if index:
|
||
block[key] = index
|
||
if "date" not in block:
|
||
found = next((i for i, v in enumerate(header) if clean(v).lower() == "date"), None)
|
||
if found is None:
|
||
print(f" {label}: 第 {forced_row} 行未找到 Date 列，跳过")
|
||
return {}
|
||
block["date"] = found
|
||
if "install" not in block:
|
||
found = next((i for i, v in enumerate(header) if is_install_header(v)), None)
|
||
if found is not None:
|
||
block["install"] = [found]
|
||
if "spend" not in block:
|
||
found = next((i for i, v in enumerate(header) if is_spend_header(v)), None)
|
||
if found is not None:
|
||
block["spend"] = [found]
|
||
# 防护：Revenue 列不能是 FFR、比率或本币列
|
||
for index in list(block.get("spend") or []):
|
||
name = clean(cell(header, index)).lower()
|
||
if any(word in name for word in ("ffr", "%", "rate", "r$")):
|
||
print(f" {label}: 拒绝把 {letter(index)}({name}) 当作 Revenue，已剔除")
|
||
block["spend"].remove(index)
|
||
if not block.get("spend"):
|
||
block.pop("spend", None)
|
||
|
||
def describe(key):
|
||
indexes = block.get(key)
|
||
if not indexes:
|
||
return f" {key}=缺失"
|
||
parts = "+".join(f"{letter(i)}({clean(cell(header, i))})" for i in indexes)
|
||
return f" {key}={parts}"
|
||
|
||
blocks = [block]
|
||
described = f"date={letter(block['date'])}" + describe("install") + describe("spend")
|
||
print(f" {label}: 指定表头第 {forced_row} 行，{described}")
|
||
return accumulate(rows, header_row, blocks, start_date, end_date, label)
|
||
|
||
header_row, blocks = find_blocks(rows)
|
||
if header_row is None:
|
||
print(f" {label}: 未识别到表头，跳过")
|
||
return {}
|
||
described = "; ".join(
|
||
f"date={letter(b['date'])}"
|
||
+ (f" install={letter(b['install'][0])}" if b.get("install") else "")
|
||
+ (f" spend={letter(b['spend'][0])}" if b.get("spend") else "")
|
||
for b in blocks
|
||
)
|
||
print(f" {label}: 表头第 {header_row + 1} 行，{len(blocks)} 个区块 [{described}]")
|
||
return accumulate(rows, header_row, blocks, start_date, end_date, label)
|
||
|
||
|
||
def accumulate(rows, header_row, blocks, start_date, end_date, label=""):
|
||
daily = defaultdict(lambda: {"installs": 0.0, "revenue": 0.0})
|
||
parsed = 0
|
||
in_range = 0
|
||
sample = []
|
||
found_dates = []
|
||
for row in rows[header_row + 1:]:
|
||
for block in blocks:
|
||
raw = clean(cell(row, block["date"]))
|
||
day = parse_date(raw)
|
||
if day is None:
|
||
if raw and len(sample) < 3:
|
||
sample.append(raw)
|
||
continue
|
||
parsed += 1
|
||
found_dates.append(day)
|
||
if not (start_date <= day <= end_date):
|
||
continue
|
||
in_range += 1
|
||
item = daily[day.isoformat()]
|
||
for index in block.get("install") or []:
|
||
value = parse_number(cell(row, index))
|
||
if value is not None:
|
||
item["installs"] += value
|
||
for index in block.get("spend") or []:
|
||
value = parse_number(cell(row, index))
|
||
if value is not None:
|
||
item["revenue"] += value
|
||
if label and in_range == 0:
|
||
header = rows[header_row] if header_row < len(rows) else []
|
||
column = blocks[0]["date"] if blocks else None
|
||
title = clean(cell(header, column)) if column is not None else ""
|
||
print(
|
||
f" 诊断 {label}: 日期列 {letter(column)}({title or '无表头'})，"
|
||
f"可解析 {parsed} 行，落在区间 0 行"
|
||
)
|
||
if found_dates:
|
||
print(f" 解析到的日期范围 {min(found_dates)} ~ {max(found_dates)}")
|
||
if sample:
|
||
print(f" 无法解析的样例 {sample}")
|
||
return daily
|
||
|
||
|
||
def read_king(sources, sheets, start_date, end_date):
|
||
"""King 为长表：Date / Campaign Name / Installs / Cost 每行一条。"""
|
||
rows, source = fetch_rows(sheets[sources["sheet"]], sources["client_gid"])
|
||
header_row = None
|
||
columns = {}
|
||
for index, row in enumerate(rows[:12]):
|
||
found = {}
|
||
for column, value in enumerate(row):
|
||
name = clean(value).lower()
|
||
if name == "date":
|
||
found.setdefault("date", column)
|
||
elif name in {"campaign name", "campaign"}:
|
||
found.setdefault("campaign", column)
|
||
elif is_install_header(value):
|
||
found.setdefault("install", column)
|
||
elif name == "cost":
|
||
found.setdefault("spend", column)
|
||
if len(found) == 4:
|
||
header_row, columns = index, found
|
||
break
|
||
if header_row is None:
|
||
raise SystemExit("King 表未识别到 Date / Campaign Name / Install / Cost")
|
||
|
||
print(f" King: 表头第 {header_row + 1} 行，来源={source}")
|
||
per_campaign = defaultdict(lambda: defaultdict(lambda: {"installs": 0.0, "revenue": 0.0}))
|
||
for row in rows[header_row + 1:]:
|
||
name = clean(cell(row, columns["campaign"]))
|
||
if not name or any(mark in name.lower() for mark in SKIP_MARKS):
|
||
continue
|
||
day = parse_date(cell(row, columns["date"]))
|
||
if day is None or not (start_date <= day <= end_date):
|
||
continue
|
||
item = per_campaign[name][day.isoformat()]
|
||
install = parse_number(cell(row, columns["install"]))
|
||
revenue = parse_number(cell(row, columns["spend"]))
|
||
if install is not None:
|
||
item["installs"] += install
|
||
if revenue is not None:
|
||
item["revenue"] += revenue
|
||
return per_campaign
|
||
|
||
|
||
def read_existing_report(sources, start_date, end_date):
|
||
"""Amazon：直接读取已有报表的 data.json。"""
|
||
url = sources["report_url"]
|
||
request = urllib.request.Request(url, headers={"User-Agent": "AgentStage-Top10/1.0"})
|
||
with urllib.request.urlopen(request, timeout=90) as response:
|
||
data = json.loads(response.read().decode("utf-8"))
|
||
revenue_field = sources.get("revenue_field", "payable_spend")
|
||
install_field = sources.get("install_field", "install")
|
||
daily = defaultdict(lambda: {"installs": 0.0, "revenue": 0.0})
|
||
cutoff = None
|
||
for row in data.get("daily", []):
|
||
day = parse_date(row.get("date"))
|
||
if day is None:
|
||
continue
|
||
if row.get("is_actual") is False:
|
||
continue
|
||
if cutoff is None or day > cutoff:
|
||
cutoff = day
|
||
if not (start_date <= day <= end_date):
|
||
continue
|
||
item = daily[day.isoformat()]
|
||
item["installs"] += float(row.get(install_field) or 0)
|
||
item["revenue"] += float(row.get(revenue_field) or 0)
|
||
print(f" Amazon: 使用 {revenue_field} 作为 Revenue，实际数据截止 {cutoff}")
|
||
return daily, cutoff
|
||
|
||
|
||
def read_targets(rows):
|
||
"""Monthly FFR：上一行是分组名（Target/Actual），下一行是 Rev/GP/GP%。"""
|
||
metric_row = None
|
||
for index, row in enumerate(rows[:12]):
|
||
names = [norm(v) for v in row]
|
||
if "rev" in names and "gp" in names:
|
||
metric_row = index
|
||
break
|
||
if metric_row is None:
|
||
raise LookupError("未找到 Rev / GP 指标行")
|
||
|
||
group_row = rows[metric_row - 1] if metric_row > 0 else []
|
||
columns = {}
|
||
current = ""
|
||
for column, value in enumerate(rows[metric_row]):
|
||
label = clean(cell(group_row, column))
|
||
if label:
|
||
current = label.lower()
|
||
metric = norm(value)
|
||
if metric in {"rev", "gp", "gp%"} and current:
|
||
columns.setdefault((current, metric), column)
|
||
|
||
targets = {}
|
||
for row in rows[metric_row + 1:]:
|
||
name = clean(cell(row, 0)).lower()
|
||
if not name:
|
||
continue
|
||
|
||
def pick(metric):
|
||
for (group, key), column in columns.items():
|
||
if key == metric and "target" in group:
|
||
return parse_number(cell(row, column))
|
||
return None
|
||
|
||
targets[name] = {"revenue": pick("rev"), "gp": pick("gp"), "gp_rate": pick("gp%")}
|
||
return targets
|
||
|
||
|
||
def load_sands():
|
||
candidates = [BASE / "sands_cost.json"] + [p / "sands_cost.json" for p in BASE.parents]
|
||
path = next((item for item in candidates if item.exists()), None)
|
||
if path is None:
|
||
previous = BASE / "data.json"
|
||
if previous.exists():
|
||
try:
|
||
sands = json.loads(previous.read_text(encoding="utf-8")).get("sands")
|
||
except Exception:
|
||
sands = None
|
||
if sands:
|
||
sands = dict(sands)
|
||
sands["carried_over"] = True
|
||
print(f"Sands：沿用上一版（抓取于 {sands.get('fetched_at')}）")
|
||
return sands
|
||
print("Sands：未找到 sands_cost.json，Cost 与 GP 将留空")
|
||
return None
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
by_pkg = {name: entry.get("cost") for name, entry in (data.get("by_pkg") or {}).items()}
|
||
print(f"Sands：{len(by_pkg)} 个包，抓取于 {data.get('fetched_at')}")
|
||
return {
|
||
"fetched_at": data.get("fetched_at"),
|
||
"source_url": data.get("source_url"),
|
||
"by_pkg": by_pkg,
|
||
"total_cost": data.get("total_cost"),
|
||
}
|
||
|
||
|
||
def account_cost(account, sands):
|
||
if not sands or not sands.get("by_pkg"):
|
||
return None, []
|
||
by_pkg = sands["by_pkg"]
|
||
total = 0.0
|
||
missing = []
|
||
found = False
|
||
for pkg in account.get("pkgs", []):
|
||
value = by_pkg.get(pkg)
|
||
if value is None:
|
||
missing.append(pkg)
|
||
else:
|
||
total += float(value)
|
||
found = True
|
||
return (round(total, 4) if found else None), missing
|
||
|
||
|
||
def build_account(account, sheets, start_date, end_date, sands, targets):
|
||
key, name, layout = account["key"], account["name"], account.get("layout")
|
||
print(f"\n[{name}] layout={layout}")
|
||
campaigns = {}
|
||
cutoff = None
|
||
|
||
if layout == "king":
|
||
per_campaign = read_king(account["sources"], sheets, start_date, end_date)
|
||
campaigns = {campaign: dict(days) for campaign, days in per_campaign.items()}
|
||
elif layout == "placeholder":
|
||
print(" 占位账号，无数据")
|
||
else:
|
||
# 同一个 KA 可以同时有已有报表和源表两种来源
|
||
sources = account.get("sources") or {}
|
||
if sources.get("report_url"):
|
||
daily, cutoff = read_existing_report(sources, start_date, end_date)
|
||
if daily:
|
||
campaigns[sources.get("campaign_name", name)] = dict(daily)
|
||
for campaign in account.get("campaigns", []):
|
||
try:
|
||
rows, source = fetch_rows(sheets[campaign["sheet"]], campaign["gid"])
|
||
except Exception as error:
|
||
print(f" {campaign['name']}: 读取失败 {type(error).__name__}: {error or '(无消息)'}")
|
||
continue
|
||
daily = read_campaign_sheet(
|
||
rows, start_date, end_date, f"{campaign['name']}({source})", campaign
|
||
)
|
||
if daily:
|
||
campaigns[campaign["name"]] = dict(daily)
|
||
|
||
# CPM 单的曝光不能与安装相加，单独统计
|
||
impression_campaigns = {
|
||
campaign.get("name"): campaign.get("metric_label")
|
||
for campaign in account.get("campaigns", [])
|
||
if str(campaign.get("metric_label", "")).lower().startswith("impression")
|
||
}
|
||
|
||
records = []
|
||
campaign_summary = []
|
||
for campaign, days in campaigns.items():
|
||
is_impression = campaign in impression_campaigns
|
||
volume = round(sum(item["installs"] for item in days.values()), 2)
|
||
revenue = round(sum(item["revenue"] for item in days.values()), 4)
|
||
campaign_summary.append({
|
||
"campaign": campaign,
|
||
"metric": "impression" if is_impression else "install",
|
||
"installs": 0 if is_impression else volume,
|
||
"impressions": volume if is_impression else 0,
|
||
"revenue": revenue,
|
||
"ecpi": None if is_impression or not volume else round(revenue / volume, 4),
|
||
"ecpm": round(revenue / volume * 1000, 4) if is_impression and volume else None,
|
||
})
|
||
for date, item in sorted(days.items()):
|
||
value = round(item["installs"], 2)
|
||
records.append({
|
||
"date": date,
|
||
"campaign": campaign,
|
||
"metric": "impression" if is_impression else "install",
|
||
"installs": 0 if is_impression else value,
|
||
"impressions": value if is_impression else 0,
|
||
"revenue": round(item["revenue"], 4),
|
||
})
|
||
|
||
campaign_summary.sort(key=lambda item: -item["revenue"])
|
||
revenue = round(sum(item["revenue"] for item in campaign_summary), 4)
|
||
installs = round(sum(item["installs"] for item in campaign_summary), 2)
|
||
impressions = round(sum(item["impressions"] for item in campaign_summary), 2)
|
||
cost, missing_pkgs = account_cost(account, sands)
|
||
gp = round(revenue - cost, 4) if cost is not None else None
|
||
gp_rate = round(gp / revenue * 100, 2) if gp is not None and revenue else None
|
||
|
||
target = dict(targets.get(account.get("target_ka", "").lower()) or {})
|
||
if target.get("gp_rate") is not None and target["gp_rate"] <= 1:
|
||
target["gp_rate"] = round(target["gp_rate"] * 100, 2)
|
||
|
||
def ratio(actual, goal):
|
||
if actual is None or not goal:
|
||
return None
|
||
return round(actual / goal * 100, 2)
|
||
|
||
print(
|
||
f" Revenue={revenue} Install={installs} Impression={impressions} "
|
||
f"Cost={cost} GP={gp} Target Rev={target.get('revenue')}"
|
||
)
|
||
if missing_pkgs:
|
||
print(f" Sands 缺少包：{missing_pkgs}")
|
||
|
||
return {
|
||
"key": key,
|
||
"name": name,
|
||
"layout": layout,
|
||
"note": account.get("note", ""),
|
||
"pkgs": account.get("pkgs", []),
|
||
"data_cutoff": cutoff.isoformat() if cutoff else None,
|
||
"actual": {
|
||
"installs": installs,
|
||
"impressions": impressions,
|
||
"revenue": revenue,
|
||
"cost": cost,
|
||
"gp": gp,
|
||
"gp_rate": gp_rate,
|
||
},
|
||
"target": {
|
||
"revenue": target.get("revenue"),
|
||
"gp": target.get("gp"),
|
||
"gp_rate": target.get("gp_rate"),
|
||
},
|
||
"completion": {
|
||
"revenue": ratio(revenue, target.get("revenue")),
|
||
"gp": ratio(gp, target.get("gp")),
|
||
},
|
||
"missing_pkgs": missing_pkgs,
|
||
"campaigns": campaign_summary,
|
||
"records": records,
|
||
}
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
raise SystemExit("用法: update_report_data.py <输出文件>")
|
||
config = json.loads(CONFIG_FILE.read_text(encoding="utf-8"))
|
||
sheets = config["sheets"]
|
||
|
||
now = datetime.now(TIMEZONE)
|
||
# 默认当月 1 号至昨天；--month YYYY-MM 可指定月份（如已结束的月份则取整月）
|
||
month_arg = next((item.split("=", 1)[-1] for item in sys.argv[2:] if item.startswith("--month")), None)
|
||
if month_arg:
|
||
year, month = (int(part) for part in month_arg.split("-")[:2])
|
||
start_date = datetime(year, month, 1).date()
|
||
else:
|
||
start_date = (now.date() - timedelta(days=1)).replace(day=1)
|
||
month_end = (start_date.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
|
||
end_date = min(month_end, now.date() - timedelta(days=1))
|
||
if end_date < start_date:
|
||
raise SystemExit(f"指定月份 {start_date:%Y-%m} 尚无已结束的日期")
|
||
days = (end_date - start_date).days + 1
|
||
month_days = month_end.day
|
||
print(f"取数范围：{start_date} 至 {end_date}（{days}/{month_days} 天）")
|
||
|
||
sands = load_sands()
|
||
# 判断 Cost 是否滞后于数据周期：抓取日在周期结束日当天或更早，说明未覆盖完整周期
|
||
if sands and sands.get("fetched_at"):
|
||
try:
|
||
fetched = datetime.fromisoformat(sands["fetched_at"]).date()
|
||
except ValueError:
|
||
fetched = None
|
||
if fetched:
|
||
sands["fetched_date"] = fetched.isoformat()
|
||
lag = (end_date - fetched).days + 1
|
||
sands["lag_days"] = max(0, lag)
|
||
sands["stale"] = lag > 0
|
||
if sands["stale"]:
|
||
print(f"警告：Sands Cost 抓取于 {fetched}，可能未覆盖到 {end_date}，滞后约 {lag} 天")
|
||
try:
|
||
target_rows, _ = fetch_rows(config["target_spreadsheet_id"], config["target_gid"])
|
||
targets = read_targets(target_rows)
|
||
print(f"Target：命中 {len(targets)} 个 KA")
|
||
except Exception as error:
|
||
targets = {}
|
||
print(f"Target 读取失败（{error}）")
|
||
|
||
accounts = [
|
||
build_account(account, sheets, start_date, end_date, sands, targets)
|
||
for account in config["accounts"]
|
||
]
|
||
active = [item for item in accounts if item["actual"]["revenue"] or item["actual"]["cost"]]
|
||
|
||
totals = {
|
||
"installs": round(sum(item["actual"]["installs"] for item in active), 2),
|
||
"impressions": round(sum(item["actual"].get("impressions") or 0 for item in active), 2),
|
||
"revenue": round(sum(item["actual"]["revenue"] for item in active), 4),
|
||
"cost": round(sum(item["actual"]["cost"] or 0 for item in active), 4),
|
||
"target_revenue": round(sum(item["target"]["revenue"] or 0 for item in active), 4),
|
||
"target_gp": round(sum(item["target"]["gp"] or 0 for item in active), 4),
|
||
}
|
||
totals["gp"] = round(totals["revenue"] - totals["cost"], 4)
|
||
totals["gp_rate"] = round(totals["gp"] / totals["revenue"] * 100, 2) if totals["revenue"] else None
|
||
totals["completion"] = {
|
||
"revenue": round(totals["revenue"] / totals["target_revenue"] * 100, 2) if totals["target_revenue"] else None,
|
||
"gp": round(totals["gp"] / totals["target_gp"] * 100, 2) if totals["target_gp"] else None,
|
||
}
|
||
|
||
payload = {
|
||
"title": "Top10 KA · 当月 FFR 与 GP",
|
||
"generated_at": now.isoformat(timespec="seconds"),
|
||
"timezone": "Asia/Shanghai",
|
||
"period": {
|
||
"start": start_date.isoformat(),
|
||
"end": end_date.isoformat(),
|
||
"days": days,
|
||
"month_days": month_days,
|
||
"time_progress_percent": round(days / month_days * 100, 2),
|
||
},
|
||
"sands": sands,
|
||
"totals": totals,
|
||
"accounts": accounts,
|
||
"definitions": {
|
||
"revenue": "各 KA 日报表的 Daily Spend / Spend / Cost，客户口径消耗，即我们的 Revenue",
|
||
"cost": "Sands Report 2.0 按包名汇总的 Cost，需本地抓取",
|
||
"gp": "GP = Revenue − Cost；GP rate = GP ÷ Revenue × 100",
|
||
"target": "Oversea FY2627 Monthly FFR 表对应 KA 行的 Target Rev / GP / GP%",
|
||
"completion": "完成率 = 实际 ÷ Target × 100，仅用于 Revenue 与 GP",
|
||
"impressions": "CPM 单（如 CapitecFuneral）的曝光量，单独统计，不并入 Install",
|
||
},
|
||
}
|
||
|
||
with open(sys.argv[1], "w", encoding="utf-8") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
print(
|
||
f"\nOK accounts={len(accounts)} active={len(active)} "
|
||
f"revenue={totals['revenue']} cost={totals['cost']} gp={totals['gp']}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|