From 22e63ebf35f47f93c4266b1ba234c0dae4d9c9fc Mon Sep 17 00:00:00 2001 From: Muhammad Aizat Date: Sun, 13 Sep 2026 01:58:36 +0800 Subject: [PATCH] v0.5: capture lane, sqlite store, refresher wired to both MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit capture.py — read the platform's own JSON instead of the pixels it painted. Network capture on X (/i/api/graphql HomeLatestTimeline) and Threads (/graphql/query feedData); in-document read on Instagram, whose feed is prefetched into a RelayPrefetchedStreamCache script and never crosses the wire. 190 X rows in 0.9s against the DOM lane's 10 in 11.1s, with exact integer counts and view numbers the DOM never exposes. Waits for Network.loadingFinished, not responseReceived — the body is not there yet at responseReceived. Nothing captured raises; an empty list would say the platform posted nothing when the truth is our listener missed it. store.py — SQLite, stdlib only. platform_checks / items / metrics. id = sha1(platform|kind|url or text), first_seen written once and ts under COALESCE so a derived Threads marker cannot decay across passes. FEED_PER_PLATFORM = 10 caps any one platform before the merge, or X's ~185 rows a pass sort Instagram's 3 off the page entirely. dashboard.json is now a derived view of the database; dashboard.html did not have to change for it. refresher.py — check_all runs the background-safe platforms in parallel (19.0s sequential to 6.1s), tiktok still sequential and foreground. --skip-foreground leaves it out of a timed pass and store.latest_checks keeps returning its last real result with its own older checkedAt; stale is honest, a fabricated row is not. Browser down stays the third state and exits 0. .gitignore: socialmag.db (regenerable), deploy/kdmirah/ (carries the box's VNC password and addresses — stays in the lab, out of git). Co-Authored-By: Claude Opus 5 (1M context) --- .gitignore | 2 + capture.py | 354 +++++++++++++++++++++++++++++++++++++++++++++++++++ refresher.py | 285 +++++++++++++++++++++++++++++++++++++++++ store.py | 322 ++++++++++++++++++++++++++++++++++++++++++++++ 4 files changed, 963 insertions(+) create mode 100644 capture.py create mode 100644 refresher.py create mode 100644 store.py diff --git a/.gitignore b/.gitignore index cc00901..8143aa8 100644 --- a/.gitignore +++ b/.gitignore @@ -2,3 +2,5 @@ __pycache__/ *.pyc state.json Archived/ +socialmag.db +deploy/kdmirah/ diff --git a/capture.py b/capture.py new file mode 100644 index 0000000..92f106a --- /dev/null +++ b/capture.py @@ -0,0 +1,354 @@ +"""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 diff --git a/refresher.py b/refresher.py new file mode 100644 index 0000000..3d22e74 --- /dev/null +++ b/refresher.py @@ -0,0 +1,285 @@ +#!/usr/bin/env python3 +"""socialmag.refresher — one pass of check_login for every platform, merged +into dashboard.json. + +Run as a one-shot; cadence belongs to a systemd .timer, not to a sleep loop. + + python3 -m socialmag.refresher + +Three states per platform, never two: + + {"loggedIn": true, "checkedAt": ...} signed in + {"loggedIn": false, "checkedAt": ...} signed out + {"error": "...", "checkedAt": ...} could not check + +"Brave is not up" is the third state, not the second. Collapsing them says an +account died when the browser simply was not running. dashboard.html already +reads all three (`loginState()` — a row with no boolean `loggedIn` and an +`error` renders as unknown). + +Owned here: `platforms`, and — unless `--no-content` is passed — `notifications` +(Threads Activity) and `feed` (the X timeline, via `capture`). `posts` is left +exactly as found, because nothing but a real `create_post` can produce it. + +Flags: + + --no-content login states only, no feed or activity read + --skip-foreground leave out platforms that still activate a tab, so a + timed pass never pulls focus (today: tiktok) + +Threads returns no timestamp, only a relative marker in the row text ("11h"). +Those rows get a `ts` derived from that marker against fetch time, flagged +`"_tsDerived": true`. A row with no marker gets no `ts` at all rather than a +guessed one — and `store` writes `ts` once and never rewrites it, so the +derivation cannot decay across passes. + +This activates each tab it checks, which pulls the browser's focus — the same +cost every socialmag call pays (see README, "Tab reuse and focus"). On a +dedicated box nobody is watching that display; do not put this on a timer on a +desktop someone is using. +""" + +import asyncio +import json +import os +import re +import sys +import time + +from .cdp import CDPClient +from .queue import PostQueue +from . import capture, store +from .platforms import threads, twitter # noqa: F401 (twitter kept for the post lane) +from .server import CDP_ENDPOINT, MODULES, STATE_PATH # noqa: F401 (STATE_PATH unused, kept explicit) + +DASHBOARD_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "dashboard.json") + +# Threads writes age, not time: "11h", "2d", "45m". This is the only clock the +# Activity page gives us. +REL_MARKER = re.compile(r"\b(\d+)\s*([smhdw])\b") +REL_SECONDS = {"s": 1, "m": 60, "h": 3600, "d": 86400, "w": 604800} + + +def derive_ts(text, fetched_epoch): + """ISO stamp implied by a relative marker, or None if the row has none.""" + match = REL_MARKER.search(text or "") + if not match: + return None + seconds = int(match.group(1)) * REL_SECONDS[match.group(2)] + return time.strftime("%Y-%m-%dT%H:%M:%S%z", time.localtime(fetched_epoch - seconds)) + + +def log(*args): + print(f"[refresher {time.strftime('%H:%M:%S')}]", *args, file=sys.stderr, flush=True) + + +def now(): + return time.strftime("%Y-%m-%dT%H:%M:%S%z") + + +def load_dashboard(): + """Existing dashboard.json, or a fresh skeleton if there is none. + + A file that exists but does not parse is an error, not an excuse to start + over — overwriting it would destroy posts/notifications/feed that nothing + else holds. + """ + if not os.path.exists(DASHBOARD_PATH): + log(f"no {DASHBOARD_PATH} yet — creating one") + return {"platforms": {}, "posts": [], "notifications": [], "feed": []} + with open(DASHBOARD_PATH) as fh: + return json.load(fh) + + +def write_dashboard(data): + """Atomic: a poller reading every 10s must never see half a file.""" + tmp = DASHBOARD_PATH + ".tmp" + with open(tmp, "w") as fh: + json.dump(data, fh, indent=2) + fh.write("\n") + os.replace(tmp, DASHBOARD_PATH) + + +#: TikTok is the one platform whose check still activates its tab, so it cannot +#: run beside the others without yanking focus mid-pass. Everything else reads a +#: JS-state or background-safe signal and goes in one `gather`. +FOREGROUND_ONLY = {"tiktok"} + + +async def check_all(skip_foreground=False): + """{platform: row} for every platform. Raises only if CDP itself is gone — + the caller turns that into the third state for all six at once. + + Background-safe checks run **in parallel**: CDP multiplexes one WebSocket + with a sessionId per tab, and `cdp.py` already keys its pending futures by + message id, so concurrent calls need nothing new. Measured 2026-09-12: + 19.0s sequential → 6.1s parallel across six platforms. + + The queue's 3-6s inter-platform gap is deliberately **not** applied here. + That pacing exists for the post flows, which write; these only read, and six + tabs restoring at once is an ordinary browser thing to do. Posting stays + sequential and paced — see PostQueue. + + `skip_foreground=True` leaves out the platforms that still activate a tab, + so a timed pass never pulls focus. Those platforms are simply not checked + this pass; `store.latest_checks` keeps returning their last real result with + its own older `checkedAt`. Stale is honest. A fabricated row is not. + """ + rows = {} + cdp = CDPClient(CDP_ENDPOINT) + await cdp.connect() + log(f"connected to {cdp.browser_version} at {CDP_ENDPOINT}") + + async def check(name): + try: + ok = bool(await MODULES[name].check_login(cdp)) + log(f"{name}: {'in' if ok else 'OUT'}") + return name, {"loggedIn": ok, "checkedAt": now()} + except Exception as exc: # noqa: BLE001 — a failed check is not a False + log(f"{name}: ERROR {type(exc).__name__}: {exc}") + return name, {"error": f"{type(exc).__name__}: {exc}", "checkedAt": now()} + + try: + parallel = sorted(set(MODULES) - FOREGROUND_ONLY) + for name, row in await asyncio.gather(*(check(n) for n in parallel)): + rows[name] = row + foreground = sorted(set(MODULES) & FOREGROUND_ONLY) + if skip_foreground: + log(f"skipping (needs foreground): {', '.join(foreground)}") + else: + for name in foreground: + _, rows[name] = await check(name) + finally: + await cdp.close() + return rows + + +async def fetch_content(count=10): + """Fresh Threads activity and X timeline. + + The X timeline comes from `capture` — the platform's own GraphQL JSON, read + off the network. That is 184 rows in about a second against the DOM lane's + 10 in eleven, with exact integer counts and view numbers the DOM never + exposes. Threads activity is still a DOM scrape: capture covers the Threads + *feed*, and the Activity page is a different surface nobody has probed. + + Returns (rows_by_key, errors). A key missing from rows_by_key means that + read failed — the caller keeps whatever was in dashboard.json rather than + replacing real data with an empty list, which would read as "no activity". + """ + queue = await PostQueue().start() + rows, errors = {}, {} + cdp = CDPClient(CDP_ENDPOINT) + await cdp.connect() + try: + try: + fetched = time.time() + raw = await queue.submit( + "notif:threads", lambda: threads.get_notifications(cdp, count=count) + ) + stamp = now() + out = [] + for item in raw or []: + row = { + "platform": "threads", + "text": item.get("text", ""), + "url": item.get("url"), + "fetchedAt": stamp, + } + ts = derive_ts(row["text"], fetched) + if ts: + row["ts"] = ts + row["_tsDerived"] = True + out.append(row) + rows["notifications"] = out + log(f"threads activity: {len(out)} rows" + f" ({sum(1 for r in out if 'ts' not in r)} with no time marker)") + except Exception as exc: # noqa: BLE001 — a failed scrape must not erase good rows + errors["notifications"] = f"{type(exc).__name__}: {exc}" + log(f"threads activity: ERROR {errors['notifications']}") + + # Every platform capture lands in one `feed`, tagged by platform. The + # feed is cross-platform, not "Twitter intel" — a row knows where it + # came from, so the page can show them together without merging away + # which is which. Everything captured goes to the store; trimming + # belongs to the view (store.VIEW_LIMIT), not to the record. + captured, capture_errors = await capture.capture_all(cdp) + if captured: + merged = [] + for platform in sorted(captured): + merged.extend(captured[platform]) + log(f"capture {platform}: {len(captured[platform])} rows") + rows["feed"] = merged + if capture_errors: + # Partial is not failure. Platforms that answered are kept; the ones + # that did not are named, so a missing rail is never a silent one. + errors["feed"] = "; ".join(f"{k}: {v}" for k, v in capture_errors.items()) + if not captured: + log(f"capture: every platform failed — {errors['feed']}") + finally: + await queue.stop() + await cdp.close() + return rows, errors + + +async def main(): + with_content = "--no-content" not in sys.argv + skip_foreground = "--skip-foreground" in sys.argv + data = load_dashboard() + try: + rows = await check_all(skip_foreground=skip_foreground) + reachable = True + except Exception as exc: # noqa: BLE001 — browser down is a state, not a crash + reason = f"{type(exc).__name__}: {exc}" + log(f"CDP unreachable at {CDP_ENDPOINT} — {reason}") + stamp = now() + rows = { + name: {"error": f"browser unreachable at {CDP_ENDPOINT}: {reason}", + "checkedAt": stamp} + for name in sorted(MODULES) + } + reachable = False + + db = store.open_db() + store.record_checks(db, rows) + + content_errors = {} + if with_content and reachable: + fresh, content_errors = await fetch_content() + for key, kind in (("feed", "feed"), ("notifications", "notification")): + if key in fresh: + added, seen_again = store.record_items(db, kind, fresh[key]) + log(f"store {kind}: {added} new, {seen_again} seen again") + elif with_content: + content_errors = {"notifications": "browser unreachable", "feed": "browser unreachable"} + + # The database is the record; dashboard.json is a picture of it. Anything the + # store does not own (posts, _note) is carried through from the old file. + data = store.snapshot(db, previous=data) + data["_refreshed"] = { + "at": now(), + "by": "socialmag.refresher", + "cdp": CDP_ENDPOINT, + "browserReachable": reachable, + "store": store.stats(db), + } + if content_errors: + data["_refreshed"]["contentErrors"] = content_errors + data.pop("_yt_caveat", None) # the activate fix made this stale; live data replaces it + write_dashboard(data) + db.close() + + counts = {"in": 0, "out": 0, "unknown": 0} + for row in rows.values(): + counts["in" if row.get("loggedIn") is True + else "out" if row.get("loggedIn") is False + else "unknown"] += 1 + log(f"wrote {DASHBOARD_PATH} — in={counts['in']} out={counts['out']} unknown={counts['unknown']}") + + # Exit 0 even when the browser is down: a waiting state is not a failure, + # and Restart=on-failure must not crash-loop before Brave comes up. + return 0 + + +if __name__ == "__main__": + sys.exit(asyncio.run(main())) diff --git a/store.py b/store.py new file mode 100644 index 0000000..93954f4 --- /dev/null +++ b/store.py @@ -0,0 +1,322 @@ +"""socialmag.store — the durable half. SQLite, stdlib only, one file. + +Not here for size. `dashboard.json` is 4KB and would hold that forever. It is +here because **every refresh replaced `feed` and `notifications` wholesale**, so: + +* there was no history — "how often does a session drop?", the exact question + README's VPS section says we need answered, was unanswerable; +* there was no concept of *new* — three Threads rows aged `11h` to `1d` and + nothing knew they were the same three rows; +* 🔴 the derived `ts` decayed. A row sank an hour every hour. + +`first_seen` fixes the third one outright: it is a timestamp **we own**, exact +and fixed. Threads' missing clock stops being a problem instead of being patched. + + from socialmag import store + db = store.open_db() + store.record_checks(db, rows) + store.record_items(db, "feed", captured_rows) + data = store.snapshot(db, previous) # -> the dashboard.json view + +`dashboard.json` becomes a **derived view**, so `dashboard.html` does not change. +The database is the record; the JSON is a picture of it. +""" + +import hashlib +import json +import os +import sqlite3 +import time + +DB_PATH = os.path.join(os.path.dirname(os.path.abspath(__file__)), "socialmag.db") + +#: How many rows of each kind the derived dashboard.json carries. The database +#: keeps everything; this is only what the page is asked to render. +VIEW_LIMIT = 25 + +#: Most rows one platform may contribute to the feed view. X returns ~185 fresh +#: rows a pass, Instagram 3 older ones — without a cap the "cross-platform feed" +#: is an X feed with the others sorted off the end. +FEED_PER_PLATFORM = 10 + +SCHEMA = """ +CREATE TABLE IF NOT EXISTS platform_checks ( + platform TEXT NOT NULL, + checked_at TEXT NOT NULL, + logged_in INTEGER, -- 1 / 0, NULL when the check itself failed + error TEXT, -- set only for the third state + PRIMARY KEY (platform, checked_at) +); + +CREATE TABLE IF NOT EXISTS items ( + id TEXT PRIMARY KEY, -- sha1(platform|kind|url or text) + platform TEXT NOT NULL, + kind TEXT NOT NULL, -- 'feed' | 'notification' | 'post' + author TEXT, + text TEXT, + ts TEXT, -- the platform's own time, when it gives one + url TEXT, + likes INTEGER, + replies INTEGER, + reposts INTEGER, + views INTEGER, + first_seen TEXT NOT NULL, -- ours. exact, fixed, never guessed + last_seen TEXT NOT NULL, + times_seen INTEGER NOT NULL +); + +CREATE INDEX IF NOT EXISTS items_by_kind ON items(kind, first_seen DESC); + +-- One row per item per pass where a count actually moved. This is the +-- "snapshot history" a paid product sells; it costs a WHERE clause. +CREATE TABLE IF NOT EXISTS metrics ( + item_id TEXT NOT NULL, + seen_at TEXT NOT NULL, + likes INTEGER, + replies INTEGER, + reposts INTEGER, + views INTEGER, + PRIMARY KEY (item_id, seen_at) +); +""" + +COUNTS = ("likes", "replies", "reposts", "views") + + +def now(): + return time.strftime("%Y-%m-%dT%H:%M:%S%z") + + +def open_db(path=DB_PATH): + db = sqlite3.connect(path) + db.row_factory = sqlite3.Row + db.executescript(SCHEMA) + return db + + +def item_id(platform, kind, row): + """Stable identity for a row that the platform may not have given one. + + Prefer the platform's own id, then the permalink, then the text. Text is the + last resort and it is the reason Threads' three unchanging notification rows + collapse to three records instead of three per pass. + """ + seed = row.get("id") or row.get("url") or row.get("text") or "" + if not seed: + raise ValueError(f"store: {platform} {kind} row has no id, url or text: {row!r}") + digest = hashlib.sha1(f"{platform}|{kind}|{seed}".encode("utf-8")) + return digest.hexdigest() + + +# --------------------------------------------------------------------------- +# writes +# --------------------------------------------------------------------------- + +def record_checks(db, rows): + """One login-state row per platform per pass. Three states, all kept. + + `rows` is refresher's {platform: {loggedIn|error, checkedAt}}. + """ + stamp = now() + with db: + for platform, row in rows.items(): + logged_in = row.get("loggedIn") + db.execute( + "INSERT OR REPLACE INTO platform_checks" + " (platform, checked_at, logged_in, error) VALUES (?, ?, ?, ?)", + (platform, + row.get("checkedAt") or stamp, + None if logged_in is None else int(bool(logged_in)), + row.get("error")), + ) + return len(rows) + + +def record_items(db, kind, rows): + """Upsert rows, and append a metrics snapshot only when a count moved. + + Returns (new, updated). A row already in the table keeps its `first_seen` — + that is the whole point of the table. + """ + stamp = now() + new = updated = 0 + with db: + for row in rows: + platform = row.get("platform") + if not platform: + raise ValueError(f"store: {kind} row has no platform: {row!r}") + key = item_id(platform, kind, row) + counts = {c: row.get(c) for c in COUNTS} + existing = db.execute( + "SELECT likes, replies, reposts, views, times_seen FROM items WHERE id = ?", + (key,), + ).fetchone() + + if existing is None: + db.execute( + "INSERT INTO items (id, platform, kind, author, text, ts, url," + " likes, replies, reposts, views, first_seen, last_seen, times_seen)" + " VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,1)", + (key, platform, kind, row.get("author"), row.get("text"), + row.get("ts"), row.get("url"), + counts["likes"], counts["replies"], counts["reposts"], counts["views"], + stamp, stamp), + ) + new += 1 + moved = any(v is not None for v in counts.values()) + else: + # 🔴 `ts` is written once and never rewritten. A Threads + # notification's time is derived from a relative marker ("1d"), + # so re-deriving it every pass moves the row forward an hour + # every hour — exactly the decay this table exists to stop. The + # first derivation was made closest to first sighting and is the + # most accurate one we will ever have. + db.execute( + "UPDATE items SET author=?, text=?, ts=COALESCE(ts, ?), url=?," + " likes=?, replies=?, reposts=?, views=?," + " last_seen=?, times_seen=times_seen+1 WHERE id=?", + (row.get("author"), row.get("text"), row.get("ts"), row.get("url"), + counts["likes"], counts["replies"], counts["reposts"], counts["views"], + stamp, key), + ) + updated += 1 + moved = any(counts[c] is not None and counts[c] != existing[c] + for c in COUNTS) + + if moved: + db.execute( + "INSERT OR REPLACE INTO metrics" + " (item_id, seen_at, likes, replies, reposts, views)" + " VALUES (?,?,?,?,?,?)", + (key, stamp, counts["likes"], counts["replies"], + counts["reposts"], counts["views"]), + ) + return new, updated + + +# --------------------------------------------------------------------------- +# reads +# --------------------------------------------------------------------------- + +def latest_checks(db): + """The newest check per platform, in refresher's own row shape.""" + rows = {} + for r in db.execute( + "SELECT platform, checked_at, logged_in, error FROM platform_checks" + " WHERE (platform, checked_at) IN" + " (SELECT platform, MAX(checked_at) FROM platform_checks GROUP BY platform)" + ): + row = {"checkedAt": r["checked_at"]} + if r["error"]: + row["error"] = r["error"] + else: + row["loggedIn"] = bool(r["logged_in"]) + rows[r["platform"]] = row + return rows + + +def recent_items(db, kind, limit=VIEW_LIMIT, per_platform=None): + """Newest first by the platform's own time, falling back to when we met it. + + A row with no `ts` — every Threads notification — sorts by `first_seen`, + which is exact rather than derived from a decaying "1d" marker. + + `per_platform` caps how many rows any one platform may contribute before + the merge. Without it a pure time sort is not a cross-platform feed: X + returns ~185 fresh rows a pass against Instagram's 3 older ones, so IG never + reaches the page at all. The cap is representation, not ranking — rows still + sort by time against each other afterwards. + """ + if per_platform: + platforms = [r["platform"] for r in db.execute( + "SELECT DISTINCT platform FROM items WHERE kind = ?", (kind,))] + rows = [] + for platform in platforms: + rows.extend(db.execute( + "SELECT * FROM items WHERE kind = ? AND platform = ?" + " ORDER BY COALESCE(ts, first_seen) DESC LIMIT ?", + (kind, platform, per_platform), + ).fetchall()) + rows.sort(key=lambda r: r["ts"] or r["first_seen"], reverse=True) + rows = rows[:limit] + else: + rows = db.execute( + "SELECT * FROM items WHERE kind = ?" + " ORDER BY COALESCE(ts, first_seen) DESC LIMIT ?", + (kind, limit), + ).fetchall() + + out = [] + for r in rows: + row = { + "platform": r["platform"], "author": r["author"], "text": r["text"], + "url": r["url"], "firstSeen": r["first_seen"], "timesSeen": r["times_seen"], + } + if r["ts"]: + row["ts"] = r["ts"] + else: + # No platform clock. Say which stamp this is instead of passing our + # own off as theirs. + row["ts"] = r["first_seen"] + row["_tsIsFirstSeen"] = True + for c in COUNTS: + if r[c] is not None: + row[c] = r[c] + out.append(row) + return out + + +def uptime(db, platform, days=7): + """(checks, drops) for one platform — the number the VPS question needs. + + A "drop" is a check that came back signed-out. Errors are not drops: a + browser that was not running says nothing about the account. + """ + since = time.strftime( + "%Y-%m-%dT%H:%M:%S%z", time.localtime(time.time() - days * 86400) + ) + row = db.execute( + "SELECT COUNT(*) AS checks," + " SUM(CASE WHEN logged_in = 0 THEN 1 ELSE 0 END) AS drops" + " FROM platform_checks WHERE platform = ? AND checked_at >= ?" + " AND error IS NULL", + (platform, since), + ).fetchone() + return row["checks"], (row["drops"] or 0) + + +def snapshot(db, previous=None, limit=VIEW_LIMIT): + """Build the dashboard.json view from the database. + + `previous` carries forward keys the store does not own — `posts`, which only + a real create_post can produce, and `_note`. + """ + data = dict(previous or {}) + data["platforms"] = latest_checks(db) + data["feed"] = recent_items(db, "feed", limit, per_platform=FEED_PER_PLATFORM) + data["notifications"] = recent_items(db, "notification", limit) + data.setdefault("posts", []) + return data + + +def stats(db): + """One line per table — what is actually in there.""" + return { + "checks": db.execute("SELECT COUNT(*) FROM platform_checks").fetchone()[0], + "items": db.execute("SELECT COUNT(*) FROM items").fetchone()[0], + "metrics": db.execute("SELECT COUNT(*) FROM metrics").fetchone()[0], + "kinds": { + r["kind"]: r["n"] for r in db.execute( + "SELECT kind, COUNT(*) AS n FROM items GROUP BY kind" + ) + }, + } + + +if __name__ == "__main__": + db = open_db() + print(json.dumps(stats(db), indent=2)) + for platform in sorted({r["platform"] for r in db.execute( + "SELECT DISTINCT platform FROM platform_checks")}): + checks, drops = uptime(db, platform) + print(f" {platform:<8} {checks:>4} checks, {drops} signed-out")