581 lines
22 KiB
Python
581 lines
22 KiB
Python
"""生成 King 各 Campaign 报告数据。
|
||
|
||
取数范围：当月 1 号至昨天。
|
||
指标：Install、Cost（消耗）、Monthly Budget、FFR、CPI。
|
||
|
||
数据来源（同一份 Spreadsheet）：
|
||
- Cost_Client（gid=0）长表：A=Date, B=Campaign Name, G=Installs, H=Cost, I=App
|
||
- Breakdown_cost【day】（gid=1217689915）宽表：按天消耗，作为消耗主来源
|
||
- forPrebill（gid=450129149）：B 列 Campaign name，Z 列当月预算
|
||
|
||
读取优先使用公开 CSV 导出；失败时回退到本地 credentials.json（仅本机可用）。
|
||
脚本不包含任何账号、Token 或私钥。
|
||
"""
|
||
|
||
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
|
||
|
||
SPREADSHEET_ID = "1zb6UR9d9j5Xn3ip6LcWEAsXBjLuPhaaQCXXI7C0gyg0"
|
||
CLIENT_GID = 0
|
||
COST_GID = 1217689915
|
||
BUDGET_GID = 450129149
|
||
TARGET_SPREADSHEET_ID = "1Bnt94y2ri3ISxF07tbzfVOLKfxaHAo__swNhmSOdccw"
|
||
TARGET_GID = 0
|
||
TARGET_KA = "king"
|
||
BUDGET_NAME_COL = 1
|
||
BUDGET_VALUE_COL = 25
|
||
TIMEZONE = ZoneInfo("Asia/Shanghai")
|
||
SKIP_MARKS = ("总计", "合计", "total")
|
||
|
||
|
||
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 column_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_subtotal(text):
|
||
lowered = clean(text).lower()
|
||
return any(mark in lowered for mark in SKIP_MARKS)
|
||
|
||
|
||
def fetch_public(gid, spreadsheet_id=SPREADSHEET_ID):
|
||
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-King-Report/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(gid, spreadsheet_id=SPREADSHEET_ID):
|
||
import gspread
|
||
from google.oauth2.service_account import Credentials
|
||
|
||
here = Path(__file__).resolve()
|
||
candidates = [here.parent / "credentials.json"] + [
|
||
parent / "credentials.json" for parent in here.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(gid, label, spreadsheet_id=SPREADSHEET_ID):
|
||
try:
|
||
rows = fetch_public(gid, spreadsheet_id)
|
||
if rows:
|
||
print(f" {label}: 公开导出成功，{len(rows)} 行")
|
||
return rows, "public"
|
||
reason = "空表"
|
||
except Exception as error:
|
||
reason = error
|
||
rows = fetch_private(gid, spreadsheet_id)
|
||
print(f" {label}: 公开导出不可用（{reason}），已使用本地凭据，{len(rows)} 行")
|
||
return rows, "credentials"
|
||
|
||
|
||
def find_header(rows, required):
|
||
"""在前若干行中寻找同时包含指定表头的行，返回列索引映射。"""
|
||
for index, row in enumerate(rows[:12]):
|
||
found = {}
|
||
for column, value in enumerate(row):
|
||
name = clean(value).lower()
|
||
for key, aliases in required.items():
|
||
if name in aliases and key not in found:
|
||
found[key] = column
|
||
if all(key in found for key in required):
|
||
return index, found
|
||
return None, {}
|
||
|
||
|
||
def read_client_long(rows, start_date, end_date):
|
||
"""Cost_Client 长表：每行一个 Campaign 的某一天数据。"""
|
||
header_row, columns = find_header(
|
||
rows,
|
||
{
|
||
"date": {"date"},
|
||
"campaign": {"campaign name", "campaign"},
|
||
"installs": {"installs", "install"},
|
||
"cost": {"cost"},
|
||
},
|
||
)
|
||
if header_row is None:
|
||
raise SystemExit("Cost_Client 未识别到 Date / Campaign Name / Installs / Cost 表头")
|
||
app_column = next(
|
||
(i for i, value in enumerate(rows[header_row]) if clean(value).lower() == "app"), None
|
||
)
|
||
print(
|
||
f" 表头第 {header_row + 1} 行："
|
||
f"日期={column_letter(columns['date'])} Campaign={column_letter(columns['campaign'])} "
|
||
f"Install={column_letter(columns['installs'])} Cost={column_letter(columns['cost'])}"
|
||
)
|
||
|
||
installs = defaultdict(float)
|
||
costs = defaultdict(float)
|
||
apps = {}
|
||
display = {}
|
||
latest = None
|
||
for row in rows[header_row + 1:]:
|
||
name = clean(cell(row, columns["campaign"]))
|
||
if not name or is_subtotal(name):
|
||
continue
|
||
day = parse_date(cell(row, columns["date"]))
|
||
if day is None:
|
||
continue
|
||
if latest is None or day > latest:
|
||
latest = day
|
||
if not (start_date <= day <= end_date):
|
||
continue
|
||
key = norm(name)
|
||
display.setdefault(key, name)
|
||
if app_column is not None:
|
||
app = clean(cell(row, app_column))
|
||
if app and not is_subtotal(app):
|
||
apps.setdefault(key, app)
|
||
install = parse_number(cell(row, columns["installs"]))
|
||
cost = parse_number(cell(row, columns["cost"]))
|
||
if install is not None:
|
||
installs[(key, day.isoformat())] += install
|
||
if cost is not None:
|
||
costs[(key, day.isoformat())] += cost
|
||
print(f" 数据截止 {latest}，命中 {len(display)} 个 Campaign")
|
||
return installs, costs, apps, display
|
||
|
||
|
||
def read_cost_wide(rows, start_date, end_date):
|
||
"""Breakdown_cost【day】宽表：日期横向排列。"""
|
||
date_row = None
|
||
date_columns = {}
|
||
for index, row in enumerate(rows[:12]):
|
||
found = {}
|
||
for column, value in enumerate(row):
|
||
day = parse_date(value)
|
||
if day is not None:
|
||
found[column] = day
|
||
if len(found) >= 3:
|
||
date_row, date_columns = index, found
|
||
break
|
||
if date_row is None:
|
||
print(" 未识别到日期行，跳过该来源")
|
||
return {}, {}, None
|
||
|
||
campaign_col = next(
|
||
(i for i, value in enumerate(rows[date_row]) if "campaign" in clean(value).lower()), None
|
||
)
|
||
if campaign_col is None:
|
||
print(" 未识别到 Campaign 列，跳过该来源")
|
||
return {}, {}, None
|
||
app_col = next(
|
||
(i for i, value in enumerate(rows[date_row]) if clean(value).lower() == "app"), None
|
||
)
|
||
in_range = {c: d for c, d in date_columns.items() if start_date <= d <= end_date}
|
||
latest = max(date_columns.values())
|
||
print(
|
||
f" 日期行第 {date_row + 1} 行，Campaign={column_letter(campaign_col)}，"
|
||
f"命中 {len(in_range)} 个日期列，数据截止 {latest}"
|
||
)
|
||
|
||
costs = defaultdict(float)
|
||
apps = {}
|
||
for row in rows[date_row + 1:]:
|
||
name = clean(cell(row, campaign_col))
|
||
if not name or is_subtotal(name):
|
||
continue
|
||
key = norm(name)
|
||
if app_col is not None:
|
||
app = clean(cell(row, app_col))
|
||
if app and not is_subtotal(app):
|
||
apps.setdefault(key, app)
|
||
for column, day in in_range.items():
|
||
number = parse_number(cell(row, column))
|
||
if number is not None:
|
||
costs[(key, day.isoformat())] += number
|
||
return costs, apps, latest
|
||
|
||
|
||
def split_name(name):
|
||
"""从 Campaign 名提取品牌前缀与国家代码，用于跨命名规则匹配。"""
|
||
tokens = [item for item in re.split(r"[_\s]+", clean(name)) if item]
|
||
brand = tokens[0].lower() if tokens else ""
|
||
geo = ""
|
||
for token in tokens[1:3]:
|
||
if re.fullmatch(r"[A-Za-z]{2}", token):
|
||
geo = token.lower()
|
||
break
|
||
if not geo and re.fullmatch(r"[A-Za-z]{2}", brand):
|
||
# 只写了国家代码的行，例如 B 列为 "DE"
|
||
return "", brand
|
||
return brand, geo
|
||
|
||
|
||
def read_budgets(rows):
|
||
header = clean(cell(rows[0], BUDGET_VALUE_COL)) if rows else ""
|
||
exact = {}
|
||
by_brand_geo = defaultdict(list)
|
||
by_geo = defaultdict(list)
|
||
for row in rows[1:]:
|
||
name = clean(cell(row, BUDGET_NAME_COL))
|
||
if not name or is_subtotal(name):
|
||
continue
|
||
number = parse_number(cell(row, BUDGET_VALUE_COL))
|
||
if number is None or number <= 0:
|
||
continue
|
||
exact.setdefault(norm(name), number)
|
||
brand, geo = split_name(name)
|
||
if brand and geo:
|
||
by_brand_geo[(brand, geo)].append(number)
|
||
elif geo:
|
||
by_geo[geo].append(number)
|
||
print(
|
||
f" Z 列表头={header or '(空)'}，精确 {len(exact)} 条，"
|
||
f"品牌+国家 {len(by_brand_geo)} 条，仅国家 {len(by_geo)} 条"
|
||
)
|
||
return {"exact": exact, "brand_geo": by_brand_geo, "geo": by_geo}, header
|
||
|
||
|
||
def resolve_budget(name, budgets):
|
||
"""三级匹配：精确名 -> 品牌+国家 -> 仅国家（唯一时）。"""
|
||
value = budgets["exact"].get(norm(name))
|
||
if value is not None:
|
||
return value, "exact"
|
||
brand, geo = split_name(name)
|
||
if brand and geo:
|
||
values = budgets["brand_geo"].get((brand, geo))
|
||
if values and len(set(values)) == 1:
|
||
return values[0], "brand+geo"
|
||
if geo:
|
||
values = budgets["geo"].get(geo)
|
||
if values and len(set(values)) == 1:
|
||
return values[0], "geo"
|
||
return None, None
|
||
|
||
|
||
def read_targets(rows):
|
||
"""解析 Monthly FFR 表：上一行是分组名（Target/Actual），下一行是 Rev/GP/GP%。"""
|
||
metric_row = None
|
||
for index, row in enumerate(rows[:12]):
|
||
names = [clean(value).lower().replace(" ", "") for value 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 = clean(value).lower().replace(" ", "")
|
||
if metric in {"rev", "gp", "gp%"} and current:
|
||
columns.setdefault((current, metric), column)
|
||
|
||
target_row = None
|
||
for row in rows[metric_row + 1:]:
|
||
if clean(cell(row, 0)).lower() == TARGET_KA:
|
||
target_row = row
|
||
break
|
||
if target_row is None:
|
||
raise LookupError(f"未找到 KA 为 {TARGET_KA} 的行")
|
||
|
||
def pick(group, metric):
|
||
for (name, key), column in columns.items():
|
||
if key == metric and group in name:
|
||
return parse_number(cell(target_row, column))
|
||
return None
|
||
|
||
revenue = pick("target", "rev")
|
||
gp = pick("target", "gp")
|
||
gp_rate = pick("target", "gp%")
|
||
print(f" Target: revenue={revenue} gp={gp} gp_rate={gp_rate}")
|
||
return {"revenue": revenue, "gp": gp, "gp_rate": gp_rate}
|
||
|
||
|
||
def load_sands_cost():
|
||
"""读取本地抓取的 Sands Cost（登录站点，服务器无法自动获取）。"""
|
||
here = Path(__file__).resolve()
|
||
candidates = [here.parent / "sands_cost.json"] + [
|
||
parent / "sands_cost.json" for parent in here.parents
|
||
]
|
||
path = next((item for item in candidates if item.exists()), None)
|
||
if path is None:
|
||
# 服务器上没有 sands_cost.json（Sands 需登录），沿用上一版 data.json 里的值
|
||
previous = here.parent / "data.json"
|
||
if previous.exists():
|
||
try:
|
||
sands = (json.loads(previous.read_text(encoding="utf-8")).get("performance") or {}).get("sands")
|
||
except Exception:
|
||
sands = None
|
||
if sands and sands.get("cost"):
|
||
sands = dict(sands)
|
||
sands["carried_over"] = True
|
||
print(f" Sands Cost：沿用上一版 {sands['cost']}（抓取于 {sands.get('fetched_at') or '未知'}）")
|
||
return sands
|
||
print(" Sands Cost：未找到 sands_cost.json，将留空")
|
||
return None
|
||
data = json.loads(path.read_text(encoding="utf-8"))
|
||
king = data.get("king_total") or {}
|
||
# 优先使用 King 两个包的合计；没有则退回整表合计
|
||
if king.get("cost"):
|
||
cost, scope = king["cost"], f"King 包合计 {king.get('pkgs')}"
|
||
missing = king.get("missing_pkgs") or []
|
||
if missing:
|
||
print(f" 警告：Sands 结果中缺少包 {missing}，Cost 可能不完整")
|
||
else:
|
||
cost, scope = data.get("total_cost"), "整表合计"
|
||
print(f" Sands Cost：{cost}（{scope}，抓取于 {data.get('fetched_at') or '未知'}）")
|
||
return {
|
||
"cost": cost,
|
||
"scope": scope,
|
||
"pkgs": king.get("pkgs"),
|
||
"missing_pkgs": king.get("missing_pkgs"),
|
||
"fetched_at": data.get("fetched_at"),
|
||
"source_url": data.get("source_url"),
|
||
"detail_rows": data.get("detail_rows"),
|
||
}
|
||
|
||
|
||
def build_performance(revenue, sands, targets, period_days, month_days):
|
||
"""Revenue 取日报 Cost；Cost 取 Sands；GP = Revenue - Cost。"""
|
||
cost = sands.get("cost") if sands else None
|
||
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
|
||
|
||
def ratio(actual, target):
|
||
if actual is None or not target:
|
||
return None
|
||
return round(actual / target * 100, 2)
|
||
|
||
target_gp_rate = targets.get("gp_rate") if targets else None
|
||
if target_gp_rate is not None and target_gp_rate <= 1:
|
||
target_gp_rate = round(target_gp_rate * 100, 2)
|
||
|
||
return {
|
||
"actual": {"revenue": round(revenue, 4), "cost": cost, "gp": gp, "gp_rate": gp_rate},
|
||
"target": {
|
||
"revenue": targets.get("revenue") if targets else None,
|
||
"gp": targets.get("gp") if targets else None,
|
||
"gp_rate": target_gp_rate,
|
||
},
|
||
"completion": {
|
||
"revenue": ratio(revenue, targets.get("revenue") if targets else None),
|
||
"gp": ratio(gp, targets.get("gp") if targets else None),
|
||
"gp_rate": ratio(gp_rate, target_gp_rate),
|
||
},
|
||
"time_progress_percent": round(period_days / month_days * 100, 2) if month_days else None,
|
||
"month_days": month_days,
|
||
"elapsed_days": period_days,
|
||
"sands": sands,
|
||
}
|
||
|
||
|
||
def build_payload(now):
|
||
end_date = now.date() - timedelta(days=1)
|
||
start_date = end_date.replace(day=1)
|
||
days = (end_date - start_date).days + 1
|
||
print(f"取数范围：{start_date} 至 {end_date}（{days} 天）")
|
||
|
||
client_rows, client_source = fetch_rows(CLIENT_GID, "Cost_Client")
|
||
installs, client_costs, client_apps, display = read_client_long(client_rows, start_date, end_date)
|
||
|
||
cost_rows, cost_source = fetch_rows(COST_GID, "Breakdown_cost")
|
||
wide_costs, wide_apps, wide_latest = read_cost_wide(cost_rows, start_date, end_date)
|
||
|
||
budget_rows, budget_source = fetch_rows(BUDGET_GID, "forPrebill")
|
||
budgets, budget_header = read_budgets(budget_rows)
|
||
|
||
targets = {"revenue": None, "gp": None, "gp_rate": None}
|
||
target_source = "unavailable"
|
||
try:
|
||
target_rows, target_source = fetch_rows(TARGET_GID, "Monthly FFR", TARGET_SPREADSHEET_ID)
|
||
targets = read_targets(target_rows)
|
||
except Exception as error:
|
||
print(f" Target 读取失败（{error}）；请把该表共享给服务账号或开放链接查看")
|
||
sands = load_sands_cost()
|
||
|
||
# 消耗表里可能出现 Cost_Client 未覆盖的 Campaign，一并纳入
|
||
for key, _day in wide_costs:
|
||
display.setdefault(key, key)
|
||
|
||
dates = [(start_date + timedelta(days=offset)).isoformat() for offset in range(days)]
|
||
keys = sorted(display, key=lambda item: display[item].lower())
|
||
|
||
records = []
|
||
for key in keys:
|
||
for day in dates:
|
||
cost = wide_costs.get((key, day))
|
||
if cost is None:
|
||
cost = client_costs.get((key, day))
|
||
install = installs.get((key, day))
|
||
if cost is None and install is None:
|
||
continue
|
||
records.append({
|
||
"date": day,
|
||
"campaign": display[key],
|
||
"installs": round(install or 0, 2),
|
||
"cost": round(cost or 0, 4),
|
||
})
|
||
|
||
summary = []
|
||
for key in keys:
|
||
install_total = sum(installs.get((key, day), 0.0) for day in dates)
|
||
cost_total = sum(
|
||
wide_costs.get((key, day), client_costs.get((key, day), 0.0)) for day in dates
|
||
)
|
||
if install_total == 0 and cost_total == 0:
|
||
continue
|
||
budget, match = resolve_budget(display[key], budgets)
|
||
summary.append({
|
||
"campaign": display[key],
|
||
"app": client_apps.get(key) or wide_apps.get(key) or "",
|
||
"installs": round(install_total, 2),
|
||
"cost": round(cost_total, 4),
|
||
"budget": budget,
|
||
"budget_match": match,
|
||
"ffr_percent": round(cost_total / budget * 100, 2) if budget else None,
|
||
"cpi": round(cost_total / install_total, 4) if install_total else None,
|
||
})
|
||
|
||
approximate = [item["campaign"] for item in summary if item["budget_match"] not in (None, "exact")]
|
||
missing = [item["campaign"] for item in summary if item["budget"] is None]
|
||
if approximate:
|
||
print(f" 预算近似匹配 {len(approximate)} 个：")
|
||
for item in summary:
|
||
if item["budget_match"] not in (None, "exact"):
|
||
print(f" {item['campaign']} <- {item['budget_match']} = {item['budget']}")
|
||
if missing:
|
||
print(f" 未匹配到预算 {len(missing)} 个：")
|
||
for name in missing:
|
||
print(f" {name}")
|
||
|
||
total_install = round(sum(item["installs"] for item in summary), 2)
|
||
total_cost = round(sum(item["cost"] for item in summary), 4)
|
||
month_days = (start_date.replace(day=28) + timedelta(days=4)).replace(day=1) - timedelta(days=1)
|
||
performance = build_performance(total_cost, sands, targets, days, month_days.day)
|
||
budget_values = [item["budget"] for item in summary if item["budget"] is not None]
|
||
total_budget = round(sum(budget_values), 4) if budget_values else None
|
||
|
||
return {
|
||
"title": "King · 当月 Campaign 表现",
|
||
"generated_at": now.isoformat(timespec="seconds"),
|
||
"timezone": "Asia/Shanghai",
|
||
"period": {"start": start_date.isoformat(), "end": end_date.isoformat(), "days": days},
|
||
"sources": {
|
||
"client": client_source,
|
||
"cost": cost_source,
|
||
"budget": budget_source,
|
||
"budget_column_header": budget_header,
|
||
"target": target_source,
|
||
},
|
||
"performance": performance,
|
||
"kpis": {
|
||
"TOTAL": {
|
||
"installs": total_install,
|
||
"cost": total_cost,
|
||
"budget": total_budget,
|
||
"ffr_percent": round(total_cost / total_budget * 100, 2) if total_budget else None,
|
||
"cpi": round(total_cost / total_install, 4) if total_install else None,
|
||
}
|
||
},
|
||
"campaigns": [item["campaign"] for item in summary],
|
||
"summary": summary,
|
||
"records": records,
|
||
"definitions": {
|
||
"period": "当月 1 号至昨天",
|
||
"installs": "Cost_Client 表 Installs 列按天汇总",
|
||
"cost": "Breakdown_cost【day】按天消耗；缺失时用 Cost_Client 的 Cost 兜底（USD）",
|
||
"budget": "forPrebill 表 B 列 Campaign name 对应 Z 列当月预算",
|
||
"budget_match": "exact 为同名匹配；brand+geo 或 geo 为近似匹配（预算表命名与投放表不一致时使用）",
|
||
"ffr_percent": "累计消耗 ÷ 当月预算 × 100",
|
||
"cpi": "累计消耗 ÷ 累计安装",
|
||
"excluded": "带“总计/合计/Total”的小计行已排除",
|
||
"performance_revenue": "Revenue 取本报表的 Cost 合计（客户口径消耗）",
|
||
"performance_cost": "Cost 取 Sands Report 2.0 的 Cost 合计，需本地抓取",
|
||
"performance_gp": "GP = Revenue − Cost；GP rate = GP ÷ Revenue × 100",
|
||
"performance_target": "Target 来自 Oversea FY2627 Monthly FFR 表 King 行",
|
||
"performance_completion": "完成率 = 实际 ÷ Target × 100；Target 为整月目标",
|
||
},
|
||
}
|
||
|
||
|
||
def main():
|
||
if len(sys.argv) < 2:
|
||
raise SystemExit("用法: update_report_data.py <输出文件>")
|
||
payload = build_payload(datetime.now(TIMEZONE))
|
||
with open(sys.argv[1], "w", encoding="utf-8") as handle:
|
||
json.dump(payload, handle, ensure_ascii=False, indent=2)
|
||
kpis = payload["kpis"]["TOTAL"]
|
||
print(
|
||
f"OK campaigns={len(payload['summary'])} records={len(payload['records'])} "
|
||
f"installs={kpis['installs']} cost={kpis['cost']} budget={kpis['budget']}"
|
||
)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
main()
|