api/connectors_tt.py

"""connectors_tt.py — the TIKTOK connector (wave 29 · item 7 · DEBT D-9 · rulings R1 + R2).

Everything in this file knows what a VENDOR's TikTok row looks like. Nothing in it knows what an
automation is. That split is `connectors_ig.py`'s (wave 27 item 23) and it is the reason this file
exists at all rather than another thousand lines inside the engine.

⛔ **EVERY VENDOR FIELD NAME HERE WAS PROBED, NOT GUESSED.** The whole schema — 40 profile / 43
post / 17 comment fields, each with the vendor's own type, description and `pii` flag — was read
live from `GET /datasets/{id}/metadata` for **$0.00** and written down in
`.claude/wiki/waves/wave29/proto/tiktok-schema.md` (promoted to `tiktok-capture.md` at
close-out). That document is the AUTHORITY: do not re-probe it, and do not invent a key. Where a
name below reads through a candidate list it is because the vendor has two names for one fact
(`biography`/`signature`, `region`/`country`), never because the name is uncertain.

⛔ **NOTHING HERE EVER AUTHENTICATES TO TIKTOK.** No login, no cookie, no account to get banned —
public data through a supplier, exactly the rail `connectors_ig.py` states for Instagram. The
vendor key is a key to a SUPPLIER.

⭐ **THE TRANSPORT IS `connectors_bd.py` — SHARED, VENDOR-NAMED, AND NO LONGER BORROWED FROM THE
OTHER PLATFORM'S CONNECTOR** (WAVE 30 · T09, DEBT D-128). `bd_call`, `bd_scrape` and
`bd_filter_start` take the dataset id as a PARAMETER — they are Bright Data's wire, not
Instagram's — and re-implementing them here would be a second copy of the deferral handling, the
truncation guard, the SSRF rail and the snapshot-progress reader, i.e. five places for one bug.
Until wave 30 the code was right and the NAME was wrong: this file imported thirteen symbols from
`connectors_ig`, which read as a dependency on Instagram and was really a dependency on a supplier.
⚠ **This file now imports ZERO names from `connectors_ig`, and a gate check asserts that**, because
the sentence above is the kind that quietly stops being true.

⚠ **WHAT $0 COULD NOT BUY, so nobody reads this file as more measured than it is:**
  1. the real ROW shape — `/metadata` describes a DATASET, and Instagram's rows carry undeclared
     envelope keys (`timestamp`, `input`) that no metadata call mentions;
  2. that a declared field POPULATES — Bright Data's Instagram Reels *declares* `views: number`
     and delivers an account-grain wrong number (§4e). **Declared is not delivered**, and the one
     TikTok claim that matters most (`play_count`) is exactly a declaration.
"""
from __future__ import annotations

import automation_engine as engine
# ⚠ T09 — FOUR NAMES CAME OFF THIS LIST AND NOTHING BROKE, which is the point of deriving an
# import block from the AST both ways. `bd_call`, `bd_filter_start`, `bd_key` and `bd_ready` were
# imported here under a comment claiming they were *"re-exported for the runners"*; no runner ever
# read them off this module (the engine imports them from the transport itself), so they were four
# lines of dependency nobody was paying for. `[[artifact-with-no-importer]]` in its smallest form.
from connectors_bd import (
    _bd_first_url,
    _bd_flag,
    _bd_list,
    _bd_source_payload,
    _first,
    _ig_int,
    bd_scrape,
)

# ---------------------------------------------------------------------------------------------
# THE DATASETS
# ---------------------------------------------------------------------------------------------
# ⚠ CATALOGUE PRESENCE IS NOT ENTITLEMENT — the same finding Instagram produced. `GET
# /datasets/list` returned 1,735 rows of which 12 are TikTok; the three below answer 200 with a
# full field list, and the two DISCOVERY halves answer **404 for our key**:
#   `gd_lj71gn6l68bz7y9hc` (posts by profile) and `gd_lilwhto81z415d9mdl` (posts by keyword).
# ⇒ TikTok discovery routes through the PROFILES dataset's corpus filter, exactly as Instagram's
# does. A ticket that reaches for a by-keyword endpoint is reaching for a 404.
TT_DS_PROFILES = "gd_l1villgoiiidt09ci"       # TikTok - Profiles. 40 fields, 152,000,000 records
TT_DS_POSTS = "gd_lu702nij2f790tmv9h"         # TikTok - Posts. 43 fields
TT_DS_COMMENTS = "gd_lkf2st302ap89utw5k"      # TikTok - Comments. 17 fields

