Blame for src/formatter.py

"""
固定消息模板：钉钉 Markdown 周报。
结构不变，数据填充，专家预警分级展示。
"""
from typing import Any

ALERT_ICON = {
    "red": "🔴",
    "yellow": "🟡",
    "green": "🟢",
    # 兼容旧格式
    "critical": "🔴",
    "warning": "🟡",
    "positive": "🟢",
}


# ─── 数值格式化 ───

def _n(val: float | None, prefix: str = "", suffix: str = "", d: int = 2) -> str:
    if val is None or val != val:
        return "—"
    if abs(val) >= 1_000_000:
        return f"{prefix}{val / 1_000_000:,.{d}f}M{suffix}"
    if abs(val) >= 1_000:
        return f"{prefix}{val / 1_000:,.{d}f}K{suffix}"
    return f"{prefix}{val:,.{d}f}{suffix}"


def _pct(val: float | None) -> str:
    if val is None or val != val:
        return "—"
    return f"{val * 100:.2f}%" if abs(val) < 1 else f"{val:.2f}%"


def _arrow(change: float | None) -> str:
    """根据变化幅度返回方向箭头。"""
    if change is None or change != change:
        return ""
    if change >= 10:
        return "↑↑"
    if change > 0:
        return "↑"
    if change <= -10:
        return "↓↓"
    if change < 0:
        return "↓"
    return "→"


def _deltas_zh(wow_val: float | None, yoy_val: float | None) -> str:
    """环比/同比（中文标签），适合钉钉 Markdown；大幅波动加粗。"""
    parts: list[str] = []
    if wow_val is not None and wow_val == wow_val:
        sign = "+" if wow_val >= 0 else ""
        ar = _arrow(wow_val)
        s = f"环比 {ar}{sign}{wow_val:.1f}%"
        if abs(wow_val) >= 10:
            s = f"**{s}**"
        parts.append(s)
    if yoy_val is not None and yoy_val == yoy_val:
        sign = "+" if yoy_val >= 0 else ""
        ar = _arrow(yoy_val)
        s = f"同比 {ar}{sign}{yoy_val:.1f}%"
        if abs(yoy_val) >= 10:
            s = f"**{s}**"
        parts.append(s)
    if not parts:
        return ""
    return "　｜ " + " ｜ ".join(parts)


def _append_funnel_dim_lines(L: list[str], m: dict[str, Any], cw: dict, cy: dict) -> None:
    """漏斗维度：每指标一行，钉钉端更易扫读。"""
    L.append(f"- **花费** {_n(m.get('Cost'), prefix='$')}{_deltas_zh(cw.get('Cost'), cy.get('Cost'))}")
    L.append(f"- **销售额** {_n(m.get('Ads.Sales'), prefix='$')}{_deltas_zh(cw.get('Ads.Sales'), cy.get('Ads.Sales'))}")
    L.append(f"- **ROAS** {m.get('ROAS', 0):.2f}{_deltas_zh(cw.get('ROAS'), cy.get('ROAS'))}")
    L.append(f"- **展示量** {_n(m.get('Impressions'), d=0)}{_deltas_zh(cw.get('Impressions'), cy.get('Impressions'))}")
    L.append(f"- **CTR** {_pct(m.get('CTR'))}{_deltas_zh(cw.get('CTR'), cy.get('CTR'))}")
    L.append(f"- **CPC** {_n(m.get('CPC'), prefix='$')}{_deltas_zh(cw.get('CPC'), cy.get('CPC'))}")
    if m.get("ATC"):
        L.append(f"- **加购 ATC** {_n(m.get('ATC'), d=0)}{_deltas_zh(cw.get('ATC'), cy.get('ATC'))}")
    if m.get("GA4.Sessions"):
        L.append(f"- **会话** {_n(m.get('GA4.Sessions'), d=0)}{_deltas_zh(cw.get('GA4.Sessions'), cy.get('GA4.Sessions'))}")


