api/connectors_ig.py

"""connectors_ig.py — THE INSTAGRAM CONNECTOR. Every wire that reaches a vendor for IG data.

Wave 27 item 23, and the split is along one line: **this module knows what Instagram and its
vendors look like; `automation_engine.py` knows what an AUTOMATION is.** The engine used to know
both, which is how a 10,500-line file came to carry Bright Data's snapshot namespaces, Apify's
actor id and the shape of an `og:description` string. Nothing about the automation runtime needed
any of that, and nothing here needs to know what a trigger is.

WHAT LIVES HERE
  * the ANONYMOUS ladder's readers — Instagram's own public surfaces, $0, no key of any kind;
  * BRIGHT DATA — the paid rung: the scraper route, the pre-collected CORPUS route, their two
    mutually-404ing snapshot namespaces, and the row mappers that turn either into our shape;
  * APIFY — the second provider, wired for the one capability Bright Data is measured incapable of;
  * the CORPUS FILTER wire — the discovery query's shape guards (the vendor's 4-rule group cap and
    its 3-level nesting ceiling) and the start/status/download calls;
  * `pull_profile` — the orchestrator that walks paid rung then anonymous rungs for one profile.

WHAT DELIBERATELY DOES NOT
  * **Table schema.** `SNAPSHOT_FIELDS`, `POST_FIELDS`, `COMMENT_FIELDS`, `POST_SNAPSHOT_FIELDS`
    and the `ut_ig_*` presets stay in the engine. They declare COLUMNS A TENANT'S VIEWS BIND TO —
    product schema that must not move when a vendor does. That is the same boundary
    `providers.CANONICAL_POST_KEYS` draws one level down, and drawing it twice is the point.
  * **Routing between vendors.** `providers.py` owns which vendor answers which capability, in
    what order, and at what cost. This module supplies the RUNNERS that routing calls.
  * **Discovery vocabulary.** Which fields a person may filter on, what the operators are called,
    which ones actually narrow a 620M-row corpus — that is a product surface and it stays in the
    engine beside the runner that spends money on it.

⛔⛔ THE IMPORT DIRECTION, WHICH IS THE ONE DECISION IN THIS FILE THAT IS NOT ABOUT INSTAGRAM.
`automation_engine` imports THIS module at its top level (it needs `bd_ready`, `pull_profile` and
the mappers, and so do `routes_automation` and `routes_connectors` through it). So a module-level
`import automation_engine` HERE would close a cycle — and a cycle's failure depends on which side
a caller reached first. The server imports the engine first and would have worked; a gate or a
probe doing `import connectors_ig` would have got a half-built engine and an AttributeError from
inside a vendor call. **An intermittent-by-import-order failure is the worst shape available**, so
every reach-back is a `from automation_engine import …` INSIDE the function that needs it, where
the module is always fully built. There are exactly seven — `_bd_why`, `bd_call`,
`apify_posts`, `apify_reels`, `apify_profile`, `_tag_metric_deferrals`, `pull_profile_bd`,
`pull_profile` — each
marked `# lazy — see the module header`, and they reach for four kinds of thing:

    fetch / fetch_json / Refused   the SSRF-guarded HTTP rail (the scrape runner shares it)
    PACE_SECONDS                   the pacing floor — ⚠ read at CALL time ON PURPOSE: the gate
                                   sets `engine.PACE_SECONDS = 0` to run in seconds instead of
                                   minutes, and a copy imported at module load would silently
                                   stop honouring that while every test still passed
    DEFAULT_POSTS_PER_PULL         `clean_max_posts`' default lives with the config that validates
    _s / _iso / clean_tier         small shared helpers with one definition each, deliberately

⚠ A CREDENTIAL NEVER LEAVES THIS FILE, AND NEVER APPEARS IN A RETURN VALUE. Keys are read lazily
from the environment inside the call that uses them (`bd_key`, `apify_key`), travel in a header or
a query string, and every error path below quotes a STATUS or an exception TYPE NAME — never the
request, never `str(e)`. The readiness booleans (`bd_ready`, `apify_ready`) are what a surface is
allowed to know.
"""
from __future__ import annotations

import hashlib
import json
import os
import re
import time
from urllib.parse import urlparse

import requests

import providers

# ⭐⭐ WAVE 30 · T09 (DEBT D-128) — THE VENDOR WIRE LIVES IN `connectors_bd.py` NOW, AND THIS IS AN
# IMPORT, NOT A RE-EXPORT-FOR-EVERYONE. Only the names THIS file's own code still calls are listed:
# `verify_automation.section_split` derives the list both ways from the AST, so an import that
# outlived its last call site and a use with no import are each a red. Everything else that used to
# live here — `bd_ready`, `bd_key`, the whole corpus filter, `bd_group`, `expand_predicates` — is
# reached by its callers directly from `connectors_bd`.
#
# ⛔ **WHY THE NAMES ARE IMPORTED RATHER THAN CALLED AS `bd.<name>`, deliberately:** the functions
# below that still call `bd_call` and `bd_scrape` resolve them from THIS module's globals, which is
# what makes `connectors_ig.bd_call = <stub>` continue to seal them exactly as it did before the
# split. Changing the call style would have silently retired six working patch sites in the gate.
# ⚠ **AND THE HALF THAT DOES NOT SURVIVE, because it is the money one:** a function that MOVED
# (`bd_scrape`, `bd_snapshot_progress`, the corpus trio) now resolves `bd_call` from
# `connectors_bd`'s globals, so patching it HERE no longer seals it. `verify_automation._seal_bd`
# patches every module that holds a binding and then sweeps `sys.modules` to prove none was missed.
from connectors_bd import (                                          # noqa: F401
    BD_MAX_KB,
    BD_METRIC_SCRAPE_WAIT,
    BD_PATH_FILTER,
    BD_PATH_FILTER_SNAPSHOT,
    BD_PATH_SNAPSHOT,
    BD_PATH_TRIGGER,
    BD_SCRAPE_POLL,
    BD_SCRAPE_WAIT,
    _bd_deferral,
    _bd_first_url,
    _bd_flag,
    _bd_list,
    _bd_rows,
    _bd_source_payload,
    _first,
    _ig_int,
    bd_call,
    bd_scrape,
    bd_snapshot_progress
)


# ---------------------------------------------------------------------------------------------
# HANDLE + COUNT READERS (no vendor, no key — pure parsing)
# ---------------------------------------------------------------------------------------------
# ⛔ THE ANONYMOUS LADDER THAT USED TO LIVE HERE IS DELETED (wave 28, owner ruling R5), and this
# note is the reason it must not come back on a "we could read this for free" impulse. Its two
# rungs — Instagram's `web_profile_info` endpoint and the profile HTML's `og:description` —
# returned counts the PAGE had already rounded ("204K"), which landed in the same `followers`
# column as a measured 204,318 under an `approx: "1"` flag nobody sees until after they have
# averaged it. Free was never the problem; a column you cannot do arithmetic on was.
# Gone with it: `_ig_get`, `IG_APP_ID`, `IG_MEDIA_QUERY_HASH`, `_paginate`, `_posts_from_edges`,
# `_profile_from_web_api`, `_OG_COUNTS`, `_loose_count`.

def ig_handle(url):
    """The handle out of a profile URL (or a bare handle). '' when it is not one."""
    raw = str(url or "").strip()
    if not raw:
        return ""
    if raw.startswith("@"):
        return re.sub(r"[^A-Za-z0-9._]", "", raw[1:])[:40]
    if "instagram.com" not in raw:
        return re.sub(r"[^A-Za-z0-9._]", "", raw)[:40] if "/" not in raw else ""
    path = urlparse(raw if "//" in raw else "https://" + raw).path.strip("/")
    first = (path.split("/") or [""])[0]
    if first in ("p", "reel", "reels", "explore", "accounts", "stories", ""):
        return ""
    return re.sub(r"[^A-Za-z0-9._]", "", first)[:40]




def _ig_zero_is_blank(v):
    """Like `_ig_int`, but a vendor `0` reads as NOT MEASURED rather than as a measurement.

    ⛔ FOR THE PAID VIEW/PLAY RUNG ONLY. `POST_FIELDS` states the rule — *"BLANK, NEVER ZERO … a 0
    here would claim a post nobody watched"* — but `_ig_int(0)` returns `0`, so the rule was
    prose with nothing enforcing it. MEASURED 2026-08-08: `sriyynntt` came back with `views: 0` on
    all twelve reels and those zeros were published as measurements; they were removed by the
    REPEAT rule, not the zero rule, i.e. by luck. A single vendor `0` on one reel still landed.
    ⚠ DO NOT reach for this on `likes` or `comments`. Zero likes is a real, common and useful
    measurement; blanking it would destroy data to satisfy a rule that was never about it.
    """
    n = _ig_int(v)
    return None if n == 0 else n


BD_DS_PROFILES = "gd_l1vikfch901nx3by4"       # Instagram – Profiles. 36 fields, 620M records
BD_DS_POSTS = "gd_lk5ns7kz21pck8jpis"         # Instagram – Posts. 43 fields
BD_DS_REELS = "gd_lyclm20il4r5helnj"           # Instagram Reels: views and play counts
BD_DS_COMMENTS = "gd_ltppn085pokosxh13"        # Instagram Comments: opt-in full comment engagement














#: Source-data retention is intentionally unfiltered: if Bright Data returned it in an already
#: paid Profile, Post, Reel, or Comment response, it belongs in that record's source document.
#: The one cost boundary is the *request*: the full Comments dataset is only called when an owner
#: enables `commentMetrics`.


#: The vendor's content vocabulary → ours. MEASURED values: profiles' `posts[].content_type` is
#: `Image`/`Reel`/`Carousel`; the Posts dataset adds `product_type` `clips`/`carousel_container`.
_BD_TYPES = {"image": "image", "images": "image", "photo": "image", "photos": "image",
             "video": "video", "videos": "video", "reel": "video", "reels": "video",
             "clip": "video", "clips": "video", "igtv": "video", "carousel": "carousel",
             "carousel_container": "carousel", "sidecar": "carousel", "posts": "image"}


def _bd_type(*raw):
    """The first recognisable content type among `raw`, defaulting to `image`."""
    for v in raw:
        hit = _BD_TYPES.get(str(v or "").strip().lower())
        if hit:
            return hit
    return "image"


def ig_shortcode(url):
    """The shortcode out of a post/reel permalink. '' when it is not one.

    The `ut_ig_posts` identity key. The vendor gives a post URL (sometimes without its trailing
    slash) and a numeric `pk`; the shortcode is what both datasets agree on and what a human can
    paste into a browser, so it stays the key.
    """
    raw = str(url or "").strip()
    if not raw:
        return ""
    path = urlparse(raw if "//" in raw else "https://" + raw).path.strip("/").split("/")
    if len(path) >= 2 and path[0] in ("p", "reel", "reels", "tv"):
        return re.sub(r"[^A-Za-z0-9_-]", "", path[1])[:40]
    return ""




