socialmag/capture.py

"""socialmag.capture — structured JSON straight from the platform's own app.

The page already fetches typed JSON to render itself. This reads *that* instead
of reading the pixels it painted. No credentials, no unofficial client, no new
IP — the traffic is genuinely abang's; we only listen.

    from socialmag import capture
    rows = await capture.capture_feed(cdp, "twitter")

One principle, two mechanisms (measured 2026-09-12 against live Brave):

    twitter   network   /i/api/graphql/.../HomeLatestTimeline   190 rows in 0.9s
    threads   network   /graphql/query -> feedData                4 rows, 22.9KB
    ig        document  RelayPrefetchedStreamCache               3 rows, 150KB script

X and Threads must be *heard*. Instagram must be *read* — Meta prefetches the
feed into the document, so nothing crosses the wire to capture.

Three things this does not do, on purpose:

* It never activates a tab. Every signal here is JS state or the network, and
  both work with `document.visibilityState == "hidden"`.
* It never returns an empty list quietly. Nothing captured raises, because
  "the platform posted nothing" and "our listener missed it" are different
  facts and an empty list tells you the wrong one.
* It does not write. Reading is solved here; posting still belongs to
  `platforms/`, sequential and paced, where the queue's gap is the disguise.
"""

import asyncio
import json
import time
from datetime import datetime

from .cdp import CDPError

#: How long to listen after navigate before giving up on a platform.
LISTEN_SECONDS = 16.0

#: Poll gap while draining the session's event queue.
EVENT_POLL = 1.0


def log(*args):
    import sys
    print(f"[capture {time.strftime('%H:%M:%S')}]", *args, file=sys.stderr, flush=True)


# ---------------------------------------------------------------------------
# normalisation — one row shape across every platform
# ---------------------------------------------------------------------------

def _row(platform, **kw):
    row = {
        "platform": platform,
        "id": None, "author": None, "text": "", "ts": None,
        "likes": None, "replies": None, "reposts": None, "views": None,
        "url": None,
    }
    row.update({k: v for k, v in kw.items() if k in row})
    return row


def _epoch_to_iso(seconds):
    if seconds is None:
        return None
    return time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime(int(seconds)))


#: X ships `created_at` in Twitter's own 2010-era format, e.g.
#: "Fri Sep 11 19:41:10 +0000 2026". Meta ships an epoch. Both become ISO *with
#: an explicit offset* here, so one `ts` field means one thing everywhere.
#: 🔴 Do not route this through time.strptime/strftime: %z reads the offset in
#: but struct_time cannot carry it back out, so the stamp emerges naked and
#: JS Date.parse() then reads it as local time — every X row silently shifted
#: by the machine's UTC offset.
_X_TIME = "%a %b %d %H:%M:%S %z %Y"


def _x_time_to_iso(raw):
    if not raw:
        return None
    try:
        return datetime.strptime(raw, _X_TIME).isoformat()
    except ValueError:
        return raw          # keep the platform's own string over inventing one


def _int(value):
    """Counts arrive as ints here, unlike the DOM lane's '76.9K' strings."""
    if isinstance(value, bool) or value is None:
        return None
    if isinstance(value, int):
        return value
    try:
        return int(value)
    except (TypeError, ValueError):
        return None


# ---------------------------------------------------------------------------
# extractors — one per platform, each walks its own payload shape
# ---------------------------------------------------------------------------

def _walk(node, visit):
    """Depth-first over a decoded JSON document, calling visit() on every dict.

    The payloads are deeply nested GraphQL envelopes whose paths change between
    releases; walking for a recognisable *shape* survives that, where a fixed
    path like data.home.home_timeline_urt.instructions[0] does not.
    """
    if isinstance(node, dict):
        visit(node)
        for value in node.values():
            _walk(value, visit)
    elif isinstance(node, list):
        for value in node:
            _walk(value, visit)