def _append_country_dim_lines(L: list[str], m: dict[str, Any], cw: dict, cy: dict, *, full: bool) -> None:
    """市场维度：full=True 与周报一致；False 仅花费+ROAS（日报/简版）。"""
    L.append(f"- **花费** {_n(m.get('Cost'), prefix='$')}{_deltas_zh(cw.get('Cost'), cy.get('Cost'))}")
    if full:
        L.append(f"- **销售额** {_n(m.get('Ads.Sales'), prefix='$')}{_deltas_zh(cw.get('Ads.Sales'), cy.get('Ads.Sales'))}")
    L.append(f"- **ROAS** {m.get('ROAS', 0):.2f}{_deltas_zh(cw.get('ROAS'), cy.get('ROAS'))}")
    if full:
        L.append(f"- **会话** {_n(m.get('GA4.Sessions'), d=0)}{_deltas_zh(cw.get('GA4.Sessions'), cy.get('GA4.Sessions'))}")
        L.append(f"- **加购** {_n(m.get('ATC'), d=0)}{_deltas_zh(cw.get('ATC'), cy.get('ATC'))}")
        L.append(f"- **转化率** {_pct(m.get('CVR'))}{_deltas_zh(cw.get('CVR'), cy.get('CVR'))}")


# ─── 固定模板 ───

def format_report(data: dict[str, Any]) -> str:
    cp = data.get("current_period", {})
    pp = data.get("previous_period")
    year, week = cp.get("year"), cp.get("week")

    L: list[str] = []
    L.append(f"## DM 周报 · {year} W{week}")
    cmp_bits: list[str] = []
    if pp:
        cmp_bits.append(f"环比：{pp['year']} 年第 {pp['week']} 周")
    if data.get("yoy_period"):
        yp = data["yoy_period"]
        cmp_bits.append(f"同比：{yp['year']} 年第 {yp['week']} 周")
    if cmp_bits:
        L.append("> " + "　｜ ".join(cmp_bits))
    L.append("")

    curr = data.get("current", {})
    wow_t = data.get("wow_total") or {}
    yoy_t = data.get("yoy_total") or {}
    total = curr.get("total", {})

    L.append("### 一、整体表现")
    L.append("")
    _block_metrics(L, total, wow_t, yoy_t if yoy_t else None)
    L.append("")

    by_plat = curr.get("by_platform", {})
    wow_plat = data.get("wow_by_platform") or {}
    yoy_plat = data.get("yoy_by_platform") or {}
    L.append("### 二、平台对比")
    L.append("")
    for plat in ["Google", "Meta"]:
        m = by_plat.get(plat)
        if not m:
            continue
        pw = wow_plat.get(plat, {})
        py = yoy_plat.get(plat, {})
        L.append(f"**{plat}**")
        _block_metrics(L, m, pw, py if py else None)
        L.append("")

    by_ct = curr.get("by_campaign_type", {})
    wow_ct = data.get("wow_by_campaign") or {}
    yoy_ct = data.get("yoy_by_campaign") or {}
    L.append("### 三、广告漏斗")
    L.append("")
    for ct in ["Awareness", "Consideration", "Conversion"]:
        m = by_ct.get(ct)
        if not m:
            continue
        cw = wow_ct.get(ct, {})
        cy = yoy_ct.get(ct, {})
        L.append(f"**{ct}**")
        _append_funnel_dim_lines(L, m, cw, cy)
        L.append("")

    by_country = curr.get("by_country", {})
    wow_country = data.get("wow_by_country") or {}
    yoy_country = data.get("yoy_by_country") or {}
    L.append("### 四、市场表现")
    L.append("")
    for c in sorted(by_country.keys()):
        m = by_country[c]
        cw = wow_country.get(c, {})
        cy = yoy_country.get(c, {})
        L.append(f"**{c}**")
        _append_country_dim_lines(L, m, cw, cy, full=True)
        L.append("")

    _append_expert_section(L, data.get("expert", {}))

    return "\n".join(L).strip()


