Blame for src/meta_api.py

"""
Meta Marketing API 数据拉取引擎。
从 Meta Ads API 拉取 ad（素材）级日度数据，转换为统一的列结构。

支持 campaign / adset / ad 三个粒度，默认拉取 ad 级别。

注意：部分云环境（如 HF Spaces）可能无法直连 graph.facebook.com，
此时 Meta 数据将依赖 Google Sheets 中已有的数据（手动或外部同步）。
"""
import json
import logging
import os
from datetime import date, timedelta
from typing import Any

import pandas as pd
import requests
from dotenv import load_dotenv

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

API_VERSION = "v21.0"
BASE_URL = f"https://graph.facebook.com/{API_VERSION}"

OBJECTIVE_MAP = {
    "OUTCOME_AWARENESS": "Awareness",
    "BRAND_AWARENESS": "Awareness",
    "REACH": "Awareness",
    "OUTCOME_ENGAGEMENT": "Consideration",
    "OUTCOME_TRAFFIC": "Consideration",
    "TRAFFIC": "Consideration",
    "POST_ENGAGEMENT": "Consideration",
    "VIDEO_VIEWS": "Consideration",
    "LINK_CLICKS": "Consideration",
    "OUTCOME_SALES": "Conversion",
    "OUTCOME_LEADS": "Conversion",
    "CONVERSIONS": "Conversion",
    "PRODUCT_CATALOG_SALES": "Conversion",
}

INSIGHTS_FIELDS = [
    "campaign_name",
    "campaign_id",
    "adset_name",
    "adset_id",
    "ad_name",
    "ad_id",
    "objective",
    "spend",
    "impressions",
    "clicks",
    "cpc",
    "cpm",
    "ctr",
    "actions",
    "action_values",
    "video_30_sec_watched_actions",
    "video_p50_watched_actions",
    "video_p100_watched_actions",
]


def _get_action_value(actions: list[dict], action_type: str) -> float | None:
    if not actions:
        return None
    for a in actions:
        if a["action_type"] == action_type:
            return float(a["value"])
    return None


def _fetch_insights(
    account_id: str, token: str, since: str, until: str,
    level: str = "ad",
) -> list[dict]:
    """分页拉取一个广告账户的日度数据。level 可选 campaign / adset / ad。"""
    url = f"{BASE_URL}/act_{account_id}/insights"
    params = {
        "access_token": token,
        "fields": ",".join(INSIGHTS_FIELDS),
        "level": level,
        "time_range": json.dumps({"since": since, "until": until}),
        "time_increment": 1,
        "limit": 500,
    }

    all_rows: list[dict] = []
    while url:
        try:
            resp = requests.get(url, params=params, timeout=60)
        except requests.exceptions.ConnectionError as e:
            if "NameResolutionError" in str(e) or "getaddrinfo" in str(e):
                raise RuntimeError(
                    "无法连接 graph.facebook.com（DNS 解析失败）。"
                    "当前云环境可能限制了 Facebook 域名访问。"
                    "Meta 数据请通过 Google Sheets 手动更新，或在无网络限制的环境中运行同步。"
                ) from e
            raise
        if resp.status_code != 200:
            log.error(f"Meta API error ({resp.status_code}): {resp.text[:300]}")
            raise RuntimeError(f"Meta API 请求失败: {resp.status_code}")
        data = resp.json()
        all_rows.extend(data.get("data", []))
        paging = data.get("paging", {})
        url = paging.get("next")
        params = {}

    return all_rows