def extract_twitter(payload):
    rows = []

    def visit(node):
        legacy = node.get("legacy")
        if not isinstance(legacy, dict) or "full_text" not in legacy:
            return
        user = (((node.get("core") or {}).get("user_results") or {}).get("result") or {})
        handle = ((user.get("core") or {}).get("screen_name")
                  or (user.get("legacy") or {}).get("screen_name"))
        tweet_id = node.get("rest_id") or legacy.get("id_str")
        rows.append(_row(
            "twitter",
            id=tweet_id,
            author=handle,
            text=legacy.get("full_text") or "",
            ts=_x_time_to_iso(legacy.get("created_at")),
            likes=_int(legacy.get("favorite_count")),
            replies=_int(legacy.get("reply_count")),
            reposts=_int(legacy.get("retweet_count")),
            views=_int((node.get("views") or {}).get("count")),
            url=(f"https://x.com/{handle}/status/{tweet_id}"
                 if handle and tweet_id else None),
        ))

    _walk(payload, visit)
    return rows


def _meta_post(node, platform):
    """Shared shape for Threads and Instagram — both are Meta media objects."""
    user = node.get("user") or {}
    caption = node.get("caption") or {}
    handle = user.get("username")
    code = node.get("code")
    app_info = node.get("text_post_app_info") or {}
    if platform == "threads":
        url = f"https://www.threads.com/@{handle}/post/{code}" if handle and code else None
    else:
        url = f"https://www.instagram.com/p/{code}/" if code else None
    return _row(
        platform,
        id=node.get("pk") or code,
        author=handle,
        text=(caption.get("text") if isinstance(caption, dict) else None) or "",
        ts=_epoch_to_iso(node.get("taken_at")),
        likes=_int(node.get("like_count")),
        replies=_int(app_info.get("direct_reply_count") if app_info
                     else node.get("comment_count")),
        views=None,
        url=url,
    )


def _extract_meta(payload, platform):
    rows, seen = [], set()

    def visit(node):
        if "pk" not in node or "user" not in node:
            return
        if "caption" not in node and "like_count" not in node:
            return
        key = node.get("pk") or node.get("code")
        if key in seen:
            return
        seen.add(key)
        rows.append(_meta_post(node, platform))

    _walk(payload, visit)
    return rows


def extract_threads(payload):
    return _extract_meta(payload, "threads")


def extract_instagram(payload):
    return _extract_meta(payload, "ig")


# ---------------------------------------------------------------------------
# sources
# ---------------------------------------------------------------------------

#: platform -> (tab match, url to open, response-URL substrings, extractor)
NETWORK_SOURCES = {
    "twitter": ("x.com", "https://x.com/home",
                ("/i/api/graphql/",), extract_twitter),
    "threads": ("threads.com", "https://www.threads.com/",
                ("/graphql/query", "/api/graphql"), extract_threads),
}

#: platform -> (tab match, url to open, inline-script marker, extractor)
DOCUMENT_SOURCES = {
    "ig": ("instagram.com", "https://www.instagram.com/",
           "xdt_api__v1__feed__timeline", extract_instagram),
}