def _append_expert_section(L: list[str], expert: dict[str, Any]) -> None:
    """将专家分析追加到报告末尾。兼容新格式（verdict）和旧格式（summary）。"""
    if not expert:
        return

    L.append("### 陈策点评")
    L.append("")

    verdict = expert.get("verdict") or expert.get("summary", "")
    if verdict:
        L.append(f"> 💡 {verdict}")
        L.append("")

    findings = expert.get("key_findings") or []
    if not findings:
        for key in ["efficiency", "platform", "funnel", "market"]:
            findings.extend(expert.get(key, []))
    if findings:
        for i, f in enumerate(findings, 1):
            L.append(f"**{i}.** {f}")
        L.append("")

    alerts = expert.get("alerts", [])
    if alerts:
        for a in alerts:
            level = a.get("level", "yellow")
            icon = ALERT_ICON.get(level, "🟡")
            msg = a.get("msg") or f"{a.get('title', '')}：{a.get('detail', '')}"
            L.append(f"{icon} **{msg}**" if level == "red" else f"{icon} {msg}")
        L.append("")

    action = expert.get("next_action", "")
    if not action:
        recs = expert.get("recommendations", [])
        if recs:
            action = recs[0]
    if action:
        L.append(f"> 👉 **下一步：{action}**")
        L.append("")


def _flag(wow_val: float | None, yoy_val: float | None) -> str:
    """大幅异动时加前缀标记，让读者快速定位重点。"""
    for v in [wow_val, yoy_val]:
        if v is not None and v == v:
            if v <= -20:
                return "⚠️ "
            if v >= 20:
                return "🔥 "
    return ""


def _block_metrics(
    L: list[str],
    m: dict[str, float],
    wow: dict[str, float | None],
    yoy: dict[str, float | None] | None = None,
) -> None:
    """每个核心指标独立一行，大幅异动自动标记。"""
    y = yoy or {}
    rows = [
        ("花费", _n(m.get("Cost", 0), prefix="$"), wow.get("Cost"), y.get("Cost")),
        ("销售额", _n(m.get("Ads.Sales", 0), prefix="$"), wow.get("Ads.Sales"), y.get("Ads.Sales")),
        ("ROAS", f"{m.get('ROAS', 0):.2f}", wow.get("ROAS"), y.get("ROAS")),
        ("展示量", _n(m.get("Impressions", 0), d=0), wow.get("Impressions"), y.get("Impressions")),
        ("CTR", _pct(m.get("CTR", 0)), wow.get("CTR"), y.get("CTR")),
        ("转化率", _pct(m.get("CVR", 0)), wow.get("CVR"), y.get("CVR")),
        ("CPC", _n(m.get("CPC", 0), prefix="$"), wow.get("CPC"), y.get("CPC")),
        ("会话", _n(m.get("GA4.Sessions", 0), d=0), wow.get("GA4.Sessions"), y.get("GA4.Sessions")),
    ]
    if m.get("ATC", 0):
        rows.append(("加购 ATC", _n(m["ATC"], d=0), wow.get("ATC"), y.get("ATC")))
    for name, val, w, yr in rows:
        prefix = _flag(w, yr)
        L.append(f"- {prefix}**{name}** {val}{_deltas_zh(w, yr)}")


def format_title(data: dict[str, Any]) -> str:
    cp = data.get("current_period", {})
    return f"DM Weekly Report {cp.get('year')} W{cp.get('week')}"


# ─────────────────────────────────────────────
# 月度报告模板（与周报结构一致）
# ─────────────────────────────────────────────

