felhom.eu/documentation/audits/forum-research-2026-10-12/parse.py

#!/usr/bin/env python3
"""Parse the cached topic pages into posts.jsonl (in the cache folder, never in git).

Read-only research (forum-research-2026-10-12). Stdlib only.

Per post: num, date, reply_to, author_id, off, text.
- author_id = first 12 hex of sha256(salt + username). The salt is made once and kept in the cache
  folder; it is never committed.
- text = the post's OWN words. The quoted parent sits in <div class="message-content-replied"> and is
  skipped whole; only <div class="rtif-content message-content"> outside it is kept.
- off = the site marked the post off-topic (<li class="message-off">). A post with no parent has
  data-rplid="" (an earlier pattern that required digits silently dropped every such post).
- Posts before WINDOW_START are dropped; a post number seen twice (a redirected page) is kept once.

Then it runs the parser controls (spec 7 B) and prints them. Exit 1 if a control fails.
Usage: python3 parse.py
"""
import glob
import hashlib
import html
import json
import os
import random
import re
import secrets
import sys
from html.parser import HTMLParser

CACHE = os.path.expanduser("~/research-cache/prohardver-hazi-szerver")
PAGES = os.path.join(CACHE, "pages")
OUT = os.path.join(CACHE, "posts.jsonl")
SALT_FILE = os.path.join(CACHE, "salt.txt")
WINDOW_START = "2023-10-12 00:00:00"

LI_RE = re.compile(r'<li (?:class="([^"]*)" )?data-id="(\d+)" data-rplid="(\d*)">')
TIME_RE = re.compile(r'<time class="message-time">(.*?)</time>', re.S)
TAG_RE = re.compile(r"<[^>]+>")
# The site shows this year's dates without a year ("aug. 9., 12:34:56"), and "ma"/"tegnap" for
# today/yesterday. FETCH_DAY is the day the pages were fetched (fetch.log).
FETCH_DAY = (2026, 10, 12)
HU_MONTHS = {"jan.": 1, "febr.": 2, "márc.": 3, "ápr.": 4, "máj.": 5, "jún.": 6, "júl.": 7,
             "aug.": 8, "szept.": 9, "okt.": 10, "nov.": 11, "dec.": 12}


def norm_date(raw):
    """Return 'YYYY-MM-DD HH:MM:SS' or '' if the shape is unknown. Edited posts carry a pencil
    <span> inside <time>; the tags are stripped first (an earlier pattern dropped those posts)."""
    import datetime
    t = " ".join(TAG_RE.sub(" ", raw).split())
    if re.match(r"^\d{4}-\d\d-\d\d \d\d:\d\d:\d\d$", t):
        return t
    m = re.match(r"^(ma|tegnap) (\d\d:\d\d:\d\d)$", t)
    if m:
        d = datetime.date(*FETCH_DAY) - datetime.timedelta(days=1 if m.group(1) == "tegnap" else 0)
        return "%s %s" % (d.isoformat(), m.group(2))
    m = re.match(r"^(\S+\.) (\d+)\., (\d\d:\d\d:\d\d)$", t)
    if m and m.group(1) in HU_MONTHS:
        return "%04d-%02d-%02d %s" % (FETCH_DAY[0], HU_MONTHS[m.group(1)], int(m.group(2)), m.group(3))
    return ""
USER_RE = re.compile(r'<div class="message-head-user">.*?<a class="user-link"[^>]*>([^<]+)</a>', re.S)


class OwnText(HTMLParser):
    """Collect text inside rtif-content message-content divs that are NOT inside
    message-content-replied. Emoji images become their alt text; <br>/<p> become breaks."""

    def __init__(self):
        super().__init__(convert_charrefs=True)
        self.stack = []  # per open div: (is_replied, is_content)
        self.out = []

    def _skip(self):
        return any(r for r, _ in self.stack)

    def _in_content(self):
        return any(c for _, c in self.stack)

    def handle_starttag(self, tag, attrs):
        a = dict(attrs)
        if tag == "div":
            cls = (a.get("class") or "").split()
            self.stack.append(("message-content-replied" in cls,
                               "rtif-content" in cls and "message-content" in cls))
            return
        if self._skip() or not self._in_content():
            return
        if tag in ("p", "br", "li"):
            self.out.append("\n")
        elif tag == "img" and a.get("alt"):
            self.out.append(" " + a["alt"] + " ")

    def handle_endtag(self, tag):
        if tag == "div" and self.stack:
            self.stack.pop()

    def handle_data(self, data):
        if self._in_content() and not self._skip():
            self.out.append(data)

    def text(self):
        t = "".join(self.out)
        t = re.sub(r"[ \t\r\f\v]+", " ", t)
        t = re.sub(r"\n\s*\n+", "\n", t)
        return t.strip()


def salt():
    if not os.path.exists(SALT_FILE):
        with open(SALT_FILE, "w") as f:
            f.write(secrets.token_hex(32))
        os.chmod(SALT_FILE, 0o600)
    return open(SALT_FILE).read().strip()


def parse_page(path, s):
    h = open(path, encoding="utf-8").read()
    marks = list(LI_RE.finditer(h))
    posts = []
    for i, m in enumerate(marks):
        end = marks[i + 1].start() if i + 1 < len(marks) else len(h)
        block = h[m.start():end]
        t = TIME_RE.search(block)
        u = USER_RE.search(block)
        p = OwnText()
        p.feed(block)
        user = html.unescape(u.group(1)).strip() if u else ""
        posts.append({
            "num": int(m.group(2)),
            "date": norm_date(t.group(1)) if t else "",
            "edited": bool(t and "Szerkesztve" in t.group(1)),
            "reply_to": int(m.group(3)) if m.group(3) else None,
            "author_id": hashlib.sha256((s + user).encode()).hexdigest()[:12] if user else "",
            "off": "message-off" in (m.group(1) or ""),
            "text": p.text(),
        })
    return posts


