378 lines
17 KiB
Python
378 lines
17 KiB
Python
#!/usr/bin/env python3
|
||
"""Render normalized SW Ads report data to Wise-styled HTML and PNG."""
|
||
|
||
from __future__ import annotations
|
||
|
||
import argparse
|
||
import html
|
||
import json
|
||
import os
|
||
from pathlib import Path
|
||
import shutil
|
||
import subprocess
|
||
import sys
|
||
import tempfile
|
||
from typing import Any
|
||
|
||
|
||
SKILL_DIR = Path(__file__).resolve().parent.parent
|
||
CSS_PATH = SKILL_DIR / "assets" / "report.css"
|
||
|
||
|
||
def esc(value: Any) -> str:
|
||
return html.escape("" if value is None else str(value), quote=True)
|
||
|
||
|
||
def number(value: Any) -> float | None:
|
||
if value is None or value == "":
|
||
return None
|
||
try:
|
||
return float(value)
|
||
except (TypeError, ValueError):
|
||
return None
|
||
|
||
|
||
def money(value: Any, currency: str) -> str:
|
||
value = number(value)
|
||
return "N/A" if value is None else f"{currency} {value:,.2f}"
|
||
|
||
|
||
def ratio(value: Any) -> str:
|
||
value = number(value)
|
||
return "N/A" if value is None else f"{value:.2f}×"
|
||
|
||
|
||
def percent(value: Any) -> str:
|
||
value = number(value)
|
||
return "N/A" if value is None else f"{value:.1f}%"
|
||
|
||
|
||
def integer(value: Any) -> str:
|
||
value = number(value)
|
||
return "N/A" if value is None else f"{int(round(value)):,}"
|
||
|
||
|
||
def decimal(value: Any) -> str:
|
||
value = number(value)
|
||
return "N/A" if value is None else f"{value:,.2f}"
|
||
|
||
|
||
def text_or(value: Any, fallback: str = "N/A") -> str:
|
||
return fallback if value is None or str(value).strip() == "" else str(value)
|
||
|
||
|
||
def render_kpi(label: str, value: str, note: str, *, accent: bool = False, tone: str = "") -> str:
|
||
classes = "kpi accent" if accent else "kpi"
|
||
value_class = f"kpi-value {tone}".strip()
|
||
return (
|
||
f'<div class="{classes}"><span class="kpi-label">{esc(label)}</span>'
|
||
f'<span class="{value_class}">{esc(value)}</span>'
|
||
f'<span class="kpi-note">{esc(note)}</span></div>'
|
||
)
|
||
|
||
|
||
def render_campaign_table(items: list[dict[str, Any]], *, empty_label: str) -> str:
|
||
if not items:
|
||
rows = f'<tr><td colspan="5">{esc(empty_label)}</td></tr>'
|
||
else:
|
||
rendered = []
|
||
for item in items[:5]:
|
||
reason = text_or(item.get("reason"), "")
|
||
suffix = f'<span class="row-note">{esc(reason)}</span>' if reason else ""
|
||
rendered.append(
|
||
"<tr>"
|
||
f'<td>{esc(text_or(item.get("name"), item.get("id", "Unnamed campaign")))}{suffix}</td>'
|
||
f'<td>{esc(decimal(item.get("spend")))}</td>'
|
||
f'<td>{esc(decimal(item.get("revenue")))}</td>'
|
||
f'<td>{esc(integer(item.get("orders")))}</td>'
|
||
f'<td><strong>{esc(ratio(item.get("roas")))}</strong></td>'
|
||
"</tr>"
|
||
)
|
||
rows = "".join(rendered)
|
||
return (
|
||
'<div class="table-wrap"><table><thead><tr>'
|
||
'<th>Campaign</th><th>Spend</th><th>GMV</th><th>订单</th><th>ROAS</th>'
|
||
f'</tr></thead><tbody>{rows}</tbody></table></div>'
|
||
)
|
||
|
||
|
||
def render_creative_table(items: list[dict[str, Any]], *, anomaly: bool) -> str:
|
||
if not items:
|
||
return '<div class="table-wrap"><table><tbody><tr><td>暂无足够数据</td></tr></tbody></table></div>'
|
||
rendered = []
|
||
for item in items[:5]:
|
||
title = text_or(item.get("name"), "Untitled creative")
|
||
creator = text_or(item.get("creator"), "Unknown creator")
|
||
creative_id = text_or(item.get("id"), "")
|
||
meta = " · ".join(part for part in (creator, creative_id) if part)
|
||
if anomaly:
|
||
rendered.append(
|
||
"<tr>"
|
||
f'<td>{esc(title)}<span class="row-note">{esc(meta)}</span></td>'
|
||
f'<td>{esc(decimal(item.get("spend")))}</td>'
|
||
f'<td>{esc(integer(item.get("clicks")))}</td>'
|
||
f'<td>{esc(integer(item.get("orders")))}</td>'
|
||
f'<td class="bad"><strong>{esc(ratio(item.get("roas")))}</strong></td>'
|
||
"</tr>"
|
||
)
|
||
else:
|
||
rendered.append(
|
||
"<tr>"
|
||
f'<td>{esc(title)}<span class="row-note">{esc(meta)}</span></td>'
|
||
f'<td>{esc(decimal(item.get("spend")))}</td>'
|
||
f'<td>{esc(decimal(item.get("revenue")))}</td>'
|
||
f'<td>{esc(integer(item.get("orders")))}</td>'
|
||
f'<td class="good"><strong>{esc(ratio(item.get("roas")))}</strong></td>'
|
||
"</tr>"
|
||
)
|
||
if anomaly:
|
||
headings = "<th>素材 / 发布账号</th><th>Spend</th><th>点击</th><th>订单</th><th>ROAS</th>"
|
||
else:
|
||
headings = "<th>素材 / 发布账号</th><th>Spend</th><th>GMV</th><th>订单</th><th>ROAS</th>"
|
||
return f'<div class="table-wrap"><table><thead><tr>{headings}</tr></thead><tbody>{"".join(rendered)}</tbody></table></div>'
|
||
|
||
|
||
def render_html(data: dict[str, Any], css: str) -> str:
|
||
meta = data.get("meta") or {}
|
||
summary = data.get("summary") or {}
|
||
campaigns = data.get("campaigns") or {}
|
||
creative = data.get("creative") or {}
|
||
currency = text_or(meta.get("currency"), "USD")
|
||
account_name = text_or(meta.get("account_name"), "SW Ads Account")
|
||
period_label = text_or(meta.get("period_label"), "Latest complete business day")
|
||
comparison = text_or(meta.get("comparison_label"), "Previous 7 complete days")
|
||
|
||
kpis = "".join(
|
||
[
|
||
render_kpi("Spend", money(summary.get("spend"), currency), "报告期总消耗"),
|
||
render_kpi("GMV", money(summary.get("revenue"), currency), f'{integer(summary.get("orders"))} 单 · 平台归因'),
|
||
render_kpi("ROAS", ratio(summary.get("roas")), "GMV ÷ Spend", accent=True, tone="good"),
|
||
render_kpi("简化广告 ROI", percent(summary.get("simple_roi_percent")), "未扣商品与履约成本", accent=True, tone="good"),
|
||
render_kpi("CTR", percent(summary.get("ctr_percent")), "无有效分母时显示 N/A", tone="warn"),
|
||
render_kpi("CPI", money(summary.get("cpi"), currency), "电商账户通常不适用"),
|
||
]
|
||
)
|
||
|
||
trend = data.get("trend") or []
|
||
trend_values = [number(item.get("roas")) or 0 for item in trend]
|
||
trend_max = max(trend_values, default=1) or 1
|
||
trend_items = []
|
||
for item in trend[:7]:
|
||
value = number(item.get("roas")) or 0
|
||
height = 24 + (126 * value / trend_max)
|
||
partial = " partial" if item.get("partial") else ""
|
||
trend_items.append(
|
||
'<div class="trend-item">'
|
||
f'<div class="bar{partial}" style="height:{height:.1f}px"></div>'
|
||
f'<span class="trend-value">{esc(ratio(value))}</span>'
|
||
f'<span class="trend-label">{esc(text_or(item.get("label"), ""))}</span></div>'
|
||
)
|
||
|
||
sources = creative.get("source_groups") or []
|
||
max_source_roas = max([number(item.get("roas")) or 0 for item in sources], default=1) or 1
|
||
source_rows = []
|
||
for item in sources:
|
||
width = max(3, 100 * (number(item.get("roas")) or 0) / max_source_roas)
|
||
source_rows.append(
|
||
'<div class="source-row">'
|
||
f'<span>{esc(text_or(item.get("label"), "Unknown source"))}</span>'
|
||
f'<div class="meter"><span style="width:{width:.1f}%"></span></div>'
|
||
f'<span class="source-value">{esc(ratio(item.get("roas")))}</span></div>'
|
||
)
|
||
if not source_rows:
|
||
source_rows.append('<p>发布账号归因信息不可用。</p>')
|
||
|
||
insights = creative.get("insights") or []
|
||
insight_rows = []
|
||
for index, insight in enumerate(insights[:4], start=1):
|
||
insight_rows.append(
|
||
f'<div class="insight"><span class="insight-num">{index}</span>'
|
||
f'<div><strong>归因线索</strong><p>{esc(insight)}</p></div></div>'
|
||
)
|
||
if not insight_rows:
|
||
insight_rows.append('<div class="insight"><span class="insight-num">1</span><div><strong>归因线索</strong><p>当前样本不足以形成稳定判断。</p></div></div>')
|
||
|
||
recommendations = data.get("recommendations") or []
|
||
recommendation_rows = []
|
||
labels = {"required": "需人工确认", "partial": "部分需确认", "none": "无需投放确认"}
|
||
badge_classes = {"required": "badge-bad", "partial": "badge-warn", "none": "badge-good"}
|
||
for index, recommendation in enumerate(recommendations[:3], start=1):
|
||
confirmation = text_or(recommendation.get("confirmation"), "required")
|
||
recommendation_rows.append(
|
||
'<article class="recommendation">'
|
||
f'<div class="recommendation-top"><span class="recommendation-num">{index:02d}</span>'
|
||
f'<span class="badge {badge_classes.get(confirmation, "badge-bad")}">{esc(labels.get(confirmation, "需人工确认"))}</span></div>'
|
||
f'<h3>{esc(text_or(recommendation.get("title"), "优化建议"))}</h3>'
|
||
f'<p>{esc(text_or(recommendation.get("detail"), ""))}</p></article>'
|
||
)
|
||
|
||
confirmations = data.get("manual_confirmations") or []
|
||
confirmation_rows = []
|
||
for item in confirmations:
|
||
required = bool(item.get("required"))
|
||
icon = "✓" if required else "○"
|
||
tone = "bad" if required else "good"
|
||
confirmation_rows.append(
|
||
'<div class="confirmation">'
|
||
f'<span class="{tone}"><strong>{icon}</strong></span><div>'
|
||
f'<strong>{esc(text_or(item.get("action"), "Manual review"))}</strong>'
|
||
f'<span>{esc(text_or(item.get("reason"), ""))}</span></div></div>'
|
||
)
|
||
|
||
notes = [text_or(meta.get("attribution_label"), "Platform attribution")]
|
||
notes.extend(str(note) for note in (data.get("notes") or []))
|
||
note_items = "".join(f"<li>{esc(note)}</li>" for note in notes if note)
|
||
monitoring_note = text_or(data.get("monitoring_note"), "")
|
||
callout = ""
|
||
if monitoring_note:
|
||
callout = (
|
||
'<div class="callout"><span>⚠</span><div><strong>监控提示</strong>'
|
||
f'<p>{esc(monitoring_note)}</p></div></div>'
|
||
)
|
||
|
||
coverage = number(creative.get("coverage_percent"))
|
||
coverage_text = "N/A" if coverage is None else f"{coverage:.1f}%"
|
||
freshness = text_or(meta.get("freshness_label"), "Freshness unavailable")
|
||
status = text_or(meta.get("status_label"), "Needs review")
|
||
account_id = text_or(meta.get("account_id"), "")
|
||
|
||
return f"""<!doctype html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="utf-8">
|
||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||
<title>{esc(account_name)} 每日投放报告</title>
|
||
<style>{css}</style>
|
||
</head>
|
||
<body>
|
||
<main class="report">
|
||
<header class="hero">
|
||
<div><span class="eyebrow">SW ADS · DELIVERY INTELLIGENCE</span><h1>{esc(account_name)}<br>每日投放报告</h1>
|
||
<p class="subtitle">{esc(period_label)} · {esc(meta.get("reporting_timezone"))} · {esc(meta.get("attribution_label"))}</p></div>
|
||
<div class="status"><strong>整体状态：{esc(status)}</strong><span>{esc(comparison)} · {esc(freshness)}{f" · {esc(account_id)}" if account_id else ""}</span></div>
|
||
</header>
|
||
<div class="kpis">{kpis}</div>
|
||
{callout}
|
||
|
||
<section><div class="section-head"><h2>七日 ROAS 趋势</h2><p>{esc(comparison)}；不完整业务日只作为监控信号。</p></div>
|
||
<div class="panel trend-panel"><div class="trend" role="img" aria-label="Seven-day ROAS trend">{"".join(trend_items)}</div></div>
|
||
</section>
|
||
|
||
<section><div class="section-head"><h2>Campaign 分层</h2><p>强弱判断同时考虑效率、消耗与订单样本，不把 Campaign 名称中的 ROI 当作真实配置。</p></div>
|
||
<div class="panel-grid">
|
||
<div class="panel"><div class="panel-head"><span class="panel-title">相对优质 Campaign</span><span class="badge badge-good">观察放量</span></div>{render_campaign_table(campaigns.get("winners") or [], empty_label="暂无足够样本")}</div>
|
||
<div class="panel"><div class="panel-head"><span class="panel-title">异常 / 低效 Campaign</span><span class="badge badge-bad">优先检查</span></div>{render_campaign_table(campaigns.get("anomalies") or [], empty_label="未发现明显异常")}</div>
|
||
</div>
|
||
</section>
|
||
|
||
<section><div class="section-head"><h2>素材表现与归因线索</h2><p>{esc(integer(creative.get("row_count")))} 条有消耗素材；素材层覆盖账户消耗约 {esc(coverage_text)}。</p></div>
|
||
<div class="panel-grid">
|
||
<div class="panel"><div class="panel-head"><span class="panel-title">发布账号贡献对比</span><span class="badge badge-good">平台相关性</span></div><div class="source-list">{"".join(source_rows)}</div></div>
|
||
<div class="panel insights">{"".join(insight_rows)}</div>
|
||
</div>
|
||
<div class="panel-grid" style="margin-top:24px">
|
||
<div class="panel"><div class="panel-head"><span class="panel-title">高价值素材</span><span class="badge badge-good">保留 / 复用</span></div>{render_creative_table(creative.get("winners") or [], anomaly=False)}</div>
|
||
<div class="panel"><div class="panel-head"><span class="panel-title">高消耗低转化素材</span><span class="badge badge-bad">降权 / 排查</span></div>{render_creative_table(creative.get("anomalies") or [], anomaly=True)}</div>
|
||
</div>
|
||
</section>
|
||
|
||
<section><div class="section-head"><h2>三条优化建议</h2><p>建议只描述下一步；本报告不执行任何平台写入。</p></div><div class="recommendations">{"".join(recommendation_rows)}</div></section>
|
||
<section><div class="section-head"><h2>人工确认清单</h2><p>预算、目标、状态、素材授权与发布均保持人工控制。</p></div><div class="confirmations">{"".join(confirmation_rows)}</div></section>
|
||
<footer class="notes"><strong>数据与口径</strong><ul>{note_items}</ul></footer>
|
||
</main>
|
||
</body>
|
||
</html>
|
||
"""
|
||
|
||
|
||
def find_chrome(explicit: str | None) -> str | None:
|
||
candidates = [
|
||
explicit,
|
||
os.environ.get("CHROME_BIN"),
|
||
shutil.which("google-chrome"),
|
||
shutil.which("google-chrome-stable"),
|
||
"/Applications/Google Chrome.app/Contents/MacOS/Google Chrome",
|
||
"/Applications/Chromium.app/Contents/MacOS/Chromium",
|
||
shutil.which("chromium"),
|
||
shutil.which("chromium-browser"),
|
||
]
|
||
for candidate in candidates:
|
||
if candidate and Path(candidate).is_file():
|
||
return str(candidate)
|
||
return None
|
||
|
||
|
||
def render_png(html_path: Path, png_path: Path, *, chrome: str | None, width: int, height: int) -> None:
|
||
chrome_bin = find_chrome(chrome)
|
||
if not chrome_bin:
|
||
raise RuntimeError("Chromium/Google Chrome was not found; HTML was created but PNG rendering was skipped")
|
||
png_path.parent.mkdir(parents=True, exist_ok=True)
|
||
command = [
|
||
chrome_bin,
|
||
"--headless=new",
|
||
"--disable-gpu",
|
||
"--hide-scrollbars",
|
||
"--force-device-scale-factor=1",
|
||
f"--window-size={width},{height}",
|
||
f"--screenshot={png_path}",
|
||
html_path.resolve().as_uri(),
|
||
]
|
||
result = subprocess.run(command, capture_output=True, text=True, timeout=90)
|
||
if result.returncode != 0 or not png_path.exists():
|
||
detail = (result.stderr or result.stdout or "unknown Chromium error").strip()
|
||
raise RuntimeError(f"PNG rendering failed: {detail}")
|
||
|
||
magick = shutil.which("magick")
|
||
if magick:
|
||
with tempfile.TemporaryDirectory(prefix="swads-report-trim-") as temp_dir:
|
||
trimmed = Path(temp_dir) / "trimmed.png"
|
||
trim_result = subprocess.run(
|
||
[magick, str(png_path), "-fuzz", "2%", "-trim", "+repage", "-bordercolor", "#e8ebe6", "-border", "24", str(trimmed)],
|
||
capture_output=True,
|
||
text=True,
|
||
timeout=90,
|
||
)
|
||
if trim_result.returncode == 0 and trimmed.exists():
|
||
shutil.copy2(trimmed, png_path)
|
||
|
||
|
||
def parse_args() -> argparse.Namespace:
|
||
parser = argparse.ArgumentParser(description=__doc__)
|
||
parser.add_argument("--data", required=True, type=Path, help="Normalized report JSON")
|
||
parser.add_argument("--html", required=True, type=Path, help="Destination HTML")
|
||
parser.add_argument("--png", type=Path, help="Optional destination PNG")
|
||
parser.add_argument("--chrome", help="Explicit Chrome/Chromium executable")
|
||
parser.add_argument("--image-width", type=int, default=1440)
|
||
parser.add_argument("--image-height", type=int, default=5200)
|
||
return parser.parse_args()
|
||
|
||
|
||
def main() -> int:
|
||
args = parse_args()
|
||
data = json.loads(args.data.read_text(encoding="utf-8"))
|
||
if not isinstance(data, dict) or not isinstance(data.get("meta"), dict) or not isinstance(data.get("summary"), dict):
|
||
raise SystemExit("report JSON must contain object fields: meta and summary")
|
||
recommendations = data.get("recommendations") or []
|
||
if len(recommendations) != 3:
|
||
raise SystemExit("report JSON must contain exactly three recommendations")
|
||
css = CSS_PATH.read_text(encoding="utf-8")
|
||
rendered = render_html(data, css)
|
||
args.html.parent.mkdir(parents=True, exist_ok=True)
|
||
args.html.write_text(rendered, encoding="utf-8")
|
||
|
||
output: dict[str, str] = {"html": str(args.html.resolve())}
|
||
if args.png:
|
||
try:
|
||
render_png(args.html, args.png, chrome=args.chrome, width=args.image_width, height=args.image_height)
|
||
output["png"] = str(args.png.resolve())
|
||
except RuntimeError as error:
|
||
print(str(error), file=sys.stderr)
|
||
print(json.dumps(output, ensure_ascii=False))
|
||
return 2
|
||
print(json.dumps(output, ensure_ascii=False))
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
raise SystemExit(main())
|