api/providers.py

"""THE PROVIDER LAYER — many vendors behind one capability, chosen per FIELD and per cost.

⭐⭐ OWNER RULING 2026-08-08: *"Let's do a combination scraper. Do Bright Data first, and if
anything fails, we use APIfy or vice versa. Make this dynamic somehow. remember to cost optimize
when we scale to thousands+ customer … make sure our code is customizable and modular."*

⛔ THE DESIGN DECISION THAT MATTERS, AND IT IS NOT "TRY A, THEN TRY B".
A provider-level failover ("if Bright Data errors, re-run the whole thing on Apify") is the obvious
shape and it is the expensive one. MEASURED: Bright Data answers profile, likes and comments
correctly and CANNOT answer view counts on any route; Apify answers view counts correctly. A
provider-level fallback would notice the missing views and re-buy the profile, the likes and the
comments from Apify as well — **paying twice for the 90% that already worked**. At one profile that
is noise; at thousands of customers × twelve posts each it is the whole bill.

So routing is per CAPABILITY, and a capability is the smallest independently-billable thing:

    ig_profile       -> brightdata   (36 fields, works)
    ig_post_metrics  -> brightdata   (likes + comments, works)
    ig_post_views    -> apify        (Bright Data is INCAPABLE — see `capable=False` below)
    ig_comments      -> brightdata

Each capability has an ORDERED list of providers. The first one that is configured AND capable AND
succeeds wins; the rest are never called, so the common path costs exactly one vendor record.

⚠ "FAILURE" INCLUDES ANSWERING WITH THE FIELD BLANK. A vendor that returns HTTP 200 and a null in
the one column you asked for has failed for our purposes, and a chain that only catches exceptions
would stop at it forever. `run()` takes a `satisfied` predicate and falls through on an unsatisfying
answer exactly as it would on a 500.

⚠ ORDER IS DATA, NOT CODE. `AIOS_PROVIDER_ORDER` overrides any chain at deploy time
(`ig_post_views=apify,brightdata;ig_profile=apify`), which is what "or vice versa" means and what
lets a tenant be moved off a vendor without a release.
"""
from __future__ import annotations

import json
import os
import time
from dataclasses import dataclass, field
from typing import Callable

#: ⚠ APPROXIMATE, AND DELIBERATELY SO. These are list prices per RECORD in USD, used only to RANK
#: providers and to estimate a run's spend for the operator. They are not billing truth — the
#: vendor's own dashboard is. Wrong by 2x still ranks correctly; wrong by 100x does not, which is
#: why they are named and dated rather than guessed silently.
#: Bright Data Instagram datasets ~ $0.0015/record (2026-08 list). Apify instagram-scraper
#: ~ $0.0027/result (2026-08 list). Both re-checked when a provider is added.
#: ⚠ WAVE 30 · T16 — APIFY CORRECTED 0.0023 -> 0.0027, and the direction matters: the old number
#: made the fallback look CHEAPER than it is, and `estimate()` is what a person is shown before
#: they authorise a run. An under-stated price is the one rounding error a cost guard cannot catch.
#: ⚠ Both are LIST rates used to rank and to estimate. They are not invoices, and nothing here
#: reads a live price — a modelled figure that says so is honest; one that pretends is not.
_COST_BRIGHTDATA = float(os.environ.get("AIOS_COST_BRIGHTDATA") or 0.0015)
_COST_APIFY = float(os.environ.get("AIOS_COST_APIFY") or 0.0027)


@dataclass(frozen=True)
class Capability:
    """What ONE provider can do for ONE capability, and what it costs to ask."""
    #: ⛔ `False` means MEASURED INCAPABLE, not "untried". A provider declared incapable is never
    #: called for this capability at all — it cannot be reached by a fallback, cannot be put first
    #: by an env override, and cannot silently start being billed because somebody reordered a
    #: list. Bright Data's `ig_post_views` is False on the strength of §4e-§4h: every route
    #: (Posts x /p/, Posts x /reel/, Reels x /p/, Reels x /reel/, discover-by-profile) on four
    #: accounts from 13K to 268M followers returned an account-grain constant or a null.
    capable: bool = True
    cost_per_record: float = 0.0
    #: Free-text, shown to an operator deciding where their money went.
    note: str = ""


