api/connectors_bd.py

"""connectors_bd.py -- THE BRIGHT DATA TRANSPORT. One vendor's wire, and nothing else.

⭐⭐ WAVE 30 · T09 (DEBT D-128). Every function below was CORRECT CODE UNDER A WRONG NAME: it lived
in `connectors_ig.py` and took the dataset id as a PARAMETER, which is the definition of not being
Instagram's. `connectors_tt.py` therefore had to import thirteen names from the Instagram connector
to reach a wire neither platform owns -- and the module header there said so in as many words
(*"THE TRANSPORT IS BORROWED FROM `connectors_ig`, ON PURPOSE ... the name is wrong and the code is
right"*). This file is that sentence resolved.

⛔⛔ **THIS MODULE IS THE ONE PLACE A SOCKET IS OPENED TO BRIGHT DATA, AND THAT IS A TESTABILITY
PROPERTY BEFORE IT IS AN ARCHITECTURAL ONE.** `bd_call` is the single door: `bd_scrape`,
`bd_snapshot_progress` and the three corpus functions all reach the network THROUGH it, resolving it
from THIS module's globals. So one monkeypatch on `connectors_bd.bd_call` seals every paid path in
the product, for both platforms, whatever the caller. Before the split there was no such point --
`connectors_ig.bd_call` sealed Instagram's rungs and TikTok's, but only because TikTok's connector
had borrowed Instagram's function object, which is a coincidence of an import rather than a
guarantee. [[seal-the-transport-not-the-rung]] is the standing rule; this module is where it becomes
structural. ⚠ `verify_automation.py:_seal_bd` patches here AND sweeps `sys.modules` for any other
binding of the same name, because a re-export elsewhere is a second door that reads exactly as green.

⛔ **WHAT IS DELIBERATELY *NOT* HERE: ANY PLATFORM'S VOCABULARY.** No `BD_DS_*` dataset id (those are
per-platform and stay with their connector), no field map, no profile/post/comment mapper. The test
for whether something belongs here is not the `bd_` prefix -- it is whether the function could serve
a THIRD platform tomorrow without an edit. ⚠ One Instagram assumption was found INSIDE the moved code
and is now a parameter rather than a literal: `bd_filter_start`'s engine-added exclusion rule named
the column `account`, which B-20 MEASURED as `account_id` in Bright Data's TikTok Profiles dataset.
It rides on every discovery run, is invisible to the condition builder, and would have reached the
vendor on the first TikTok search that had anything to exclude. See `handle_field`.

⚠ **THE ROW PRIMITIVES IN SECTION 3 ARE HERE FOR A REASON AND IT IS NOT TIDINESS.** They are the
readers every Bright Data mapper needs (a vendor row is JSON with two or three names for one fact),
both connectors used them, and leaving them in `connectors_ig.py` would have kept the import line
this ticket exists to delete. They know nothing about Instagram. ⚠ `_ig_int` KEEPS ITS NAME on
purpose: it is referenced by name in the engine and in exact-match gate fixtures, and a rename buys
nothing but a diff.
"""
from __future__ import annotations

import json
import os
import time

import requests


# =============================================================================================
# 1. THE WIRE -- auth, one HTTP door, the scraper endpoints
# =============================================================================================
BD_BASE_DEFAULT = "https://api.brightdata.com"


BD_PATH_SCRAPE = "/datasets/v3/scrape"        # SYNC: rows come back inline. param: dataset_id


BD_PATH_TRIGGER = "/datasets/v3/trigger"      # ASYNC: -> {"snapshot_id": "sd_…"}


BD_PATH_SNAPSHOT = "/datasets/v3/snapshot"    # /<sd_id>?format=json -> the rows


#: ⭐⭐ 2026-08-09 — THE SNAPSHOT'S OWN STATUS DOCUMENT, and asking it is the difference between
#: "the vendor is still working" and "the vendor finished and collected nothing". MEASURED on
#: `sd_msl80v7l1ti14rhcd6` (nurilab's `roxyfoxypinky`): `{"status": "ready", "records": 0,
#: "errors": 1, "error_codes": {"crawl_error": 1}, "collection_duration": 320513}`. The rows
#: endpoint answers `[]` for that snapshot forever, which `bd_scrape`'s old `if rows:` poll could
#: not distinguish from a snapshot mid-build — so it burned its whole budget and then reported
#: *"the records are collected, not lost"* about a batch that had collected nothing at all.
#: ⛔ SCRAPER NAMESPACE ONLY (`sd_…`). The corpus has no twin here; its status is
#: `BD_PATH_FILTER_SNAPSHOT/<snap_id>`, and crossing the two 404s (§2c).
BD_PATH_PROGRESS = "/datasets/v3/progress"    # /<sd_id> -> {status, records, errors, error_codes}


BD_PATH_FILTER = "/datasets/filter"           # ⚠ NO /v3/ — the CORPUS query (discovery)


