socialmag/store.py

"""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")