@dataclass
class Provider:
    key: str
    label: str
    #: The env var holding this provider's credential. NEVER the credential itself — this module
    #: is imported by surfaces that serialise their config.
    key_env: str = ""
    caps: dict = field(default_factory=dict)

    def configured(self) -> bool:
        return bool((os.environ.get(self.key_env) or "").strip()) if self.key_env else True

    def cap(self, capability: str) -> Capability | None:
        return self.caps.get(capability)

    def can(self, capability: str) -> bool:
        c = self.cap(capability)
        return bool(c and c.capable and self.configured())


#: ⚠ ADDING A PROVIDER IS A REGISTRY ENTRY PLUS A RUNNER — no change to any caller. That is the
#: "modular/customizable" half of the ruling, and the reason the chains below name STRINGS.
PROVIDERS: dict[str, Provider] = {
    "brightdata": Provider(
        key="brightdata", label="Bright Data", key_env="AIOS_BRIGHTDATA_KEY",
        caps={
            "ig_profile": Capability(True, _COST_BRIGHTDATA, "36 fields incl. bio/email/category"),
            "ig_post_metrics": Capability(True, _COST_BRIGHTDATA, "likes + comments + captions"),
            "ig_comments": Capability(True, _COST_BRIGHTDATA, "separate paid dataset, opt-in"),
            # ⛔ THE ONE THAT COSTS US NOTHING TO GET RIGHT AND EVERYTHING TO GET WRONG.
            "ig_post_views": Capability(
                False, _COST_BRIGHTDATA,
                "MEASURED INCAPABLE: returns one account-grain number for every reel of a "
                "creator (identical for two shortcodes in one call, ticking upward between "
                "calls) and never populates video_play_count. Their own documented example "
                "account no longer reproduces it."),
            # ⭐⭐ WAVE 29 (item 7 / D-9 / R1) — TIKTOK. Three datasets, all reachable with our
            # existing key, all schema-probed for $0.00 (40/43/17 fields with the vendor's own
            # types): `gd_l1villgoiiidt09ci` · `gd_lu702nij2f790tmv9h` · `gd_lkf2st302ap89utw5k`.
            "tt_profile": Capability(True, _COST_BRIGHTDATA,
                                     "40 fields incl. bio/engagement rates/region"),
            # ⛔ THERE IS NO `tt_post_views` CAPABILITY, AND ITS ABSENCE IS THE CLAIM.
            # `TikTok - Posts` declares `play_count: number` and the vendor's own sample says
            # "no empty values or zeros" — so on TikTok the view count arrives INSIDE the post
            # record and a separate view rung would be a second bill for a number we already have.
            # ⚠ DECLARED, NOT MEASURED, and that distinction is the whole scar tissue of §4e:
            # Bright Data's Instagram Reels also DECLARES `views: number` and delivers an
            # account-grain wrong one. One live pull settles it (W29-T06, spend-gated). If it
            # fails, the honest repair is a `tt_post_views` capability routed elsewhere — never a
            # quiet fallback bolted onto this one.
            "tt_post_metrics": Capability(
                True, _COST_BRIGHTDATA,
                "likes + comments + shares + saves, and play_count inline (DECLARED "
                "no-empties-or-zeros, UNPROVEN until one live pull)"),
            "tt_comments": Capability(True, _COST_BRIGHTDATA,
                                      "separate paid dataset, opt-in — 17 fields"),
        }),
    "apify": Provider(
        key="apify", label="Apify", key_env="AIOS_APIFY_KEY",
        caps={
            # ⭐ MEASURED 2026-08-08 against the public Reels grid: `videoPlayCount` 137,684 and
            # 299,493 vs a browser-read ground truth of 134K-137K and 299K. Exact.
            "ig_post_views": Capability(True, _COST_APIFY,
                                        "videoPlayCount — matches Instagram's displayed views"),
            "ig_post_metrics": Capability(True, _COST_APIFY, "likes + comments (fallback)"),
            "ig_profile": Capability(True, _COST_APIFY, "profile fields (fallback)"),
        }),
}

