Blame for src/p3_analyzer.py

from __future__ import annotations

"""
P3 投放分析引擎 — L1-L4 四层聚合。

L1 市场: Budget Pacing + Revenue vs Target
L2 funnel×优化目标: Spend Share + 各目标效率指标
L3 平台: Spend/Revenue Share + 通用效率指标
L4 Campaign: 单个 campaign 明细 + 环比
"""
import logging
from datetime import date, timedelta
from pathlib import Path
from typing import Any

import pandas as pd

from .loader import load_config, load_daily_data
from .campaign_parser import enrich_dataframe
from .p3_metrics import compute_group_metrics, get_core_metrics_for_objective
from .p3_config import load_all_p3_config

log = logging.getLogger("ad-agent")


def _safe_div(n: float, d: float) -> float:
    if d == 0 or pd.isna(d):
        return 0.0
    return n / d


def _filter_period(
    df: pd.DataFrame,
    period_type: str,
    target_year: int | None,
    target_period: int | None,
    start_date: date | None,
    end_date: date | None,
) -> tuple[pd.DataFrame, str]:
    """根据周期类型筛选数据，返回 (filtered_df, period_label)。"""
    if period_type == "range" and start_date and end_date:
        mask = (df["date"].dt.date >= start_date) & (df["date"].dt.date <= end_date)
        label = f"{start_date} ~ {end_date}"
        return df[mask].copy(), label

    if period_type == "day" and start_date:
        mask = df["date"].dt.date == start_date
        label = str(start_date)
        return df[mask].copy(), label

    if period_type == "month" and target_year and target_period:
        cal_year = df["date"].dt.year
        mask = (cal_year == target_year) & (df["month"] == target_period)
        label = f"{target_year}年{target_period}月"
        return df[mask].copy(), label

    # default: week
    if target_year and target_period:
        mask = (df["year"] == target_year) & (df["week"] == target_period)
        label = f"{target_year} W{target_period}"
        return df[mask].copy(), label

    # latest week
    latest_year = df["year"].max()
    latest_week = df[df["year"] == latest_year]["week"].max()
    mask = (df["year"] == latest_year) & (df["week"] == latest_week)
    label = f"{latest_year} W{latest_week}"
    return df[mask].copy(), label


def _get_prev_period(
    df: pd.DataFrame,
    period_type: str,
    target_year: int | None,
    target_period: int | None,
    start_date: date | None,
    end_date: date | None,
) -> tuple[pd.DataFrame, str | None]:
    """获取上一期数据用于环比。"""
    if period_type == "range" and start_date and end_date:
        days = (end_date - start_date).days + 1
        prev_end = start_date - timedelta(days=1)
        prev_start = prev_end - timedelta(days=days - 1)
        return _filter_period(df, "range", None, None, prev_start, prev_end)

    if period_type == "day" and start_date:
        prev = start_date - timedelta(days=1)
        return _filter_period(df, "day", None, None, prev, None)

    if period_type == "month" and target_year and target_period:
        pm = target_period - 1
        py = target_year
        if pm < 1:
            pm = 12
            py -= 1
        return _filter_period(df, "month", py, pm, None, None)

    # week
    weeks = (
        df[["year", "week"]]
        .drop_duplicates()
        .sort_values(["year", "week"])
        .values.tolist()
    )
    curr = [target_year, target_period] if target_year and target_period else None
    if not curr and weeks:
        curr = weeks[-1]
    if curr and curr in weeks:
        idx = weeks.index(curr)
        if idx > 0:
            py, pw = weeks[idx - 1]
            return _filter_period(df, "week", py, pw, None, None)
    return pd.DataFrame(), None


