#!/usr/bin/env python3 """Fact Block 0.1: generate and gate the txtfirst dialect from source records. A Fact Block is eight fields in a fixed order inside ordinary markdown. It is not a new syntax. The point is that the shape never varies, so a model copies it instead of interpreting it, and a regex can extract it. The `answer` line is composed deterministically from the other fields. No language model writes it, because a model with no source invents one. python3 factblock.py --site # generate from Viator python3 factblock.py --check # self-test """ import argparse, datetime, json, os, re, sys, urllib.request FIELDS = ["answer", "where", "price", "duration", "rating", "booking", "source", "checked"] MAX_AGE_DAYS = 180 VIATOR = "https://api.viator.com/partner" # ---------- the gate ---------- class Rejected(Exception): pass def gate(block, today=None): """Refuse to publish a block that cannot be trusted. Raises Rejected.""" today = today or datetime.date.today() got = dict(re.findall(r"^(\w+):[ \t]*(.*)$", block, re.M)) if not got.get("source", "").strip(): raise Rejected("no source") if not re.match(r"https?://", got["source"]): raise Rejected("source is not a URL: %r" % got["source"]) if not got.get("checked"): raise Rejected("no checked date") try: checked = datetime.date.fromisoformat(got["checked"]) except ValueError: raise Rejected("checked is not an ISO date: %r" % got["checked"]) if checked > today: raise Rejected("checked is in the future: %s" % checked) age = (today - checked).days if age > MAX_AGE_DAYS: raise Rejected("stale by %d days (limit %d)" % (age - MAX_AGE_DAYS, MAX_AGE_DAYS)) present = list(got) expected = [f for f in FIELDS if f in got] if present != expected: raise Rejected("fields out of order: got %s, expected %s" % (present, expected)) missing = [f for f in FIELDS if f not in got] if missing: raise Rejected("missing fields: %s" % missing) # An answer that carries none of the numbers is a sentence, not an answer. nums = re.findall(r"\d[\d,.]*", got.get("answer", "")) if len(nums) < 2: raise Rejected("answer carries %d numbers, needs 2 (it must be quotable alone)" % len(nums)) if re.search(r"not stated|unknown", got.get("answer", "")): raise Rejected("answer contains a gap, so it is not quotable on its own") # Every number in the answer must appear in a VALUE field below it. source and # checked are excluded on purpose: a Viator slug like d903-288497P1 and an ISO # date are full of digits, and including them let the answer corroborate invented # numbers against a URL. The rule is only worth anything against value fields. VALUE_FIELDS = ("where", "price", "duration", "rating", "booking") body = "\n".join("%s: %s" % (k, got[k]) for k in VALUE_FIELDS if k in got) for n in nums: if n not in body: raise Rejected("answer cites %r, which appears in no field" % n) return got # ---------- rendering ---------- def render(name, where, price, currency, minutes, rating, reviews, instant, free_cancel, source, checked): dur = ("%d min" % minutes) if minutes and minutes < 90 else ( "%.1f h" % (minutes / 60) if minutes else "not stated") booking = ("instant confirmation" if instant else "on request") + \ (", free cancellation" if free_cancel else ", no free cancellation") answer = "from %s %s, rated %s out of 5 from %s reviews, %s" % ( _n(price), currency, _n(rating), _n(reviews), dur) return "\n".join([ "## %s" % name, "answer: %s" % answer, "where: %s" % where, "price: from %s %s" % (_n(price), currency), "duration: %s" % dur, "rating: %s/5 from %s reviews" % (_n(rating), _n(reviews)), "booking: %s" % booking, "source: %s" % source, "checked: %s" % checked, ]) def _n(v): if v is None: return "not stated" # one vocabulary for a gap, everywhere if isinstance(v, float) and v == int(v): v = int(v) return "{:,}".format(v) if isinstance(v, int) else str(v) # ---------- Viator source ---------- def viator(path, payload): key = open(os.path.expanduser("~/.config/viator/api_key")).read().strip() req = urllib.request.Request(VIATOR + path, data=json.dumps(payload).encode(), method="POST") for h, v in (("exp-api-key", key), ("Accept", "application/json;version=2.0"), ("Accept-Language", "en-US"), ("Content-Type", "application/json")): req.add_header(h, v) return json.load(urllib.request.urlopen(req, timeout=60)) def from_site(config_path, count=20): cfg = json.load(open(config_path)) dest = cfg["viator_destination_id"] where = ", ".join(x for x in (cfg.get("city"), cfg.get("region"), cfg.get("country")) if x) pid, mcid = cfg.get("viator_pid"), cfg.get("viator_mcid") today = datetime.date.today() res = viator("/products/search", { "filtering": {"destination": str(dest), "startDate": str(today), "endDate": str(today + datetime.timedelta(days=180))}, "sorting": {"sort": "TRAVELER_RATING", "order": "DESCENDING"}, "pagination": {"start": 1, "count": count}, "currency": "USD"}) products = res.get("products") or [] if not products: # A source that returns nothing is a broken query until proven otherwise. raise SystemExit("destination %s returned 0 products. Check the id against " "/destinations before assuming the destination is empty." % dest) blocks, rejected = [], [] for p in products: rev = p.get("reviews") or {} pr = (p.get("pricing") or {}) url = p.get("productUrl") or "" # productUrl already carries pid and mcid when the key is an affiliate key. # Appending them again produces a URL with the parameter twice. if pid and url and "pid=" not in url: url += ("&" if "?" in url else "?") + "pid=%s&mcid=%s" % (pid, mcid) b = render( name=p.get("title"), where=where, price=(pr.get("summary") or {}).get("fromPrice"), currency=pr.get("currency") or "USD", minutes=(p.get("duration") or {}).get("fixedDurationInMinutes") or (p.get("duration") or {}).get("variableDurationFromMinutes"), rating=rev.get("combinedAverageRating"), reviews=rev.get("totalReviews"), instant=p.get("confirmationType") == "INSTANT", free_cancel="FREE_CANCELLATION" in (p.get("flags") or []), source=url, checked=str(today)) try: gate(b) blocks.append(b) except Rejected as e: rejected.append((p.get("productCode"), str(e))) return blocks, rejected # ---------- self-test ---------- def check(): good = render("Snorkel Tour", "San Juan, Puerto Rico", 69.0, "USD", 180, 4.8, 1204, True, True, "https://viator.com/x", "2026-09-04") got = gate(good, today=datetime.date(2026, 9, 4)) assert got["price"] == "from 69 USD", got["price"] assert list(got) == FIELDS, list(got) def rejects(b, frag, today=datetime.date(2026, 9, 4)): try: gate(b, today=today) except Rejected as e: assert frag in str(e), "expected %r, got %r" % (frag, str(e)) return raise AssertionError("should have been rejected: %s" % frag) rejects(good.replace("source: https://viator.com/x", "source: "), "no source") rejects("\n".join(l for l in good.split("\n") if not l.startswith("duration:")), "missing fields") swapped = good.replace("where: San Juan, Puerto Rico\nprice: from 69 USD", "price: from 69 USD\nwhere: San Juan, Puerto Rico") rejects(swapped, "out of order") rejects(good.replace("checked: 2026-09-04", "checked: 2026-01-01"), "stale by") rejects(good.replace("checked: 2026-09-04", "checked: 2027-01-01"), "in the future") rejects(good.replace("checked: 2026-09-04", "checked: last week"), "not an ISO date") rejects(good.replace("answer: from 69 USD, rated 4.8 out of 5 from 1,204 reviews, 3.0 h", "answer: a great snorkel tour"), "needs 2") rejects(good.replace("rated 4.8 out of 5", "rated 4.9 out of 5"), "appears in no field") rejects(good.replace("source: https://viator.com/x", "source: viator.com/x"), "not a URL") # the answer must not be corroborated by digits in the URL or the date rejects(good.replace("answer: from 69 USD", "answer: from 903 USD") .replace("source: https://viator.com/x", "source: https://viator.com/d903-288497P1"), "appears in no field") rejects(good.replace("answer: from 69 USD, rated 4.8 out of 5 from 1,204 reviews, 3.0 h", "answer: from 2026 USD, rated 4.8 out of 5 from 1,204 reviews, 3.0 h"), "appears in no field") # a gap must not reach a quotable line assert _n(None) == "not stated", _n(None) rejects(render("X", "San Juan", None, "USD", 180, 4.8, 1204, True, True, "https://viator.com/x", "2026-09-04"), "gap") # the page states the file length, so the file checks it n = sum(1 for _ in open(os.path.abspath(__file__))) assert n == 231, "dialect.md says 231 lines, file has %d. Update both." % n # a stale block is rejected even though every other field is perfect rejects(good, "stale by", today=datetime.date(2027, 9, 4)) print("check ok (gate rejects: missing source, non-URL source, stale, future, " "malformed date, missing fields, out of order, unquotable answer, gaps in " "the answer, and numbers corroborated only by the URL or the date)") if __name__ == "__main__": ap = argparse.ArgumentParser() ap.add_argument("--site", help="path to a site _config.json") ap.add_argument("--count", type=int, default=20) ap.add_argument("--check", action="store_true") a = ap.parse_args() if a.check: check() elif a.site: blocks, rejected = from_site(a.site, a.count) print("\n\n".join(blocks)) print("\n" % (len(blocks), len(rejected)), file=sys.stderr) for code, why in rejected: print(" rejected %s: %s" % (code, why), file=sys.stderr) else: ap.print_help()