BD_PATH_FILTER_SNAPSHOT = "/datasets/snapshot"        # /<snap_id> -> status


#: A sync scrape MEASURED at 15–23 s for one URL and ~40 s for two, so the timeout is generous;
#: it is a ceiling against a hung socket, not a latency budget.
BD_TIMEOUT = 300.0


#: The READ ceiling for one vendor answer, in KB. `fetch` truncates SILENTLY at this size, so it
#: is load-bearing arithmetic rather than a round number: MEASURED `file_size` was 175,617 bytes
#: for 5 corpus rows ⇒ **~35 KB per profile row**, and `BD_MAX_RECORDS` is set from it with
#: headroom (500 × 35 KB ≈ 17 MB against 32 MB). A truncated body is REPORTED as truncated in
#: `bd_call` — never allowed to surface as "that is not JSON".
BD_MAX_KB = 32768


#: How long a run waits on a scrape batch the vendor DEFERRED (see `_bd_deferral`). Much shorter
#: than the corpus filter's budget because a scrape snapshot is minutes, not tens of minutes —
#: and unlike the corpus path there is no handoff to a later run, so this is the whole wait.
BD_SCRAPE_WAIT = float(os.environ.get("AIOS_BD_SCRAPE_WAIT") or 180)


BD_SCRAPE_POLL = float(os.environ.get("AIOS_BD_SCRAPE_POLL") or 10)


# A Profile response is the enrichment's identity half, so it retains the normal bounded
# synchronous wait. Post/Reel engagement can create several snapshots per profile; action
# runners hand those off immediately and the scheduler collects the already-paid work later.
BD_METRIC_SCRAPE_WAIT = float(os.environ.get("AIOS_BD_METRIC_SCRAPE_WAIT") or 0)


#: ⚠ SPEC, NOT MEASURED — off the pricing page, never returned by the API (the funds gate fires
#: BEFORE a price is quoted, and `price: 0` means "not priced", not "free"). Every surface that
#: shows a number derived from this MUST say it is an estimate. `/customer/balance` answers 403
#: for our token, so there is no way to check a balance or a spend from here either.
BD_RECORD_PRICE_SPEC = 0.0025


def bd_key():
    """The key, or ''. Read fresh every call — see the module note above on why."""
    return (os.environ.get("AIOS_BRIGHTDATA_KEY") or "").strip()


def bd_base():
    return (os.environ.get("AIOS_BRIGHTDATA_BASE") or BD_BASE_DEFAULT).strip().rstrip("/")


def bd_ready():
    """Is the paid rung configured at all? The UI reads this to say so honestly."""
    return bool(bd_key())


def _bd_why(raw, key=""):
    """The vendor's OWN words for a refusal, as `" — …"`, or `""` when it gave none.

    ⛔ THIS WAS THROWN AWAY, AND THE COST WAS MEASURED RATHER THAN IMAGINED. nurilab's live
    discovery automation failed four runs across two days with `the profile source answered 400`
    while the body it discarded said, in full:

        {"validation_errors":["Filter logical groups can have a maximum of 4 rules."]}

    — the exact instruction needed to fix it, in our hands, deleted one line before the customer.
    A bare status code sends somebody to read a vendor's schema for a fault we were already
    holding the answer to; it is the same defect as a green dot over nothing, pointed the other
    way. Every non-2xx says WHY now, and the named cases above (402/401/403/429) keep their own
    plainer sentences because those are ours to explain, not the vendor's.
    """
    from automation_engine import _s                             # lazy — see the module header
    try:
        text = bytes(raw or b"").decode("utf-8", "replace").strip()[:600]
    except Exception:                                             # noqa: BLE001
        return ""
    if not text:
        return ""
    why = ""
    try:
        d = json.loads(text)
    except Exception:                                             # noqa: BLE001
        d = None
    if isinstance(d, dict):
        errs = d.get("validation_errors") or d.get("errors")
        if isinstance(errs, list) and errs:
            why = "; ".join(str(e) for e in errs)
        else:
            why = str(d.get("error") or d.get("message") or d.get("detail") or "")
    why = why or text
    # ⚠ A BODY WE DID NOT WRITE GETS THE SAME TREATMENT AS AN EXCEPTION'S str(). `bd_call` already
    # refuses to put the latter on the wire in case it carries the key; promising that a vendor
    # never echoes an Authorization header back is not a guarantee this module can make.
    if key and key in why:
        why = why.replace(key, "***")
    return f" — {_s(why, 300)}"