def _parse_row(row: dict, country: str) -> dict[str, Any]:
    """将一条 Meta API 返回数据转换为与 Google Sheets 相同的列结构。"""
    actions = row.get("actions") or []
    action_vals = row.get("action_values") or []

    spend = float(row.get("spend") or 0)
    impressions = int(row.get("impressions") or 0)
    clicks = int(row.get("clicks") or 0)
    ctr_raw = float(row.get("ctr") or 0)

    purchase_revenue = _get_action_value(action_vals, "omni_purchase") or _get_action_value(action_vals, "purchase") or 0
    purchase_count = _get_action_value(actions, "omni_purchase") or _get_action_value(actions, "purchase") or 0
    atc = _get_action_value(actions, "omni_add_to_cart") or _get_action_value(actions, "add_to_cart") or 0
    lpv = _get_action_value(actions, "landing_page_view") or 0
    video_views = _get_action_value(actions, "video_view") or 0
    link_clicks = _get_action_value(actions, "link_click") or 0

    roas = purchase_revenue / spend if spend > 0 else None
    cpc = float(row.get("cpc") or 0)
    cpm = float(row.get("cpm") or 0)
    ctr = ctr_raw / 100  # Meta returns CTR as percentage, convert to decimal
    cvr = purchase_count / link_clicks if link_clicks > 0 else None

    cp_atc = spend / atc if atc > 0 else None
    cplpv = spend / lpv if lpv > 0 else None
    cp_video = spend / video_views if video_views > 0 else None

    video_p50 = row.get("video_p50_watched_actions")
    video_p100 = row.get("video_p100_watched_actions")
    p50_count = float(video_p50[0]["value"]) if video_p50 else 0
    p100_count = float(video_p100[0]["value"]) if video_p100 else 0
    play_50_rate = p50_count / impressions if impressions > 0 and p50_count else None
    play_100_rate = p100_count / impressions if impressions > 0 and p100_count else None

    hook_rate = (video_views / impressions) if impressions > 0 and video_views > 0 else None

    campaign_name = row.get("campaign_name", "")
    name_parts = campaign_name.split("_")
    name_funnel = name_parts[1] if len(name_parts) > 1 else ""
    if name_funnel in ("Awareness", "Consideration", "Conversion"):
        campaign_type = name_funnel
    else:
        objective = row.get("objective", "")
        campaign_type = OBJECTIVE_MAP.get(objective, "Conversion")

    return {
        "country": country,
        "platform": "Meta",
        "campaign_type": campaign_type,
        "campaign_name": campaign_name,
        "adset_name": row.get("adset_name", ""),
        "ad_name": row.get("ad_name", ""),
        "ad_id": row.get("ad_id", ""),
        "date": row.get("date_start", ""),
        "Cost": spend,
        "Ads.Sales": purchase_revenue,
        "Ads.Sales(1d+7d)": None,
        "SPF.Sales": None,
        "ROAS": roas,
        "GA4.Sessions": None,
        "SPF.Sessions": None,
        "LPV": lpv,
        "CTR": ctr,
        "CVR": cvr,
        "CPC": cpc,
        "CPSE": None,
        "CPLPV": cplpv,
        "CPM": cpm,
        "Impressions": impressions,
        "Bounce Rate": None,
        "Average Duration": None,
        "ATC": atc,
        "CP-ATC": cp_atc,
        "Email": None,
        "CP-Email": None,
        "Video Views": video_views,
        "CP-Video Views": cp_video,
        "50% Play Rate": play_50_rate,
        "100% Play Rate": play_100_rate,
        "Hook Rate": hook_rate,
    }


def fetch_meta_data(
    since: date | None = None,
    until: date | None = None,
    days_back: int = 90,
    level: str = "ad",
) -> pd.DataFrame:
    """
    从所有配置的 Meta 广告账户拉取数据，返回统一结构的 DataFrame。
    level: "campaign" / "adset" / "ad"（默认 ad，素材级别）。
    默认拉取最近 90 天。
    """
    token = os.getenv("META_ACCESS_TOKEN")
    if not token:
        raise RuntimeError("未配置 META_ACCESS_TOKEN")

    accounts_str = os.getenv("META_AD_ACCOUNTS", "")
    if not accounts_str:
        raise RuntimeError("未配置 META_AD_ACCOUNTS (格式: account_id:country,...)")

    accounts = []
    for pair in accounts_str.split(","):
        pair = pair.strip()
        if ":" in pair:
            aid, country = pair.split(":", 1)
            accounts.append((aid.strip(), country.strip()))

    if not accounts:
        raise RuntimeError("META_AD_ACCOUNTS 格式错误")

    if until is None:
        until = date.today() - timedelta(days=1)
    if since is None:
        since = until - timedelta(days=days_back)

    since_str = since.strftime("%Y-%m-%d")
    until_str = until.strftime("%Y-%m-%d")

    all_parsed: list[dict] = []
    for account_id, country in accounts:
        log.info(f"拉取 Meta 数据: account={account_id} country={country} level={level} {since_str}~{until_str}")
        try:
            rows = _fetch_insights(account_id, token, since_str, until_str, level=level)
            log.info(f"  获取 {len(rows)} 条记录")
            for row in rows:
                parsed = _parse_row(row, country)
                all_parsed.append(parsed)
        except Exception as e:
            log.error(f"  拉取失败: {e}")
            raise

    if not all_parsed:
        return pd.DataFrame()

    df = pd.DataFrame(all_parsed)
    df["date"] = pd.to_datetime(df["date"])
    return df


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--days", type=int, default=7)
    parser.add_argument("--level", default="ad", choices=["campaign", "adset", "ad"])
    args = parser.parse_args()

    logging.basicConfig(level=logging.INFO)
    df = fetch_meta_data(days_back=args.days, level=args.level)
    print(f"\n拉取到 {len(df)} 条数据 (level={args.level})")
    print(f"日期范围: {df['date'].min()} ~ {df['date'].max()}")
    print(f"国家: {df['country'].unique()}")
    print(f"Campaign Types: {df['campaign_type'].unique()}")
    if "ad_name" in df.columns:
        print(f"唯一 Ad 数: {df['ad_name'].nunique()}")
    print(f"\n汇总:")
    print(f"  总花费: ${df['Cost'].sum():,.2f}")
    print(f"  总销售: ${df['Ads.Sales'].sum():,.2f}")
    total_cost = df["Cost"].sum()
    total_sales = df["Ads.Sales"].sum()
    print(f"  ROAS: {total_sales/total_cost:.2f}" if total_cost > 0 else "  ROAS: N/A")
    if args.level == "ad" and len(df) > 0:
        print(f"\n前 5 条 Ad 数据:")
        cols = ["ad_name", "campaign_name", "Cost", "Ads.Sales", "ROAS", "CTR", "Hook Rate"]
        print(df[cols].head().to_string())