def format_monthly_report(data: dict[str, Any]) -> str:
    """月度汇总报告，与周报结构一致，含环比+同比。"""
    period = data.get("period_label", "")
    prev = data.get("prev_label")
    yoy_label = data.get("yoy_label")
    error = data.get("error")

    L: list[str] = []
    L.append(f"## DM 月报 · {period}")
    cmp_parts = []
    if prev:
        cmp_parts.append(f"环比：{prev}")
    if yoy_label:
        cmp_parts.append(f"同比：{yoy_label}")
    if cmp_parts:
        L.append("> " + "　｜ ".join(cmp_parts))
    L.append("")

    if error:
        L.append(f"> {error}")
        return "\n".join(L).strip()

    curr = data.get("current", {})
    wow_t = data.get("wow_total") or {}
    yoy_t = data.get("yoy_total") or {}
    total = curr.get("total", {})

    L.append("### 一、整体表现")
    L.append("")
    _block_metrics(L, total, wow_t, yoy_t if yoy_t else None)
    L.append("")

    by_plat = curr.get("by_platform", {})
    wow_plat = data.get("wow_by_platform") or {}
    yoy_plat = data.get("yoy_by_platform") or {}
    L.append("### 二、平台对比")
    L.append("")
    for plat in ["Google", "Meta"]:
        m = by_plat.get(plat)
        if not m:
            continue
        pw = wow_plat.get(plat, {})
        py = yoy_plat.get(plat, {})
        L.append(f"**{plat}**")
        _block_metrics(L, m, pw, py if py else None)
        L.append("")

    by_ct = curr.get("by_campaign_type", {})
    wow_ct = data.get("wow_by_campaign") or {}
    yoy_ct = data.get("yoy_by_campaign") or {}
    L.append("### 三、广告漏斗")
    L.append("")
    for ct in ["Awareness", "Consideration", "Conversion"]:
        m = by_ct.get(ct)
        if not m:
            continue
        cw = wow_ct.get(ct, {})
        cy = yoy_ct.get(ct, {})
        L.append(f"**{ct}**")
        _append_funnel_dim_lines(L, m, cw, cy)
        L.append("")

    by_country = curr.get("by_country", {})
    wow_country = data.get("wow_by_country") or {}
    yoy_country = data.get("yoy_by_country") or {}
    L.append("### 四、市场表现")
    L.append("")
    for c in sorted(by_country.keys()):
        m = by_country[c]
        cw = wow_country.get(c, {})
        cy = yoy_country.get(c, {})
        L.append(f"**{c}**")
        _append_country_dim_lines(L, m, cw, cy, full=True)
        L.append("")

    _append_expert_section(L, data.get("expert", {}))

    return "\n".join(L).strip()


def format_monthly_title(data: dict[str, Any]) -> str:
    return f"DM Monthly Report {data.get('period_label', '')}"


# ─────────────────────────────────────────────
# 日报模板（复用月报结构）
# ─────────────────────────────────────────────

def format_daily_report(data: dict[str, Any]) -> str:
    """日度报告，结构与月报一致，标签改为 DoD/YoY。"""
    period = data.get("period_label", "")
    prev = data.get("prev_label")
    yoy_label = data.get("yoy_label")
    error = data.get("error")

    L: list[str] = []
    L.append(f"## DM 日报 · {period}")
    cmp_parts = []
    if prev:
        cmp_parts.append(f"环比：{prev}")
    if yoy_label:
        cmp_parts.append(f"同比：{yoy_label}")
    if cmp_parts:
        L.append("> " + "　｜ ".join(cmp_parts))
    L.append("")

    if error:
        L.append(f"> {error}")
        return "\n".join(L).strip()

    curr = data.get("current", {})
    wow_t = data.get("wow_total") or {}
    yoy_t = data.get("yoy_total") or {}
    total = curr.get("total", {})

    L.append("### 整体表现")
    L.append("")
    _block_metrics(L, total, wow_t, yoy_t if yoy_t else None)
    L.append("")

    by_plat = curr.get("by_platform", {})
    wow_plat = data.get("wow_by_platform") or {}
    yoy_plat = data.get("yoy_by_platform") or {}
    L.append("### 平台对比")
    L.append("")
    for plat in ["Google", "Meta"]:
        m = by_plat.get(plat)
        if not m:
            continue
        pw = wow_plat.get(plat, {})
        py = yoy_plat.get(plat, {})
        L.append(f"**{plat}**")
        _block_metrics(L, m, pw, py if py else None)
        L.append("")

    by_country = curr.get("by_country", {})
    wow_country = data.get("wow_by_country") or {}
    yoy_country = data.get("yoy_by_country") or {}
    if by_country:
        L.append("### 市场表现")
        L.append("")
        for c in sorted(by_country.keys()):
            m = by_country[c]
            cw = wow_country.get(c, {})
            cy = yoy_country.get(c, {})
            L.append(f"**{c}**")
            _append_country_dim_lines(L, m, cw, cy, full=False)
            L.append("")

    _append_expert_section(L, data.get("expert", {}))

    return "\n".join(L).strip()