def _bd_posts_count(node):
    """`posts_count`, or None — **and a vendor ZERO IS DISCARDED.**

    ⛔ MEASURED 2026-08-05: the Profiles dataset returned `posts_count: 0` for BOTH probe accounts
    — `inayma` (203,800 followers, 12 posts returned in the same row) and `utopian_events` (20,924
    followers, 12 posts). The POSTS dataset carries the true value for the same two accounts
    (1,215 and 751), so the zero is not the account being empty; it is a measurement failure
    wearing a number. Writing it would be the invented measurement this module's honest-status
    contract exists to forbid — and a `_first`-style candidate list cannot help, because the
    vendor DID answer, with the wrong answer.

    ⚠ THE ASYMMETRY IS DELIBERATE AND IT COSTS SOMETHING: an account that genuinely has zero posts
    also reads blank. "Not read" is honest about an unknown; "0" is a claim, and on the only
    evidence we have that claim is wrong 2 times out of 2.
    ⛔ `len(posts)` IS NOT A SUBSTITUTE. The profile row carries the TOP 12 — a cap, not a count.
    """
    v = _ig_int(_first(node, "posts_count", "media_count", "post_count"))
    return None if v == 0 else v






def _bd_tagged_location(node):
    """Bright Data's optional Post `location` -> one filterable content-context string.

    The documented Posts response is an ordered array such as
    `["Capanema", "Pará", "Brasil"]`; retain the order rather than guessing which component is
    a city or country. Reels do not promise this field, so a blank means it was not returned.
    """
    raw = _first(node, "location", "location_details")
    if isinstance(raw, dict):
        raw = [raw.get(k) for k in ("name", "city", "region", "state", "country", "slug")]
    if not isinstance(raw, list):
        raw = [raw]
    parts = []
    for value in raw:
        if isinstance(value, dict):
            value = _first(value, "name", "city", "region", "state", "country", "slug")
        text = str(value or "").strip()
        if text and text not in parts:
            parts.append(text)
    return ", ".join(parts)[:400] or None




def _bd_comment(row, influencer_key="", shortcode="", fallback_key=""):
    """One Bright Data Comment response -> one linked Comment record with engagement.

    The dedicated Comments dataset is invoked only by the default-off `commentMetrics` switch.
    This mapper is also used for comment arrays embedded in an already-paid Post/Reel response,
    which adds no request. The full provider object is preserved in Source data; Likes and Replies
    are promoted for the grid and future rollups.
    """
    if not isinstance(row, dict):
        return None
    url = str(_first(row, "url", "post_url", default="") or "")
    code = str(shortcode or "") or ig_shortcode(url)
    comment_id = str(_first(row, "comment_id", "id", "pk", default="") or fallback_key or "")
    if not code or not comment_id:
        return None
    posted = str(_first(row, "comment_date", "date_of_comment", "created_at", "created_time",
                        default="") or "")
    return {
        "comment_key": comment_id,
        "shortcode": code,
        "influencer_key": str(influencer_key or _first(row, "post_user", default="") or ""),
        # ⭐⭐ OWNER RULING 2026-08-12 — the comment's CONTENT is a column on BOTH networks now.
        # `comment` is the key Bright Data's Comments dataset sends (and the one this module's own
        # fixture has always carried); the rest are defensive, in the house `_first` style, because
        # the EMBEDDED comment arrays on a paid Post/Reel response are a different shape from the
        # dedicated dataset and this mapper serves both.
        # ⛔ TEXT ONLY — `comment_user` / `comment_user_url` are vendor-flagged PII and stay in
        # `source_payload`, which the 2026-08-07 retention ruling keeps whole.
        "text": str(_first(row, "comment", "comment_text", "text", default="") or ""),
        "commented_at": posted.replace("T", " ")[:16],
        "likes": _ig_int(_first(row, "likes_number", "likes", "like_count", "likes_count")),
        "replies": _ig_int(_first(row, "replies_number", "num_replies", "replies",
                                   "reply_count", "replies_count")),
        "source_payload": _bd_source_payload(row),
    }


def _bd_embedded_comments(post, influencer_key=""):
    """Comment rows carried inside a paid Profile/Post/Reel record, with no extra request."""
    if not isinstance(post, dict):
        return []
    shortcode = str(post.get("shortcode") or "") or ig_shortcode(post.get("url"))
    if not shortcode:
        return []
    out, seen = [], set()
    # Known Bright Data arrays plus any future list-valued comment field. Source data still keeps
    # unknown shapes even if it cannot be normalised into a Comment row.
    arrays = [(str(k), v) for k, v in post.items()
              if isinstance(v, list) and (str(k) in ("latest_comments", "top_comments")
                                           or "comment" in str(k).lower())]
    for source_key, values in arrays:
        for index, value in enumerate(values):
            if not isinstance(value, dict):
                continue
            try:
                raw = json.dumps(value, ensure_ascii=False, sort_keys=True,
                                 separators=(",", ":"))
            except (TypeError, ValueError):
                continue
            fallback = "embedded:" + hashlib.sha256(
                f"{shortcode}:{source_key}:{index}:{raw}".encode("utf-8")).hexdigest()[:24]
            got = _bd_comment(value, influencer_key=influencer_key, shortcode=shortcode,
                              fallback_key=fallback)
            if got and got["comment_key"] not in seen:
                seen.add(got["comment_key"])
                out.append(got)
    return out


def _bd_profile(node, handle):
    """A Profiles row → our snapshot shape. Every name below is MEASURED off a real row
    (2026-08-05, 35 keys); the candidates guard renames, not guesses.

    ⚠ Several of these were **null on both probe rows** (`avg_engagement`, `category_name`,
    `business_category_name`, `is_professional_account`, `bio_hashtags`, `post_hashtags`). They
    are mapped ANYWAY — see the note on `SNAPSHOT_FIELDS`: n=2 is not evidence enough to drop a
    column permanently, and history is unbuyable, so a field omitted today is lost for today.
    ⚠ They ARE populated on the CORPUS (`/datasets/filter`) rows, which is a different path with
    a different population; that is a reason to capture them, not a licence to promise a
    scrape-side feature before D-25 measures the fill rate.
    """
    out = {
        "username": str(_first(node, "account", "username", default=handle) or handle),
        "full_name": str(_first(node, "full_name", "profile_name", default="") or ""),
        "bio": str(_first(node, "biography", "bio", default="") or ""),
        "followers": _ig_int(_first(node, "followers", "follower_count")),
        "following": _ig_int(_first(node, "following", "following_count")),
        "posts_count": _bd_posts_count(node),
        "verified": "1" if _first(node, "is_verified", default=False) else "",
        "external_url": _bd_first_url(_first(node, "external_url", "external_urls",
                                             "external_url_title")),
        # ⛔ NO `approx` KEY. These counts are exact, and carrying the anonymous rung's rounding
        # tag onto them would be a lie in the other direction (R1: "drop the approx tag when
        # exact"). Its ABSENCE is what tells the snapshot row it is trustworthy.
        "ig_id": str(_first(node, "id", "fbid", default="") or ""),
        "profile_url": str(_first(node, "profile_url", "url", default="") or ""),
        "avg_engagement": _first(node, "avg_engagement"),
        "category": str(_first(node, "category_name", default="") or ""),
        "business_category": str(_first(node, "business_category_name", default="") or ""),
        "is_business": _bd_flag(node, "is_business_account"),
        "is_professional": _bd_flag(node, "is_professional_account"),
        "is_private": _bd_flag(node, "is_private"),
        "highlights_count": _ig_int(_first(node, "highlights_count")),
        "bio_hashtags": _bd_list(node, "bio_hashtags"),
        "pronouns": str(_first(node, "pronouns", default="") or ""),
        # ⭐ 2026-08-07 (owner instruction: *"have ALL Fields available to us from Bright Data to
        # be pre-set Fields for us and populated"*) — THE REST OF THE PROFILES SCHEMA.
        #
        # Every key below was MEASURED on a real row (2026-08-05, 35 row keys vs 36 declared;
        # `wave20-split.md`), and the first four were explicitly logged as *"populated on scrape
        # and NOT in any previous list"* — so these are not speculative names, they are fields the
        # vendor was already sending and this map was already throwing away.
        # Every unpromoted provider key — including contact and image fields — stays complete in
        # Source data. This keeps the locked grid readable without throwing away paid data.
        "profile_name": str(_first(node, "profile_name", default="") or ""),
        "is_joined_recently": _bd_flag(node, "is_joined_recently"),
        "has_channel": _bd_flag(node, "has_channel"),
        "partner_id": str(_first(node, "partner_id", default="") or ""),
        # ⚠ An OBJECT on the real row (`external_url_title`), which is why it goes through the
        # same reader as the url itself rather than `str()` — a `str(dict)` would write the
        # literal `{'title': …}` into a column somebody reads.
        "external_url_title": _bd_first_url(_first(node, "external_url_title")),
        # ⚠ `fbid` is the Facebook-side id, a SECOND opaque identifier beside `ig_id`. Text for
        # the same reason `ig_id` is: it is a name, not a quantity, and some exceed 2^53.
        "fbid": str(_first(node, "fbid", default="") or ""),
        # ⚠ MEASURED `None` (not `[]`) on the real row — the graph-walk discovery seed is dead
        # (§2c). Mapped anyway: history is unbuyable, so a field not captured today is lost for
        # today, and n=1 is not evidence enough to drop a column permanently.
        "related_accounts": _bd_list(node, "related_accounts"),
        # ⚠ `country_code` is the ONE field the corpus filter API REJECTS as a predicate, and it
        # was `None` on all five corpus rows. It is a row field regardless, and reading it costs
        # nothing; a blank here means not read, like every other blank in this map.
        "country_code": str(_first(node, "country_code", default="") or ""),
        "source_payload": _bd_source_payload(node),
    }
    return out


def _bd_post_identity(p):
    """One entry of a Profiles row's `posts[]` → our post shape. None when it has no shortcode.

    ⚠ IDENTITY ONLY — AND THAT IS THE FINDING THIS WHOLE RUNG IS SHAPED AROUND. MEASURED
    2026-08-05 across 24 posts on two accounts: `caption`, `content_type`, `id`, `image_url`,
    `url` are populated 24/24, and **`likes`, `comments` and `datetime` are `None` 24/24.** So the
    profile scrape can fill `ut_ig_posts` and CANNOT fill `ut_ig_post_snapshots` — the engagement
    time series §3's append law exists for needs the POSTS dataset, one record per post. That is
    why `postMetrics` is a separate, opt-in, ~13×-cost rung rather than a free by-product.
    ⚠ `is_pinned` was `True` on all 24 — a constant, not a measurement. Not mapped.
    ⚠ Blank-valued keys are OMITTED rather than written, so a later engagement pull that DOES know
    `posted_at` fills it in instead of being overwritten by this rung's silence.
    """
    code = ig_shortcode(p.get("url")) if isinstance(p, dict) else ""
    if not code:
        return None
    kind = _bd_type(p.get("content_type"), p.get("type"))
    # The URL route is part of Bright Data's endpoint contract. A Reel sent through the Posts
    # scraper can return likes/comments while omitting its view fields; `/reel/{shortcode}` is
    # the input documented for the Reels scraper.
    route = "reel" if kind == "video" else "p"
    out = {"shortcode": code, "url": f"https://www.instagram.com/{route}/{code}/",
           "type": kind}
    caption = _first(p, "caption", "description", default="")
    if isinstance(caption, dict):
        caption = _first(caption, "text", default="")
    if caption:
        out["caption"] = str(caption)
    out["source_payload"] = _bd_source_payload(p)
    embedded = _bd_embedded_comments(p)
    if embedded:
        out["embedded_comments"] = embedded
    return out