async def _capture_network(cdp, platform, seconds):
    match, url, patterns, extract = NETWORK_SOURCES[platform]
    tab, _ = await cdp.get_or_open_tab(url, match, activate=False)
    events = cdp._events[tab.session_id]
    while not events.empty():          # drop anything from a previous caller
        events.get_nowait()

    await tab.send("Network.enable")
    await tab.eval("window.onbeforeunload = null")
    await tab.navigate(url)

    pending, best = {}, None
    deadline = time.time() + seconds
    while time.time() < deadline:
        try:
            method, params = await asyncio.wait_for(events.get(), timeout=EVENT_POLL)
        except asyncio.TimeoutError:
            continue

        if method == "Network.responseReceived":
            response_url = params["response"]["url"]
            if any(p in response_url for p in patterns):
                pending[params["requestId"]] = response_url
            continue

        # A body is only readable once loading has FINISHED. encodedDataLength
        # at responseReceived time reported 149 bytes for a response that was
        # 22,697 characters — that is reading a letter still in the post.
        if method == "Network.loadingFinished" and params["requestId"] in pending:
            try:
                body = await tab.send(
                    "Network.getResponseBody", {"requestId": params["requestId"]}
                )
            except CDPError:
                continue          # buffer already evicted; a later one will do
            text = body.get("body") or ""
            if not text.lstrip().startswith(("{", "[")):
                continue
            try:
                rows = extract(json.loads(text))
            except (ValueError, TypeError):
                continue
            if rows and (best is None or len(rows) > len(best)):
                best = rows
                log(f"{platform}: {len(rows)} rows from "
                    f"{pending[params['requestId']].split('?')[0][-46:]}")

    if best is None:
        raise CDPError(
            f"capture: nothing matched {patterns} on {platform} within {seconds}s "
            f"— the listener may have been armed after the request fired"
        )
    return best


async def _capture_document(cdp, platform, seconds):
    match, url, marker, extract = DOCUMENT_SOURCES[platform]
    tab, _ = await cdp.get_or_open_tab(url, match, activate=False)
    await tab.eval("window.onbeforeunload = null")
    await tab.navigate(url)

    find = (
        "(() => { const s = [...document.querySelectorAll('script')]"
        ".map(x => x.textContent || '');"
        " return s.find(x => x.includes(%s)) || ''; })()" % json.dumps(marker)
    )
    deadline = time.time() + seconds
    raw = ""
    while time.time() < deadline:
        raw = await tab.eval(find)
        if raw:
            break
        await asyncio.sleep(0.5)

    if not raw:
        raise CDPError(
            f"capture: no inline script containing {marker!r} on {platform} "
            f"within {seconds}s"
        )
    try:
        rows = extract(json.loads(raw))
    except (ValueError, TypeError) as exc:
        raise CDPError(
            f"capture: {platform} inline script did not parse as JSON: {exc}"
        ) from exc
    if not rows:
        raise CDPError(
            f"capture: {platform} inline script parsed but held no posts "
            f"({len(raw)} chars) — the payload shape moved"
        )
    log(f"{platform}: {len(rows)} rows from inline script ({len(raw)} chars)")
    return rows


PLATFORMS = sorted(set(NETWORK_SOURCES) | set(DOCUMENT_SOURCES))


async def capture_feed(cdp, platform, seconds=LISTEN_SECONDS):
    """Structured rows for one platform. Raises rather than returning [].

    TikTok and YouTube are absent because they have not been tried on this lane
    — absent, not broken. Asking for one says so instead of guessing.
    """
    if platform in NETWORK_SOURCES:
        return await _capture_network(cdp, platform, seconds)
    if platform in DOCUMENT_SOURCES:
        return await _capture_document(cdp, platform, seconds)
    raise ValueError(
        f"capture_feed: no capture source for {platform!r}; have {PLATFORMS}"
    )


async def capture_all(cdp, seconds=LISTEN_SECONDS):
    """Every platform that has a source, in parallel.

    Read-only and unpaced, same reasoning as the parallel login pass: the
    PostQueue gap exists for writes.

    Returns (rows, errors). A platform in `errors` kept none of its old data —
    the caller decides what to do with what it already had.
    """
    async def one(name):
        try:
            return name, await capture_feed(cdp, name, seconds), None
        except Exception as exc:  # noqa: BLE001 — one dead platform is not six
            log(f"{name}: ERROR {type(exc).__name__}: {exc}")
            return name, None, f"{type(exc).__name__}: {exc}"

    rows, errors = {}, {}
    for name, got, err in await asyncio.gather(*(one(p) for p in PLATFORMS)):
        if err:
            errors[name] = err
        else:
            rows[name] = got
    return rows, errors