#: The vendor's two post-type tokens, verbatim from the dataset's own `ai_description`
#: (*"strictly these two"*, video = 99.5% of rows). Ours are `image`/`video`/`carousel`.
TT_POST_TYPE_VIDEO = "video"
TT_POST_TYPE_CONTENT = "content"

#: What a TikTok profile URL looks like, for the runner that has a handle and needs a URL. Kept
#: beside the dataset ids because it is the same class of vendor fact.
#: ⚠ ONE SPELLING. `platform/core/user_tables.profile_url(handle, 'tiktok')` builds the identical
#: string from `_PROFILE_RULES` (contract C2, E's half) — this exists for the connector's own
#: batch calls, and the gate asserts the two agree rather than trusting that they do.
TT_PROFILE_URL = "https://www.tiktok.com/@{handle}"


def tt_profile_url(handle):
    """`nurilab` → `https://www.tiktok.com/@nurilab`. `''` for a blank handle, never a bare `@`."""
    h = str(handle or "").strip().lstrip("@")
    return TT_PROFILE_URL.format(handle=h) if h else ""


# ---------------------------------------------------------------------------------------------
# THE FIELD MAPS
# ---------------------------------------------------------------------------------------------
# Each function turns ONE vendor row into the cell dict for one of the `ut_tt_*` schemas declared
# in `automation_engine`. The rules they all obey, stated once:
#
#   * **BLANK MEANS NOT READ, NEVER "THEY HAVE NONE".** A key the vendor did not send is OMITTED,
#     so a later, richer pull fills it instead of being overwritten by this one's silence. `_first`
#     returns `None` (never 0) when nothing matches, which is what makes that possible.
#   * **A zero from the vendor is a MEASUREMENT and survives.** (Instagram's `_ig_zero_is_blank`
#     rule is scoped to a paid rung whose zeros were proven fictional; nothing here has earned it.)
#   * **Every unpromoted vendor key stays whole in `source_payload`.** A schema addition on the
#     vendor's side is preserved rather than silently discarded while our column model catches up.
#   * **Nothing here writes a session token.** `tt_chain_token`, `secu_id` (~85% null), `short_id`
#     (100% null in the sample), `ftc` (100% null) and `relation` are deliberately unmapped; they
#     ride in `source_payload` where they make no claim.


def _tt_str(node, *names):
    """The first non-empty string among `names`, or None when the vendor sent nothing.

    ⛔ `None`, NOT `""`. The callers below drop `None` keys, which is what keeps a blank honest —
    an empty string written into a cell claims "we looked and it is empty".
    """
    v = _first(node, *names)
    if v is None:
        return None
    s = str(v).strip()
    return s or None


def _tt_pct(node, *names):
    """A vendor 0–1 engagement fraction → our stored 0–100 percentage, or None.

    ⛔ THE ×100 IS NOT COSMETIC (wave 26, amendment C1-a). Our `pct` renderer appends the sign to
    the STORED number, so writing the vendor's raw 0.0656 would print a 6.6% creator as `0.0%` —
    measured on the Instagram side, and TikTok sends the same shape on all three of its rates.
    """
    v = _first(node, *names)
    if v is None:
        return None
    out = engine._pct100(v)
    return out or None


def _tt_day(node, *names):
    """A vendor stamp → `YYYY-MM-DD`, or None. Our `date` columns store a day."""
    v = _first(node, *names)
    if v is None:
        return None
    return engine._day(v) or None


def _drop_blanks(row):
    """The one place a mapped row loses its `None`s — see the BLANK MEANS NOT READ rule above."""
    return {k: v for k, v in row.items() if v is not None and v != ""}