def _bd_post_metrics(row):
    """One POSTS-dataset row → the engagement half. None when it has no shortcode.

    MEASURED 2026-08-05 on three real post rows: `likes` (308 / 3,736), `num_comments` (37 / 134)
    and `date_posted` (`2026-05-04T17:51:58.000Z`) all populated.
    ⚠ NO VIEW COUNT CAME BACK — `video_play_count` / `video_view_count` are declared in the schema
    and the KEY WAS ABSENT from every row, including a Reel. So `views` reads None and the column
    stays blank, which is the honest answer; it is not folded into likes and it is not zeroed.
    ⚠ The candidates below are all COUNTS OF VIEWS in different vendor typings (`video_view_count`
    is declared `text`, the others `number`). A candidate list may span types only when the
    MEANING is identical — never the way `like_and_view_counts_disabled`, a boolean, would have
    turned "likes are hidden" into "1 like".
    Embedded comment arrays are retained in Source data and converted to linked Comment rows
    without triggering the separate full Comments-dataset request.
    """
    if not isinstance(row, dict):
        return None
    code = str(_first(row, "shortcode", "content_id", default="") or "") or \
        ig_shortcode(row.get("url"))
    if not code:
        return None
    posted = str(_first(row, "date_posted", "taken_at", default="") or "")
    partner = row.get("partnership_details")
    out = {
        "shortcode": code,
        "url": str(_first(row, "url", default=f"https://www.instagram.com/p/{code}/")),
        "influencer_key": str(_first(row, "user_posted", default="") or ""),
        "posted_at": posted.replace("T", " ")[:16],
        "type": _bd_type(row.get("content_type"), row.get("product_type")),
        "caption": str(_first(row, "description", "caption", default="") or ""),
        "likes": _ig_int(_first(row, "likes", "like_count")),
        "comments": _ig_int(_first(row, "num_comments", "comment_count", "comments")),
        # ⛔⛔ `views` IS NOT READ HERE, AND THE FIELD NAME IS THE WHOLE TRAP (2026-08-08, §4e of
        # instagram-capture.md). Bright Data's `views` is an ACCOUNT-level number: it rides in the
        # vendor's own profile-context block —
        #     "input": {"url": …, "posts_count": 386, "followers": 6133, "following": 2850,
        #               "video_play_count": null, "views": null, "country_code": null}
        # — next to followers/following/posts_count, and on the delivered row it is CONSTANT
        # across every reel exactly as those are. MEASURED: one value across 12 distinct reels
        # whose likes ran 3→271, on 18 of 19 creators, and 9753→9754 between two captures two
        # hours apart (so it is a live account counter, not a stale cache).
        # ⚠ THIS IS NOT A COLLECTION-MODE PROBLEM AND MUST NOT BE "FIXED" BY REWRITING THE CALL.
        # The documented discover-by-profile contract was bought and tested: same constant, same
        # null play count. Reading `views` here promoted a profile metric into a post column, and
        # that — not a vendor fault — is what put one number on many unrelated posts.
        # `video_play_count` is the genuine per-reel field. It is NULL on 91 of 91 records across
        # 20 accounts in this cohort and BOTH collection modes, so Plays stays blank; blank is the
        # honest answer for a paid rung that did not answer, and always was.
        "plays": _ig_zero_is_blank(_first(row, "video_play_count", "play_count")),
        # Sponsored-post detection. MEASURED populated (`True`, with the brand alongside) — and
        # it is the one field here that answers a commercial question the counts cannot.
        "paid_partnership": _bd_flag(row, "is_paid_partnership"),
        "partner": (str(_first(partner, "username", "profile_id", default="") or "")
                    if isinstance(partner, dict) else ""),
        "hashtags": _bd_list(row, "hashtags"),
        "alt_text": str(_first(row, "alt_text", default="") or ""),
        "tagged_location": _bd_tagged_location(row),
        "source_payload": _bd_source_payload(row),
    }
    embedded = _bd_embedded_comments(row, influencer_key=out["influencer_key"])
    if embedded:
        out["embedded_comments"] = embedded
    return out


def bd_true_posts_count(rows):
    """The TRUE `posts_count`, read out of a Posts/Reels row's profile-context block. None if
    none of them carries one.

    ⭐⭐ DEBT D-82, AND THE FIX WAS ALREADY IN A ROW WE HAD PAID FOR. The Profiles dataset returns
    `posts_count: 0` for accounts with thousands of posts (MEASURED 2026-08-05: `inayma` 203,800
    followers and `utopian_events` 20,924, both `0`), so `_bd_posts_count` discards it — correctly,
    because a fabricated zero is a measurement failure wearing a number. The POSTS dataset carries
    the truth for those same two accounts (1,215 and 751), and it carries it in the vendor's own
    `input` echo:

        "input": {"url": …, "posts_count": 386, "followers": 6133, "following": 2850,
                  "video_play_count": null, "views": null, "country_code": null}

    ⚠ THIS IS THE SAME BLOCK THAT PRODUCED THE `views` DEFECT, read the right way round. That
    field is ACCOUNT-grain, which is exactly why mapping it onto a POST was wrong and why reading
    `posts_count` from it is right: we are asking an account-level question of an account-level
    block ([[bd-reels-views-account-level]]).
    ⛔ A ZERO IS STILL DISCARDED HERE. The Profiles dataset's zero is not trusted; the same number
    from a second endpoint is not evidence, it is the same failure arriving twice.
    """
    for row in rows or []:
        blk = (row or {}).get("input") if isinstance(row, dict) else None
        if not isinstance(blk, dict):
            continue
        v = _ig_int(_first(blk, "posts_count", "media_count", "post_count"))
        if v:
            return v
    return None










#: How many profile URLs ride in ONE `/datasets/v3/scrape` call.
#:
#: ⭐⭐ 2026-08-09 — THE WIRE WAS NEVER THE ONE-AT-A-TIME; THE CALLER WAS. `bd_scrape` has always
#: sent `[{"url": u} for u in urls]`, and its own docstring says one call carrying many URLs is
#: "what keeps the per-profile pacing floor from turning a 20-profile run into an hour" — while
#: the only profile caller passed a SINGLE-element list from inside a per-record loop with a
#: `PACE_SECONDS` floor between iterations. MEASURED on nurilab: a 25-record walk still running at
#: 67 minutes, one billed snapshot per record, one vendor email per record.
#:
#: ⚠ FIVE, NOT TWENTY-FIVE, AND THE NUMBER IS THE VENDOR'S OPINION, NOT A GUESS. `bd_scrape`'s own
#: deferral sentence asks for "a smaller batch", and a batch of ONE has already been measured
#: deferring at a vendor-reported `collection_duration` of 320 s against a 180 s budget. Five turns
#: 25 serial round trips into 5 while keeping each chunk near a size the vendor has been seen to
#: answer inline. Raise it only against a measurement, never against a hope.
PROFILE_BATCH_URLS = max(1, int(os.environ.get("AIOS_BD_PROFILE_BATCH") or 5))


#: The widest post window a single enrichment may buy. A ceiling on SPEND, not a vendor limit.
#: ⭐ MEASURED 2026-08-09 (@theresalearns): `num_of_posts: 30` on the documented discover-by-url
#: route delivered exactly 30 rows in 90 s — 23 Reels + 7 Carousels. The long-standing "12" is the
#: PROFILE route's cap (asked for 40 there, got 12), never the vendor's.
MAX_POST_WINDOW = int(os.environ.get("AIOS_IG_MAX_POST_WINDOW") or 60)


def _bd_mapped_posts(rows, raw_out=None):
    """Posts-dataset rows -> our canonical post rows, drops what will not map.

    ⚠ `raw_out` IS NOT A CONVENIENCE. `bd_true_posts_count` reads the vendor's profile-context
    echo, which lives on the RAW row and is destroyed by mapping — so a caller that only kept the
    mapped rows would lose the FREE `posts_count` read and fall through to the branch that BUYS a
    record for it. The sink keeps a window route exactly as cheap as the per-permalink one it
    replaces. Same shape as `deferred=` and `applied=` elsewhere in this module.

    ⚠ `_bd_post_metrics`, the SAME mapper `pull_profile_bd` feeds its per-permalink reads through.
    A window route with its own mapper would be a second answer to "which vendor key is `views`",
    and this repo has already paid for one of those ([[one-question-two-normalizers]]).
    """
    out = []
    for r in (rows or []):
        if isinstance(raw_out, list) and isinstance(r, dict):
            raw_out.append(r)
        got = _bd_post_metrics(r) if isinstance(r, dict) else None
        if got:
            out.append(got)
    return out


def bd_profile_posts(handle, count, deferred=None, wait=None, post_type="", raw_out=None):
    """A profile's LAST `count` posts via discover-by-url. `(rows, note)` — rows already MAPPED.

    ⭐⭐ WAVE 30 · T16 (carried W29-T10) — `post_type` IS THE WHOLE FIX AND IT IS ONE KEY.
    MEASURED 2026-08-10 on `@theresalearns`, same window, same hour, so no probability argument is
    needed: this route WITHOUT the key returned **8 Reels + 3 Carousels** for a 12-post ask; WITH
    `post_type="reel"` it returned **12/12**. Asking for "the last 12 reels" and being handed eight
    is not a filter that under-delivers, it is a filter applied to the wrong set — the shortfall
    scales with a creator's carousel share and nothing in the run says so.
    ⚠ It is HONOURED, not silently ignored (the failure mode that looks like success): the returned
    rows carry `content_type: "Reel"`. Same dataset, same route, same mapper, same price.
    ⛔ **AND IT DOES NOT REVIVE VIEWS ON BRIGHT DATA.** This measured `product_type`, never `views`;
    the Reels dataset stays type-capable and views-INCAPABLE ([[bd-reels-views-account-level]]).
    Views still come from Apify. Conflating the two questions about one dataset cost a day once.

    ⭐ ROWS ARE MAPPED HERE NOW, and that is the second half of why this had no caller. It returned
    raw `_bd_rows()` — a vendor-shaped list every caller would have had to normalise itself, which
    is a second mapper waiting to disagree with `_bd_post_metrics`. It is the boundary's job.

    ⛔ A DIFFERENT ROUTE FROM `pull_profile`, AND THAT IS THE WHOLE POINT.

    ⛔ A DIFFERENT ROUTE FROM `pull_profile`, AND THAT IS THE WHOLE POINT. The profile record
    embeds at most 12 posts however many you ask for — measured, not assumed. This is the
    documented discovery contract (`type=discover_new&discover_by=url`, `num_of_posts: N`), the
    only route that can answer "the last 30 posts", which is what makes a per-group filter
    ("last 12 reels AND last 12 images") expressible at all.

    ⚠ IT COSTS PER RECORD. 30 posts is ~2.5x the spend of 12, so no caller should reach for this
    because wider is nicer — `MAX_POST_WINDOW` bounds it and the automation's own config decides.
    """
    try:
        n = max(1, min(int(count), MAX_POST_WINDOW))
    except (TypeError, ValueError):
        n = 12
    _row = {"url": f"https://www.instagram.com/{ig_handle(handle) or handle}/",
            "num_of_posts": n}
    if str(post_type or "").strip():
        # ⚠ SENT ONLY WHEN ASKED FOR. An unconditional `post_type: "post"` would change what every
        # future caller buys, and the vendor's documented values are exactly `post` and `reel`.
        _row["post_type"] = str(post_type).strip()
    payload, note = bd_call(BD_PATH_TRIGGER,
                            {"dataset_id": BD_DS_POSTS, "type": "discover_new",
                             "discover_by": "url"},
                            body=[_row])
    if note:
        return [], note
    # ⛔ `isinstance`, NOT `(payload or {}).get(...)` — WAVE 30 · T16, and this is a crash that
    # shipped because the function had no caller. The very next line handles the case where the
    # vendor answered with ROWS INLINE instead of a snapshot id, so a list payload is an EXPECTED
    # shape here; `.get` on it raises `AttributeError` before that line can run. Whole, correct-
    # looking, unreachable code is not tested code — the first real caller found this in one run
    # ([[artifact-with-no-importer]], the failure mode rather than the waste).
    sid = _bd_deferral(payload) or (str(payload.get("snapshot_id") or "")
                                    if isinstance(payload, dict) else "")
    if not sid:
        return _bd_mapped_posts(_bd_rows(payload), raw_out), ""
    budget = BD_SCRAPE_WAIT if wait is None else float(wait)
    waited = 0.0
    while True:
        state, records, empty_note = bd_snapshot_progress(sid)
        if state in ("done", "failed") and not records:
            return [], empty_note
        if state != "running":
            got, gnote = bd_call(f"{BD_PATH_SNAPSHOT}/{sid}", {"format": "json"})
            if not gnote:
                rows = _bd_rows(got)
                if rows and not _bd_deferral(got) and not (
                        len(rows) == 1 and str(rows[0].get("status") or "") in
                        ("running", "building", "collecting")):
                    return _bd_mapped_posts(rows, raw_out), ""
        if waited >= budget:
            break
        time.sleep(BD_SCRAPE_POLL)
        waited += BD_SCRAPE_POLL
    if isinstance(deferred, list):
        deferred.append({"snapshotId": sid, "datasetId": str(BD_DS_POSTS),
                         "urls": [f"https://www.instagram.com/{handle}/"], "kind": "posts"})
    return [], (f"the post source deferred this request to {sid} and it was not ready within "
                f"{int(budget)}s — the records are collected, not lost")


