git / moonbite-network / forum.py · raw
"""MoonBite community forum — a small SQLite-backed discussion board.Threaded discussions with free-text display names (no accounts, no passwords).The server stores raw text only; every value is HTML-escaped at render time byJinja's autoescaping, so stored posts can never inject markup into a page.Storage is a single SQLite file (MOONBITE_FORUM_DB, default ``forum.db``). On anephemeral filesystem (e.g. Railway) the file resets on redeploy; on the dropletit persists. SQLite is stdlib, so this adds no dependency. WAL mode lets theread-heavy board tolerate concurrent gunicorn workers."""from __future__ import annotationsimport osimport reimport sqlite3import timefrom typing import Optionalfrom markupsafe import Markup, escapeimport forum_emoji# --- limits (also enforced client-side, but the server is the source of truth) #MAX_TITLE = 120MAX_AUTHOR = 40MAX_BODY = 8000MIN_BODY = 1DEFAULT_AUTHOR = "anon"MAX_QUERY = 80# Fixed board sections. The slug is what's stored; order is display order.CATEGORIES = [ {"slug": "general", "name": "General", "desc": "Anything about MoonBite that doesn't fit elsewhere."}, {"slug": "mining", "name": "Mining", "desc": "Setting up the miner, hashrate, finding blocks."}, {"slug": "wallets", "name": "Wallets", "desc": "Addresses, sending, backups and recovery."}, {"slug": "nodes", "name": "Nodes & code", "desc": "Running MoonBite Core, building from source, bugs."}, {"slug": "ideas", "name": "Ideas", "desc": "Proposals and what the project should do next."}, {"slug": "network", "name": "Network", "official_only": True, "desc": "Automatic reports from the verified MoonBite account, built from the live chain."},]CATEGORY_NAMES = {c["slug"]: c["name"] for c in CATEGORIES}OFFICIAL_ONLY = {c["slug"] for c in CATEGORIES if c.get("official_only")}# The verified moderator account, and names people could use to pass as the# project. Nobody can post under these from the web; only server-side code# (official=True, e.g. forum_bot.py) posts as OFFICIAL_NAME, and pages show# those posts with a verified badge.OFFICIAL_NAME = "MoonBite"_RESERVED_EXACT = {"admin", "administrator", "mod", "moderator", "official", "staff", "support", "team", "bot", "verified"}def _reserved(name: str) -> bool: key = re.sub(r"[^a-z0-9]", "", name.lower()) return "moonbite" in key or key in _RESERVED_EXACTDEFAULT_CATEGORY = "general"SORTS = ("hot", "new", "top", "unanswered")COMMENT_SORTS = ("best", "new", "old")MAX_DEPTH = 8 # replies deeper than this attach to the deepest allowed parent_HOT_WINDOW = 1000 # hot ranking scores the newest N matching threads in Python_DB_PATH = os.environ.get("MOONBITE_FORUM_DB", "").strip() or "forum.db"def _connect() -> sqlite3.Connection: """Open a short-lived connection with the schema guaranteed to exist. Each request opens and closes its own connection; WAL mode keeps concurrent readers from blocking on a writer. `CREATE TABLE IF NOT EXISTS` makes first use self-initialising, so there is no separate migration step to forget. """ conn = sqlite3.connect(_DB_PATH, timeout=5.0) conn.row_factory = sqlite3.Row conn.execute("PRAGMA journal_mode=WAL") conn.execute("PRAGMA foreign_keys=ON") conn.execute( """CREATE TABLE IF NOT EXISTS threads ( id INTEGER PRIMARY KEY AUTOINCREMENT, title TEXT NOT NULL, author TEXT NOT NULL, body TEXT NOT NULL, created_at INTEGER NOT NULL, bumped_at INTEGER NOT NULL, reply_count INTEGER NOT NULL DEFAULT 0 )""" ) conn.execute( """CREATE TABLE IF NOT EXISTS replies ( id INTEGER PRIMARY KEY AUTOINCREMENT, thread_id INTEGER NOT NULL, author TEXT NOT NULL, body TEXT NOT NULL, created_at INTEGER NOT NULL, FOREIGN KEY (thread_id) REFERENCES threads(id) ON DELETE CASCADE )""" ) # One vote per target per voter. `net` is a salted hash of the voter's # network address, so clearing cookies on the same connection doesn't # buy a second vote; see vote(). conn.execute( """CREATE TABLE IF NOT EXISTS votes ( kind TEXT NOT NULL, target_id INTEGER NOT NULL, voter TEXT NOT NULL, net TEXT NOT NULL, value INTEGER NOT NULL, created_at INTEGER NOT NULL, PRIMARY KEY (kind, target_id, voter) )""" ) conn.execute("CREATE INDEX IF NOT EXISTS votes_net ON votes (kind, target_id, net)") # forum_bot.py's memory of what it has already posted (one row per post key). conn.execute( """CREATE TABLE IF NOT EXISTS bot_state ( key TEXT PRIMARY KEY, thread_id INTEGER, created_at INTEGER NOT NULL )""" ) _add_column(conn, "threads", "is_official", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "threads", "automated", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "replies", "is_official", "INTEGER NOT NULL DEFAULT 0") # Boards created before categories, scores and nesting: add columns in place. _add_column(conn, "threads", "category", f"TEXT NOT NULL DEFAULT '{DEFAULT_CATEGORY}'") _add_column(conn, "threads", "score", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "replies", "score", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "replies", "parent_id", "INTEGER") # Moderation: pinned/locked threads, soft-removed posts, and user reports. _add_column(conn, "threads", "pinned", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "threads", "locked", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "threads", "removed", "INTEGER NOT NULL DEFAULT 0") _add_column(conn, "replies", "removed", "INTEGER NOT NULL DEFAULT 0") conn.execute( """CREATE TABLE IF NOT EXISTS reports ( id INTEGER PRIMARY KEY AUTOINCREMENT, kind TEXT NOT NULL, target_id INTEGER NOT NULL, reason TEXT NOT NULL, net TEXT NOT NULL, created_at INTEGER NOT NULL, resolved INTEGER NOT NULL DEFAULT 0, UNIQUE (kind, target_id, net) )""" ) return conndef _add_column(conn: sqlite3.Connection, table: str, col: str, ddl: str) -> None: cols = {r["name"] for r in conn.execute(f"PRAGMA table_info({table})")} if col not in cols: conn.execute(f"ALTER TABLE {table} ADD COLUMN {col} {ddl}")def _clean(value, *, field: str, maxlen: int, minlen: int = 0, default: str | None = None) -> str: """Validate one user-supplied text field: coerce, strip, bound its length. Raises ValueError on empty required fields or overrun so the route can turn it into a clean 400 (mirrors the exchange/merchants modules' convention). """ text = str(value or "").strip() if not text and default is not None: text = default if len(text) < minlen or not text: raise ValueError(f"{field} is required") if len(text) > maxlen: raise ValueError(f"{field} must be {maxlen} characters or fewer") return textdef _iso(ts: int) -> str: """Human-readable UTC stamp for display (e.g. '2026-07-26 14:30 UTC').""" return time.strftime("%Y-%m-%d %H:%M UTC", time.gmtime(ts))def _ago(ts: int) -> str: """Short relative age for lists ('now', '5m', '3h', '2d', then a date).""" s = max(0, int(time.time()) - int(ts)) if s < 60: return "now" if s < 3600: return f"{s // 60}m" if s < 86400: return f"{s // 3600}h" if s < 30 * 86400: return f"{s // 86400}d" return time.strftime("%d %b %Y", time.gmtime(ts))def _excerpt(body: str, n: int = 160) -> str: # Skip quoted lines so a reply-style opener doesn't lead the excerpt. lines = [l for l in body.splitlines() if not l.lstrip().startswith(">")] text = " ".join(" ".join(lines).split()) or " ".join(body.split()) return text if len(text) <= n else text[: n - 1].rstrip() + "…"def avatar(name: str) -> dict: """Initials and one of four neutral tones for a display name. The tone is a stable hash of the name so a poster looks the same everywhere; it is not an identity check (names aren't verified).""" name = str(name or DEFAULT_AUTHOR).strip() or DEFAULT_AUTHOR parts = re.findall(r"[A-Za-z0-9]+", name) if len(parts) >= 2: initials = parts[0][0] + parts[1][0] elif parts: initials = parts[0][:2] else: initials = name[:1] tone = sum(ord(ch) * (i + 1) for i, ch in enumerate(name)) % 4 return {"i": initials.upper(), "t": tone}_URL = re.compile(r"https?://[^\s<>\"']+")_CODE = re.compile(r"`([^`\n]{1,200})`")def _inline(line: str) -> str: """One escaped line with `code` spans, emoji and bare http(s) links made live. Works on already-escaped text, so nothing typed becomes markup. Emoji and :shortcodes: inside `code` stay literal.""" parts = _CODE.split(str(escape(line))) # split() with one group alternates: text, code, text, code, ... out = "".join( f"<code>{p}</code>" if i % 2 else forum_emoji.emojify(p) for i, p in enumerate(parts) ) def link(m): url = m.group(0) trail = "" while url and url[-1] in ".,;:!?)": trail = url[-1] + trail url = url[:-1] return f'<a href="{url}" rel="nofollow ugc noopener" target="_blank">{url}</a>{trail}' return _URL.sub(link, out)def render_body(body: str) -> Markup: """Plain text to safe HTML: paragraphs on blank lines, '> ' lines become a quote block, single newlines are kept. Every character is escaped first.""" html, para, quote = [], [], [] def flush_para(): if para: html.append("<p>" + "<br>".join(_inline(l) for l in para) + "</p>") para.clear() def flush_quote(): if quote: html.append("<blockquote>" + "<br>".join(_inline(l) for l in quote) + "</blockquote>") quote.clear() for raw in body.replace("\r\n", "\n").split("\n"): if raw.startswith(">"): flush_para() quote.append(raw[1:].lstrip()) elif raw.strip() == "": flush_para() flush_quote() else: flush_quote() para.append(raw) flush_para() flush_quote() if forum_emoji.emoji_only(body): return Markup('<p class="fx-jumbo">' + _inline(body.strip()) + "</p>") return Markup("".join(html))def emojify_text(text: str) -> Markup: """Titles and excerpts: escape, then draw emoji (no other formatting).""" return Markup(forum_emoji.emojify(str(escape(text or ""))))def _thread_row(row: sqlite3.Row) -> dict: keys = row.keys() category = row["category"] if "category" in keys else DEFAULT_CATEGORY return { "id": row["id"], "title": row["title"], "author": row["author"], "body": row["body"], "excerpt": _excerpt(row["body"]), "category": category, "category_name": CATEGORY_NAMES.get(category, category), "created_at": row["created_at"], "created_display": _iso(row["created_at"]), "created_ago": _ago(row["created_at"]), "bumped_at": row["bumped_at"], "bumped_display": _iso(row["bumped_at"]), "bumped_ago": _ago(row["bumped_at"]), "reply_count": row["reply_count"], "score": row["score"] if "score" in keys else 0, "is_official": bool(row["is_official"]) if "is_official" in keys else False, "automated": bool(row["automated"]) if "automated" in keys else False, "pinned": bool(row["pinned"]) if "pinned" in keys else False, "locked": bool(row["locked"]) if "locked" in keys else False, "removed": bool(row["removed"]) if "removed" in keys else False, }def _reply_row(row: sqlite3.Row) -> dict: keys = row.keys() return { "id": row["id"], "thread_id": row["thread_id"], "parent_id": row["parent_id"] if "parent_id" in keys else None, "author": row["author"], "body": row["body"], "score": row["score"] if "score" in keys else 0, "is_official": bool(row["is_official"]) if "is_official" in keys else False, "removed": bool(row["removed"]) if "removed" in keys else False, "created_at": row["created_at"], "created_display": _iso(row["created_at"]), "created_ago": _ago(row["created_at"]), }def _hot(t: dict, now: int) -> float: """Votes and replies, decayed by age (HN-style gravity).""" age_h = max(0, now - t["created_at"]) / 3600 return (t["score"] + 0.5 * t["reply_count"] + 1) / (age_h + 2) ** 1.5# --- auto-moderation ------------------------------------------------------- ## Runs on every post before it is stored. The seed-phrase guard uses the same# BIP39 list the wallet uses (static/moonbite-bip39.js), so there's one source.MAX_LINKS = 5DUPLICATE_WINDOW = 3600_SEED_RUN = 12def _load_bip39() -> frozenset: path = os.path.join(os.path.dirname(os.path.abspath(__file__)), "static", "moonbite-bip39.js") try: with open(path, encoding="utf-8") as fh: m = re.search(r'WORDLIST\s*=\s*"([a-z ]+)"', fh.read()) return frozenset(m.group(1).split()) if m else frozenset() except OSError: return frozenset()_BIP39 = _load_bip39()_WIF = re.compile(r"\b[5KL][1-9A-HJ-NP-Za-km-z]{50,51}\b")_XPRV = re.compile(r"\b[xt]prv[1-9A-HJ-NP-Za-km-z]{100,}\b")def looks_like_seed(text: str) -> bool: """12+ consecutive BIP39 words, or a WIF / extended private key.""" run = 0 for w in re.findall(r"[A-Za-z]+", text): run = run + 1 if w.lower() in _BIP39 else 0 if run >= _SEED_RUN: return True return bool(_WIF.search(text) or _XPRV.search(text))def _guard(*texts: str) -> None: joined = "\n".join(texts) if looks_like_seed(joined): raise ValueError("That looks like a seed phrase or private key, so it wasn't posted. " "Never share it with anyone, here or anywhere. If you already did, " "move your coins to a new wallet now.") if len(_URL.findall(joined)) > MAX_LINKS: raise ValueError(f"Too many links. Up to {MAX_LINKS} per post.")def _norm(text: str) -> str: return " ".join(text.lower().split())_SPAM_MIN = 40 # a reply this long, repeated word for word by anyone, is spamdef _duplicate(conn, body: str, thread_id: int | None, now: int, *, title: str = "", author: str = "") -> bool: """Within the last hour: a thread with the same title and body, or in this thread a reply with the same text from the same name (or from anyone, if it's long enough that two people wouldn't type it identically).""" since = now - DUPLICATE_WINDOW norm = _norm(body) if thread_id is None: rows = conn.execute("SELECT title, body FROM threads WHERE created_at >= ?", (since,)).fetchall() return any(_norm(r["body"]) == norm and _norm(r["title"]) == _norm(title) for r in rows) rows = conn.execute("SELECT author, body FROM replies WHERE thread_id = ? AND created_at >= ?", (thread_id, since)).fetchall() return any(_norm(r["body"]) == norm and (r["author"] == author or len(norm) >= _SPAM_MIN) for r in rows)def create_thread(title, author, body, category=DEFAULT_CATEGORY, *, official: bool = False, automated: bool = False) -> dict: """Open a new discussion thread. Returns the stored thread dict. official=True is server-side only (never from a form): it posts as OFFICIAL_NAME with the verified badge and may use official-only categories. automated=True additionally tags the post as an automatic report.""" title = _clean(title, field="Title", maxlen=MAX_TITLE, minlen=1) body = _clean(body, field="Message", maxlen=MAX_BODY, minlen=MIN_BODY) category = str(category or DEFAULT_CATEGORY).strip() if category not in CATEGORY_NAMES or (category in OFFICIAL_ONLY and not official): raise ValueError("Pick a category") if official: author = OFFICIAL_NAME else: author = _clean(author, field="Name", maxlen=MAX_AUTHOR, default=DEFAULT_AUTHOR) if _reserved(author): raise ValueError("That name is reserved. Pick another.") _guard(title, body) now = int(time.time()) with _connect() as conn: if _duplicate(conn, body, None, now, title=title): raise ValueError("This was already posted in the last hour.") cur = conn.execute( "INSERT INTO threads (title, author, body, created_at, bumped_at, category, is_official, automated) " "VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (title, author, body, now, now, category, int(official), int(official and automated)), ) thread_id = cur.lastrowid row = conn.execute("SELECT * FROM threads WHERE id = ?", (thread_id,)).fetchone() return _thread_row(row)def _participants(conn, threads: list[dict]) -> None: """Attach `voices` (distinct names, most recent poster first, OP last if not already in) and `last_author` to each thread dict, in one query.""" if not threads: return ids = [t["id"] for t in threads] marks = ",".join("?" * len(ids)) rows = conn.execute( f"SELECT thread_id, author, MAX(id) AS last_id FROM replies " f"WHERE thread_id IN ({marks}) GROUP BY thread_id, author ORDER BY last_id DESC", ids, ).fetchall() by_thread: dict[int, list[str]] = {} for r in rows: by_thread.setdefault(r["thread_id"], []).append(r["author"]) for t in threads: names = by_thread.get(t["id"], []) t["last_author"] = names[0] if names else t["author"] if t["author"] not in names: names = names + [t["author"]] t["voices"] = names t["voice_count"] = len(names)def list_threads(limit: int = 20, offset: int = 0, category: str | None = None, sort: str = "hot", q: str | None = None) -> dict: """Return a page of threads plus the matching total. sort: hot (votes and replies decayed by age), new (newest first), top (highest score), unanswered (no replies yet). q matches title or body.""" limit = max(1, min(int(limit), 50)) offset = max(0, int(offset)) if sort not in SORTS: sort = "hot" # Removed threads never list; pinned ones are shown above the feed instead. where, args = ["removed = 0", "pinned = 0"], [] if category in CATEGORY_NAMES: where.append("category = ?") args.append(category) if q: needle = q.strip()[:MAX_QUERY].replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_") where.append("(title LIKE ? ESCAPE '\\' OR body LIKE ? ESCAPE '\\')") args += [f"%{needle}%", f"%{needle}%"] if sort == "unanswered": where.append("reply_count = 0") order = { "hot": "created_at DESC, id DESC", "new": "created_at DESC, id DESC", "top": "score DESC, reply_count DESC, created_at DESC, id DESC", "unanswered": "created_at DESC, id DESC", }[sort] clause = (" WHERE " + " AND ".join(where)) if where else "" with _connect() as conn: total = conn.execute(f"SELECT COUNT(*) AS n FROM threads{clause}", args).fetchone()["n"] if sort == "hot": rows = conn.execute( f"SELECT * FROM threads{clause} ORDER BY {order} LIMIT ?", args + [_HOT_WINDOW], ).fetchall() now = int(time.time()) ranked = sorted((_thread_row(r) for r in rows), key=lambda t: _hot(t, now), reverse=True) threads = ranked[offset:offset + limit] else: rows = conn.execute( f"SELECT * FROM threads{clause} ORDER BY {order} LIMIT ? OFFSET ?", args + [limit, offset], ).fetchall() threads = [_thread_row(r) for r in rows] _participants(conn, threads) return { "threads": threads, "total": total, "limit": limit, "offset": offset, }def stats() -> dict: """Board-wide counts for the sidebar and category chips.""" week_ago = int(time.time()) - 7 * 86400 with _connect() as conn: threads = conn.execute("SELECT COUNT(*) AS n FROM threads WHERE removed = 0").fetchone()["n"] replies = conn.execute("SELECT COUNT(*) AS n FROM replies WHERE removed = 0").fetchone()["n"] voices = conn.execute( "SELECT COUNT(*) AS n FROM (SELECT author FROM threads WHERE removed = 0 " "UNION SELECT author FROM replies WHERE removed = 0)" ).fetchone()["n"] week = conn.execute( "SELECT (SELECT COUNT(*) FROM threads WHERE created_at >= ? AND removed = 0) + " "(SELECT COUNT(*) FROM replies WHERE created_at >= ? AND removed = 0) AS n", (week_ago, week_ago), ).fetchone()["n"] per = {r["category"]: r["n"] for r in conn.execute( "SELECT category, COUNT(*) AS n FROM threads WHERE removed = 0 GROUP BY category" )} return { "threads": threads, "replies": replies, "posts": threads + replies, "voices": voices, "week": week, "per_category": {c["slug"]: per.get(c["slug"], 0) for c in CATEGORIES}, }def _comment_key(sort: str): if sort == "new": return lambda c: (-c["created_at"], -c["id"]) if sort == "old": return lambda c: (c["created_at"], c["id"]) return lambda c: (-c["score"], c["created_at"], c["id"]) # bestdef get_thread(thread_id: int, sort: str = "best", include_removed: bool = False) -> Optional[dict]: """Return one thread with its replies, or None if missing (or removed, unless include_removed, which moderators use). `replies` is the flat list (oldest first); `comments` is the same replies as a tree (each with `children` and `depth`), siblings ordered by `sort`. Removed replies stay in the tree (flagged) so their children keep context.""" if sort not in COMMENT_SORTS: sort = "best" with _connect() as conn: row = conn.execute("SELECT * FROM threads WHERE id = ?", (thread_id,)).fetchone() if row is None or (row["removed"] and not include_removed): return None replies = conn.execute( "SELECT * FROM replies WHERE thread_id = ? ORDER BY created_at ASC, id ASC", (thread_id,), ).fetchall() thread = _thread_row(row) _participants(conn, [thread]) flat = [_reply_row(r) for r in replies] by_id = {c["id"]: dict(c, children=[]) for c in flat} roots = [] for c in by_id.values(): parent = by_id.get(c["parent_id"]) if c["parent_id"] else None (parent["children"] if parent else roots).append(c) key = _comment_key(sort) def order(nodes, depth): nodes.sort(key=key) for n in nodes: n["depth"] = depth order(n["children"], depth + 1) order(roots, 0) thread["replies"] = flat thread["comments"] = roots return threaddef add_reply(thread_id: int, author, body, parent_id=None, *, official: bool = False) -> dict: """Append a reply to a thread (or under another reply) and bump the thread. Raises ValueError if the thread or the parent reply doesn't exist, or the thread is locked. official=True (moderators only) posts as OFFICIAL_NAME.""" if official: author = OFFICIAL_NAME else: author = _clean(author, field="Name", maxlen=MAX_AUTHOR, default=DEFAULT_AUTHOR) if _reserved(author): raise ValueError("That name is reserved. Pick another.") body = _clean(body, field="Reply", maxlen=MAX_BODY, minlen=MIN_BODY) _guard(body) now = int(time.time()) with _connect() as conn: t = conn.execute( "SELECT locked, removed FROM threads WHERE id = ?", (thread_id,) ).fetchone() if t is None or t["removed"]: raise ValueError("thread not found") if t["locked"] and not official: raise ValueError("This thread is locked. A moderator closed it to new replies.") if _duplicate(conn, body, thread_id, now, author=author): raise ValueError("You already posted that here.") if parent_id: parent_id = int(parent_id) depth, cur_id = 0, parent_id chain = [] while cur_id: p = conn.execute( "SELECT id, parent_id FROM replies WHERE id = ? AND thread_id = ?", (cur_id, thread_id), ).fetchone() if p is None: raise ValueError("That comment no longer exists") chain.append(p["id"]) cur_id = p["parent_id"] depth += 1 # Past the cap, hang the reply off the deepest allowed ancestor. if depth > MAX_DEPTH: parent_id = chain[depth - MAX_DEPTH] else: parent_id = None cur = conn.execute( "INSERT INTO replies (thread_id, author, body, created_at, parent_id, is_official) VALUES (?, ?, ?, ?, ?, ?)", (thread_id, author, body, now, parent_id, int(official)), ) reply_id = cur.lastrowid conn.execute( "UPDATE threads SET reply_count = reply_count + 1, bumped_at = ? WHERE id = ?", (now, thread_id), ) row = conn.execute("SELECT * FROM replies WHERE id = ?", (reply_id,)).fetchone() return _reply_row(row)def vote(kind: str, target_id: int, voter: str, net: str, value: int) -> dict: """Set this voter's vote on a thread ('t') or reply ('r'): 1, -1, or 0 to clear. A previous vote from the same browser OR the same network is replaced, not added to. Returns the new score and the vote now standing. This is per-browser-and-network, not per-person: there are no accounts, so votes are a signal, not a poll.""" if kind not in ("t", "r"): raise ValueError("bad vote target") value = int(value) if value not in (-1, 0, 1): raise ValueError("bad vote value") table = "threads" if kind == "t" else "replies" with _connect() as conn: if conn.execute(f"SELECT 1 FROM {table} WHERE id = ?", (target_id,)).fetchone() is None: raise ValueError("not found") conn.execute( "DELETE FROM votes WHERE kind = ? AND target_id = ? AND (voter = ? OR net = ?)", (kind, target_id, voter, net), ) if value: conn.execute( "INSERT INTO votes (kind, target_id, voter, net, value, created_at) VALUES (?, ?, ?, ?, ?, ?)", (kind, target_id, voter, net, value, int(time.time())), ) score = conn.execute( "SELECT COALESCE(SUM(value), 0) AS s FROM votes WHERE kind = ? AND target_id = ?", (kind, target_id), ).fetchone()["s"] conn.execute(f"UPDATE {table} SET score = ? WHERE id = ?", (score, target_id)) return {"score": score, "mine": value}def my_votes(voter: str, net: str, kind: str, ids) -> dict: """{target_id: value} for the votes this browser/network has standing.""" ids = [int(i) for i in ids] if not ids or not (voter or net): return {} marks = ",".join("?" * len(ids)) with _connect() as conn: rows = conn.execute( f"SELECT target_id, value FROM votes WHERE kind = ? AND (voter = ? OR net = ?) " f"AND target_id IN ({marks})", [kind, voter, net] + ids, ).fetchall() return {r["target_id"]: r["value"] for r in rows}def bot_claim(key: str) -> bool: """Reserve a bot post key. True if this run should post it; False if a previous (or concurrent) run already has. The INSERT is the lock.""" with _connect() as conn: cur = conn.execute( "INSERT OR IGNORE INTO bot_state (key, created_at) VALUES (?, ?)", (key, int(time.time())), ) return cur.rowcount == 1def bot_record(key: str, thread_id: int) -> None: with _connect() as conn: conn.execute("UPDATE bot_state SET thread_id = ? WHERE key = ?", (thread_id, key))def bot_release(key: str) -> None: """Give a claimed key back after a failed post so the next run retries.""" with _connect() as conn: conn.execute("DELETE FROM bot_state WHERE key = ? AND thread_id IS NULL", (key,))# --- moderation -------------------------------------------------------------- ## Called only from moderator-authenticated routes in web_app.py._FLAGS = {("t", "pinned"), ("t", "locked"), ("t", "removed"), ("r", "removed")}REPORT_REASONS = {"spam": "Spam or ads", "scam": "Scam or phishing", "abuse": "Abuse or harassment", "secret": "Shares a seed phrase or key", "other": "Something else"}def set_flag(kind: str, target_id: int, flag: str, on: bool) -> None: if (kind, flag) not in _FLAGS: raise ValueError("bad moderation action") table = "threads" if kind == "t" else "replies" with _connect() as conn: cur = conn.execute(f"UPDATE {table} SET {flag} = ? WHERE id = ?", (int(bool(on)), int(target_id))) if cur.rowcount != 1: raise ValueError("not found") if flag == "removed" and on: conn.execute("UPDATE reports SET resolved = 1 WHERE kind = ? AND target_id = ?", (kind, int(target_id)))def pinned_threads() -> list[dict]: with _connect() as conn: rows = conn.execute( "SELECT * FROM threads WHERE pinned = 1 AND removed = 0 ORDER BY created_at DESC LIMIT 5" ).fetchall() threads = [_thread_row(r) for r in rows] _participants(conn, threads) return threadsdef report(kind: str, target_id: int, reason: str, net: str) -> bool: """File a report. One per post per network; True if it was new.""" if kind not in ("t", "r") or reason not in REPORT_REASONS: raise ValueError("bad report") table = "threads" if kind == "t" else "replies" with _connect() as conn: if conn.execute(f"SELECT 1 FROM {table} WHERE id = ?", (int(target_id),)).fetchone() is None: raise ValueError("not found") cur = conn.execute( "INSERT OR IGNORE INTO reports (kind, target_id, reason, net, created_at) VALUES (?, ?, ?, ?, ?)", (kind, int(target_id), reason, net, int(time.time())), ) return cur.rowcount == 1def open_reports() -> list[dict]: """Unresolved reports grouped per post, most-reported first, with the post.""" with _connect() as conn: rows = conn.execute( "SELECT kind, target_id, GROUP_CONCAT(reason) AS reasons, COUNT(*) AS n, MAX(created_at) AS last " "FROM reports WHERE resolved = 0 GROUP BY kind, target_id ORDER BY n DESC, last DESC LIMIT 100" ).fetchall() out = [] for r in rows: if r["kind"] == "t": p = conn.execute("SELECT id, id AS thread_id, title, author, body, removed FROM threads WHERE id = ?", (r["target_id"],)).fetchone() else: p = conn.execute("SELECT r.id, r.thread_id, t.title, r.author, r.body, r.removed FROM replies r " "JOIN threads t ON t.id = r.thread_id WHERE r.id = ?", (r["target_id"],)).fetchone() if p is None: continue out.append({"kind": r["kind"], "id": r["target_id"], "count": r["n"], "last_ago": _ago(r["last"]), "reasons": sorted(set(r["reasons"].split(","))), "thread_id": p["thread_id"], "title": p["title"], "author": p["author"], "excerpt": _excerpt(p["body"], 200), "removed": bool(p["removed"])}) return outdef resolve_reports(kind: str, target_id: int) -> None: with _connect() as conn: conn.execute("UPDATE reports SET resolved = 1 WHERE kind = ? AND target_id = ?", (kind, int(target_id)))def recent_posts(limit: int = 40) -> list[dict]: """Newest threads and replies together, for the moderator's review list.""" with _connect() as conn: rows = conn.execute( "SELECT 't' AS kind, id, id AS thread_id, title, author, body, created_at, removed, is_official FROM threads " "UNION ALL " "SELECT 'r', r.id, r.thread_id, t.title, r.author, r.body, r.created_at, r.removed, r.is_official " "FROM replies r JOIN threads t ON t.id = r.thread_id " "ORDER BY created_at DESC LIMIT ?", (int(limit),) ).fetchall() return [{"kind": r["kind"], "id": r["id"], "thread_id": r["thread_id"], "title": r["title"], "author": r["author"], "excerpt": _excerpt(r["body"], 160), "ago": _ago(r["created_at"]), "removed": bool(r["removed"]), "is_official": bool(r["is_official"])} for r in rows]def metrics(now: int | None = None) -> dict: """This week vs last week, counting only human (non-official) posts, plus how quickly new threads get a first answer. The numbers a moderator should watch to know whether the board is actually growing.""" now = int(now if now is not None else time.time()) week = 7 * 86400 def window(conn, start, end): t = conn.execute("SELECT COUNT(*) AS n FROM threads WHERE created_at >= ? AND created_at < ? " "AND removed = 0 AND is_official = 0", (start, end)).fetchone()["n"] r = conn.execute("SELECT COUNT(*) AS n FROM replies WHERE created_at >= ? AND created_at < ? " "AND removed = 0 AND is_official = 0", (start, end)).fetchone()["n"] people = conn.execute( "SELECT COUNT(*) AS n FROM (SELECT author FROM threads WHERE created_at >= ? AND created_at < ? " "AND removed = 0 AND is_official = 0 UNION SELECT author FROM replies WHERE created_at >= ? " "AND created_at < ? AND removed = 0 AND is_official = 0)", (start, end, start, end)).fetchone()["n"] return {"threads": t, "replies": r, "posters": people} with _connect() as conn: this, last = window(conn, now - week, now + 1), window(conn, now - 2 * week, now - week) # Human threads older than a day: how many got a reply within 24h? rows = conn.execute( "SELECT t.id, t.created_at, (SELECT MIN(created_at) FROM replies r WHERE r.thread_id = t.id " "AND r.removed = 0 AND r.author != t.author) AS first FROM threads t " "WHERE t.removed = 0 AND t.is_official = 0 AND t.created_at < ? AND t.created_at >= ?", (now - 86400, now - 30 * 86400)).fetchall() waiting = conn.execute( "SELECT id, title, created_at FROM threads WHERE removed = 0 AND reply_count = 0 AND is_official = 0 " "ORDER BY created_at ASC LIMIT 20").fetchall() answered = sum(1 for r in rows if r["first"] is not None and r["first"] - r["created_at"] <= 86400) return { "this_week": this, "last_week": last, "answered_24h": (answered, len(rows)), "unanswered": [{"id": w["id"], "title": w["title"], "ago": _ago(w["created_at"])} for w in waiting], }