def normalize_profile(node, handle=""):
    """A TikTok Profiles row → the `ut_tt_profile` / `ut_tt_snapshots` cell shape.

    ⚠ TWO FIELDS READ THROUGH A CANDIDATE PAIR, and both pairs are the vendor's, not a guess:
      * `biography` is PRIMARY and `signature` the FALLBACK — the probe measured `signature`
        populated on 85% of rows and they carry the same text;
      * `region` is PRIMARY and `country` the FALLBACK — `region` is the one with a documented
        two-letter-ISO description, `country` has no description at all.
    ⚠ `videos_count` is mapped to `posts_count` with a caveat recorded rather than hidden: its
    `ai_description` ranges 1-89, so it may be a WINDOW rather than a lifetime total. It is the
    only count of its kind the dataset offers.
    """
    node = node if isinstance(node, dict) else {}
    account = _tt_str(node, "account_id") or str(handle or "").strip().lstrip("@")
    return _drop_blanks({
        "platform": engine.PLATFORM_TIKTOK,
        "handle": account,
        "full_name": _tt_str(node, "nickname"),
        "tt_id": _tt_str(node, "id"),
        "profile_url": _tt_str(node, "url") or (tt_profile_url(account) or None),
        "bio": _tt_str(node, "biography", "signature"),
        "external_url": _bd_first_url(_first(node, "bio_link")),
        "verified": _bd_flag(node, "is_verified"),
        "is_private": _bd_flag(node, "is_private"),
        # ⚠ APPROXIMATE, and the word is the vendor's: `is_commerce_user` has *"many null values"*
        # on their own description. It is the closest thing TikTok has to Instagram's
        # `is_business_account`, and `_bd_flag` writes nothing at all when the key is absent — so
        # the approximation only ever fills a cell the vendor actually answered.
        "is_business": _bd_flag(node, "is_commerce_user"),
        "followers": _ig_int(_first(node, "followers")),
        "following": _ig_int(_first(node, "following")),
        "posts_count": _ig_int(_first(node, "videos_count")),
        # ⚠ `likes` on a PROFILE row is likes RECEIVED across the account's videos (18-110,200, no
        # nulls). It is not a post-level number and it is not our `likes` column, which is why it
        # is stored under a different name.
        "likes_received": _ig_int(_first(node, "likes")),
        "avg_engagement": _tt_pct(node, "awg_engagement_rate"),
        "like_engagement": _tt_pct(node, "like_engagement_rate"),
        "comment_engagement": _tt_pct(node, "comment_engagement_rate"),
        "country_code": _tt_str(node, "region", "country"),
        "region": _tt_str(node, "region"),
        "predicted_lang": _tt_str(node, "predicted_lang"),
        # ⚠ ACCOUNT AGE, NOT A MEASUREMENT STAMP — `create_time` on a profile is when the ACCOUNT
        # was made. TikTok stamps nothing with "when this number was true", exactly like Instagram,
        # which is why the append law dates a snapshot by when WE read it.
        "account_created_at": _tt_day(node, "create_time"),
        "source_payload": _bd_source_payload(node),
    })


def tt_post_type(node):
    """A TikTok post row → one of OUR three type options, or None.

    ⛔ THE FIRST OF THE PROBE DOC'S TWO NAMED BLOCKERS. The vendor's vocabulary is `"video"` /
    `"content"`; ours is `image` / `video` / `carousel` and has no `"content"`. Writing the
    vendor's token would fail `_clean_field`'s option check on the way in and would put an
    untranslated API word in front of a user on the way out.

    So: `video` is `video`, and `content` — TikTok's photo-mode post — is `image`, EXCEPT when the
    row carries more than one `carousel_images` entry, which is what a carousel IS on either
    network. ⚠ The multi-image branch is decided from the IMAGES, never from the type token: the
    token cannot express it, so inferring `carousel` from the word would be inventing a fact.
    ⚠ An UNKNOWN token returns None rather than defaulting to `video` (99.5% of rows are video, and
    that is exactly what would make the wrong default invisible).
    """
    raw = str((node or {}).get("post_type") or "").strip().lower()
    if raw == TT_POST_TYPE_VIDEO:
        return "video"
    if raw != TT_POST_TYPE_CONTENT:
        return None
    images = (node or {}).get("carousel_images")
    return "carousel" if isinstance(images, list) and len(images) > 1 else "image"