#: The DEFAULT chain per capability, cheapest-capable-first. Overridable — see `chain()`.
DEFAULT_CHAINS: dict[str, tuple] = {
    "ig_profile": ("brightdata", "apify"),
    "ig_post_metrics": ("brightdata", "apify"),
    # Only one entry, and that is the point: Bright Data is declared incapable, so listing it here
    # would be a lie that costs a wasted call on every single post.
    "ig_post_views": ("apify",),
    "ig_comments": ("brightdata",),
    # ⭐ WAVE 29 — TikTok, and every chain is deliberately SINGLE-PROVIDER.
    # ⛔ A MULTI-PROVIDER CHAIN IS A PROMISE SOMETHING WALKS IT. `ig_post_metrics` has declared a
    # two-provider fallback since wave 28 and NOTHING reaches the second name: if Bright Data
    # answers with the likes blank, Apify is never asked. `verify_automation`'s E2b pins that gap
    # BY NAME so a third unwalked chain turns it red — which is exactly what a second name here
    # would be today. Apify does sell TikTok (`clockworks/tiktok-profile-scraper`, $0.003/result,
    # 0.7% 30-day failure rate — measured from its own store record), so the fallback is buildable;
    # it is not declared until a runner walks it.
    "tt_profile": ("brightdata",),
    "tt_post_metrics": ("brightdata",),
    "tt_comments": ("brightdata",),
}


def chain(capability: str) -> list:
    """The provider order for `capability` — env override first, then the default.

    `AIOS_PROVIDER_ORDER` is a `;`-separated list of `capability=p1,p2` clauses. This is how the
    owner's "or vice versa" is expressed WITHOUT a release, and how one tenant can be moved off a
    vendor that is having a bad day.
    ⛔ An override may REORDER and may DROP, but it can never make an incapable provider capable —
    `can()` still gates every name. A chain that reads `ig_post_views=brightdata` therefore
    resolves to EMPTY rather than to a provider that would return a wrong number, because a
    plausible wrong number is worse than an honest refusal.
    """
    raw = (os.environ.get("AIOS_PROVIDER_ORDER") or "").strip()
    names = None
    for clause in raw.split(";"):
        if "=" in clause:
            cap_name, _, order = clause.partition("=")
            if cap_name.strip() == capability:
                names = [x.strip() for x in order.split(",") if x.strip()]
    if names is None:
        names = list(DEFAULT_CHAINS.get(capability) or ())
    return [PROVIDERS[n] for n in names if n in PROVIDERS and PROVIDERS[n].can(capability)]


def estimate(capability: str, records: int) -> dict:
    """What the FIRST capable provider would cost for `records` — the number an operator plans on.

    Reported per capability rather than per run because that is the unit that scales: at a thousand
    tenants the question is never "what did this run cost", it is "what does adding view counts to
    every post cost per month".
    """
    ch = chain(capability)
    if not ch:
        return {"capability": capability, "records": records, "provider": None, "usd": 0.0,
                "note": "no configured, capable provider"}
    p = ch[0]
    c = p.cap(capability)
    return {"capability": capability, "records": records, "provider": p.key,
            "usd": round((c.cost_per_record or 0.0) * max(0, int(records)), 4),
            "note": c.note}


@dataclass
class Attempt:
    provider: str
    ok: bool
    note: str = ""
    records: int = 0
    seconds: float = 0.0


