#!/usr/bin/env python3 """Re-run the primary-docs check behind the secure-photo-sharing-app post. For every row in dataset.json, fetch the page it cites, reduce it to text, and report whether the quoted sentence is still there. Code rows are read from the repository this file sits in, at origin/main, with `git show`. python3 check.py # fetch every page and compare python3 check.py --recheck # same, minus anything cached; no lab steps here Exit 0 when every row still holds, 1 otherwise. """ import argparse import html import json import os import re import subprocess import sys import time import urllib.error import urllib.request from html.parser import HTMLParser HERE = os.path.dirname(os.path.abspath(__file__)) DATASET = os.path.join(HERE, "dataset.json") REPO = os.path.abspath(os.path.join(HERE, "..", "..", "..", "..", "..")) UA = ( "Mozilla/5.0 (Macintosh; Intel Mac OS X 10_15_7) " "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0 Safari/537.36" ) class TextExtractor(HTMLParser): """Reduces an HTML document to its visible text, one block per line.""" SKIP = ("script", "style", "noscript", "svg") BREAK = ("p", "div", "br", "li", "tr", "h1", "h2", "h3", "h4", "section", "td", "th") def __init__(self): HTMLParser.__init__(self, convert_charrefs=True) self.out = [] self.skip = 0 def handle_starttag(self, tag, attrs): if tag in self.SKIP: self.skip += 1 if tag in self.BREAK: self.out.append("\n") def handle_endtag(self, tag): if tag in self.SKIP and self.skip: self.skip -= 1 if tag in ("p", "li", "tr", "h1", "h2", "h3", "h4"): self.out.append("\n") def handle_data(self, data): if not self.skip: self.out.append(data) def html_to_text(raw): parser = TextExtractor() parser.feed(raw) return "".join(parser.out) def normalise(text): """Whitespace, quote style, dash style and entities out of the comparison.""" text = html.unescape(text) for src, dst in ( ("\u2019", "'"), ("\u2018", "'"), ("\u201c", '"'), ("\u201d", '"'), ("\u2013", "-"), ("\u2014", "-"), ("\u00a0", " "), ("\u2212", "-"), ): text = text.replace(src, dst) return re.sub(r"\s+", " ", text).strip() def decompress(body, encoding): """Some sites gzip whether or not we ask; the body's magic bytes say which.""" if body[:2] == b"\x1f\x8b": import gzip return gzip.decompress(body) if encoding == "deflate": import zlib return zlib.decompress(body, -zlib.MAX_WBITS) return body def fetch(url, attempts=3): """Returns (raw_html, error).""" last = None for attempt in range(attempts): request = urllib.request.Request( url, headers={ "User-Agent": UA, "Accept-Language": "en-US,en;q=0.9", "Accept": "text/html,application/xhtml+xml", "Accept-Encoding": "gzip, deflate", }, ) try: with urllib.request.urlopen(request, timeout=60) as response: charset = response.headers.get_content_charset() or "utf-8" body = decompress(response.read(), response.headers.get("Content-Encoding", "")) return body.decode(charset, "replace"), None except urllib.error.HTTPError as exc: last = "HTTP %s" % exc.code if exc.code in (403, 404, 410): return None, last except Exception as exc: # network, timeout, TLS last = "%s: %s" % (type(exc).__name__, exc) if attempt + 1 < attempts: time.sleep(1 + attempt) return None, last def read_code(path): """Returns (text, error). Reads the file as committed on origin/main.""" try: out = subprocess.run( ["git", "-C", REPO, "show", "origin/main:%s" % path], capture_output=True, timeout=60, ) except Exception as exc: return None, "%s: %s" % (type(exc).__name__, exc) if out.returncode != 0: return None, out.stderr.decode("utf-8", "replace").strip()[:200] return out.stdout.decode("utf-8", "replace"), None def main(): parser = argparse.ArgumentParser() parser.add_argument("--recheck", action="store_true", help="re-read every cited page") parser.add_argument("--rows", action="store_true", help="print one line per row") args = parser.parse_args() with open(DATASET, encoding="utf-8") as handle: dataset = json.load(handle) pages = {} # url -> (normalised text, raw html, error) rows = dataset["results"] failures = [] print("check: %s, method %s, checked %s" % (dataset["slug"], dataset["method"], dataset["checkedAt"])) print("recheck: %s" % ("yes" if args.recheck else "no")) print("") for row in rows: label = "%s / %s" % (row["option"], row["property"]) if row.get("source") == "code": text, error = read_code(row["repo_path"]) if error and not os.path.isdir(REPO): print("SKIP %s - local clone absent, code row not verified" % label) continue where = "origin/main:%s" % row["repo_path"] raw = text or "" else: url = row["url"] if url not in pages: raw, error = fetch(url) pages[url] = (html_to_text(raw) if raw else "", raw or "", error) text, raw, error = pages[url] where = url if error: print("FAIL %s - page not read (%s): %s" % (label, error, where)) failures.append((label, "page not read: %s" % error)) continue haystack = normalise(text if row.get("source") == "code" else text) needle = normalise(row["quote"]) missing = needle not in haystack marker = row.get("marker") marker_missing = bool(marker) and marker not in raw if missing or marker_missing: reason = [] if missing: reason.append("quote not found") if marker_missing: reason.append("marker %s not in page" % marker) print("FAIL %s - %s: %s" % (label, ", ".join(reason), where)) failures.append((label, ", ".join(reason))) elif args.rows: print("OK %s -> %s" % (label, row["value"])) checked = len(rows) - len(failures) print("") print("rows: %d, quotes held: %d, failed: %d" % (len(rows), checked, len(failures))) for label, reason in failures: print(" - %s: %s" % (label, reason)) return 0 if not failures else 1 if __name__ == "__main__": sys.exit(main())