def group_types(groups):
    """`{type: limit}` for a stored `postGroups` value, ignoring anything malformed.

    ⚠ ONE READER FOR THE STORED SHAPE. `select_post_groups` built this dict inline and the two
    routing helpers below need the same answer; three copies of "what did the person ask for" is
    how a filter and the read that feeds it end up disagreeing about the ask.
    """
    wanted = {}
    for g in (groups or []):
        if not isinstance(g, dict):
            continue
        t = str(g.get("type") or "").strip().lower()
        try:
            lim = int(g.get("limit"))
        except (TypeError, ValueError):
            continue
        if t and lim > 0:
            wanted[t] = max(wanted.get(t, 0), lim)
    return wanted


def reels_only_limit(groups):
    """The reel limit when the config asks for REELS AND NOTHING ELSE, else 0.

    ⛔ THE TEST IS EXCLUSIVITY, NOT PRESENCE. "12 reels AND 6 images" cannot be served by a
    single-type route — one request cannot express it — so a `video in wanted` test would send a
    mixed ask down the reels-only branch and silently drop the images. That is the specialisation
    trap: the narrow route is right only when the ask is narrow.
    """
    wanted = group_types(groups)
    return wanted.get("video", 0) if set(wanted) == {"video"} else 0


def group_window(groups):
    """How many posts ONE wide read must return for every group to be fillable.

    ⚠ THE SUM, then bounded by `MAX_POST_WINDOW`. A mixed ask is only satisfiable if the window is
    at least as large as everything asked for, and the vendor's window is what actually limits it —
    which is why the caller REPORTS a bound it could not meet rather than quietly returning fewer
    (a short window reads downstream as "this creator posts less", which is a claim about them).
    """
    return min(sum(group_types(groups).values()), MAX_POST_WINDOW)


def select_post_groups(posts, groups):
    """Keep the newest `limit` posts of each named TYPE. Returns the kept posts, newest first.

    ⭐ 2026-08-09 (owner: *"not just last 12 but by group also"*). `groups` is
    `[{"type": "video", "limit": 12}, {"type": "image", "limit": 6}]`. A type nobody named is
    DROPPED, which is the point of a filter — asking for reels and getting carousels back is the
    behaviour this replaces.

    ⚠ NEWEST-FIRST BY `posted_at`, and a post with no date sorts LAST rather than first. A blank
    date is not "just now"; ranking it as though it were would silently prefer exactly the rows we
    know least about ([[sentinel-in-a-sort-key]] — partition, never rank a sentinel).
    """
    wanted = group_types(groups)
    if not wanted:
        return list(posts or [])
    dated, undated = [], []
    for p in (posts or []):
        (dated if str((p or {}).get("posted_at") or "").strip() else undated).append(p)
    dated.sort(key=lambda p: str(p.get("posted_at") or ""), reverse=True)
    kept, seen = [], {}
    for p in dated + undated:
        t = str((p or {}).get("type") or "").strip().lower()
        if t not in wanted:
            continue
        if seen.get(t, 0) >= wanted[t]:
            continue
        seen[t] = seen.get(t, 0) + 1
        kept.append(p)
    return kept


def bd_profiles_batch(handles, deferred=None, wait=None):
    """Read MANY profiles per vendor call. Returns `({handle: node}, note)`.

    The batch FAST PATH for `pull_profile_bd`: whatever comes back inline is handed over as a
    prefetch map so the per-record rung never asks the vendor for it again. Handles this cannot
    resolve are simply absent from the map, and the per-record path runs for them exactly as it
    does today — this function can make a run faster, never wronger.

    ⭐ ATTRIBUTION COMES OFF THE ROW, NOT OFF THE REQUEST ORDER. Each Profiles row carries its own
    `account`/`username` (it is what `_bd_profile` reads first, taking its `handle` argument only
    as a default), so a 5-URL answer is mapped back by identity. Zipping the response against the
    request list would silently mis-assign every profile the moment the vendor reorders, dedupes
    or drops one — and a mis-assigned profile is worse than a missing one, because nothing about
    it looks wrong.

    ⛔ A DEFERRED CHUNK IS HANDED BACK, NOT DROPPED. `bd_scrape` appends `{snapshotId, datasetId,
    urls}` to `deferred`, and the caller fans it out into one task per DESTINATION. Discarding it
    would re-create the exact defect fixed this morning — a paid snapshot whose id nothing kept —
    and would then bill a second time when the per-record rung asked again.
    """
    want, seen = [], set()
    for h in (handles or []):
        key = str(h or "").strip().lstrip("@").lower()
        if key and key not in seen:
            seen.add(key)
            want.append(key)
    if not want:
        return {}, ""

    out, notes = {}, []
    for i in range(0, len(want), PROFILE_BATCH_URLS):
        chunk = want[i:i + PROFILE_BATCH_URLS]
        rows, note = bd_scrape(BD_DS_PROFILES,
                               [f"https://www.instagram.com/{h}/" for h in chunk],
                               wait=wait, deferred=deferred)
        if note:
            notes.append(note)
        for node in rows or []:
            if not isinstance(node, dict):
                continue
            got = str(_first(node, "account", "username", default="") or "").strip()
            got = got.lstrip("@").lower()
            # ⚠ Only accept a row we ASKED for. A vendor answering with something adjacent is a
            # bug report, not a cell value.
            if got and got in seen:
                out[got] = node
    return out, "; ".join(n for n in notes if n)


# =============================================================================================
# APIFY — THE SECOND PROVIDER (owner ruling 2026-08-08). Registered in `providers.py`; this is
# only the wire.
#
# ⭐ IT EXISTS FOR EXACTLY ONE CAPABILITY TODAY: `ig_post_views`. Bright Data answers profile,
# likes and comments correctly and cannot answer a per-reel view count on ANY route; Apify's
# `videoPlayCount` matched a browser-read ground truth to the digit (137,684 vs ~137K and 299,493
# vs 299K). So this runs as a TOP-UP over Bright Data's rows, never as a replacement for them —
# re-buying the profile and the likes from a second vendor is the expensive mistake the capability
# split exists to prevent.
# ⛔ `videoPlayCount`, NEVER `videoViewCount`. Apify ships BOTH, and the wrong one (10,678) agrees
# with Bright Data's useless `views` (10,638) for the same reel whose true count is ~137K. Two
# vendors' junk fields agreeing is what makes picking by name so dangerous here.
# =============================================================================================

APIFY_BASE = "https://api.apify.com/v2"
#: The general Instagram scraper. `directUrls` accepts /p/ and /reel/ permalinks alike.
APIFY_ACTOR_POSTS = os.environ.get("AIOS_APIFY_ACTOR") or "apify~instagram-scraper"
#: Seconds to let a synchronous actor run before giving up. Apify bills the run either way, so a
#: timeout here is money already spent — generous on purpose.
APIFY_WAIT = float(os.environ.get("AIOS_APIFY_WAIT") or 240)


#: ⭐⭐ THE ONE SENTENCE FOR "THIS HANDLE HAS NO ACCOUNT BEHIND IT", named rather than inlined so
#: that the engine can recognise it and act on it. A vendor saying *not found* is categorically
#: different from a vendor being slow, rate-limited or out of credit: the first will never
#: succeed however many times it is retried, and every retry is billed. `pull_profile` promotes
#: a rung that returns this into `res["gone"]`, which is what lets a run stop re-buying a dead
#: handle every morning ([[flag-shipped-without-its-writer]] — the flag needs a writer AND a
#: reader, so both are in this file's call graph).
ACCOUNT_GONE_NOTE = ("Instagram has no account with that handle — the source reports it does "
                     "not exist (it may have been deleted, renamed, or suspended)")


def apify_key():
    """The key, or ''. Read fresh every call, exactly as `bd_key` is."""
    return (os.environ.get("AIOS_APIFY_KEY") or "").strip()


def apify_ready():
    return bool(apify_key())


def apify_posts(urls):
    """Post/reel permalinks -> (canonical rows, note). A note means it did NOT answer.

    Returns rows already normalised through `providers.normalize_post_apify`, so nothing
    downstream ever sees an Apify-shaped key. That is the owner's schema ruling enforced at the
    only place it can be: the boundary.
    """
    from automation_engine import Refused, fetch_json            # lazy — see the module header
    want = [str(u) for u in (urls or []) if u]
    if not want:
        return [], ""
    key = apify_key()
    if not key:
        return [], "AIOS_APIFY_KEY is not configured — the Apify rung is closed"
    url = (f"{APIFY_BASE}/acts/{APIFY_ACTOR_POSTS}/run-sync-get-dataset-items"
           f"?token={requests.utils.quote(key)}&timeout={int(APIFY_WAIT)}")
    body = {"directUrls": want, "resultsType": "posts", "resultsLimit": max(1, len(want)),
            "addParentData": False}
    try:
        status, raw = fetch_json(url, body, timeout=APIFY_WAIT + 60, max_kb=BD_MAX_KB)
    except Refused as e:
        return [], f"Apify's address was refused by the URL rail: {e}"
    except Exception as e:                                          # noqa: BLE001
        # ⚠ TYPE NAME ONLY — the token is in the query string of `url`, so a str(e) that echoed
        # the request would put a live credential in the run log.
        return [], f"Apify did not answer ({type(e).__name__})"
    if status == 402:
        return [], "the Apify account is out of credit"
    if status in (401, 403):
        return [], f"Apify refused our key ({status})"
    if status >= 400:
        return [], f"Apify answered {status}"
    try:
        items = json.loads(raw.decode("utf-8", "replace"))
    except Exception:                                               # noqa: BLE001
        return [], "Apify's answer was not readable JSON"
    if not isinstance(items, list):
        return [], "Apify answered with something other than a list of items"
    rows = [providers.normalize_post_apify(x) for x in items]
    return [r for r in rows if r], ""