def run(capability, work, satisfied=None, log=None):
    """Walk the chain for `capability` until one provider gives a SATISFYING answer.

    `work(provider) -> (result, note)`; a non-empty note means it did not answer.
    `satisfied(result) -> bool` decides whether an answer is good enough to stop. Default: any
    truthy result.

    Returns `(result, attempts)`. `attempts` is the audit trail — every provider tried, whether it
    satisfied, and how long it took — because "where did this number come from and what did it
    cost" is a question somebody asks about a bill, not about a stack trace.

    ⛔ THE `satisfied` HOOK IS THE WHOLE POINT. Without it this is an error-handler, and the
    failure it must catch is not an error: a vendor answering 200 with the one field you needed
    left blank. That is exactly how Bright Data behaves on view counts, and a chain that only
    caught exceptions would have stopped there forever and never reached Apify.
    """
    log = log or (lambda *_a: None)
    ok = satisfied or (lambda r: bool(r))
    attempts, last = [], None
    for provider in chain(capability):
        started = time.time()
        try:
            result, note = work(provider)
        except Exception as e:                                    # noqa: BLE001
            # ⚠ TYPE NAME ONLY. A vendor client's str() can carry the URL, and the credential is
            # one refactor away from being a query param; this line must not be what leaks it.
            result, note = None, f"{type(e).__name__}"
        secs = round(time.time() - started, 2)
        if note:
            attempts.append(Attempt(provider.key, False, note, 0, secs))
            log(f"  {provider.label}: {note} — falling through")
            continue
        if not ok(result):
            attempts.append(Attempt(provider.key, False, "answered without the field asked for",
                                    0, secs))
            log(f"  {provider.label}: answered, but not with what was asked for — falling through")
            last = result if last is None else last
            continue
        attempts.append(Attempt(provider.key, True, "", _count(result), secs))
        log(f"  {provider.label}: ok ({_count(result)} records, {secs}s)")
        return result, attempts
    return last, attempts


def _count(result):
    if isinstance(result, (list, tuple, set)):
        return len(result)
    if isinstance(result, dict):
        return len(result)
    return 1 if result else 0


def wire():
    """What the Settings surface renders. Booleans and labels — NEVER a credential.

    Mirrors the `hikerReady` rule the rest of the product follows: a surface is told WHETHER a
    provider is usable, never what the key is.
    """
    return {
        "providers": [
            {"key": p.key, "label": p.label, "configured": p.configured(),
             "capabilities": sorted(k for k, c in p.caps.items() if c.capable)}
            for p in PROVIDERS.values()
        ],
        "chains": {cap: [p.key for p in chain(cap)] for cap in DEFAULT_CHAINS},
        "incapable": {
            f"{p.key}:{cap}": c.note
            for p in PROVIDERS.values() for cap, c in p.caps.items() if not c.capable
        },
    }


# =============================================================================================
# THE CANONICAL SCHEMA — owner ruling 2026-08-08:
#   *"standardize the schema between Bright Data and APIfy so we keep using the same pre-set
#    database even if the underlying engine changes"*
#
# ⛔ THE PRESET TABLES ARE THE CONTRACT; A VENDOR IS AN IMPLEMENTATION DETAIL. `ut_ig_posts` and
# `ut_ig_post_snapshots` must not gain, lose or rename a column because a chain was reordered — a
# tenant's saved views, filters, rollups and forms all bind to these keys, and a schema that moves
# with the vendor turns a routing change into a data migration.
#
# So every provider normalises INTO the keys below and nothing reads a vendor row downstream.
# `verify_automation` asserts both normalisers emit exactly `CANONICAL_POST_KEYS` on a fixture, so
# a third provider cannot ship with a near-miss key like `viewCount` and silently write a column
# nobody declared.
#
# ⚠ ONE NAME PER MEASUREMENT, AND `views` IS THE MEASUREMENT INSTAGRAM DISPLAYS. Meta folded
# Impressions/Plays/Video Views into a single **Views** metric on 2025-04-10, so carrying both a
# `views` and a `plays` column would be modelling a distinction the platform deleted — and it is
# exactly the distinction that let an account-grain number wear the "Views" label for a month.
# ⛔ THE VENDOR KEY THAT LOOKS RIGHT IS THE WRONG ONE, ON BOTH VENDORS. Apify ships BOTH
# `videoViewCount` (10,678) and `videoPlayCount` (137,684) for one reel whose true displayed count
# is ~137K — and Bright Data's useless `views` is 10,638 for that same reel. The two vendors' junk
# fields AGREE with each other, which is precisely what makes picking by name so dangerous.
#   canonical `views`  <-  apify `videoPlayCount`      ✅ matches the grid
#   canonical `views`  <-  apify `videoViewCount`      ⛔ off by 13x
#   canonical `views`  <-  brightdata `views`          ⛔ off by 13x AND account-grain
# =============================================================================================