def bd_call(path, params=None, body=None):
    """One authenticated vendor call. GET when `body` is None, POST otherwise.

    Returns `(payload, note)`; a non-empty `note` means it did NOT answer, and the note is safe to
    show a user and safe to write to a log.

    Never raises. Every refusal shape — no key, a guarded base, a 402, a 429, a transport error,
    unparseable JSON — comes back as a note so the caller can record the attempt and drop to the
    next rung.
    """
    from automation_engine import Refused, fetch, fetch_json     # lazy — see the module header
    key = bd_key()
    if not key:
        # ⛔ THE FAIL-CLOSED PATH (C4). Named precisely, because "blocked" with no reason sends
        # somebody to read Instagram's status page instead of setting an env var.
        return None, "AIOS_BRIGHTDATA_KEY is not configured — the paid rung is closed"
    qs = "&".join(f"{k}={requests.utils.quote(str(v))}"
                  for k, v in (params or {}).items() if v not in (None, ""))
    url = f"{bd_base()}{path}" + (f"?{qs}" if qs else "")
    hdrs = {"Authorization": f"Bearer {key}", "Accept": "application/json"}
    try:
        if body is None:
            status, _final, raw = fetch(url, timeout=BD_TIMEOUT, max_kb=BD_MAX_KB, headers=hdrs)
        else:
            status, raw = fetch_json(url, body, timeout=BD_TIMEOUT, max_kb=BD_MAX_KB,
                                     headers=hdrs)
    except Refused as e:
        # The BASE was refused by the SSRF rail — a configuration fault, not a vendor outage.
        return None, f"the profile source's address was refused by the URL rail: {e}"
    except Exception as e:                                        # noqa: BLE001
        # ⚠ `type(e).__name__` only. A requests exception's str() can carry the full URL, and the
        # Authorization header is one refactor away from being a query param; this line must not
        # be what leaks it.
        return None, f"the profile source did not answer ({type(e).__name__})"
    if status == 402:
        return None, "the profile search is out of credit"
    if status in (401, 403):
        return None, f"the profile source refused our key ({status})"
    if status == 429:
        # ⚠ MEASURED during the discovery probe: `too_many_parallel_jobs`. The filter API has a
        # concurrency cap and the probe's most important test died on it — hence every caller
        # below is SERIAL by construction.
        return None, "the profile source is busy with another search; they run one at a time"
    if not (200 <= status < 300):
        return None, f"the profile source answered {status}{_bd_why(raw, key)}"
    # ⛔ A BODY THAT REACHED THE READ CEILING IS A TRUNCATION, AND IT MUST SAY SO. `fetch` reads
    # at most `max_kb` and returns what it got — silently. A truncated JSON document then fails
    # `json.loads`, falls through to the NDJSON branch, and comes back as "answered with
    # something that is not JSON": a sentence that sends somebody to read the vendor's schema
    # looking for a fault that is OUR byte ceiling — after a result set that has already been
    # paid for. Naming it here is the same no-silent-caps rule the row and table ceilings follow.
    if len(raw) >= BD_MAX_KB * 1024:
        return None, (f"the profile source's answer exceeded this client's {BD_MAX_KB // 1024} MB read "
                      f"ceiling and was TRUNCATED — ask for fewer records; nothing was parsed "
                      f"from a partial document")
    text = raw.decode("utf-8", "replace").strip()
    if not text:
        return None, f"the profile source answered {status} with an empty body"
    try:
        return json.loads(text), ""
    except Exception:                                             # noqa: BLE001
        pass
    # NDJSON is the other shape this API uses for row sets. Tried SECOND, so a genuine JSON error
    # is not silently reinterpreted as a one-line NDJSON document.
    rows = []
    for line in text.splitlines():
        line = line.strip()
        if not line:
            continue
        try:
            rows.append(json.loads(line))
        except Exception:                                         # noqa: BLE001
            return None, "the profile source answered with something we could not read"
    return (rows, "") if rows else (None, "the profile source answered with something we could not read")


def _bd_rows(payload):
    """A vendor answer → the list of row dicts inside it, whatever envelope it arrived in.

    MEASURED shapes: a bare list (`/v3/scrape`), NDJSON (already listified by `bd_call`), and
    `{"data": [...]}` / `{"results": [...]}` on the snapshot reads. An unreadable envelope yields
    NOTHING rather than a guess.
    """
    if isinstance(payload, list):
        return [x for x in payload if isinstance(x, dict)]
    if isinstance(payload, dict):
        for k in ("data", "results", "records", "items"):
            v = payload.get(k)
            if isinstance(v, list):
                return [x for x in v if isinstance(x, dict)]
        return [payload] if payload else []
    return []


def _bd_deferral(payload):
    """The snapshot id when the 'sync' endpoint DEFERRED, else ''.

    ⛔ MEASURED 2026-08-05, AND IT IS THE TRAP OF THIS WHOLE RUNG. `/v3/scrape` is documented and
    named as the synchronous call, and for a small batch it is — two profile URLs came back
    inline. Give it more work and it answers **200** with no rows and a note instead:

        {"snapshot_id": "sd_…", "message": "Your request is still in progress and cannot be
         retrieved in this call. Use the provided Snapshot ID to track progress…"}

    Read naively that is a 200 with a body, so a parser looking for "did it error?" sails past
    it, `_bd_rows` wraps the envelope as ONE unusable row, and the caller reports that the vendor
    returned nothing — **while a snapshot we have already been billed for sits on the account,
    finishing, and is never collected.** That is the D-26 failure shape (paid work abandoned)
    arriving through the success path.
    ⚠ It is a TIME threshold, not a row count: 2 URLs answered inline, 6 and 10 deferred.
    """
    if isinstance(payload, dict) and payload.get("snapshot_id") and not payload.get("account"):
        return str(payload["snapshot_id"])
    return ""