def normalize_post(node):
    """A TikTok Posts row → the `ut_tt_posts` cell shape. None when it carries no identity.

    ⛔ THE SECOND NAMED BLOCKER, RESOLVED HERE AND NOWHERE ELSE: `play_count` is ONE number and
    Instagram's schema has TWO columns for it (`plays` and `views`). On TikTok they are the same
    fact — `play_count` IS the count TikTok displays under a video — so it maps to `views` and
    `ut_tt_posts` HAS NO `plays` COLUMN. Copying one vendor number into two of our columns would
    manufacture a second measurement that a rollup could average or double-count, which is a worse
    outcome than the missing column it would paper over.

    ⚠ `shortcode` reads `shortcode` then `post_id`: both are 19-digit numerics on this dataset and
    the probe measured them as the same shape. That equality is what lets the comments→posts link
    join with NO normaliser, which the Instagram side never had.
    ⚠ `num_share_count` (a number) is preferred over `share_count` (typed TEXT by the vendor).
    ⚠ `commerce_info` is a business/commerce LOCATION per its own description — cities and
    countries. It is NOT a paid-partnership flag, and nothing in this dataset is: TikTok declares
    no equivalent, so `paid_partnership`/`partner` have no column on this family at all.
    """
    node = node if isinstance(node, dict) else {}
    shortcode = _tt_str(node, "shortcode", "post_id")
    if not shortcode:
        return None
    return _drop_blanks({
        "platform": engine.PLATFORM_TIKTOK,
        "shortcode": shortcode,
        # ⛔⛔ `account_id`, NOT `profile_username` — MEASURED on a real Posts row 2026-08-12.
        # The vendor's `profile_username` is the DISPLAY NAME (`"Dina"`), while `account_id` is the
        # @handle (`"d1na_th"`) — the same field `normalize_profile` already reads for `handle`, so
        # one name means one thing across both corpora. Reading the display name silently broke the
        # only join this table has: `ut_tt_posts.influencer_key` -> `ut_tt_profile.handle` matched
        # NOTHING, so a person could not filter posts by creator and a rollup would count zero.
        # ⚠ NO FALLBACK TO `profile_username`, deliberately. It is not a degraded handle, it is a
        # different fact, and filling a join key with it is worse than leaving it blank — a blank
        # is visibly missing, a display name looks like an answer [[one-question-two-normalizers]].
        # The URL is the honest second source: it carries the handle by construction.
        "influencer_key": (_tt_str(node, "account_id")
                           or tt_handle(_tt_str(node, "profile_url", "url") or "") or None),
        "posted_at": _tt_day(node, "create_time"),
        "type": tt_post_type(node),
        "caption": _tt_str(node, "description"),
        "url": _tt_str(node, "url"),
        "hashtags": _bd_list(node, "hashtags"),
        "tagged_location": _tt_str(node, "commerce_info"),
        "views": _ig_int(_first(node, "play_count")),
        "likes": _ig_int(_first(node, "digg_count")),
        "comments": _ig_int(_first(node, "comment_count")),
        "shares": _ig_int(_first(node, "num_share_count")),
        "saves": _ig_int(_first(node, "collect_count")),
        "video_duration": _ig_int(_first(node, "video_duration")),
        "source_payload": _bd_source_payload(node),
    })


