feat: add daily report skill
This commit is contained in:
Jeffrey Wu
@@ -0,0 +1,377 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user