#!/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()))