def main():
    s = salt()
    by_num, per_page = {}, {}
    for path in sorted(glob.glob(os.path.join(PAGES, "hsz_*.html"))):
        posts = parse_page(path, s)
        per_page[os.path.basename(path)] = posts
        for p in posts:
            by_num.setdefault(p["num"], p)
    # Hand-made quotes: a user pastes the parent's words into the own text (often in
    # <small><b><i>). No markup marks them reliably, so drop every own-text line of 20+ characters
    # that also occurs in the parent's text. Spec 8: an app named in a quote is not counted.
    def squash(t):
        return " ".join(t.lower().replace('"', " ").split())
    hand_quotes = 0
    for p in by_num.values():
        par = by_num.get(p["reply_to"]) if p["reply_to"] else None
        if not par:
            continue
        ptxt = squash(par["text"])
        keep = []
        for line in p["text"].split("\n"):
            sq = squash(line).strip(" :.,")
            if len(sq) >= 20 and sq in ptxt:
                hand_quotes += 1
                continue
            # A partial paste: drop each sentence (25+ chars) of the line found in the parent.
            parts = re.split(r'(?<=[.?!:"\]])\s+', line)
            kept = [x for x in parts
                    if not (len(squash(x).strip(' :.,[]')) >= 25 and squash(x).strip(' :.,[]') in ptxt)]
            hand_quotes += len(parts) - len(kept)
            if kept:
                keep.append(" ".join(kept))
        p["text"] = "\n".join(keep).strip()
    print("== hand-made quote lines removed: %d" % hand_quotes)
    bad = [p["num"] for p in by_num.values() if not p["date"]]
    if bad:
        print("UNPARSED DATES (fix norm_date before tagging): %s" % bad[:20])
        sys.exit(1)
    window = sorted((p for p in by_num.values() if p["date"] >= WINDOW_START), key=lambda p: p["num"])
    with open(OUT, "w", encoding="utf-8") as f:
        for p in window:
            f.write(json.dumps(p, ensure_ascii=False) + "\n")

    ok = True
    print("== control 1: counts per page (pages holding window posts)")
    lo, hi = window[0]["num"], window[-1]["num"]
    for name, posts in sorted(per_page.items(), key=lambda kv: int(kv[0].split("_")[1].split("-")[0])):
        nums = [p["num"] for p in posts]
        first = int(name.split("_")[1].split("-")[0])
        expected = set(range(first, first + 100))
        stray = [n for n in nums if n not in expected]
        dup = len(nums) - len(set(nums))
        print("  %-22s posts=%3d range=%s stray=%d dup=%d" %
              (name, len(nums), "%d-%d" % (min(nums), max(nums)) if nums else "-", len(stray), dup))
    have = set(by_num)
    missing = [n for n in range(lo, hi + 1) if n not in have]
    print("  window posts=%d  first=#%d (%s)  last=#%d (%s)" %
          (len(window), lo, window[0]["date"], hi, window[-1]["date"]))
    print("  missing numbers in window: %d" % len(missing))
    runs, start = [], None
    for n in missing:
        if start is None or n != prev + 1:
            if start is not None:
                runs.append((start, prev))
            start = n
        prev = n
    if start is not None:
        runs.append((start, prev))
    for a, b in runs:
        print("    #%d-#%d (%d)" % (a, b, b - a + 1))
    empty_text = sum(1 for p in window if not p["text"])
    print("  posts with empty own text: %d; off-topic flagged: %d; distinct authors: %d" %
          (empty_text, sum(p["off"] for p in window), len({p["author_id"] for p in window})))

    print("== control 3: #8099 own text present, quoted #8096 text absent")
    p = by_num.get(8099)
    own_ok = p and "proxmox" in p["text"].lower() and "7.x" in p["text"]
    # #8096 asks about moving from a Raspberry Pi 4 to an i5-6500T box.
    quote_absent = p and "6500" not in p["text"] and "raspberry" not in p["text"].lower()
    q = by_num.get(8096)
    neg_ctrl = q and ("6500" in q["text"] or "raspberry" in q["text"].lower())
    print("  own-text positive: %s; quote absent: %s; #8096 itself carries the marker (control works): %s"
          % (bool(own_ok), bool(quote_absent), bool(neg_ctrl)))
    print("  #8099 text: %r" % (p["text"][:200] if p else None))
    ok = ok and bool(own_ok and quote_absent and neg_ctrl)

    # Generic quote-leak check: no post's text may contain its parent's full text (when long enough).
    leaks = []
    for p in window:
        par = by_num.get(p["reply_to"]) if p["reply_to"] else None
        if par and len(par["text"]) >= 40 and par["text"][:40] in p["text"]:
            leaks.append(p["num"])
    print("== quote-leak sweep: posts whose text contains the parent's first 40 chars: %d %s"
          % (len(leaks), leaks[:20]))

    print("== control 2: five random posts for an eye check")
    rnd = random.Random(20261012)
    for p in rnd.sample(window, 5):
        print("  #%d reply_to=%s date=%s\n    %r" % (p["num"], p["reply_to"], p["date"], p["text"][:300]))
    sys.exit(0 if ok else 1)


if __name__ == "__main__":
    main()