#: Every key a normalised POST row may carry. Absent > blank: a key is omitted when the provider
#: did not answer, because `upsert_rows` merges and an empty string would ERASE what an earlier
#: paid run learned.
CANONICAL_POST_KEYS = (
    "shortcode", "url", "influencer_key", "posted_at", "type", "caption",
    "likes", "comments", "views", "paid_partnership", "partner", "hashtags",
    "alt_text", "tagged_location", "source_payload",
    # ⭐ 2026-08-09 — the three the SECOND provider answers and the first does not. Each has a
    # matching `field_def` in the engine's POST_FIELDS; a key here without a column there is a
    # value that normalises cleanly and is then dropped by the write door, silently.
    "plays", "video_duration", "comments_disabled",
)


def _int_or_none(v):
    """`-1` IS NOT A COUNT. Apify returns `likesCount: -1` when the creator HIDES their like
    count — a real state that is not a measurement. Writing -1 would render as a negative like
    count; writing 0 would claim nobody liked it. Both are lies, so the key is omitted."""
    try:
        n = int(v)
    except (TypeError, ValueError):
        return None
    return None if n < 0 else n


def normalize_post_apify(row):
    """One Apify `instagram-scraper` item -> the canonical post row."""
    if not isinstance(row, dict):
        return None
    code = str(row.get("shortCode") or "").strip()
    if not code:
        return None
    out = {
        "shortcode": code,
        "url": str(row.get("url") or f"https://www.instagram.com/reel/{code}/"),
        "influencer_key": str(row.get("ownerUsername") or ""),
        "posted_at": str(row.get("timestamp") or "").replace("T", " ")[:16],
        "type": "video" if str(row.get("type") or "").lower() == "video" else
                ("carousel" if row.get("childPosts") else "image"),
        "caption": str(row.get("caption") or ""),
        # ⭐ THE FIELD THIS WHOLE PROVIDER EXISTS FOR. `videoPlayCount`, never `videoViewCount`:
        # the wrong one agreed with Bright Data's junk (10,678 vs 10,638) on a reel whose true
        # count was ~137K, and `videoPlayCount` matched a browser read to the digit.
        "views": _int_or_none(row.get("videoPlayCount")),
        # ⭐⭐ 2026-08-09 (owner: *"Video plays (# Plays) field isn't in APIfy? i believe it is"*).
        # They were right, and the `plays` COLUMN had been empty on all 816 rows because nothing
        # ever wrote it — the field existed with no writer. MEASURED on two of their own reels:
        # `videoPlayCount` 216,904 / 95,331 and `videoViewCount` 115,929 / 30,871. Two different
        # real numbers, and we were storing only one of them.
        # ⚠ `plays` TAKES THE PLAY COUNT, which is also what `views` carries today — so the two
        # columns will agree until somebody decides otherwise, and that decision is the OWNER'S:
        # re-sourcing `views` to `videoViewCount` would change what the ~480 rows already captured
        # mean, and a column whose meaning changes halfway down is the one thing worse than a
        # column with no data. Flagged rather than done.
        "plays": _int_or_none(row.get("videoPlayCount")),
        "likes": _int_or_none(row.get("likesCount")),
        "comments": _int_or_none(row.get("commentsCount")),
        # ⭐ FIELDS BRIGHT DATA DOES NOT RETURN AT ALL, kept because they are already paid for in
        # this same response (owner: *"whatever APIfy has more than BD pls use it"*).
        "video_duration": _int_or_none(row.get("videoDuration")),
        "comments_disabled": "1" if row.get("isCommentsDisabled") else "",
        "hashtags": ", ".join(str(h) for h in (row.get("hashtags") or []) if h),
        "alt_text": str(row.get("alt") or ""),
        "paid_partnership": "1" if row.get("paidPartnership") else "",
        "partner": ", ".join(str((s or {}).get("username") or "")
                             for s in (row.get("sponsors") or []) if isinstance(s, dict)),
        "tagged_location": str((row.get("locationName") or "")),
        "source_payload": json.dumps(row, default=str)[:32_000_000],
    }
    return {k: v for k, v in out.items() if v not in (None, "")}


