#!/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'
{esc(label)}' f'{esc(value)}' f'{esc(note)}
' ) def render_campaign_table(items: list[dict[str, Any]], *, empty_label: str) -> str: if not items: rows = f'{esc(empty_label)}' else: rendered = [] for item in items[:5]: reason = text_or(item.get("reason"), "") suffix = f'{esc(reason)}' if reason else "" rendered.append( "" f'{esc(text_or(item.get("name"), item.get("id", "Unnamed campaign")))}{suffix}' f'{esc(decimal(item.get("spend")))}' f'{esc(decimal(item.get("revenue")))}' f'{esc(integer(item.get("orders")))}' f'{esc(ratio(item.get("roas")))}' "" ) rows = "".join(rendered) return ( '
' '' f'{rows}
CampaignSpendGMV订单ROAS
' ) def render_creative_table(items: list[dict[str, Any]], *, anomaly: bool) -> str: if not items: return '
暂无足够数据
' 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( "" f'{esc(title)}{esc(meta)}' f'{esc(decimal(item.get("spend")))}' f'{esc(integer(item.get("clicks")))}' f'{esc(integer(item.get("orders")))}' f'{esc(ratio(item.get("roas")))}' "" ) else: rendered.append( "" f'{esc(title)}{esc(meta)}' f'{esc(decimal(item.get("spend")))}' f'{esc(decimal(item.get("revenue")))}' f'{esc(integer(item.get("orders")))}' f'{esc(ratio(item.get("roas")))}' "" ) if anomaly: headings = "素材 / 发布账号Spend点击订单ROAS" else: headings = "素材 / 发布账号SpendGMV订单ROAS" return f'
{headings}{"".join(rendered)}
' 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( '
' f'
' f'{esc(ratio(value))}' f'{esc(text_or(item.get("label"), ""))}
' ) 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( '
' f'{esc(text_or(item.get("label"), "Unknown source"))}' f'
' f'{esc(ratio(item.get("roas")))}
' ) if not source_rows: source_rows.append('

发布账号归因信息不可用。

') insights = creative.get("insights") or [] insight_rows = [] for index, insight in enumerate(insights[:4], start=1): insight_rows.append( f'
{index}' f'
归因线索

{esc(insight)}

' ) if not insight_rows: insight_rows.append('
1
归因线索

当前样本不足以形成稳定判断。

') 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( '
' f'
{index:02d}' f'{esc(labels.get(confirmation, "需人工确认"))}
' f'

{esc(text_or(recommendation.get("title"), "优化建议"))}

' f'

{esc(text_or(recommendation.get("detail"), ""))}

' ) 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( '
' f'{icon}
' f'{esc(text_or(item.get("action"), "Manual review"))}' f'{esc(text_or(item.get("reason"), ""))}
' ) notes = [text_or(meta.get("attribution_label"), "Platform attribution")] notes.extend(str(note) for note in (data.get("notes") or [])) note_items = "".join(f"
  • {esc(note)}
  • " for note in notes if note) monitoring_note = text_or(data.get("monitoring_note"), "") callout = "" if monitoring_note: callout = ( '
    监控提示' f'

    {esc(monitoring_note)}

    ' ) 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""" {esc(account_name)} 每日投放报告
    SW ADS · DELIVERY INTELLIGENCE

    {esc(account_name)}
    每日投放报告

    {esc(period_label)} · {esc(meta.get("reporting_timezone"))} · {esc(meta.get("attribution_label"))}

    整体状态:{esc(status)}{esc(comparison)} · {esc(freshness)}{f" · {esc(account_id)}" if account_id else ""}
    {kpis}
    {callout}

    七日 ROAS 趋势

    {esc(comparison)};不完整业务日只作为监控信号。

    Campaign 分层

    强弱判断同时考虑效率、消耗与订单样本,不把 Campaign 名称中的 ROI 当作真实配置。

    相对优质 Campaign观察放量
    {render_campaign_table(campaigns.get("winners") or [], empty_label="暂无足够样本")}
    异常 / 低效 Campaign优先检查
    {render_campaign_table(campaigns.get("anomalies") or [], empty_label="未发现明显异常")}

    素材表现与归因线索

    {esc(integer(creative.get("row_count")))} 条有消耗素材;素材层覆盖账户消耗约 {esc(coverage_text)}。

    发布账号贡献对比平台相关性
    {"".join(source_rows)}
    {"".join(insight_rows)}
    高价值素材保留 / 复用
    {render_creative_table(creative.get("winners") or [], anomaly=False)}
    高消耗低转化素材降权 / 排查
    {render_creative_table(creative.get("anomalies") or [], anomaly=True)}

    三条优化建议

    建议只描述下一步;本报告不执行任何平台写入。

    {"".join(recommendation_rows)}

    人工确认清单

    预算、目标、状态、素材授权与发布均保持人工控制。

    {"".join(confirmation_rows)}
    """ 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())