def normalize_comment(node):
    """A TikTok Comments row → the `ut_tt_comments` cell shape. None without a comment id.

    ⚠ THE NAME COLLISION, RESOLVED: TikTok's `replies` is an ARRAY of reply objects and OUR
    `replies` column is an INT count. The count comes from `num_replies`; the array stays whole in
    `source_payload`. Reading the array's length instead would be a second, disagreeing answer to
    a question the vendor already answers — and it would disagree, because a page of replies is not
    all of them.
    ⚠ `date_created` is typed `date` by this vendor, unlike Instagram's `comment_date` which is
    text and needs defensive parsing. It still goes through `_tt_day` — one date path, so a vendor
    that changes its mind cannot change ours.
    ⚠ The comment TEXT and every identifiable commenter field (`commenter_user_name` is flagged
    PII) stay in `source_payload` and are promoted to no column, which is the same posture the
    Instagram comment schema takes for the same D-24 reason.
    """
    node = node if isinstance(node, dict) else {}
    comment_key = _tt_str(node, "comment_id")
    if not comment_key:
        return None
    return _drop_blanks({
        "platform": engine.PLATFORM_TIKTOK,
        "comment_key": comment_key,
        "shortcode": _tt_str(node, "post_id"),
        # ⭐⭐ OWNER RULING 2026-08-12: the comment's CONTENT gets a column. `comment_text` is the
        # vendor's own key and `comment_text_only` its stripped variant (the probe recorded both);
        # primary first, so a row carrying the rich form is not silently served the plain one.
        # ⛔ The commenter's identity is deliberately NOT promoted — see `TT_COMMENT_FIELDS`.
        "text": _tt_str(node, "comment_text", "comment_text_only"),
        "commented_at": _tt_day(node, "date_created"),
        "likes": _ig_int(_first(node, "num_likes")),
        "replies": _ig_int(_first(node, "num_replies")),
        "source_payload": _bd_source_payload(node),
    })


#: ⭐ The three maps, addressable by name — so a gate (and the runners in T04-T06) can walk them
#: rather than naming three functions, and so adding a fourth dataset is one entry.
TT_NORMALIZERS = {
    "tt_profile": normalize_profile,
    "tt_post_metrics": normalize_post,
    "tt_comments": normalize_comment,
}


# ---------------------------------------------------------------------------------------------
# THE FETCH — wave 30 · W30-T08 (carrying wave-29's dropped T05)
# ---------------------------------------------------------------------------------------------

def tt_handle(url):
    """A TikTok profile URL **or** a bare handle → the handle. `''` when it is neither.

    Deliberately permissive about the input and strict about the output, because the two callers
    hand it different things: an automation stores whatever a person typed in the profile column
    (`@nurilab`, `nurilab`, or the full URL), while the discovery runner already holds a clean
    `account_id`. One normaliser, so a row found by discovery and a row typed by hand cannot
    resolve to two different handles.
    """
    s = str(url or "").strip()
    if not s:
        return ""
    if "tiktok.com" in s.lower():
        # Everything after the first `@`, up to the next path segment or query.
        tail = s.split("@", 1)[1] if "@" in s else ""
        s = tail.split("/")[0].split("?")[0].split("#")[0]
    s = s.strip().lstrip("@").strip()
    # A handle is the vendor's `account_id` shape: alphanumerics, dots and underscores.
    return s if s and all(c.isalnum() or c in "._" for c in s) else ""