def bd_snapshot_progress(sid):
    """One scraper snapshot's status. `(state, records, note)`.

    state ∈ `running` | `done` | `failed` | `unknown`:
      · `running` — the vendor is still collecting; poll again.
      · `done`    — finished. `records` says whether it produced anything.
      · `failed`  — finished, produced NOTHING, and the vendor blamed the TARGET (an
                    `error_codes` entry such as `crawl_error`). On a profile request that is the
                    vendor saying it could not reach that account at all, which is a different
                    action from "we found no matches" — so it is a distinct state rather than a
                    phrase inside the note. A caller that has to grep an English sentence to
                    decide what happened is a caller whose behaviour changes when someone
                    improves the wording.
      · `unknown` — the status call itself did not answer; fall back to probing the rows.

    ⭐⭐ 2026-08-09 — THE QUESTION `bd_scrape` COULD NOT ASK. Its poll loop advanced only on
    `if rows:`, so a snapshot the vendor had FINISHED with zero records looked byte-for-byte like
    one still building: the rows endpoint answers `[]` in both cases. MEASURED on nurilab's one
    pending handle — `status: ready, records: 0, errors: 1, crawl_error: 1` after 320 s — while
    the enrich run reported *"deferred … the records are collected, not lost"*, which was false
    about that batch and had been re-reported on three separate runs.

    ⛔ `unknown` IS NOT `done`, and the distinction is the whole reason this returns three states
    rather than a boolean. When the status call itself fails we must fall back to the old
    row-probing behaviour, not conclude the snapshot is empty — an unreachable status endpoint
    would otherwise turn every deferral into a confident "the vendor found nothing".

    ⚠ THE NOTE IS DELIBERATELY VENDOR-NEUTRAL about WHAT was being read. This serves profile,
    post, reel and comment snapshots alike; a sentence naming "the account" would be wrong on
    three of the four, and the caller knows which it asked for.
    """
    payload, err = bd_call(f"{BD_PATH_PROGRESS}/{sid}", None, None)
    if err or not isinstance(payload, dict):
        return "unknown", 0, ""
    status = str(payload.get("status") or "").strip().lower()

    def _n(key):
        try:
            return int(payload.get(key) or 0)
        except (TypeError, ValueError):
            return 0

    records, errors = _n("records"), _n("errors")
    if status not in ("ready", "done", "failed", "error"):
        return "running", records, ""
    if records > 0:
        return "done", records, ""
    codes = payload.get("error_codes")
    named = (", ".join(f"{str(k).replace('_', ' ')} x{v}"
                       for k, v in sorted(codes.items()))
             if isinstance(codes, dict) and codes else "")
    if errors or named or status in ("failed", "error"):
        return "failed", 0, ("the source finished this request and collected nothing"
                             + (f" — it reported {named}" if named else "")
                             + "; the target is unreachable, private, or no longer exists")
    return "done", 0, "the source finished this request and found no records for it"


def bd_scrape(dataset_id, urls, wait=None, deferred=None):
    """Scrape a batch. `(rows, note)`; a note means it did not answer.

    Sends the SYNC call and, when the vendor defers it (see `_bd_deferral`), **collects the
    snapshot it handed back** rather than discarding it. One call carries many URLs, which is
    what keeps the per-profile pacing floor from turning a 20-profile run into an hour.
    """
    requested_urls = [str(u) for u in (urls or []) if u]
    payload, note = bd_call(BD_PATH_SCRAPE, {"dataset_id": dataset_id},
                            body=[{"url": u} for u in requested_urls])
    if note:
        return [], note
    sid = _bd_deferral(payload)
    if not sid:
        return _bd_rows(payload), ""

    # --- THE DEFERRED PATH. Poll the scraper namespace (`sd_…` at `/datasets/v3/…` — crossing it
    # with the corpus namespace is a flat 404 about a snapshot that is alive).
    budget = BD_SCRAPE_WAIT if wait is None else float(wait)
    waited = 0.0
    while True:
        # ⭐ THE STATUS DOCUMENT IS ASKED FIRST, because it is the only surface that can say
        # "finished, and there was nothing". Fetching rows while the vendor is still collecting
        # is also a wasted call on every single poll.
        state, records, empty_note = bd_snapshot_progress(sid)
        if state in ("done", "failed") and not records:
            return [], empty_note          # a definitive EMPTY, never our timeout sentence
        if state != "running":
            got, gnote = bd_call(f"{BD_PATH_SNAPSHOT}/{sid}", {"format": "json"})
            if not gnote:
                rows = _bd_rows(got)
                # The status document is itself JSON, so "did it deliver?" is decided by whether
                # what came back looks like ROWS — never by the HTTP code.
                if rows and not _bd_deferral(got) and not (
                        len(rows) == 1 and str(rows[0].get("status") or "") in
                        ("running", "building", "collecting")):
                    return rows, ""
        if waited >= budget:
            break
        time.sleep(BD_SCRAPE_POLL)
        waited += BD_SCRAPE_POLL
    # ⛔ NAME THE SNAPSHOT. It has been paid for and it is still finishing; a note that said only
    # "no rows" would throw away both the data and the way to get it.
    if isinstance(deferred, list):
        deferred.append({"snapshotId": sid, "datasetId": str(dataset_id),
                         "urls": requested_urls})
    return [], (f"the profile source deferred this batch to {sid} and it was not ready within "
                f"{int(budget)}s — the records are collected, not lost; ask for a smaller batch "
                f"or collect {sid} from the control panel")