def apify_reels(handle, limit):
    """A profile's LAST `limit` REELS, with view counts INLINE. `(rows, note)`.

    ⭐⭐ WAVE 30 · T16 (carried W29-T11) — ONE CALL FOR WHAT CURRENTLY TAKES THREE.
    MEASURED 2026-08-10 (`@theresalearns`): `resultsType: "reels"` + `resultsLimit: 12` returned
    **12/12 `productType: clips`, every one carrying a distinct `videoPlayCount`** (2,113 →
    418,232). The shipped path for the same answer is a Bright Data profile read, a client-side
    type filter, and a SEPARATE per-permalink Apify `top_up_views` — three steps, two vendors, and
    one request's worth of data.

    ⛔ A NEW BRANCH, NEVER A RE-POINTING OF THE DEFAULT CHAIN. `ig_post_metrics` does not move and
    Bright Data stays first for it; this is the specialisation for a single-type ask, and it is
    reached only when the config's groups name reels and nothing else.
    ⚠ ITS CALLER MUST NOT THEN RUN `top_up_views`. The views are already here; asking Apify again
    per permalink would buy the same numbers twice, which is the shape a "top-up" is meant to
    avoid rather than cause.
    ⛔ THE ACTOR PAGE CONTRADICTS ITS OWN INPUT SCHEMA — the page says `resultsType` has no
    `"reels"` value. It does; the response is the evidence ([[measure-the-real-call]]).
    """
    from automation_engine import Refused, fetch_json            # lazy — see the module header
    h = ig_handle(handle) or str(handle or "").strip().lstrip("@")
    if not h:
        return [], "no handle to read reels for"
    key = apify_key()
    if not key:
        return [], "AIOS_APIFY_KEY is not configured — the Apify rung is closed"
    try:
        n = max(1, min(int(limit), MAX_POST_WINDOW))
    except (TypeError, ValueError):
        n = 12
    url = (f"{APIFY_BASE}/acts/{APIFY_ACTOR_POSTS}/run-sync-get-dataset-items"
           f"?token={requests.utils.quote(key)}&timeout={int(APIFY_WAIT)}")
    body = {"directUrls": [f"https://www.instagram.com/{h}/"], "resultsType": "reels",
            "resultsLimit": n, "addParentData": False}
    try:
        status, raw = fetch_json(url, body, timeout=APIFY_WAIT + 60, max_kb=BD_MAX_KB)
    except Refused as e:
        return [], f"Apify's address was refused by the URL rail: {e}"
    except Exception as e:                                          # noqa: BLE001
        # ⚠ TYPE NAME ONLY — the token is in the query string of `url`.
        return [], f"Apify did not answer ({type(e).__name__})"
    if status == 402:
        return [], "the Apify account is out of credit"
    if status in (401, 403):
        return [], f"Apify refused our key ({status})"
    if status >= 400:
        return [], f"Apify answered {status}"
    try:
        items = json.loads(raw.decode("utf-8", "replace"))
    except Exception:                                               # noqa: BLE001
        return [], "Apify's answer was not readable JSON"
    if not isinstance(items, list):
        return [], "Apify answered with something other than a list of items"
    rows = [r for r in (providers.normalize_post_apify(x) for x in items) if r]
    if not rows:
        return [], "Apify answered with no reels for this profile"
    return rows, ""


def apify_profile(handle):
    """One handle -> (canonical profile | None, note). A note means it did NOT answer.

    ⭐ THE SECOND PROVIDER FOR `ig_profile`, and the rung that was MISSING (owner report
    2026-08-09). `providers.py` has declared Apify capable of `ig_profile` and the chain
    `("brightdata", "apify")` since 2026-08-08, but no runner existed and nothing walked that
    chain — so a profile Bright Data could not scrape fell straight past a configured provider
    to the anonymous rungs, and reported `blocked` when those failed too.

    ⚠ `resultsType: "details"`, NOT `"posts"` — the same actor, a different question. Asking for
    posts and reading the parent profile off `addParentData` would buy a page of media we did not
    want on every profile read, which is the expensive mistake the capability split exists to
    prevent (see the Apify header note above).
    """
    from automation_engine import Refused, fetch_json            # lazy — see the module header
    h = str(handle or "").strip().lstrip("@")
    if not h:
        return None, "no handle"
    key = apify_key()
    if not key:
        return None, "AIOS_APIFY_KEY is not configured — the Apify rung is closed"
    url = (f"{APIFY_BASE}/acts/{APIFY_ACTOR_POSTS}/run-sync-get-dataset-items"
           f"?token={requests.utils.quote(key)}&timeout={int(APIFY_WAIT)}")
    body = {"directUrls": [f"https://www.instagram.com/{h}/"],
            "resultsType": "details", "resultsLimit": 1, "addParentData": False}
    try:
        status, raw = fetch_json(url, body, timeout=APIFY_WAIT + 60, max_kb=BD_MAX_KB)
    except Refused as e:
        return None, f"Apify's address was refused by the URL rail: {e}"
    except Exception as e:                                          # noqa: BLE001
        # ⚠ TYPE NAME ONLY — the token rides in the query string of `url`.
        return None, f"Apify did not answer ({type(e).__name__})"
    if status == 402:
        return None, "the Apify account is out of credit"
    if status in (401, 403):
        return None, f"Apify refused our key ({status})"
    if status >= 400:
        return None, f"Apify answered {status}"
    try:
        items = json.loads(raw.decode("utf-8", "replace"))
    except Exception:                                               # noqa: BLE001
        return None, "Apify's answer was not readable JSON"
    if not isinstance(items, list) or not items:
        return None, "Apify answered with no profile for that handle"
    # ⭐⭐ 2026-08-09 — THE VENDOR'S ERROR ENVELOPE IS AN ANSWER, AND IT WAS BEING THROWN AWAY.
    # MEASURED on nurilab's one pending handle: Apify returned
    # `{"url": …, "username": "roxyfoxypinky", "error": "not_found",
    #   "errorDescription": "Post does not exist"}` — a 200, a well-formed item, and the single
    # most useful sentence any provider produced about that account. `normalize_profile_apify`
    # mapped it into a "profile" carrying username + url, this function returned `(prof, "")`
    # meaning SUCCESS, and the caller — finding no follower count on it — recorded the useless
    # `apify: answered without follower counts`. Three runs asked, three runs were told, and the
    # owner still could not find out why the row was blank.
    # ⚠ `error` RIDES EVERY ITEM AS `null` on the success path (measured on `sriyynntt`), so the
    # test must be truthiness, never key presence.
    # ⚠ AND THE VENDOR'S OWN WORDING IS NOT REPEATED VERBATIM: it says "Post does not exist" for
    # a PROFILE url, which describes the wrong kind of object to anyone reading a profile row.
    err = str((items[0] or {}).get("error") or "").strip()
    if err:
        desc = str((items[0] or {}).get("errorDescription") or "").strip()
        if err in ("not_found", "not-found", "notfound"):
            return None, ACCOUNT_GONE_NOTE
        return None, f"the source could not read that account ({desc or err})"
    prof = providers.normalize_profile_apify(items[0])
    if not prof:
        return None, "Apify's profile row was not readable"
    return prof, ""


#: ⭐⭐ 2026-08-07 — HOW LONG THE CORPUS RUNG MAY WAIT. A corpus query is a scan over 620M rows
#: and takes MINUTES, not seconds (§2d MEASURED ~4 min for a narrow one). It is a FALLBACK behind
#: a scrape that already failed, so the budget is generous — but bounded, because an enrich step
#: that never returns is an automation that never finishes.
BD_CORPUS_WAIT = float(os.environ.get("AIOS_BD_CORPUS_WAIT") or 420)
BD_CORPUS_POLL = float(os.environ.get("AIOS_BD_CORPUS_POLL") or 15)


def bd_corpus_profile(handle, log=print):
    """One profile out of the PRE-COLLECTED corpus. `(node|None, note)`.

    ⭐⭐ WHY THIS RUNG EXISTS, and it is measured rather than defensive. On 2026-08-07 the SCRAPE
    path answered `Parse error: Invalid URL` for `inayma` on **four consecutive attempts and
    through both input shapes** (a URL with and without its trailing slash, and the `user_name`
    discovery input, which the vendor converts to the same URL internally) — while a control
    profile in the SAME batch came back with 36 keys. So the URL form was not the problem and
    neither was the account: the anonymous ladder read it fine. The vendor simply could not
    scrape that profile.
    ⇒ **The same dataset answered instantly by the OTHER route.** `POST /datasets/filter` with
    `account = inayma` returned the full 33-key row — exact followers (203,818 against the
    anonymous rung's rounded 204,000), biography, `is_verified`, `avg_engagement`, `fbid`,
    `highlights_count`. One dataset, two routes, and only one of them was broken.

    ⚠ THE ROWS ARE PRE-COLLECTED, so this is not a live read — it is the vendor's own last
    capture, and its population differs from the scrape path's (§2d: `avg_engagement` 88%
    populated on the corpus vs 0% on scrape). That is why it is a FALLBACK and not the default:
    fresher-but-thinner beats staler-but-fuller when both work.
    ⛔ THE SNAPSHOT NAMESPACES ARE NOT INTERCHANGEABLE. A corpus query mints `snap_…` and is read
    at `/datasets/snapshot/{id}`; the scraper mints `sd_…` and is read at
    `/datasets/v3/snapshot/{id}`. Crossing them returns a flat 404 about a snapshot that is alive
    and building — the trap §2c records, and the reason these paths are spelled out here.
    """
    h = str(handle or "").strip().lstrip("@").lower()
    if not h:
        return None, "no handle to look up"
    payload, note = bd_call(BD_PATH_FILTER, None,
                            {"dataset_id": BD_DS_PROFILES, "records_limit": 1,
                             "filter": {"name": "account", "operator": "=", "value": h}})
    if note:
        return None, note
    snap = str((payload or {}).get("snapshot_id") or "").strip()
    if not snap:
        return None, "the profile corpus accepted the query but named no snapshot"
    deadline = time.time() + BD_CORPUS_WAIT
    while time.time() < deadline:
        time.sleep(BD_CORPUS_POLL)
        got, perr = bd_call(f"{BD_PATH_FILTER_SNAPSHOT}/{snap}", None, None)
        status = str((got or {}).get("status") or "").strip()
        if perr or status in ("failed", "error"):
            return None, (perr or f"the corpus query failed ({(got or {}).get('error') or status})")
        if status == "ready":
            # ⚠ READY IS NOT DELIVERABLE. §2d MEASURED `/download` answering "Snapshot is
            # building. Try again in a few minutes" while status already said `ready` — delivery
            # lags readiness. So the download is polled too, inside the same budget.
            while time.time() < deadline:
                rows, derr = bd_call(f"{BD_PATH_FILTER_SNAPSHOT}/{snap}/download",
                                     {"format": "json"}, None)
                if derr:
                    time.sleep(BD_CORPUS_POLL)
                    continue
                if isinstance(rows, dict):
                    rows = [rows]
                if isinstance(rows, list) and rows:
                    return rows[0], ""
                time.sleep(BD_CORPUS_POLL)
            break
    return None, (f"the profile corpus did not deliver {h!r} within "
                  f"{int(BD_CORPUS_WAIT)}s (snapshot {snap} is still building — it is not lost)")


