"""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, "")]