def normalize_profile_apify(row):
    """One Apify `instagram-scraper` PROFILE item (`resultsType: "details"`) -> our snapshot shape.

    ⭐ WHY THIS EXISTS (owner report 2026-08-09: *"I need Bright Data and Apify to work correctly
    in tandem"*). `PROVIDERS["apify"]` has declared `ig_profile` capable since 2026-08-08 and
    `DEFAULT_CHAINS["ig_profile"]` has read `("brightdata", "apify")` — but **nothing ever ran
    that chain.** `connectors_ig.pull_profile` went Bright Data -> anonymous HTML rungs and Apify
    was never asked, so a profile Bright Data cannot scrape came back `blocked` while a
    configured, declared-capable provider sat unused. A registry entry with no runner is a
    promise the product does not keep ([[flag-shipped-without-its-writer]]).

    ⚠ THE KEYS ARE APIFY'S camelCase, and mapping them HERE is the boundary rule this module
    already enforces for posts: nothing downstream may ever see a vendor-shaped key, so the
    snapshot writer cannot tell which vendor answered — which is what makes the fallback
    invisible to every consumer instead of a second schema.

    ⛔ OMIT, NEVER ZERO. A field Apify did not send is dropped, exactly as `normalize_post_apify`
    drops a missing count: writing 0 followers would claim we measured an empty account, and the
    caller's `satisfied` hook reads absence as "did not answer" and falls through. A zero would
    stop the chain on a lie.
    """
    if not isinstance(row, dict):
        return None
    # ⛔ AN ERROR ENVELOPE IS NOT A PROFILE (measured 2026-08-09). Apify answers a dead handle
    # with a 200 and a well-formed item — `{"username": …, "error": "not_found",
    # "errorDescription": "Post does not exist"}` — and this function used to build a "profile"
    # out of it: a username, a url and a source_payload, with every measured field absent. It
    # then read to the caller as a vendor that answered, so the reason was replaced by a shrug.
    # Refused HERE as well as in `connectors_ig.apify_profile` on purpose: the boundary rule this
    # module exists for is that nothing downstream ever sees a vendor-shaped key, and a vendor's
    # ERROR shape is the one that must never become a row.
    # ⚠ TRUTHINESS, NOT KEY PRESENCE — a successful item carries `error: null` (measured on
    # `sriyynntt`), so `"error" in row` would refuse every good profile.
    if row.get("error"):
        return None
    handle = str(row.get("username") or "").strip()
    if not handle:
        return None
    out = {
        "username": handle,
        "full_name": str(row.get("fullName") or ""),
        "bio": str(row.get("biography") or ""),
        "followers": _int_or_none(row.get("followersCount")),
        "following": _int_or_none(row.get("followsCount")),
        "posts_count": _int_or_none(row.get("postsCount")),
        "verified": "1" if row.get("verified") else "",
        "external_url": str(row.get("externalUrl") or ""),
        "ig_id": str(row.get("id") or ""),
        "profile_url": str(row.get("url") or f"https://www.instagram.com/{handle}/"),
        "business_category": str(row.get("businessCategoryName") or ""),
        "is_business": "1" if row.get("isBusinessAccount") else "",
        "is_private": "1" if row.get("private") else "",
        "highlights_count": _int_or_none(row.get("highlightReelCount")),
        "source_payload": json.dumps(row, default=str)[:32_000_000],
    }
    # ⚠ `followers`/`following` are ints and 0 is a LEGITIMATE value for them, so the filter
    # below must not treat 0 as absent the way the post normaliser can — an account really can
    # have zero followers. Only None and "" are dropped.
    return {k: v for k, v in out.items() if v is not None and v != ""}


def canonical_gaps(row, want=("views",)):
    """Which requested canonical fields this row does NOT carry — the `satisfied` input.

    Named rather than inlined because "did the vendor actually answer the question" is the whole
    fallback trigger, and a chain that asks it differently in two places will drift.
    """
    row = row if isinstance(row, dict) else {}
    return [k for k in want if row.get(k) in (None, "")]