def tt_post_urls(node, limit=0):
    """⭐ WAVE 30 · T10 — the profile row's own post permalinks, newest-first as the vendor sends.

    ⛔ THIS IS WHY TIKTOK POST CAPTURE COSTS NO EXTRA DISCOVERY. `top_videos` rides the PROFILE row
    we have already bought, so the posts read is a scrape of links we hold, never a search for them.
    The two TikTok DISCOVERY datasets (posts-by-profile, posts-by-keyword) are **404 for our key**,
    so a design that reached for either would not merely be dearer, it would not work.

    ⛔⛔ CORRECTED 2026-08-12 — THIS DOCSTRING USED TO SAY *"the probe MEASURED `top_videos` as an
    array of video permalinks, NO empties"*, AND THAT SENTENCE IS WHAT SHIPPED THE BUG. The probe
    read `/datasets/{id}/metadata` — a DATASET description — and the phrase quoted was the field's
    `ai_description`, not an observation of a row. The real row sends **dicts keyed `video_url`**
    (measured below), so the reader built against the quoted sentence found nothing, forever, in
    silence. ⭐ The transferable half: *"the probe measured X"* and *"the probe read a declaration
    of X"* are different claims, and prose cannot be told apart by a reader downstream — which is
    why the correction names the method, not just the value.

    ⚠ `top_posts_data` is deliberately NOT read: the probe calls it *"a thin dup of `top_videos`"*,
    and preferring whichever happened to be longer is how one creator's window silently differs
    from another's.
    ⚠ `limit <= 0` means "everything the row carried". The CAP IS THE CALLER'S — `config.maxPosts`,
    validated 1..12 — and it is applied here rather than after the scrape so an unwanted post is
    never bought. [[a-constant-two-features-share]]: the 12 is the vendor's measured profile window,
    not a number this function may invent.
    """
    raw = (node or {}).get("top_videos")
    out = []
    for item in raw if isinstance(raw, list) else []:
        # ⛔⛔ MEASURED ON A REAL ROW 2026-08-12, AND IT IS NOT WHAT THE SCHEMA SAID.
        # `top_videos` is NOT an array of permalink strings. One paid TikTok Profiles scrape of a
        # live handle returns **19 DICTS**, keyed
        # `video_url · video_id · playcount · diggcount · commentcount · share_count ·
        #  favorites_count · create_date · cover_image`.
        # The docstring above cites the $0 probe as having "measured" permalinks — it had not, and
        # could not: `/datasets/{id}/metadata` describes a DATASET, and the probe's own verdict says
        # so in terms (*"what $0 cannot buy … the real ROW shape … that a declared field
        # POPULATES"*, `wave29/proto/tiktok-schema.md`). This is the SECOND time this vendor's
        # declaration has diverged from its delivery on this exact axis; BD's IG Reels `views` was
        # the first. [[reachable-is-not-the-same-as-built]]
        # ⇒ The consequence, live: `TT_DS_POSTS` was never reached, because this returned an EMPTY
        # list on every real profile — post capture could not have worked for anybody, and the
        # T10 gate stayed green because its canned fixture encoded the DECLARED shape. A fixture
        # written from a schema tests the schema.
        # ⚠ `video_url` is FIRST because it is the key the vendor actually sends; `url` is kept
        # because it costs nothing and is what a future corpus revision would most likely use. The
        # bare-string branch stays for the same reason — this widens what is ACCEPTED and invents
        # nothing: a shape that yields no `http…` value still degrades to "no posts", exactly as
        # before, rather than to a URL built out of a guess.
        # ⚠ `top_posts_data` is STILL not read (it carries `post_url` and would work): preferring
        # whichever array happened to be longer is how one creator's window silently differs from
        # another's, and that reasoning is unchanged by this correction.
        if isinstance(item, dict):
            u = str(item.get("video_url") or item.get("url") or "").strip()
        else:
            u = str(item or "").strip()
        if u.lower().startswith("http") and u not in out:
            out.append(u)
    return out[:limit] if limit and limit > 0 else out


def pull_posts_tt(post_urls, log=print, deferred=None):
    """The TikTok Posts dataset for a list of permalinks → `(rows, note)`, already normalised.

    ⚠ ONE CALL FOR THE WHOLE WINDOW. `bd_scrape` has always taken a list, and the Instagram side
    measured what happens when a caller forgets: 25 records, one billed snapshot each, a walk still
    running at 67 minutes. Nothing here loops per URL.
    """
    urls = [str(u) for u in (post_urls or []) if str(u or "").strip()]
    if not urls:
        return [], ""
    rows, note = bd_scrape(TT_DS_POSTS, urls, deferred=deferred)
    if note:
        log(f"[aios-tt] posts: {note}")
        return [], note
    out = [r for r in (normalize_post(n) for n in rows) if r]
    return out, ""


def pull_comments_tt(post_urls, log=print, deferred=None):
    """The TikTok Comments dataset for a list of POST permalinks → `(rows, note)`, normalised.

    ⛔ THE MOST EXPENSIVE THING THIS PRODUCT BUYS, and the reason `commentMetrics` defaults OFF on
    both networks: a comments scrape ingests identifiable third parties who never entered anybody's
    list (D-24). The mapper already keeps every commenter field in `source_payload` and promotes
    none of them to a column; this function adds no new exposure, it just has to be asked for.
    """
    urls = [str(u) for u in (post_urls or []) if str(u or "").strip()]
    if not urls:
        return [], ""
    rows, note = bd_scrape(TT_DS_COMMENTS, urls, deferred=deferred)
    if note:
        log(f"[aios-tt] comments: {note}")
        return [], note
    out = [r for r in (normalize_comment(n) for n in rows) if r]
    return out, ""