def _tag_metric_deferrals(items, start, kind, influencer_key):
    """Attach only the context a later snapshot collector needs to preserve the IG graph."""
    from automation_engine import _iso                           # lazy — see the module header
    for item in items[start:]:
        if isinstance(item, dict):
            item["kind"] = kind
            item["influencer"] = str(influencer_key or "").strip().lstrip("@").lower()
            item["requestedAt"] = _iso()


def top_up_views(posts, log=print):
    """Fill `views` on video posts from the `ig_post_views` capability. MUTATES `posts`; returns
    a note for the run (`""` when there was nothing to do).

    ⭐⭐ CAPABILITY ROUTING, NOT PROVIDER FAILOVER (owner ruling 2026-08-08, `providers.py`).
    Bright Data answers profile + likes + comments and is declared INCAPABLE of `ig_post_views`,
    so the chain for that ONE capability resolves to Apify and only the video permalinks are
    re-bought. Re-running the whole profile on the second vendor would pay twice for the 90% that
    already worked, which at thousands of tenants x twelve posts is the entire bill.

    ⛔⛔ WHY THIS IS A FUNCTION AND NOT INLINE, which is the whole point of the 2026-08-09 fix.
    It used to live inside `pull_profile_bd`'s post-metrics block, so it ran ONLY when the Posts
    scrape answered within the wait budget. When that batch DEFERRED — routine, and the normal
    outcome on a busy account — the posts came back through
    `automation_engine.collect_pending_metric_snapshots` instead, which never called it. MEASURED
    on nurilab's `theresalearns`: profile, 12 posts and 154 comments all landed, and **every
    Views cell was blank**, because the deferred path never asked Apify at all. One capability,
    two collection paths, one of them wired: the same shape as the `ig_profile` rung that was
    declared and never called ([[flag-shipped-without-its-writer]]).
    ⇒ ONE implementation, TWO callers. A copy in the collector would have been a second thing to
    keep in step, and the two would drift on the next ruling.

    ⚠ VIDEO ONLY. A carousel or image has no view count, so sending it would buy a record that
    can only come back blank.
    """
    video_urls = [p["url"] for p in (posts or [])
                  if p.get("type") == "video" and p.get("url")]
    if not video_urls:
        return ""

    def _views_work(prov, _urls=tuple(video_urls)):
        if prov.key == "apify":
            return apify_posts(list(_urls))
        # A provider in the chain with no runner here is a configuration error, not a vendor
        # outage — say so rather than returning an empty list that reads as "the vendor had
        # nothing".
        return None, f"no {prov.key} runner is wired for ig_post_views"

    got_views, attempts = providers.run(
        "ig_post_views", _views_work,
        # ⛔ THE SATISFIED PREDICATE IS THE FALLBACK TRIGGER. Rows that come back with every
        # `views` blank are a FAILURE for this capability even at HTTP 200 — which is exactly how
        # Bright Data behaves, and why a chain that only caught exceptions would never have
        # reached a second provider.
        satisfied=lambda rows: bool(rows) and any(r.get("views") for r in rows),
        log=log)
    for row in (got_views or []):
        target = next((p for p in posts
                       if p.get("shortcode") == row.get("shortcode")), None)
        # ⛔ TAKE ONLY THE CAPABILITY THAT WAS ASKED FOR. This provider also returns
        # likes/comments/caption, and letting them land would silently switch the source of
        # columns Bright Data already answered — the schema would be stable but the PROVENANCE
        # would flip halfway through a row.
        if target and row.get("views") not in (None, ""):
            target["views"] = row["views"]
        # ⭐⭐ 2026-08-09 — THE FIELDS ONLY THIS PROVIDER ANSWERS, taken from the record that has
        # ALREADY been bought for the view count. `plays` had a column and a type and no writer at
        # all: empty on every one of 816 rows because nothing ever assigned it.
        # ⚠ THE "TAKE ONLY WHAT WAS ASKED FOR" RULE ABOVE STILL HOLDS FOR CONTESTED COLUMNS —
        # likes/comments/caption are deliberately NOT copied here, because the primary source
        # answers those and flipping their provenance mid-row is the drift that rule prevents.
        # These three are different: the primary returns them for nobody, so there is no
        # provenance to flip.
        for _extra in ("plays", "video_duration", "comments_disabled"):
            if target and row.get(_extra) not in (None, ""):
                target[_extra] = row[_extra]
        # ⭐ LOCATION RIDES ALONG FOR FREE, AND ONLY INTO A GAP (2026-08-08).
        # MEASURED: Bright Data's `location_details` is RICHER when present — it carries real
        # coordinates (`lat -6.9246, lng 106.9292, name "Sukabumi"`) which Apify does not return
        # at all — but it is populated on only 44 of 227 posts, and on `DblAkEbv0ry` it returned
        # nothing while Apify returned "Jakarta, Indonesia". So Bright Data stays the source and
        # this fills only what it left BLANK.
        # ⛔ THE COST ARGUMENT IS WHY IT IS HERE AND NOT ITS OWN CAPABILITY: this Apify record has
        # ALREADY been bought for the view count and carries `locationName` in the same payload.
        # Routing location as a separate capability would buy a second record for a field that is
        # already sitting in this response.
        if target and not str(target.get("tagged_location") or "").strip() \
                and str(row.get("tagged_location") or "").strip():
            target["tagged_location"] = row["tagged_location"]
    used = next((a.provider for a in attempts if a.ok), "")
    if used:
        # ⛔ NO VENDOR NAME IN AN OPERATOR-FACING SENTENCE (owner instruction 2026-08-09). Which
        # company answered is a routing detail the product owns; the person reading a run log
        # wants to know the view counts arrived. The provider key is still on `attempts` and in
        # the server log for debugging — it just stops being product copy. See `public_source`.
        return "view counts collected"
    if attempts:
        # The REASONS still ride, because a person needs to know why a column is blank — only the
        # vendor's NAME is dropped, never its explanation.
        reasons = "; ".join(a.note for a in attempts if a.note)
        return f"view counts could not be collected{': ' + reasons if reasons else ''}"
    return ""


#: ⛔ THE ONE PLACE A VENDOR KEY BECOMES OPERATOR-FACING TEXT (owner instruction 2026-08-09:
#: *"Remove any mention of APIfy or Bright Data from the user interface, places like Logs etc."*).
#:
#: The keys stay REAL everywhere they do work — `providers.py` chains, `AIOS_PROVIDER_ORDER`, the
#: env var names, `res["via"]` in the call graph, and the server log. What changes is only what a
#: person reads: which company we bought a record from is a routing decision the product owns, and
#: naming it in a run log invites "why did this one use the other one" about a choice nobody made.
#: ⚠ ROUTING, NOT QUALITY. "Primary"/"Backup" says which rung answered — the thing the column
#: labelled *Read via* exists to record — without the brand. An unknown key falls through to a
#: generic word rather than leaking itself.
PUBLIC_SOURCE = {
    "brightdata": "Primary",
    "brightdata:deferred": "Primary (deferred)",
    # ⭐ 2026-08-10 — DISCOVERY IS A RUNG, and until now it was the only one that read numbers and
    # recorded no observation. A discovery run reads followers/following/engagement off the
    # vendor's PRE-COLLECTED corpus and writes them onto the profile row; the fall-through would
    # have spelled this "Primary (discovery)", which is accurate about the vendor and useless
    # about the thing an operator needs to know — that this number was read off a corpus at an
    # unknown collection time rather than measured now. One word, its own row in this table, so
    # the "Read via" column keeps ONE vocabulary and `public_source` stays the only evaluator.
    "brightdata:discovery": "Discovery",
    "apify": "Backup",
}


def public_source(via):
    """A vendor key -> the word an operator reads. Never returns the key itself."""
    key = str(via or "").strip().lower()
    if not key:
        return ""
    if key in PUBLIC_SOURCE:
        return PUBLIC_SOURCE[key]
    # A compound we have not mapped ("x:deferred") still must not leak the vendor half.
    base = key.split(":", 1)[0]
    if base in PUBLIC_SOURCE:
        suffix = key.split(":", 1)[1] if ":" in key else ""
        return f"{PUBLIC_SOURCE[base]} ({suffix})" if suffix else PUBLIC_SOURCE[base]
    return "Automated"


#: Marker a caller puts in the prefetch map for a handle whose BATCH chunk was deferred.
#: ⛔ ITS JOB IS TO STOP A SECOND PURCHASE. The snapshot is paid for and already queued for
#: collection; a rung that saw "not in the prefetch" and scraped again would bill the same handle
#: twice per run — the batch making things cheaper AND more expensive at once.
DEFERRED_MARK = "__aios_deferred_snapshot__"