def _infer_month(period_df: pd.DataFrame) -> int | None:
    """从当期数据推断所属月份（取中位日期的月份）。"""
    if period_df.empty or "date" not in period_df.columns:
        return None
    dates = period_df["date"].dropna()
    if dates.empty:
        return None
    mid = dates.sort_values().iloc[len(dates) // 2]
    return mid.month


# ────────────────────────────────────────────────
# L1 市场层
# ────────────────────────────────────────────────

def _build_l1(
    curr_df: pd.DataFrame,
    prev_df: pd.DataFrame | None,
    p3_config: dict,
    target_month: int | None,
) -> list[dict[str, Any]]:
    if curr_df.empty or "market" not in curr_df.columns:
        return []

    targets = p3_config.get("monthly_targets", {})
    rows = []

    for mkt, grp in curr_df.groupby("market"):
        m = compute_group_metrics(grp)
        cost = m["Cost"]
        revenue = m["Revenue"]
        roas = m["ROAS"]

        row: dict[str, Any] = {
            "market": mkt,
            "spend": cost,
            "revenue": revenue,
            "roas": roas,
            "cpa": m["CPA"],
            "impressions": m.get("Impressions", 0),
        }

        mkt_targets = targets.get(mkt, {}).get(target_month, {}) if target_month else {}
        budget = mkt_targets.get("dm_budget")
        dm_target = mkt_targets.get("dm_target")
        target_roas = mkt_targets.get("target_roas")

        row["budget"] = budget
        row["budget_pacing"] = _safe_div(cost, budget) * 100 if budget else None
        row["dm_target"] = dm_target
        row["revenue_vs_target"] = _safe_div(revenue, dm_target) * 100 if dm_target else None
        row["target_roas"] = target_roas

        if prev_df is not None and not prev_df.empty:
            prev_grp = prev_df[prev_df["market"] == mkt] if "market" in prev_df.columns else pd.DataFrame()
            if not prev_grp.empty:
                pm = compute_group_metrics(prev_grp)
                row["prev_spend"] = pm["Cost"]
                row["prev_revenue"] = pm["Revenue"]
                row["prev_roas"] = pm["ROAS"]
                row["spend_chg"] = _safe_div(cost - pm["Cost"], pm["Cost"]) * 100 if pm["Cost"] else None
                row["revenue_chg"] = _safe_div(revenue - pm["Revenue"], pm["Revenue"]) * 100 if pm["Revenue"] else None
                row["roas_chg"] = _safe_div(roas - pm["ROAS"], pm["ROAS"]) * 100 if pm["ROAS"] else None

        rows.append(row)

    return rows


# ────────────────────────────────────────────────
# L2 funnel×优化目标层
# ────────────────────────────────────────────────

def _build_l2(
    curr_df: pd.DataFrame,
    prev_df: pd.DataFrame | None,
    p3_config: dict,
    target_month: int | None,
) -> list[dict[str, Any]]:
    needed_cols = {"ad_objective", "audience_type", "objective_group", "funnel"}
    if not needed_cols.issubset(curr_df.columns):
        return []

    total_cost = curr_df["Cost"].fillna(0).sum() if "Cost" in curr_df.columns else 0
    rows = []

    for obj_group, grp in curr_df.groupby("objective_group"):
        objective = grp["ad_objective"].iloc[0] if not grp.empty else None
        audience = grp["audience_type"].iloc[0] if not grp.empty else "cold"
        funnel = grp["funnel"].iloc[0] if not grp.empty else ""

        m = compute_group_metrics(grp, objective=objective)
        cost = m["Cost"]

        row: dict[str, Any] = {
            "objective_group": obj_group,
            "ad_objective": objective,
            "audience_type": audience,
            "funnel": funnel,
            "spend": cost,
            "spend_share": _safe_div(cost, total_cost) * 100,
            **{k: m.get(k) for k in get_core_metrics_for_objective(objective or "")},
            "impressions": m.get("Impressions", 0),
            "clicks": m.get("Clicks", 0),
        }

        if prev_df is not None and not prev_df.empty and "objective_group" in prev_df.columns:
            prev_grp = prev_df[prev_df["objective_group"] == obj_group]
            if not prev_grp.empty:
                pm = compute_group_metrics(prev_grp, objective=objective)
                for key in get_core_metrics_for_objective(objective or ""):
                    curr_val = m.get(key, 0)
                    prev_val = pm.get(key, 0)
                    row[f"{key}_chg"] = (
                        _safe_div(curr_val - prev_val, prev_val) * 100
                        if prev_val else None
                    )

        rows.append(row)

    rows.sort(key=lambda r: r.get("spend", 0), reverse=True)
    return rows


# ────────────────────────────────────────────────
# L3 平台层
# ────────────────────────────────────────────────

def _build_l3(
    curr_df: pd.DataFrame,
    prev_df: pd.DataFrame | None,
    p3_config: dict,
    target_month: int | None,
) -> list[dict[str, Any]]:
    plat_col = "parsed_platform" if "parsed_platform" in curr_df.columns else "platform"
    if plat_col not in curr_df.columns:
        return []

    total_cost = curr_df["Cost"].fillna(0).sum() if "Cost" in curr_df.columns else 0
    total_revenue = curr_df["Ads.Sales"].fillna(0).sum() if "Ads.Sales" in curr_df.columns else 0

    channel_budgets = p3_config.get("channel_budgets", {})
    rows = []

    for plat, grp in curr_df.groupby(plat_col):
        m = compute_group_metrics(grp)
        cost = m["Cost"]
        revenue = m["Revenue"]

        row: dict[str, Any] = {
            "platform": plat,
            "spend": cost,
            "spend_share": _safe_div(cost, total_cost) * 100,
            "revenue": revenue,
            "revenue_share": _safe_div(revenue, total_revenue) * 100,
            "roas": m["ROAS"],
            "impressions": m.get("Impressions", 0),
            "cpm": m["CPM"],
            "clicks": m["Clicks"],
            "ctr": m["CTR"],
            "cpc": m["CPC"],
            "cpa": m["CPA"],
            "conversions": m["Conversions"],
        }

        # 与渠道预算对照
        us_budgets = channel_budgets.get("US", {}).get("channel_budgets", {})
        if target_month and target_month in us_budgets:
            planned = us_budgets[target_month].get(plat)
            row["planned_budget"] = planned
            row["budget_pacing"] = _safe_div(cost, planned) * 100 if planned else None

        if prev_df is not None and not prev_df.empty and plat_col in prev_df.columns:
            prev_grp = prev_df[prev_df[plat_col] == plat]
            if not prev_grp.empty:
                pm = compute_group_metrics(prev_grp)
                row["prev_spend"] = pm["Cost"]
                row["prev_roas"] = pm["ROAS"]
                row["spend_chg"] = _safe_div(cost - pm["Cost"], pm["Cost"]) * 100 if pm["Cost"] else None
                row["roas_chg"] = _safe_div(m["ROAS"] - pm["ROAS"], pm["ROAS"]) * 100 if pm["ROAS"] else None
                row["cpc_chg"] = _safe_div(m["CPC"] - pm["CPC"], pm["CPC"]) * 100 if pm["CPC"] else None

        rows.append(row)

    rows.sort(key=lambda r: r.get("spend", 0), reverse=True)
    return rows


# ────────────────────────────────────────────────
# L4 Campaign 层
# ────────────────────────────────────────────────

def _build_l4(
    curr_df: pd.DataFrame,
    prev_df: pd.DataFrame | None,
) -> list[dict[str, Any]]:
    if "campaign_name" not in curr_df.columns:
        return []

    rows = []
    dim_cols = [
        "campaign_name", "market", "funnel", "parsed_platform",
        "ad_objective", "product_line", "audience_type", "objective_group",
    ]
    existing_dims = [c for c in dim_cols if c in curr_df.columns]

    for name, grp in curr_df.groupby("campaign_name"):
        objective = grp["ad_objective"].iloc[0] if "ad_objective" in grp.columns else None
        m = compute_group_metrics(grp, objective=objective)

        row: dict[str, Any] = {"campaign_name": name}
        for c in existing_dims:
            if c != "campaign_name":
                row[c] = grp[c].iloc[0] if c in grp.columns else None

        row.update({
            "spend": m["Cost"],
            "impressions": m.get("Impressions", 0),
            "cpm": m["CPM"],
            "clicks": m["Clicks"],
            "ctr": m["CTR"],
            "cpc": m["CPC"],
            "conversions": m["Conversions"],
            "revenue": m["Revenue"],
            "cpa": m["CPA"],
            "roas": m["ROAS"],
        })

        if objective:
            for key in get_core_metrics_for_objective(objective):
                if key not in row:
                    row[key] = m.get(key)

        if prev_df is not None and not prev_df.empty and "campaign_name" in prev_df.columns:
            prev_grp = prev_df[prev_df["campaign_name"] == name]
            if not prev_grp.empty:
                pm = compute_group_metrics(prev_grp, objective=objective)
                row["prev_roas"] = pm["ROAS"]
                row["roas_chg"] = (
                    _safe_div(m["ROAS"] - pm["ROAS"], pm["ROAS"]) * 100
                    if pm["ROAS"] else None
                )
                row["prev_spend"] = pm["Cost"]
                row["spend_chg"] = (
                    _safe_div(m["Cost"] - pm["Cost"], pm["Cost"]) * 100
                    if pm["Cost"] else None
                )

        rows.append(row)

    rows.sort(key=lambda r: r.get("spend", 0), reverse=True)
    return rows


# ────────────────────────────────────────────────
# 主入口
# ────────────────────────────────────────────────

def run_p3_analysis(
    config_path: Path | None = None,
    period_type: str = "week",
    target_year: int | None = None,
    target_period: int | None = None,
    start_date: date | None = None,
    end_date: date | None = None,
    market: str | None = None,
) -> dict[str, Any]:
    """
    P3 投放分析主入口。
    加载数据 -> 解析 campaign name -> 四层聚合 -> 返回 L1-L4 结果。
    """
    config_path = config_path or (Path(__file__).resolve().parent.parent / "config.yaml")
    df = load_daily_data(config_path)
    df = enrich_dataframe(df)

    if market and "market" in df.columns:
        df = df[df["market"] == market].copy()

    curr_df, period_label = _filter_period(
        df, period_type, target_year, target_period, start_date, end_date
    )

    if curr_df.empty:
        return {
            "period_label": period_label,
            "error": f"{period_label} 无可用数据",
            "L1": [], "L2": [], "L3": [], "L4": [],
        }

    prev_df, prev_label = _get_prev_period(
        df, period_type, target_year, target_period, start_date, end_date
    )
    prev_enriched = enrich_dataframe(prev_df) if prev_df is not None and not prev_df.empty else None

    p3_config = load_all_p3_config(config_path)

    target_month = _infer_month(curr_df)

    l1 = _build_l1(curr_df, prev_enriched, p3_config, target_month)
    l2 = _build_l2(curr_df, prev_enriched, p3_config, target_month)
    l3 = _build_l3(curr_df, prev_enriched, p3_config, target_month)
    l4 = _build_l4(curr_df, prev_enriched)

    parse_ok = curr_df["parse_ok"].sum() if "parse_ok" in curr_df.columns else 0
    parse_total = len(curr_df)

    return {
        "period_label": period_label,
        "prev_label": prev_label,
        "period_type": period_type,
        "target_month": target_month,
        "market_filter": market,
        "parse_rate": _safe_div(parse_ok, parse_total) * 100 if parse_total else 0,
        "row_count": parse_total,
        "L1": l1,
        "L2": l2,
        "L3": l3,
        "L4": l4,
        "p3_config": p3_config,
    }