juxtapo/socialmag
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.5: capture lane, sqlite store, refresher wired to both

Browse Source 
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) <noreply@anthropic.com>
This commit is contained in:
juxtapo
2026-09-13 01:58:36 +08:00
co-authored by Claude Opus 5
parent 5419b08264
commit 22e63ebf35
4 changed files with 963 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2
View File
@@ -2,3 +2,5 @@ __pycache__/
*.pyc
state.json
Archived/
socialmag.db
deploy/kdmirah/
capture.py
+354
View File
@@ -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
refresher.py
+285
View File
@@ -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()))
store.py
+322
View File
@@ -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")