# =============================================================================================
# 2. THE CORPUS FILTER -- `POST /datasets/filter`, a DIFFERENT namespace from the scraper
# =============================================================================================
# ⚠ Snapshots minted here are `snap_…` and are read at `/datasets/snapshot/…`; the scraper's are
# `sd_…` at `/datasets/v3/snapshot/…`, and the two 404 each other. Both halves live in this module
# so that pairing is visible in one place instead of being rediscovered per platform.
#: ⛔ THE VENDOR'S HARD CAP ON ONE LOGICAL GROUP — MEASURED 2026-08-06 against the live filter API
#: with nurilab's own seven-condition search, which had failed every run for two days:
#:
#:     400 {"validation_errors":["Filter logical groups can have a maximum of 4 rules."]}
#:
#: ⚠ EVERY ONE OF THOSE SEVEN CONDITIONS IS ACCEPTED ALONE (all seven probed individually, all
#: 200). The refusal is about the SHAPE of the group and nothing else — so no field, operator or
#: value in the builder is at fault, and no amount of editing the conditions would have found it.
#: What it means without `bd_group` below: the FIFTH condition anybody adds turns a working
#: discovery automation into one that fails forever, and the surface blames the search.
BD_GROUP_MAX = 4


def bd_group(nodes, operator):
    """`nodes` → an equivalent list in which NO logical group exceeds `BD_GROUP_MAX` rules.

    ⭐ SAFE BECAUSE AND AND OR ARE ASSOCIATIVE. Folding a run of same-operator rules into a nested
    group of the SAME operator cannot change what the filter matches — `a∧b∧c∧d∧e` and
    `(a∧b∧c∧d)∧e` are one predicate written two ways. That is why this is a re-SHAPING of the
    owner's search rather than a reinterpretation of it, and why it needs no ruling: a condition
    list means the same thing before and after. Nesting itself is already proven on this API
    (`expand_predicates`' or-groups, `snap_msh14ix81n8hlpo3nr`).

    ⛔ A ONE-RULE CHUNK IS NEVER WRAPPED. `expand_predicates` already carries that law for the
    one-value case — a one-element group is a shape nothing has ever been billed against — and a
    balanced split reaches it whenever a level divides with a remainder of one.
    """
    nodes = list(nodes or [])
    if len(nodes) <= BD_GROUP_MAX:
        return nodes
    # ⚠ BALANCED, NOT "take three and nest the rest". The tail-recursive shape costs one level of
    # DEPTH per extra rule (13 conditions ⇒ 5 deep) against a vendor whose nesting limit we have
    # not measured and would discover the same way we discovered this one: in production, on
    # somebody's automation. Chunking makes the depth log₄(n) — 16 conditions fit in two levels.
    out = []
    for i in range(0, len(nodes), BD_GROUP_MAX):
        chunk = nodes[i:i + BD_GROUP_MAX]
        out.append(chunk[0] if len(chunk) == 1 else {"operator": operator, "filters": chunk})
    return bd_group(out, operator)


#: ⛔ DEBT D-68 — THE VENDOR'S NESTING CEILING, MEASURED IN PRODUCTION 2026-08-06:
#:     400 — "filter" failed custom validation because logical operators cannot be more than
#:           3 levels deep
#: ⭐ AND IT IS NOT AN INDEPENDENT BUDGET FROM `BD_GROUP_MAX`. A >4-rule group is fixed by
#: NESTING, so the 4-rule cap is what MANUFACTURES depth: five top-level conditions carrying
#: multi-value lists goes `AND`(split) → `OR`(values) → `OR`(split) and is refused. The real
#: envelope is about 4 top-level conditions x <=4 values each, which is why
#: `MAX_PREDICATE_VALUES = 12` was never reachable at a realistic condition count.
BD_MAX_DEPTH = 3