def pull_profile_bd(url, max_posts=None, post_metrics=False,
                    comment_metrics=False, log=print, pending_metrics=None,
                    pending_profile=None, prefetch=None, post_groups=None):
    """The paid rung: one profile, EXACT counts, plus the top posts the anonymous surface hides.

    Same return contract as `pull_profile` (`{state, profile, posts, via, note}`) so the runner
    treats every rung identically. `blocked` here means the VENDOR rung did not answer — the
    caller decides whether to drop to the anonymous ladder.

    `post_metrics=True` adds the paid Post/Reel calls. `comment_metrics=True` independently adds
    the separate paid Comments-dataset call. Both default off; the latter is never inferred from
    an enabled post-metrics switch. Embedded comment previews returned by either paid record are
    retained without that extra Comments request.

    ⚠ `max_posts=None` MEANS THE DEFAULT, and the `None` is a consequence of the module split
    rather than a new option. A default argument is evaluated when this `def` runs — i.e. at
    import — so naming `DEFAULT_POSTS_PER_PULL` in the signature would be exactly the
    module-level reach into `automation_engine` the header forbids. Every caller in the engine
    already writes `cfg.get("maxPosts") or DEFAULT_POSTS_PER_PULL`, so the value is unchanged.
    """
    from automation_engine import (DEFAULT_POSTS_PER_PULL, PACE_SECONDS,  # lazy — module header
                                   _s)
    max_posts = DEFAULT_POSTS_PER_PULL if max_posts is None else max_posts
    handle = ig_handle(url)
    if not handle:
        return {"state": "error", "note": f"{url!r} is not an Instagram profile URL",
                "profile": {}, "posts": [], "comments": [], "via": ""}
    # ⭐⭐ 2026-08-09 — THE PROFILE SCRAPE NOW HANDS ITS DEFERRAL OVER, and until today it was the
    # ONLY `bd_scrape` caller that did not. Every post, reel and comment call below passes
    # `deferred=pending_metrics`; this one passed nothing, so a profile the vendor took longer
    # than `BD_SCRAPE_WAIT` to collect was BILLED and its snapshot id discarded — on every run,
    # forever. MEASURED on nurilab: two runs, two fresh `sd_…` snapshots, both abandoned, the
    # second at a vendor-reported `collection_duration` of 320 s against our 180 s budget.
    # That is [[artifact-with-no-importer]] in a function signature: the parameter existed, the
    # machinery to collect it existed, and nothing handed the list in.
    _deferred = []
    # ⭐ 2026-08-09 — THE BATCH FAST PATH. `prefetch` is `{handle: node}` filled by ONE
    # `bd_profiles_batch` call for the whole selection. A hit here is a vendor round trip that
    # does not happen; a miss falls through to the single-URL scrape below, unchanged.
    cached = prefetch.get(handle) if isinstance(prefetch, dict) else None
    if isinstance(cached, dict) and cached.get(DEFERRED_MARK):
        # The batch chunk holding this handle was deferred and its snapshot is ALREADY queued
        # against this record. Report exactly what a deferred single scrape reports, minus the
        # second purchase — the shape is identical so nothing downstream learns a new state.
        sid = str(cached.get(DEFERRED_MARK))
        return {"state": "blocked", "profile": {}, "posts": [], "comments": [],
                "via": "brightdata", "deferredProfile": [sid],
                "note": (f"the profile source deferred this batch to {sid} — the record is "
                         f"collected, not lost; the next run picks it up")}
    if isinstance(cached, dict) and cached:
        rows, note = [cached], ""
    else:
        rows, note = bd_scrape(BD_DS_PROFILES, [f"https://www.instagram.com/{handle}/"],
                               deferred=_deferred)
    node = rows[0] if rows else {}
    profile = _bd_profile(node, handle) if node else {}
    unreadable = profile.get("followers") is None and profile.get("following") is None
    via = "brightdata"
    if note or unreadable:
        # ⭐⭐ 2026-08-07 — THE CORPUS IS TRIED BEFORE THIS RUNG GIVES UP, and it is the difference
        # between a blank row and a full one. MEASURED: the scraper answered `Parse error: Invalid
        # URL` for `inayma` four times running and through both input shapes, while a control
        # profile in the same batch returned 36 keys — and `POST /datasets/filter` on the SAME
        # dataset returned that profile's complete 33-key row. A rung that reports `blocked`
        # while the vendor is holding the answer on another route is a rung that gave up early.
        # ⚠ The FIRST failure is kept in the note either way: if the corpus also misses, the
        # person needs to know the scrape was tried and what it said, not only the fallback.
        # ⚠ THE UNREADABLE-ROW DIAGNOSTIC IS CARRIED, NOT DROPPED. "It answered 200 with something
        # we could not parse" and "it refused" are different problems with different fixes, and
        # the field-names hint is the one that tells the next reader to go and diff a real row
        # against the schema. Losing it behind the corpus fallback would make a mapping
        # regression look like a vendor outage.
        scrape_note = note or ("the scrape answered, but no follower/following counts were "
                               "readable in it (the field names may have moved — see "
                               "instagram-capture.md)")
        cnode, cnote = bd_corpus_profile(handle, log=log)
        if cnode:
            node = cnode
            profile = _bd_profile(node, handle)
            via = "brightdata:corpus"
            # ⚠ The corpus is PRE-COLLECTED, so the row is the vendor's last capture rather than
            # a read taken now. Said out loud on the attempt, because "when was this true" is the
            # one question a stored measurement must always be able to answer.
            note = (f"the live scrape did not answer ({_s(scrape_note, 120)}), so this came from "
                    f"the vendor's pre-collected profile corpus — the values are its last "
                    f"capture, not a read taken just now")
        else:
            # ⭐ THE DEFERRAL IS REGISTERED ONLY WHEN THE PAID RUNG GIVES UP. If the corpus
            # answered we already have the profile, and queueing a second write of the same
            # identity would add churn for a row we did not need. Here we have nothing — and the
            # snapshot is both paid for and the freshest answer that will ever exist for this
            # handle, so it is the one thing worth carrying forward.
            if isinstance(pending_profile, list):
                for d in _deferred:
                    pending_profile.append({**d, "kind": "profile", "influencer": handle})
            return {"state": "blocked", "profile": {}, "posts": [], "comments": [], "via": "brightdata",
                    "deferredProfile": [d.get("snapshotId") for d in _deferred],
                    "note": f"{scrape_note}; the profile corpus did not answer either ({cnote})"}
    if profile.get("followers") is None and profile.get("following") is None:
        # It answered 200 with something we could not read. Say THAT — not "0 followers".
        # ⚠ `posts_count` is deliberately NOT part of this test: it arrives as a fabricated 0 and
        # is discarded, so a row where it is the only "readable" field is a row we cannot read.
        return {"state": "blocked", "profile": {}, "posts": [], "comments": [], "via": "brightdata",
                "note": "the profile source answered, but no follower/following counts were readable in "
                        "it (the field names may have moved — see instagram-capture.md)"}
    posts = []
    for p in (node.get("posts") or [])[:max_posts]:
        got = _bd_post_identity(p)
        if got:
            posts.append(got)

    # ⭐⭐ WAVE 30 · T16 — THE WINDOW COMES BEFORE THE PRICE, WHICH IS WHY THIS SITS HERE AND NOT
    # IN THE ENGINE. `select_post_groups` filters what we already bought, and what we already
    # bought is the PROFILE's embed — at most 12 posts, MIXED. So "the last 12 reels" has been
    # landing eight on a creator who posts carousels, forever, with a green run and a summary that
    # says twelve. The engine's own comment named this and deferred it: *"widening belongs ahead of
    # that step, in the paid path, and that is its own change"*. This is that change.
    #
    # ⛔ GATED ON `post_metrics`, DELIBERATELY. Both routes below BUY records. `post_metrics` is
    # the switch that already means "I am paying per post"; widening without it would spend money
    # on a run whose config says not to. ⚠ And when a group config cannot be honoured for that
    # reason the run SAYS SO rather than filtering a window it knows is too narrow — a short answer
    # that reads as a fact about the creator is the failure this note exists to prevent.
    # ⚠ `rows2` IS INITIALISED HERE, not inside the metric block, and that is a defect this
    # ticket introduced and caught in one run: the block that defines it is now SKIPPED when a
    # window was bought, so `bd_true_posts_count(rows2 ...)` below raised `UnboundLocalError` on
    # the new path. A name defined in a branch, read outside it, is a crash waiting for whoever
    # adds the next branch.
    rows2 = []
    views_inline, window_bought, window_note = False, False, ""
    if post_groups and not post_metrics:
        window_note = ("post groups were configured but post data is switched off, so they filtered "
                       "the profile's own preview (at most 12 mixed posts) rather than a window "
                       "bought for them")
    elif post_groups:
        _reels_n = reels_only_limit(post_groups)
        if _reels_n and apify_ready():
            # ⭐ ONE CALL: last N reels WITH `videoPlayCount` inline. `views_inline` is what stops
            # `top_up_views` buying the same numbers a second time, per permalink.
            wide, wnote = apify_reels(handle, _reels_n)
            if wide:
                posts, views_inline, window_bought = wide, True, True
            else:
                window_note = f"the reels route did not answer ({_s(wnote, 90)})"
        else:
            # ⚠ THE FALLBACK IS NOT A DEGRADED COPY — for a reels-only ask it sends Bright Data's
            # own `post_type: "reel"`, MEASURED at 12/12 where the unfiltered route gives 8. For a
            # MIXED ask no single-type route exists, so it buys the SUM and lets
            # `select_post_groups` keep N of each — which is the one path that clause still owns.
            wide, wnote = bd_profile_posts(handle, group_window(post_groups),
                                           post_type=("reel" if _reels_n else ""),
                                           deferred=pending_metrics, raw_out=rows2)
            if wide:
                posts, window_bought = wide, True
            else:
                window_note = f"the post window was not readable ({_s(wnote, 90)})"

    # ⛔ `not window_bought` IS THE DOUBLE-SPEND GUARD, and it is the whole reason the routes
    # above return MAPPED rows. Both of them buy records that already carry likes, comments,
    # captions and (on the Apify branch) views; re-reading the same permalinks below would pay
    # twice for numbers already in hand — the exact thing a "top-up" exists to avoid causing.
    if post_metrics and posts and not window_bought:
        time.sleep(min(PACE_SECONDS, 1.0))    # the vendor is paid, but it is still someone's API
        # Posts and Reels are different Bright Data endpoints. Sending every permalink to the
        # Posts endpoint was why Inayma's five video rows got likes/comments but no views: the
        # documented Reels response is where `views` and `video_play_count` live. Split the batch
        # without duplicating any record, so enabling post metrics still buys one row per post.
        ordinary = [p for p in posts if p.get("type") != "video"]
        videos = [p for p in posts if p.get("type") == "video"]
        metric_notes, ordinary_rows = [], []
        # `None` keeps direct callers backwards-compatible: they wait for their answer.  An
        # automation passes one run-level list, which makes a deferred metric batch durable and
        # lets this profile finish rather than parking the whole enrichment on it.
        metric_wait = BD_METRIC_SCRAPE_WAIT if isinstance(pending_metrics, list) else None
        if ordinary:
            at = len(pending_metrics) if isinstance(pending_metrics, list) else 0
            got_rows, got_note = bd_scrape(BD_DS_POSTS, [p["url"] for p in ordinary],
                                           wait=metric_wait, deferred=pending_metrics)
            if isinstance(pending_metrics, list):
                _tag_metric_deferrals(pending_metrics, at, "posts", profile.get("username") or handle)
            rows2.extend(got_rows)
            ordinary_rows.extend(got_rows)
            if got_note:
                metric_notes.append(f"posts: {got_note}")
        if videos:
            at = len(pending_metrics) if isinstance(pending_metrics, list) else 0
            got_rows, got_note = bd_scrape(BD_DS_REELS, [p["url"] for p in videos],
                                           wait=metric_wait, deferred=pending_metrics)
            if isinstance(pending_metrics, list):
                _tag_metric_deferrals(pending_metrics, at, "posts", profile.get("username") or handle)
            rows2.extend(got_rows)
            if got_note:
                metric_notes.append(f"reels: {got_note}")
        # Bright Data's Profiles feed has used more than one media spelling. If one of those
        # unrecognised labels sent a Reel to Posts, the Posts response itself identifies the
        # mistake (`content_type: Reel`) but cannot supply Views. Correct it immediately with a
        # Reels read, then let the richer response win below. This costs a replacement record only
        # for a proven misroute; correctly classified images/carousels are never re-read.
        repair_reels = []
        for row in ordinary_rows:
            mapped = _bd_post_metrics(row)
            if mapped and mapped.get("type") == "video":
                repair_reels.append(f"https://www.instagram.com/reel/{mapped['shortcode']}/")
        if repair_reels:
            at = len(pending_metrics) if isinstance(pending_metrics, list) else 0
            got_rows, got_note = bd_scrape(BD_DS_REELS, list(dict.fromkeys(repair_reels)),
                                           wait=metric_wait, deferred=pending_metrics)
            if isinstance(pending_metrics, list):
                _tag_metric_deferrals(pending_metrics, at, "posts", profile.get("username") or handle)
            rows2.extend(got_rows)
            if got_note:
                metric_notes.append(f"reel recovery: {got_note}")
        if metric_notes and not rows2:
            # The engagement half refusing does NOT lose the identity half — the profile and the
            # post rows still land, and the run says which part was not readable.
            note = f"post metrics unavailable: {'; '.join(metric_notes)}"
        else:
            by_code = {}
            for r in rows2:
                got = _bd_post_metrics(r)
                if got:
                    by_code[got["shortcode"]] = got
            for p in posts:
                extra = by_code.get(p["shortcode"])
                if extra:
                    # The metrics row is RICHER (it knows posted_at); merge it over the identity
                    # row, dropping keys it could not read so a blank never overwrites a value.
                    p.update({k: v for k, v in extra.items() if v not in (None, "")})
            if not by_code:
                note = "post metrics were requested but the Posts dataset returned no rows"
            elif metric_notes:
                note = f"some post metrics were unavailable: {'; '.join(metric_notes)}"

    # ⭐⭐ T16 — THE ONE PATH THAT LEGITIMATELY SKIPS THE VIEWS TOP-UP, and it is skipped because
    # the numbers are already here rather than because nobody wants them. `apify_reels` returns
    # `videoPlayCount` INLINE on every row; `top_up_views` would then buy the same figure again,
    # per permalink, from the same vendor. Every other route still tops up — Bright Data's Reels
    # dataset is MEASURED views-incapable, so a BD window is exactly as short of views as the
    # profile embed was. ⚠ Lifted OUT of the per-permalink block on purpose: leaving it there
    # would have silently dropped the top-up for a Bright Data WINDOW too, which is the quiet
    # half-fix this comment exists to refuse.
    if post_metrics and posts and not views_inline:
        note = "; ".join(x for x in (note, top_up_views(posts, log=log)) if x)
    if window_note:
        note = "; ".join(x for x in (note, window_note) if x)

    # ⭐⭐ DEBT D-82 (owner ruling R12: *"price, then wire the Posts-dataset call so posts_count
    # fills with the true value"*). The Profiles dataset's `posts_count: 0` is discarded as the
    # measurement failure it is, so "Posts" was blank on the PAID rung while the free anonymous
    # rung filled it — the opposite of what somebody choosing "Paid provider" expects.
    #
    # ⛔ THE CHEAP HALF FIRST, AND IT IS FREE. When post metrics were bought above, the vendor's
    # own profile-context echo is already sitting in those rows; reading it costs nothing.
    if profile.get("posts_count") is None and post_metrics:
        true_count = bd_true_posts_count(rows2 if post_metrics else [])
        if true_count:
            profile["posts_count"] = true_count
            log("  posts_count read from the Posts rows already bought - no extra spend")
    # ⛔ AND WHEN NOTHING WAS BOUGHT, IT COSTS ONE RECORD AND SAYS SO BEFORE SPENDING IT. One post
    # URL is the smallest unit that carries the account block, so this is the floor rather than a
    # batch. `providers.estimate` is asked for the number rather than `BD_RECORD_PRICE_SPEC`
    # multiplied by hand — the router knows which vendor would serve it and at what list rate, and
    # a second copy of that arithmetic is how two prices end up disagreeing.
    elif profile.get("posts_count") is None and posts:
        quote = providers.estimate("ig_post_metrics", 1)
        log(f"  buying 1 {quote.get('provider') or 'provider'} post record for the true "
            f"posts_count (~${quote.get('usd', 0)}; the Profiles dataset's 0 is false)")
        extra_rows, extra_note = bd_scrape(BD_DS_POSTS, [posts[0]["url"]],
                                           wait=BD_METRIC_SCRAPE_WAIT
                                           if isinstance(pending_metrics, list) else None,
                                           deferred=pending_metrics)
        true_count = bd_true_posts_count(extra_rows)
        if true_count:
            profile["posts_count"] = true_count
            # ⛔ THE SPEND IS NAMED WHERE A PERSON SEES IT, not only in `log()`. This branch fires
            # on EVERY paid enrich whose post metrics are off — which is the default — because the
            # Profiles dataset's `posts_count` is `0` on every row we have measured. One record
            # per profile is small; at R1's 1k-profiles-daily it is ~1k records a day, and a cost
            # that only appears in a server console is a cost nobody is deciding about.
            note = "; ".join(x for x in (
                note, "bought 1 extra post record for the true post count") if x)
        elif extra_note:
            # ⚠ NOT AN ERROR. The identity half is already read and written; a post count we could
            # not buy leaves the column blank, which is what it was before this branch existed.
            note = "; ".join(x for x in (note, f"post count unavailable: {extra_note}") if x)

    comments = []
    if comment_metrics and posts:
        # This is the cost boundary: do not make the full comment request unless the stored
        # enrichment action explicitly opted in. A batched call still bills provider records, but
        # it prevents unnecessary HTTP fan-out and preserves every returned field per comment.
        time.sleep(min(PACE_SECONDS, 1.0))
        comment_wait = BD_METRIC_SCRAPE_WAIT if isinstance(pending_metrics, list) else None
        at = len(pending_metrics) if isinstance(pending_metrics, list) else 0
        comment_source, comment_note = bd_scrape(BD_DS_COMMENTS, [p["url"] for p in posts],
                                                  wait=comment_wait, deferred=pending_metrics)
        if isinstance(pending_metrics, list):
            _tag_metric_deferrals(pending_metrics, at, "comments", profile.get("username") or handle)
        for row in comment_source:
            got = _bd_comment(row, influencer_key=profile.get("username") or handle)
            if got:
                comments.append(got)
        if comment_note:
            note = "; ".join(x for x in (note, f"comment engagement unavailable: {comment_note}") if x)
        elif not comments:
            note = "; ".join(x for x in (note,
                              "comment engagement was requested but the Comments dataset returned no rows") if x)

    # IDENTITY WITHOUT MEDIA IS STILL `partial`, on the paid rung too. The rule does not soften
    # because we are paying: a run that wrote a follower count and no posts must not paint green
    # over a posts table that did not grow.
    # ⚠ `via` CARRIES WHICH ROUTE ANSWERED — `brightdata` or `brightdata:corpus`. It lands in the
    # snapshot row's `source` column, so a stored measurement can always say whether it was read
    # live or taken from the vendor's pre-collected corpus. Two routes writing one indisting-
    # uishable `source` would make the series unauditable exactly where it matters most.
    return {"state": "ok" if posts else "partial", "profile": profile, "posts": posts,
            "comments": comments,
            "via": via,
            "note": note or ("" if posts else "profile read; no posts were returned")}





















