Blame for src/bing_api.py

"""
Microsoft Advertising (Bing Ads) 数据拉取引擎。
通过 SOAP API 直接调用 Reporting Service v13，无需 bingads SDK。
拉取 campaign 级日度数据，转换为与 Google Sheets 相同的列结构。
"""
import csv
import io
import logging
import os
import re
import time
import zipfile
from datetime import date, timedelta
from typing import Any
from xml.etree import ElementTree as ET

import pandas as pd
import requests
from dotenv import load_dotenv

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

REPORTING_URL = (
    "https://reporting.api.bingads.microsoft.com"
    "/Api/Advertiser/Reporting/v13/ReportingService.svc"
)
TOKEN_URL = "https://login.microsoftonline.com/common/oauth2/v2.0/token"

NS = "https://bingads.microsoft.com/Reporting/v13"
NS_ARR = "http://schemas.microsoft.com/2003/10/Serialization/Arrays"

REPORT_COLUMNS = [
    "TimePeriod",
    "CampaignId",
    "CampaignName",
    "CampaignType",
    "Impressions",
    "Clicks",
    "Ctr",
    "AverageCpc",
    "Spend",
    "Conversions",
    "Revenue",
]

CAMPAIGN_TYPE_KEYWORDS = {
    "Awareness": ["awareness", "branding", "reach", "display", "video", "audience"],
    "Consideration": ["consideration", "traffic", "pageview", "engagement"],
    "Conversion": ["conversion", "purchase", "retarget", "shopping", "pmax", "search",
                    "performance max", "smart", "dynamic"],
}


def _infer_campaign_type(name: str, bing_type: str | None = None) -> str:
    if bing_type:
        bt = bing_type.lower()
        if bt in ("shopping", "performancemax", "search", "dynamicsearchads"):
            return "Conversion"
        if bt in ("audience",):
            return "Awareness"
    lower = name.lower()
    for ctype, keywords in CAMPAIGN_TYPE_KEYWORDS.items():
        if any(kw in lower for kw in keywords):
            return ctype
    return "Conversion"


def _refresh_access_token() -> str:
    """用 Refresh Token 换取新的 Access Token。"""
    client_id = os.getenv("BING_CLIENT_ID")
    client_secret = os.getenv("BING_CLIENT_SECRET")
    refresh_token = os.getenv("BING_REFRESH_TOKEN")
    if not all([client_id, client_secret, refresh_token]):
        raise RuntimeError("未配置 BING_CLIENT_ID / BING_CLIENT_SECRET / BING_REFRESH_TOKEN")

    resp = requests.post(TOKEN_URL, data={
        "client_id": client_id,
        "client_secret": client_secret,
        "grant_type": "refresh_token",
        "refresh_token": refresh_token,
        "scope": "https://ads.microsoft.com/msads.manage offline_access",
    })
    if resp.status_code != 200:
        raise RuntimeError(f"Bing token 刷新失败 ({resp.status_code}): {resp.text[:300]}")

    data = resp.json()
    new_refresh = data.get("refresh_token")
    if new_refresh and new_refresh != refresh_token:
        log.info("Bing refresh token 已更新，请手动更新 .env")

    return data["access_token"]


def _soap_headers(access_token: str, action: str, account_id: str) -> str:
    customer_id = os.getenv("BING_CUSTOMER_ID", "")
    developer_token = os.getenv("BING_DEVELOPER_TOKEN", "")
    return f"""<s:Header xmlns="{NS}">
    <Action mustUnderstand="1">{action}</Action>
    <AuthenticationToken>{access_token}</AuthenticationToken>
    <CustomerAccountId>{account_id}</CustomerAccountId>
    <CustomerId>{customer_id}</CustomerId>
    <DeveloperToken>{developer_token}</DeveloperToken>
  </s:Header>"""