def filter_depth(node):
    """How many logical levels a vendor `filters` structure nests to. A bare rule is depth 1."""
    if isinstance(node, list):
        return max((filter_depth(n) for n in node), default=0)
    if isinstance(node, dict) and isinstance(node.get("filters"), list):
        return 1 + filter_depth(node["filters"])
    return 1


def depth_refusal(predicates, operator="and"):
    """⭐ D-68 — refuse a search whose SHAPE the vendor will reject, before it is sent.

    Returns a sentence or `""`.

    ⛔ WHY THIS CANNOT BE A RULE-COUNT CHECK. The failure is in the emitted STRUCTURE, not in the
    conditions: the same five conditions pass or fail depending on how many VALUES each carries,
    because every multi-value list becomes its own OR level and every group over `BD_GROUP_MAX`
    adds a split level. So the only honest test is to BUILD the shape and measure it — which is
    cheap, because we build it anyway.
    ⚠ Free to get wrong at the vendor and expensive to diagnose there: a rejected filter is never
    billed, but the 400 says nothing about which condition to drop, so a person edits keywords for
    an afternoon while the shape stays identical. That is what this sentence exists to prevent.
    """
    try:
        depth = filter_depth(bd_group(expand_predicates(predicates), operator or "and"))
    except Exception:                                                 # noqa: BLE001
        return ""                                # never block a save on this check's own bug
    if depth <= BD_MAX_DEPTH:
        return ""
    multi = [p for p in (predicates or [])
             if isinstance((p or {}).get("value"), (list, tuple)) and len(p["value"]) > 1]
    hint = (f"drop one of the {len(multi)} multi-value conditions, or shorten its list"
            if multi else "drop one condition")
    return (f"this search nests {depth} levels deep and the provider allows "
            f"{BD_MAX_DEPTH} — {hint}. Every extra condition beyond {BD_GROUP_MAX}, and every "
            f"list of values, adds a level")


def expand_predicates(predicates):
    """Stored predicates → the vendor's `filters` list, expanding a value LIST into a nested OR.

    ⭐ MEASURED 2026-08-06: the filter API ACCEPTS a nested `{operator, filters:[…]}` inside
    `filters` (`snap_msh14ix81n8hlpo3nr`), which is what makes "Bio contains any of floral,
    flower, beauty" expressible WITHOUT setting the whole search to "match any". The alternative
    — the global Match dropdown — would drag every other condition into the same union, so
    "…and at least 10,000 followers" would silently become "…or at least 10,000 followers".

    ⛔ A ONE-VALUE CONDITION IS STILL SENT FLAT. Wrapping it in a one-element group would be a
    shape nothing has ever been billed against, adopted for tidiness, on the path that spends
    money. Only the several-values case takes the new shape.
    """
    out = []
    for p in predicates or []:
        v = (p or {}).get("value")
        if isinstance(v, list) and len(v) > 1:
            # ⚠ THE OR-GROUP IS SUBJECT TO THE SAME 4-RULE CAP as the outer one, and this is the
            # leg that hits it first in practice: "Bio contains any of floral, flower, beauty,
            # bouquet, wedding" is five keywords in ONE condition — a single row in the builder,
            # and a group the vendor refuses. `bd_group` is applied to the union too.
            out.append({"operator": "or",
                        "filters": bd_group(
                            [{"name": p["name"], "operator": p["operator"], "value": one}
                             for one in v], "or")})
        elif isinstance(v, list):
            out.append({**p, "value": v[0]} if v else
                       {k: val for k, val in p.items() if k != "value"})
        else:
            out.append(dict(p))
    return out


#: ⛔ HOW MANY HANDLES ONE `not_in` MAY CARRY. W25/R1b MEASURED that the vendor accepts a flat
#: `account not_in [...]` list to this size — *"the 12-value ceiling was OURS, not theirs"*. It is
#: also why the rule below is built BY HAND instead of going through `expand_predicates`: that
#: function turns any multi-value predicate into a nested OR of one-value rules, which for 5,000
#: handles would be a 5,000-rule tree the vendor refuses on both its caps at once.
BD_EXCLUDE_MAX = 5000