#: ⭐⭐ WAVE 30 · D-156 — THE MEDIA DATASETS, AS A SET, SO THE HAND-OFF CAN FILTER ON IDENTITY.
#: `_media_deferrals` uses this to lift ONLY posts/comments snapshots out of the local deferral
#: list. That is what makes it structurally impossible to file a PROFILE snapshot in the engine's
#: metric queue — the defect a draft of T10 shipped and A-39 booked as "the wrong fix is worse
#: than the gap". A membership test cannot be got wrong by a later edit the way `if` order can.
TT_MEDIA_DATASETS = (TT_DS_POSTS, TT_DS_COMMENTS)


def _media_deferrals(deferred):
    """The POSTS/COMMENTS entries of a `bd_scrape` deferral list — never the profile's.

    ⚠ The engine, not this module, decides what a deferral MEANS: it stamps `kind` and the
    handle and files it. TikTok needs no `_tag_metric_deferrals` twin because one dataset is one
    kind here, so the id already carries everything a mapper choice depends on — and importing
    Instagram's tagger is not available anyway (W30-T09 gates ZERO `from connectors_ig` lines).
    """
    out = []
    for d in deferred or []:
        if isinstance(d, dict) and str(d.get("datasetId") or "") in TT_MEDIA_DATASETS:
            out.append(dict(d))
    return out