def _submit_report(
    access_token: str, account_id: str, since: date, until: date,
) -> str:
    """提交报告请求，返回 ReportRequestId。"""
    columns_xml = "\n".join(
        f"          <CampaignPerformanceReportColumn>{c}</CampaignPerformanceReportColumn>"
        for c in REPORT_COLUMNS
    )

    body = f"""<?xml version="1.0" encoding="utf-8"?>
<s:Envelope xmlns:i="http://www.w3.org/2001/XMLSchema-instance"
            xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">
  {_soap_headers(access_token, "SubmitGenerateReport", account_id)}
  <s:Body>
    <SubmitGenerateReportRequest xmlns="{NS}">
      <ReportRequest i:type="CampaignPerformanceReportRequest">
        <ExcludeColumnHeaders>false</ExcludeColumnHeaders>
        <ExcludeReportFooter>true</ExcludeReportFooter>
        <ExcludeReportHeader>true</ExcludeReportHeader>
        <Format>Csv</Format>
        <FormatVersion>2.0</FormatVersion>
        <Aggregation>Daily</Aggregation>
        <Columns>
{columns_xml}
        </Columns>
        <Scope>
          <AccountIds xmlns:a1="{NS_ARR}">
            <a1:long>{account_id}</a1:long>
          </AccountIds>
        </Scope>
        <Time>
          <CustomDateRangeStart>
            <Day>{since.day}</Day>
            <Month>{since.month}</Month>
            <Year>{since.year}</Year>
          </CustomDateRangeStart>
          <CustomDateRangeEnd>
            <Day>{until.day}</Day>
            <Month>{until.month}</Month>
            <Year>{until.year}</Year>
          </CustomDateRangeEnd>
          <ReportTimeZone>EasternTimeUSCanada</ReportTimeZone>
        </Time>
      </ReportRequest>
    </SubmitGenerateReportRequest>
  </s:Body>
</s:Envelope>"""

    resp = requests.post(REPORTING_URL, data=body.encode("utf-8"), headers={
        "Content-Type": "text/xml; charset=utf-8",
        "SOAPAction": "SubmitGenerateReport",
    })

    if resp.status_code != 200:
        _raise_soap_error(resp, "SubmitGenerateReport")

    root = ET.fromstring(resp.text)
    rid = root.find(".//{%s}ReportRequestId" % NS)
    if rid is None or not rid.text:
        raise RuntimeError(f"SubmitGenerateReport 未返回 ReportRequestId: {resp.text[:500]}")
    return rid.text


def _poll_report(access_token: str, account_id: str, request_id: str) -> str | None:
    """轮询报告状态，返回下载 URL 或 None（仍在 Pending）。"""
    body = f"""<?xml version="1.0" encoding="utf-8"?>
<s:Envelope xmlns:i="http://www.w3.org/2001/XMLSchema-instance"
            xmlns:s="http://schemas.xmlsoap.org/soap/envelope/">
  {_soap_headers(access_token, "PollGenerateReport", account_id)}
  <s:Body>
    <PollGenerateReportRequest xmlns="{NS}">
      <ReportRequestId>{request_id}</ReportRequestId>
    </PollGenerateReportRequest>
  </s:Body>
</s:Envelope>"""

    resp = requests.post(REPORTING_URL, data=body.encode("utf-8"), headers={
        "Content-Type": "text/xml; charset=utf-8",
        "SOAPAction": "PollGenerateReport",
    })

    if resp.status_code != 200:
        _raise_soap_error(resp, "PollGenerateReport")

    root = ET.fromstring(resp.text)
    status_el = root.find(".//{%s}Status" % NS)
    status = status_el.text if status_el is not None else "Unknown"

    if status == "Success":
        url_el = root.find(".//{%s}ReportDownloadUrl" % NS)
        if url_el is not None and url_el.text:
            return url_el.text
        raise RuntimeError("报告成功但无下载 URL（可能无数据）")
    elif status == "Pending":
        return None
    else:
        raise RuntimeError(f"报告请求失败: status={status}")


def _raise_soap_error(resp: requests.Response, operation: str):
    """从 SOAP Fault 中提取错误信息。"""
    try:
        root = ET.fromstring(resp.text)
        fault = root.find(".//{http://schemas.xmlsoap.org/soap/envelope/}Fault")
        if fault is not None:
            faultstring = fault.findtext("faultstring", "")
            detail = fault.findtext(".//Message", "") or fault.findtext(".//detail", "")
            raise RuntimeError(f"Bing {operation} SOAP 错误: {faultstring} {detail}".strip())
    except ET.ParseError:
        pass
    raise RuntimeError(
        f"Bing {operation} 请求失败 ({resp.status_code}): {resp.text[:500]}"
    )