def bd_filter_start(predicates, operator="and", records_limit=5, dataset_id="",
                    exclude_handles=(), applied=None, handle_field="account"):
    """Start ONE corpus query. Returns `(snapshot_id, note)`.

    ⛔⛔ WAVE 30 · T09 — TWO INSTAGRAM ASSUMPTIONS CAME OUT OF THIS FUNCTION WHEN IT MOVED, AND THE
    SECOND ONE WAS SPENDING MONEY.
      1. `dataset_id` defaulted to `BD_DS_PROFILES`. In a module that serves both platforms a
         default corpus is a wrong answer waiting for a caller who forgot; it now REFUSES with a
         sentence rather than searching Instagram on TikTok's behalf.
      2. ⭐ `handle_field`. The engine-added `not_in` exclusion below named the column **`account`**
         — and B-20 MEASURED that Bright Data's TikTok Profiles dataset calls it **`account_id`**
         (5 of our 21 filter names exist there at all; five more are renamed). This rule is
         invisible to the condition builder by design, so no amount of restricting what a PERSON
         may type keeps it off the wire: it rides on every discovery run that has anything to
         exclude — i.e. every run after the first. ⚠ **What Bright Data DOES with an unknown filter
         field is UNMEASURED** (refuse, or bill for an unfiltered corpus) and cannot be measured for
         $0, since every `POST /datasets/filter` that answers 200 mints a billable snapshot. Naming
         the column per platform is the half that can be fixed for nothing, so it is fixed here.

    ⚠ ALWAYS THE COMPOUND `{operator, filters:[…]}` SHAPE, even for a single condition. A bare
    `{name, operator, value}` is a plausible-looking second shape that has never been sent from
    here, and a one-element `filters` list is a trivial extension of the three-element body that
    HAS been measured end-to-end. One shape, and it is the tested one.

    ⭐⭐ WAVE 27 ITEM 7 — `exclude_handles` IS THE ENGINE'S OWN REFINEMENT AND IS NEVER A CONDITION
    ROW. Until now discovery paid for every re-found profile: a nightly search over the same
    keywords returns the same accounts, the upsert notices they are `seen_again`, and the vendor
    has already billed for all of them. This narrows the search at the VENDOR instead of after it.
    It is invisible to the condition builder deliberately (W25/R1b: `not_in` is an engine-added
    refinement, and `BD_OPS_BY_KIND` omits it so nobody can type one) — a person did not write it,
    should not have to maintain it, and would be confused by a condition that changes every night.

    ⛔ AND IT IS DROPPED RATHER THAN RISKED. Two ways a free saving could break a search that
    works, and both are checked before the body is sent:
      1. **DEPTH.** The vendor allows 3 logical levels and the 4-rule group cap MANUFACTURES
         depth, so a search already near the ceiling gains a level from this rule and starts
         answering 400 forever ([[bright-data-group-cap]]).
      2. **`or`.** Under a top-level OR, adding an exclusion as another branch would WIDEN the
         search to "…or any account not in this list" — i.e. the whole corpus. Under OR the
         existing filter is wrapped in an AND instead, which costs the level checked in (1).
    A dropped exclusion is not an error: the search runs exactly as it did before and we pay for
    the duplicates, which is the status quo, not a regression.
    """
    if not str(dataset_id or "").strip():
        # Fail-closed, and it names what is missing. A transport that guesses a corpus is a
        # transport that bills the wrong one.
        return "", "the search was not started — no corpus was named for this platform"
    nodes = bd_group(expand_predicates(predicates), operator)
    filt = {"operator": operator, "filters": nodes}
    handles = [str(h).strip().lstrip("@").lower() for h in (exclude_handles or [])]
    handles = [h for h in dict.fromkeys(handles) if h][:BD_EXCLUDE_MAX]
    if isinstance(applied, dict):
        applied["available"] = len(handles)
        applied["excluded"] = 0
    if handles:
        rule = {"name": handle_field, "operator": "not_in", "value": handles}
        cand = (bd_group([*nodes, rule], "and") if operator == "and"
                else [filt, rule])
        if filter_depth(cand) <= BD_MAX_DEPTH:
            filt = {"operator": "and", "filters": cand}
            if isinstance(applied, dict):
                applied["excluded"] = len(handles)
        elif isinstance(applied, dict):
            applied["dropped"] = ("the search is already at the provider's nesting ceiling, so "
                                  "the already-found list was left off rather than risk it")
    body = {"dataset_id": dataset_id, "records_limit": int(records_limit),
            # ⭐ `bd_group` — the top-level group obeys the vendor's 4-rule cap (see its note).
            # Applied HERE rather than inside `expand_predicates` so the expansion stays a pure
            # stored-shape → wire-shape translation and the rebalance is one readable step on the
            # one path that actually posts a filter.
            "filter": filt}
    payload, note = bd_call(BD_PATH_FILTER, body=body)
    if note:
        return "", note
    sid = str((payload or {}).get("snapshot_id") or "") if isinstance(payload, dict) else ""
    if not sid:
        return "", "the search was accepted but came back with nothing to collect it from"
    return sid, ""


def bd_filter_status(snapshot_id):
    """`(status, size, note)` for a corpus snapshot. `status` ∈ building | ready | failed | ''."""
    payload, note = bd_call(f"{BD_PATH_FILTER_SNAPSHOT}/{snapshot_id}")
    if note:
        return "", 0, note
    d = payload if isinstance(payload, dict) else {}
    status = str(d.get("status") or "")
    err = str(d.get("error") or "")
    if status == "failed" and "did not match any records" in err.lower():
        # ⭐ MEASURED 2026-08-06 (`snap_msh14ix81n8hlpo3nr`): a filter that matches NOTHING comes
        # back `failed` — "Provided filter did not match any records" — not `ready` with zero
        # rows. Passed through, that reached the customer as "the search failed", which is the
        # single most likely outcome of typing a keyword that is too specific and the one most
        # certain to be read as "the product is broken". It is an ANSWER, so it gets its own
        # status and no error note.
        return "empty", 0, ""
    if status == "failed" and err:
        # NOT_ENOUGH_FUNDS is the one everybody will hit; say what it means, not just its name.
        extra = (" — the matched set is too large for this account's balance; narrow the "
                 "conditions or lower the limit" if "FUNDS" in err.upper() else "")
        return status, 0, f"the search failed: {err}{extra}"
    return status, int(d.get("dataset_size") or 0), ""