def pull_profile(url, max_posts=None, log=print,
                 post_metrics=False, comment_metrics=False, pending_metrics=None,
                 pending_profile=None, prefetch=None, post_groups=None):
    """Everything readable about one public profile — EXACT NUMBERS OR NOTHING.

    Returns `{state, profile, posts, via, note}` with state ∈ ok | partial | blocked | error.
    `partial` means the identity was read but the media was not.

    ⭐⭐ WAVE 28 / OWNER RULING R5 — THE FREE ANONYMOUS LADDER IS RETIRED, AND THE REASON IS WHAT
    IT PUT IN THE COLUMN, not what it cost. Two rungs used to sit under the paid one: Instagram's
    own `web_profile_info` endpoint, and the profile HTML's `og:description`, which carries
    "204K Followers, 5,245 Following, 750 Posts". That last one is the page's OWN ROUNDING, and it
    landed in the same `followers` column as a measured 204,318, stamped `approx: "1"` — a flag
    that tells you the number is an abbreviation only AFTER you have already averaged it, sorted
    on it, or filtered a shortlist by it. A table where some rows are measurements and some are
    the page's shorthand is a table you cannot do arithmetic on.
    ⇒ A vendor refusal now returns `blocked` and the row KEEPS WHAT IT LAST KNEW, instead of being
    overwritten with a rounder number. The user-facing tier/fallback choice is gone with the
    rungs; there is one behaviour and it is this one.

    ⛔ THIS IS STILL CAPABILITY ROUTING, NOT A SINGLE VENDOR. `ig_profile` is declared on Bright
    Data AND Apify (`providers.py`), so a profile Bright Data cannot read is asked of Apify before
    anything reports blocked. What was retired is the FREE approximate rung, not the second
    opinion — the fallback that costs money and answers exactly is the one worth having.

    ⚠ `max_posts=None` means the default — see `pull_profile_bd` for why the constant cannot be
    named in the signature any more.
    """
    from automation_engine import (DEFAULT_POSTS_PER_PULL, PACE_SECONDS,  # lazy — module header
                                   _s)
    max_posts = DEFAULT_POSTS_PER_PULL if max_posts is None else max_posts
    handle = ig_handle(url)
    if not handle:
        return {"state": "error", "note": f"{url!r} is not an Instagram profile URL",
                "profile": {}, "posts": [], "comments": [], "via": ""}
    attempts = []
    # ⭐ 2026-08-09 — SET BY A RUNG THAT SAW A VENDOR SAY *this account does not exist*. It rides
    # out on the result so the engine can stop re-buying a dead handle every morning; it is never
    # inferred from a mere failure, only from a vendor stating it.
    gone = ""

    paid = pull_profile_bd(url, max_posts=max_posts, post_metrics=post_metrics,
                           comment_metrics=comment_metrics, log=log,
                           pending_metrics=pending_metrics,
                           pending_profile=pending_profile, prefetch=prefetch,
                           post_groups=post_groups)
    if paid["state"] in ("ok", "partial"):
        return paid
    # ⚠ 200, NOT 90. This string is the ONLY account of what the vendor said, and at 90 the
    # measured sentence truncated to *"…was not ready within"* — mid-clause, with the budget and
    # the snapshot id cut off. A reason nobody can read is the reason being discarded with extra
    # steps (D-103).
    attempts.append(f"brightdata:{_s(paid.get('note'), 200)}")
    # ⭐ THE APIFY RUNG (owner report 2026-08-09: "both Bright Data and Apify working in tandem").
    # `providers.py` has declared `ig_profile -> ("brightdata", "apify")` since 2026-08-08 and
    # nothing ever walked it: Bright Data failing went straight to the anonymous rungs, past a
    # configured provider declared capable of exactly this.
    #
    # ⚠ IT RUNS ONLY AFTER BRIGHT DATA HAS ACTUALLY FAILED, never as a top-up. Bright Data answers
    # profile+likes+comments correctly and far more cheaply; re-buying a profile from a second
    # vendor on every read is the expensive mistake the capability split exists to prevent. This
    # is a fallback, and the `attempts` trail records that it was needed.
    #
    # ⚠ AND IT IS GATED ON `configured()`, not merely on being in the chain — an unconfigured
    # provider must read as a rung that was never tried, not as one that refused.
    if providers.PROVIDERS["apify"].can("ig_profile"):
        a_prof, a_note = apify_profile(handle)
        if a_prof and a_prof.get("followers") is not None:
            log(f"  apify: profile ok ({a_prof.get('followers')} followers)")
            return {"state": "ok", "profile": a_prof, "posts": [], "comments": [],
                    "via": "apify",
                    "note": ("the primary source could not read this profile "
                             f"({_s(paid.get('note'), 90)}), so a backup source was used")}
        # ⭐ A VENDOR SAYING *not found* IS THE ANSWER, NOT A FAILED ATTEMPT. Recorded here and
        # carried to the return below, so the run can say the one thing that ends the loop
        # instead of the fourth variation on "we could not read it".
        if a_note == ACCOUNT_GONE_NOTE:
            gone = a_note
        attempts.append(f"apify:{_s(a_note or 'answered without follower counts', 200)}")
        time.sleep(PACE_SECONDS)

    # ⛔⛔ THE END OF THE WALK (R5). There were two more rungs here — `web_profile_info` and the
    # profile HTML's `og:description` — and they are DELETED, not disabled. What they returned was
    # a `partial` carrying `approx: "1"` and counts rounded by Instagram's own page furniture; the
    # docstring says why that is worse than nothing. `blocked` keeps the row's last real values.
    # ⚠ `via` STAYS "brightdata" on this path rather than "" — it names the chain that was walked,
    # which is what a person reading a blocked row needs in order to know who to ask.
    return {"state": "blocked", "profile": {}, "posts": [], "comments": [],
            "via": "brightdata", "gone": bool(gone),
            "note": (gone if gone else
                     "no provider could read this profile "
                     f"({'; '.join(attempts) or 'blocked'})")}