def _download_report(url: str) -> list[dict]:
    """下载并解压 ZIP 报告，解析 CSV 行。"""
    resp = requests.get(url, timeout=120)
    resp.raise_for_status()

    with zipfile.ZipFile(io.BytesIO(resp.content)) as zf:
        csv_name = zf.namelist()[0]
        with zf.open(csv_name) as f:
            text = f.read().decode("utf-8-sig")

    reader = csv.DictReader(io.StringIO(text))
    return list(reader)


def _parse_row(row: dict, country: str) -> dict[str, Any]:
    campaign_name = row.get("CampaignName", "")
    bing_type = row.get("CampaignType", "")
    cost = float(row.get("Spend") or 0)
    impressions = int(float(row.get("Impressions") or 0))
    clicks = int(float(row.get("Clicks") or 0))
    conversions = float(row.get("Conversions") or 0)
    revenue = float(row.get("Revenue") or 0)

    ctr_raw = row.get("Ctr", "")
    ctr = float(ctr_raw.replace("%", "")) / 100 if ctr_raw and ctr_raw != "--" else (
        clicks / impressions if impressions > 0 else None
    )
    cpc = float(row.get("AverageCpc") or 0) if row.get("AverageCpc") else (
        cost / clicks if clicks > 0 else None
    )
    cpm = (cost / impressions * 1000) if impressions > 0 else None
    roas = revenue / cost if cost > 0 else None
    cvr = conversions / clicks if clicks > 0 else None

    return {
        "country": country,
        "platform": "Bing",
        "campaign_type": _infer_campaign_type(campaign_name, bing_type),
        "campaign_name": campaign_name,
        "date": row.get("TimePeriod", ""),
        "Cost": cost,
        "Ads.Sales": revenue,
        "Ads.Sales(1d+7d)": None,
        "SPF.Sales": None,
        "ROAS": roas,
        "GA4.Sessions": None,
        "SPF.Sessions": None,
        "LPV": None,
        "CTR": ctr,
        "CVR": cvr,
        "CPC": cpc,
        "CPSE": None,
        "CPLPV": None,
        "CPM": cpm,
        "Impressions": impressions,
        "Bounce Rate": None,
        "Average Duration": None,
        "ATC": None,
        "CP-ATC": None,
        "Email": None,
        "CP-Email": None,
        "Video Views": None,
        "CP-Video Views": None,
        "50% Play Rate": None,
        "100% Play Rate": None,
        "Hook Rate": None,
    }


def fetch_bing_data(
    since: date | None = None,
    until: date | None = None,
    days_back: int = 90,
) -> pd.DataFrame:
    """
    从所有配置的 Bing 广告账户拉取数据，返回标准化 DataFrame。
    环境变量 BING_ACCOUNT_IDS 格式: account_id:country,...
    """
    accounts_str = os.getenv("BING_ACCOUNT_IDS", "")
    if not accounts_str:
        raise RuntimeError("未配置 BING_ACCOUNT_IDS (格式: 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("BING_ACCOUNT_IDS 格式错误")

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

    access_token = _refresh_access_token()

    all_parsed: list[dict] = []
    for account_id, country in accounts:
        log.info(f"拉取 Bing 数据: account={account_id} country={country} {since}~{until}")
        try:
            request_id = _submit_report(access_token, account_id, since, until)
            log.info(f"  报告已提交: {request_id}")

            max_polls = 30
            for i in range(max_polls):
                time.sleep(5 if i < 6 else 15)
                download_url = _poll_report(access_token, account_id, request_id)
                if download_url:
                    break
                log.info(f"  报告生成中... (poll {i + 1})")
            else:
                raise RuntimeError(f"报告轮询超时 (>{max_polls} 次)")

            rows = _download_report(download_url)
            log.info(f"  获取 {len(rows)} 条记录")

            for row in rows:
                all_parsed.append(_parse_row(row, country))
        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__":
    logging.basicConfig(level=logging.INFO)
    df = fetch_bing_data(days_back=7)
    print(f"\n拉取到 {len(df)} 条数据")
    if not df.empty:
        print(f"日期范围: {df['date'].min()} ~ {df['date'].max()}")
        print(f"Campaign Types: {df['campaign_type'].unique()}")
        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")