def bd_filter_rows(snapshot_id):
    """The rows of a READY corpus snapshot. `(rows, note)`.

    ⚠ `/download` answered *"Snapshot is building. Try again in a few minutes"* while the status
    endpoint ALREADY said `ready` — delivery lags readiness by minutes. So a download that does
    not parse as rows is reported as not-yet-collectable rather than as an error.
    """
    payload, note = bd_call(f"{BD_PATH_FILTER_SNAPSHOT}/{snapshot_id}/download",
                            {"format": "json"})
    if note:
        return [], note
    rows = _bd_rows(payload)
    if not rows:
        return [], "the search is ready but its rows are not downloadable yet — the vendor's " \
                   "delivery lags its own `ready` status by minutes"
    return rows, ""


# =============================================================================================
# 3. ROW PRIMITIVES -- reading one vendor row, for any platform's mapper
# =============================================================================================
#: `bd_call` refuses a partial response at this same transport ceiling. A single source document
#: therefore fits a JSON cell; Source data is never shortened by `_s`, which would corrupt JSON
#: and silently lose fields.
BD_SOURCE_PAYLOAD_MAX = BD_MAX_KB * 1024


def _ig_int(v):
    try:
        return int(v)
    except Exception:
        return None


def _first(node, *names, default=None):
    """The first of `names` present on `node` with a real value.

    ⚠ THE TOLERANCE IS THE POINT, not laziness. The vendor's `components/schemas` was not readable
    on 2026-08-04 (recorded as RECALLED in the digest), so every media field name below is a
    hypothesis. Reading through a candidate list means a rename costs one more candidate instead
    of a rung that returns nothing and cannot say why. **A field that matches nothing returns
    `default` (None), never 0** — an invented zero is a measurement, and this module's whole
    honest-status contract exists to stop those.
    """
    if not isinstance(node, dict):
        return default
    for n in names:
        v = node.get(n)
        if v not in (None, ""):
            return v
    return default


def _bd_first_url(v):
    """A link-in-bio value → one string.

    ⚠ MEASURED 2026-08-05: `external_url` is an **ARRAY** on the real row
    (`["http://linktr.ee/utopianevents/"]`), not the string the previous vendor returned. A dict
    (`external_url_title`) is also in the family. Read through the shape rather than assuming one.
    """
    if isinstance(v, list):
        v = next((x for x in v if x), "")
    if isinstance(v, dict):
        v = _first(v, "url", "link", default="")
    return str(v or "")


def _bd_flag(node, *names):
    """A vendor boolean → `'1'` / `''` / None. **None when the key is ABSENT or null**, which is
    the distinction that matters: `is_business_account: false` is a measurement and `''` records
    it; a missing key is not, and must not arrive as "false"."""
    for n in names:
        v = node.get(n) if isinstance(node, dict) else None
        if v is None:
            continue
        return "1" if v is True or str(v).strip().lower() in ("true", "1", "yes") else ""
    return None


def _bd_list(node, *names):
    """An array field → a comma-joined string, or None when it is absent/empty."""
    for n in names:
        v = node.get(n) if isinstance(node, dict) else None
        if isinstance(v, list) and v:
            return ", ".join(str(x) for x in v if x)[:400]
    return None


def _bd_source_payload(row):
    """One complete Bright Data response row as valid JSON, without field filtering.

    Source data is the lossless provider record, not a second relationship cache. A Profile's
    embedded ``posts`` array stays here *and* each parseable post becomes a linked Post record;
    keeping both preserves data while Links remain the interactive relational view. `bd_call`
    rejects partial bodies at ``BD_SOURCE_PAYLOAD_MAX``, and the table-layer JSON ceiling matches
    it, so this function never truncates a paid source response.
    """
    if not isinstance(row, dict):
        return None
    try:
        packed = json.dumps({str(k): v for k, v in row.items()}, ensure_ascii=False,
                            sort_keys=True, separators=(",", ":"))
    except (TypeError, ValueError):
        return None
    # A source row is bounded by the complete raw HTTP response. Leave this defensive assertion
    # explicit: should the transport ceiling and JSON-cell ceiling ever diverge, fail visibly
    # rather than shortening a document and pretending every paid field was retained.
    if len(packed.encode("utf-8")) > BD_SOURCE_PAYLOAD_MAX:
        raise ValueError("a Bright Data source row exceeded the guarded response ceiling")
    return packed