pipeline/ingest.py

"""Windsor.ai pull → metrics_daily upsert + creative back-mapping (SPEC_NEW §7.6) — Phase 3.

Ingest sources ad-level daily metrics from the Windsor.ai Data API (configured Meta
connector), not the Meta Graph insights endpoint. Auth via WINDSOR_API_KEY. Verify the
exact Windsor.ai endpoint + field names at build time. Publishing stays on Meta
(see meta_client.py).

Flow (SPEC_NEW §7.6, Phase 3 acceptance §8):
- DRY_RUN or no WINDSOR_API_KEY → load fixtures/windsor_sample.json (no network), so the
  whole path is testable without credentials. Live → GET the Windsor.ai Data API.
- Map Windsor field names → metrics_daily columns via FIELD_MAP (override with WINDSOR_FIELDS).
- Upsert each row ON CONFLICT (ad_id, date), storing the full source row as raw jsonb.
- Back-map: ad_ids present in publishes are KNOWN (belong to a video); unknown ads are still
  ingested (raw kept). metrics_daily has no video FK — the link is publishes.ad_id at read time.
"""

from __future__ import annotations

import json

import httpx
from psycopg.types.json import Json

from . import db
from .config import FIXTURES_DIR, get_settings

# Verify-at-build-time (same convention as VEO_MODEL_ID / META_API_VERSION, SPEC_NEW §3):
# Windsor.ai Data API. The connector path segment is s.windsor_connector ("facebook" for Meta).
#   GET https://connectors.windsor.ai/{connector}?api_key=...&fields=...&date_from=...&date_to=...
# Response shape: {"data": [ {<field>: <value>, ...}, ... ]}.
WINDSOR_BASE_URL = "https://connectors.windsor.ai"  # verify-at-build-time
WINDSOR_TIMEOUT_S = 120

FIXTURE = FIXTURES_DIR / "windsor_sample.json"

# metrics_daily column → Windsor.ai field name. Override per-deploy with WINDSOR_FIELDS
# ("col=windsor_field,col=windsor_field,..."). The Meta-connector field names below are
# best-effort — verify-at-build-time against the Windsor.ai field reference for `facebook`
# (esp. video_3s / thruplay, which Meta names drift on). The fixture uses these defaults.
FIELD_MAP: dict[str, str] = {
    "ad_id": "ad_id",
    "date": "date",
    "impressions": "impressions",
    "spend": "spend",
    "clicks": "clicks",
    "video_3s": "video_3s",      # Meta "3-second video plays" — verify-at-build-time
    "thruplay": "thruplay",      # Meta "ThruPlay" — verify-at-build-time
    "purchases": "purchases",
    "revenue": "revenue",
}

# Numeric metric columns (everything except the (ad_id, date) key) — used to build the upsert.
METRIC_COLS = ("impressions", "spend", "clicks", "video_3s", "thruplay", "purchases", "revenue")


def _field_map() -> dict[str, str]:
    """FIELD_MAP, with WINDSOR_FIELDS overrides applied ("col=field,col=field")."""
    fields = dict(FIELD_MAP)
    override = (get_settings().windsor_fields or "").strip()
    if override:
        for pair in override.split(","):
            pair = pair.strip()
            if not pair or "=" not in pair:
                continue
            col, _, name = pair.partition("=")
            col, name = col.strip(), name.strip()
            if col in fields and name:
                fields[col] = name
    return fields


def _fetch_rows(days: int) -> list[dict]:
    """The list of source rows from Windsor.ai (or the fixture in dry-run / no key)."""
    s = get_settings()
    fields = _field_map()

    if s.dry_run or not s.windsor_api_key:
        why = "DRY_RUN" if s.dry_run else "no WINDSOR_API_KEY"
        print(f"[ingest] {why}: loading fixture {FIXTURE.name} instead of Windsor.ai "
              f"(connector={s.windsor_connector}, days={days})")
        payload = json.loads(FIXTURE.read_text())
        return list(payload.get("data") or [])

    # Live: GET the Windsor.ai Data API (verify-at-build-time — endpoint/fields/date params).
    from datetime import date, timedelta

    s.require("WINDSOR_API_KEY")
    date_to = date.today()
    date_from = date_to - timedelta(days=days)
    params = {
        "api_key": s.windsor_api_key,
        "fields": ",".join(fields[c] for c in FIELD_MAP),  # request exactly the mapped fields
        "date_from": date_from.isoformat(),
        "date_to": date_to.isoformat(),
    }
    url = f"{WINDSOR_BASE_URL}/{s.windsor_connector}"
    redacted = {**params, "api_key": "***"}
    print(f"[ingest] GET {url} {redacted}")
    r = httpx.get(url, params=params, timeout=WINDSOR_TIMEOUT_S)
    if r.status_code >= 400:
        raise RuntimeError(f"Windsor.ai {r.status_code}: {r.text[:400]}")
    return list(r.json().get("data") or [])


def _coerce(row: dict, fields: dict[str, str]) -> dict | None:
    """Pull metrics_daily column values out of one Windsor row via the field map.

    Returns None (skip) if the row has no ad_id/date — those are the only required keys.
    Metric values pass through as-is (None when absent); the full source row is kept in raw.
    """
    ad_id = row.get(fields["ad_id"])
    date = row.get(fields["date"])
    if ad_id is None or date is None:
        return None
    out: dict = {"ad_id": str(ad_id), "date": date}
    for col in METRIC_COLS:
        out[col] = row.get(fields[col])
    return out


def run(days: int = 14) -> int:
    """Pull `days` of ad-level daily metrics via Windsor.ai and upsert into metrics_daily.

    Returns the number of metrics_daily rows upserted. Logs known (in publishes) vs unknown
    ad counts; unknown ads are still ingested with their raw row kept (SPEC_NEW §7.6).
    """
    fields = _field_map()
    rows = _fetch_rows(days)

    # Known = ad_ids that round-trip back to a video via publishes.ad_id (no FK; joined at read).
    known_ids = {
        r["ad_id"]
        for r in db.fetch_all("select distinct ad_id from publishes where ad_id is not null")
    }

    upserted = 0
    seen_known: set[str] = set()
    seen_unknown: set[str] = set()
    with db.connect() as conn:
        for src in rows:
            vals = _coerce(src, fields)
            if vals is None:
                continue
            conn.execute(
                """
                insert into metrics_daily
                  (ad_id, date, impressions, spend, clicks, video_3s, thruplay, purchases, revenue, raw)
                values
                  (%(ad_id)s, %(date)s, %(impressions)s, %(spend)s, %(clicks)s, %(video_3s)s,
                   %(thruplay)s, %(purchases)s, %(revenue)s, %(raw)s)
                on conflict (ad_id, date) do update set
                  impressions = excluded.impressions,
                  spend       = excluded.spend,
                  clicks      = excluded.clicks,
                  video_3s    = excluded.video_3s,
                  thruplay    = excluded.thruplay,
                  purchases   = excluded.purchases,
                  revenue     = excluded.revenue,
                  raw         = excluded.raw
                """,
                {**vals, "raw": Json(src)},
            )
            upserted += 1
            (seen_known if vals["ad_id"] in known_ids else seen_unknown).add(vals["ad_id"])

    print(f"[ingest] upserted {upserted} metrics_daily row(s) over {days}d — "
          f"{len(seen_known)} known ad(s) (linked to videos via publishes), "
          f"{len(seen_unknown)} unknown ad(s) ingested raw-only")
    return upserted