def pull_profile_tt(url, log=print, pending_profile=None, prefetch=None,
                    max_posts=0, post_metrics=False, comment_metrics=False):
    """ONE TikTok profile from the vendor. Same return contract as `pull_profile`.

    `{state, profile, posts, comments, via, note}` with `state ∈ ok | partial | blocked | error`,
    so the engine's enrich branch treats every network identically and no caller learns a new
    shape.

    ⭐ WAVE 30 · T10 — POSTS AND COMMENTS ARE REAL NOW, AND BOTH DEFAULT OFF, exactly as Instagram's
    do. `post_metrics` scrapes the profile row's own `top_videos` permalinks (see `tt_post_urls` —
    no discovery call, because both TikTok discovery datasets 404 for our key); `comment_metrics`
    then scrapes the comments of the posts that came back. ⚠ COMMENTS REQUIRE POSTS by construction
    rather than by a rule: their input IS a post permalink, so asking for comments with post capture
    off is a request with no subject, and it returns none instead of quietly buying posts nobody
    asked for.

    ⛔ `partial` IS THE SUCCESS STATE WHENEVER NO MEDIA WAS READ, and that is deliberate rather than
    pessimistic. The Instagram contract reads `ok` only when identity AND media both landed
    (`pull_profile_bd`: *"identity without media is still partial ... a run that wrote a follower
    count and no posts must not paint green over a posts table that did not grow"*). So: posts not
    ASKED for → `partial`, saying so; posts asked for and landed → `ok`; asked for and none came →
    `partial` with the vendor's reason. The state answers "did this pull deliver what it went for",
    never "did the function finish".

    ⚠ **NO FREE RUNG, AND NO FALLBACK CHAIN.** Instagram's `pull_profile` drops to Apify when the
    paid rung refuses; `providers.DEFAULT_CHAINS["tt_profile"]` is deliberately single-provider,
    with its own note explaining that a multi-provider chain is a promise something walks it and
    that nothing walks Instagram's second name today either. So a refusal here is final, and it
    says so instead of implying a retry somewhere.
    """
    handle = tt_handle(url)
    if not handle:
        return {"state": "error", "profile": {}, "posts": [], "comments": [], "via": "",
                "note": f"{url!r} is not a TikTok profile URL or handle"}

    # ⭐ THE BATCH FAST PATH, same shape as the Instagram side: `prefetch` is `{handle: node}` from
    # one multi-URL scrape covering a whole selection. A hit is a vendor round trip that does not
    # happen; a miss falls through to the single-URL call below.
    cached = prefetch.get(handle) if isinstance(prefetch, dict) else None
    _deferred = []
    if isinstance(cached, dict) and cached:
        rows, note = [cached], ""
    else:
        rows, note = bd_scrape(TT_DS_PROFILES, [tt_profile_url(handle)], deferred=_deferred)

    node = rows[0] if rows else {}
    profile = normalize_profile(node, handle) if node else {}
    # ⛔ THE READABILITY TEST IS `followers`/`following`, NOT "did we get a dict". `normalize_profile`
    # drops blanks, so an unreadable row still returns `{"platform": …, "handle": …}` — truthy, and
    # carrying nothing anybody asked for. The Instagram rung tests exactly this pair for exactly
    # this reason, and answering "0 followers" instead is the failure it exists to prevent.
    unreadable = profile.get("followers") is None and profile.get("following") is None
    if note or unreadable:
        # ⭐ THE DEFERRAL IS HANDED OVER RATHER THAN DISCARDED. A snapshot the vendor is still
        # building HAS ALREADY BEEN PAID FOR; dropping its id bills again on the next run for the
        # same record. That was live on the Instagram profile path until 2026-08-09 — measured on
        # nurilab as two runs, two fresh snapshots, both abandoned — and it is not being
        # reintroduced here by omission.
        if isinstance(pending_profile, list):
            for d in _deferred:
                pending_profile.append({**d, "kind": "profile", "influencer": handle})
        why = note or ("the scrape answered, but no follower/following counts were readable in it "
                       "(the field names may have moved - see tiktok-capture.md)")
        return {"state": "blocked", "profile": {}, "posts": [], "comments": [], "via": "brightdata",
                "deferredProfile": [d.get("snapshotId") for d in _deferred],
                "note": why}

    # --- W30-T10: THE MEDIA, ONLY WHEN IT WAS ASKED FOR. ------------------------------------
    if not post_metrics:
        return {"state": "partial", "profile": profile, "posts": [], "comments": [],
                "via": "brightdata",
                "note": note or "profile read; post capture is off for this step"}
    urls = tt_post_urls(node, limit=max_posts)
    if not urls:
        # ⚠ NOT AN ERROR AND NOT A RETRY. A creator with no `top_videos` has nothing to buy, and
        # saying so is what stops the next run paying to be told the same thing.
        return {"state": "partial", "profile": profile, "posts": [], "comments": [],
                "via": "brightdata",
                "note": note or "profile read; this account's row carried no post links"}
    posts, p_note = pull_posts_tt(urls, log=log, deferred=_deferred)
    comments, c_note = ([], "")
    if comment_metrics and posts:
        # The comments dataset is keyed on a POST permalink, so it reads the posts we just bought —
        # `url` from the mapper, never the profile's raw array, so a post the posts scrape refused
        # is not silently asked about again one rung later.
        comments, c_note = pull_comments_tt([p.get("url") for p in posts if p.get("url")],
                                            log=log, deferred=_deferred)
    # ⭐⭐ WAVE 30 · D-156 — THE MEDIA DEFERRALS ARE HANDED BACK, and the shape of the hand-off is
    # the whole lesson. An earlier draft of T10 appended every `_deferred` entry to
    # `pending_profile` tagged `kind: "profile"`. By the time control reaches here a PROFILE
    # deferral is impossible — the profile branch above returns `blocked` on any note — so **every
    # id fanned out that way was a POSTS or COMMENTS snapshot in the PROFILE queue**, whose
    # collector writes preset profile cells onto somebody's record from post rows. The engine keeps
    # the two queues apart deliberately (`_pending_profile_tasks` vs `_pending_metric_tasks`).
    # ⇒ So this returns them under their OWN key, filtered by dataset identity
    # (`_media_deferrals`), and the engine files them in the metric queue with the handle it
    # already holds. Returning rather than appending also keeps the queue's vocabulary out of a
    # connector: this module knows which CORPUS deferred, never what the engine calls it.
    # ⚠ `deferredMedia` rides BOTH returns on purpose. The empty-posts case is the one that
    # matters most — that is exactly the run where the vendor took too long, so a caller reading
    # the ids only from the success path would lose every batch it actually paid for.
    deferred_media = _media_deferrals(_deferred)
    if not posts:
        return {"state": "partial", "profile": profile, "posts": [], "comments": [],
                "via": "brightdata", "deferredMedia": deferred_media,
                "note": p_note or note or "profile read; the post source returned nothing"}
    return {"state": "ok", "profile": profile, "posts": posts, "comments": comments,
            "via": "brightdata", "deferredMedia": deferred_media,
            "note": c_note or note or ""}