def format_daily_title(data: dict[str, Any]) -> str:
    return f"DM Daily Report {data.get('period_label', '')}"


# ─────────────────────────────────────────────
# 自定义时间段报告
# ─────────────────────────────────────────────

def format_range_report(data: dict[str, Any]) -> str:
    """自定义时间段报告，结构与月报一致。"""
    period = data.get("period_label", "")
    prev = data.get("prev_label")
    yoy_label = data.get("yoy_label")
    days = data.get("days_count", "")
    error = data.get("error")

    L: list[str] = []
    L.append(f"## DM 自定义报告 · {period}")
    sub_parts = []
    if days:
        sub_parts.append(f"统计 {days} 天")
    if prev:
        sub_parts.append(f"环比：{prev}")
    if yoy_label:
        sub_parts.append(f"同比：{yoy_label}")
    if sub_parts:
        L.append("> " + "　｜ ".join(sub_parts))
    L.append("")

    if error:
        L.append(f"> {error}")
        return "\n".join(L).strip()

    curr = data.get("current", {})
    wow_t = data.get("wow_total") or {}
    yoy_t = data.get("yoy_total") or {}
    total = curr.get("total", {})

    L.append("### 整体表现")
    L.append("")
    _block_metrics(L, total, wow_t, yoy_t if yoy_t else None)
    L.append("")

    by_plat = curr.get("by_platform", {})
    wow_plat = data.get("wow_by_platform") or {}
    yoy_plat = data.get("yoy_by_platform") or {}
    L.append("### 平台对比")
    L.append("")
    for plat in ["Google", "Meta"]:
        m = by_plat.get(plat)
        if not m:
            continue
        pw = wow_plat.get(plat, {})
        py = yoy_plat.get(plat, {})
        L.append(f"**{plat}**")
        _block_metrics(L, m, pw, py if py else None)
        L.append("")

    by_ct = curr.get("by_campaign_type", {})
    wow_ct = data.get("wow_by_campaign") or {}
    yoy_ct = data.get("yoy_by_campaign") or {}
    if by_ct:
        L.append("### 广告漏斗")
        L.append("")
        for ct in ["Awareness", "Consideration", "Conversion"]:
            m = by_ct.get(ct)
            if not m:
                continue
            cw = wow_ct.get(ct, {})
            cy = yoy_ct.get(ct, {})
            L.append(f"**{ct}**")
            _append_funnel_dim_lines(L, m, cw, cy)
            L.append("")

    by_country = curr.get("by_country", {})
    wow_country = data.get("wow_by_country") or {}
    yoy_country = data.get("yoy_by_country") or {}
    if by_country:
        L.append("### 市场表现")
        L.append("")
        for c in sorted(by_country.keys()):
            m = by_country[c]
            cw = wow_country.get(c, {})
            cy = yoy_country.get(c, {})
            L.append(f"**{c}**")
            _append_country_dim_lines(L, m, cw, cy, full=False)
            L.append("")

    _append_expert_section(L, data.get("expert", {}))

    return "\n".join(L).strip()


def format_range_title(data: dict[str, Any]) -> str:
    return f"DM Custom Report {data.get('period_label', '')}"