#!/usr/bin/env python3 """wiki-llm-watcher: poll the vault's raw/ inbox and POST one webhook call per new file. Design (see README.md): * Polling is the source of truth. No filesystem events are trusted. * A file is posted only after (size, mtime) has been unchanged for WIKI_STABLE_SECONDS and across at least one full scan, so half-written clips/syncs are skipped. * Files are posted sequentially, one webhook call each, with a short delay between calls. * A file is recorded as posted in the local SQLite seen-state ONLY after n8n returns 2xx. On restart, anything not recorded is retried, so nothing is lost while n8n or the watcher is down. (n8n dedupes against its Data Table, so reposts are harmless.) * Detection is by NEW PATH only. Later edits (the librarian rewrites frontmatter) are ignored. * Standard library only. Python 3.8+. Usage: wiki-llm-watcher.py --baseline mark everything currently in raw/ as seen (no POSTs) wiki-llm-watcher.py --dry-run --once show what WOULD be posted, post nothing wiki-llm-watcher.py run forever (this is what launchd runs) wiki-llm-watcher.py --once one stabilised pass, then exit wiki-llm-watcher.py --forget raw/x.md drop a path from seen-state so it is posted again Configuration is by environment variables, optionally loaded from the KEY=VALUE file named by WIKI_ENV_FILE (keep that file chmod 600; it holds the webhook secret). """ import argparse import hashlib import json import logging import os import signal import socket import sqlite3 import sys import time import unicodedata import urllib.error import urllib.request from datetime import datetime, timezone log = logging.getLogger("wiki-llm-watcher") STOP = False # ----------------------------------------------------------------------------- config def load_env_file(path): with open(path, "r", encoding="utf-8") as fh: for line in fh: line = line.strip() if not line or line.startswith("#") or "=" not in line: continue key, _, value = line.partition("=") key, value = key.strip(), value.strip() if len(value) >= 2 and value[0] == value[-1] and value[0] in "'\"": value = value[1:-1] os.environ.setdefault(key, value) def get_config(): env_file = os.environ.get("WIKI_ENV_FILE") if env_file: load_env_file(os.path.expanduser(env_file)) cfg = { "vault_root": os.path.expanduser(os.environ.get("WIKI_VAULT_ROOT", "")), "raw_dir": os.environ.get("WIKI_RAW_DIR", "raw"), "url": os.environ.get("WIKI_WEBHOOK_URL", ""), "auth_header": os.environ.get("WIKI_AUTH_HEADER", "X-Webhook-Secret"), "auth_secret": os.environ.get("WIKI_AUTH_SECRET", ""), "poll": float(os.environ.get("WIKI_POLL_SECONDS", "20")), "stable": float(os.environ.get("WIKI_STABLE_SECONDS", "10")), "post_delay": float(os.environ.get("WIKI_POST_DELAY", "0.5")), "timeout": float(os.environ.get("WIKI_TIMEOUT", "30")), "ignore_names": {n.strip().lower() for n in os.environ.get( "WIKI_IGNORE_NAMES", "AGENTS.md,CLAUDE.md,.hermes.md,README.md,SCHEMA.md").split(",") if n.strip()}, "extensions": [e.strip().lower() for e in os.environ.get("WIKI_EXTENSIONS", ".md").split(",") if e.strip()], "state_db": os.path.expanduser( os.environ.get("WIKI_STATE_DB", "~/.local/state/wiki-llm-watcher/seen.db") ), } return cfg def validate_config(cfg, need_url): problems = [] if not cfg["vault_root"] or not os.path.isdir(cfg["vault_root"]): problems.append("WIKI_VAULT_ROOT is not set or not a directory") elif not os.path.isdir(os.path.join(cfg["vault_root"], cfg["raw_dir"])): problems.append("raw dir not found: %s" % os.path.join(cfg["vault_root"], cfg["raw_dir"])) if need_url: if not cfg["url"].startswith(("http://", "https://")): problems.append("WIKI_WEBHOOK_URL must be an http(s) URL") if not cfg["auth_secret"]: problems.append("WIKI_AUTH_SECRET is empty") return problems # ----------------------------------------------------------------------------- state def open_state(path): os.makedirs(os.path.dirname(path), exist_ok=True) db = sqlite3.connect(path) db.execute( "CREATE TABLE IF NOT EXISTS files (" " path TEXT PRIMARY KEY, sha256 TEXT, size INTEGER, mtime TEXT," " status TEXT NOT NULL, updated_at TEXT NOT NULL)" ) db.commit() return db def now_iso(): return datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") def mark(db, rel, status, sha=None, size=None, mtime=None): db.execute( "INSERT INTO files(path, sha256, size, mtime, status, updated_at) VALUES(?,?,?,?,?,?)" " ON CONFLICT(path) DO UPDATE SET sha256=excluded.sha256, size=excluded.size," " mtime=excluded.mtime, status=excluded.status, updated_at=excluded.updated_at", (rel, sha, size, mtime, status, now_iso()), ) db.commit() def known_paths(db): return {row[0] for row in db.execute("SELECT path FROM files")} # ----------------------------------------------------------------------------- scanning IGNORED_SUFFIXES = (".tmp", ".part", ".partial", ".crdownload", ".download", ".swp", ".swx", "~") def is_ignored(name, extensions, ignore_names=frozenset()): low = name.lower() if low in ignore_names: # instruction files that live in raw/ (AGENTS.md etc.) are never ingested return True if name.startswith("."): # dotfiles, .syncthing.*, .DS_Store, iCloud placeholders (.foo.icloud) return True if low.endswith(IGNORED_SUFFIXES) or low.endswith(".icloud"): return True if ".sync-conflict-" in low: return True if "*" in extensions: return False return not any(low.endswith(ext) for ext in extensions) def scan_files(cfg): """Yield (relative_path_nfc, absolute_path) for every candidate file under raw/.""" root = os.path.realpath(cfg["vault_root"]) raw = os.path.join(root, cfg["raw_dir"]) for dirpath, dirnames, filenames in os.walk(raw, followlinks=False): dirnames[:] = sorted(d for d in dirnames if not d.startswith(".")) for name in sorted(filenames): if is_ignored(name, cfg["extensions"], cfg["ignore_names"]): continue full = os.path.join(dirpath, name) if os.path.islink(full): continue rel = os.path.relpath(full, root).replace(os.sep, "/") yield unicodedata.normalize("NFC", rel), full def sha256_of(path): h = hashlib.sha256() with open(path, "rb") as fh: for chunk in iter(lambda: fh.read(1024 * 1024), b""): h.update(chunk) return h.hexdigest() # ----------------------------------------------------------------------------- HTTP def post_json(cfg, payload): """Return (status_code, body_snippet). Raises URLError/OSError when unreachable.""" req = urllib.request.Request( cfg["url"], data=json.dumps(payload).encode("utf-8"), method="POST", headers={"Content-Type": "application/json", cfg["auth_header"]: cfg["auth_secret"]}, ) try: with urllib.request.urlopen(req, timeout=cfg["timeout"]) as resp: return resp.status, resp.read(500).decode("utf-8", "replace") except urllib.error.HTTPError as err: return err.code, err.read(500).decode("utf-8", "replace") # ----------------------------------------------------------------------------- core pass class Watcher: def __init__(self, cfg, db, dry_run=False): self.cfg = cfg self.db = db self.dry_run = dry_run self.observed = {} # rel -> [(size, mtime_ns), first_seen_monotonic, seen_in_scans] self.retry = {} # rel -> (fail_count, not_before_monotonic) self.cooldown_until = 0.0 # set when n8n is unreachable or config is wrong def observe(self): """One scan: update stability tracking, return list of (rel, full, stat) eligible now.""" now = time.monotonic() seen = known_paths(self.db) present = set() eligible = [] for rel, full in scan_files(self.cfg): if rel in seen: continue try: st = os.stat(full) except OSError: continue # vanished between listing and stat present.add(rel) sig = (st.st_size, st.st_mtime_ns) rec = self.observed.get(rel) if rec is None or rec[0] != sig: self.observed[rel] = [sig, now, 1] continue rec[2] += 1 if rec[2] >= 2 and (now - rec[1]) >= self.cfg["stable"]: nb = self.retry.get(rel) if nb and now < nb[1]: continue eligible.append((rel, full, st)) for gone in set(self.observed) - present: del self.observed[gone] eligible.sort(key=lambda item: item[2].st_mtime_ns) return eligible def post_pass(self): eligible = self.observe() if eligible and time.monotonic() < self.cooldown_until: return # n8n was unreachable / misconfigured a moment ago; don't spam the log if eligible: log.info("%d stable new file(s) to post", len(eligible)) for rel, full, st in eligible: if STOP: return try: digest = sha256_of(full) st2 = os.stat(full) except OSError as exc: log.warning("skip %s (unreadable: %s)", rel, exc) continue if (st2.st_size, st2.st_mtime_ns) != (st.st_size, st.st_mtime_ns): log.info("changed while hashing, will re-check: %s", rel) self.observed.pop(rel, None) continue mtime_iso = datetime.fromtimestamp(st.st_mtime, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") payload = { "path": rel, "sha256": digest, "size": st.st_size, "mtime": mtime_iso, "source": "watcher", "host": socket.gethostname(), } if self.dry_run: log.info("[dry-run] would POST %s", json.dumps(payload, ensure_ascii=False)) continue try: code, body = post_json(self.cfg, payload) except (urllib.error.URLError, OSError) as exc: log.error("n8n unreachable (%s); pausing posts for 60s", exc) self.cooldown_until = time.monotonic() + 60 return # stop this pass; do not hammer a down endpoint if 200 <= code < 300: mark(self.db, rel, "posted", digest, st.st_size, mtime_iso) self.retry.pop(rel, None) self.observed.pop(rel, None) log.info("posted %s (HTTP %d)", rel, code) elif code in (400, 409, 413, 422): mark(self.db, rel, "rejected", digest, st.st_size, mtime_iso) self.observed.pop(rel, None) log.warning("rejected by n8n, not retrying: %s (HTTP %d: %s). Use --forget to retry.", rel, code, body[:200].replace("\n", " ")) elif code in (401, 403, 404): log.error("config problem (HTTP %d) - check URL (production, workflow active?) and secret. " "Pausing posts for 60s.", code) self.cooldown_until = time.monotonic() + 60 return else: fails = self.retry.get(rel, (0, 0))[0] + 1 delay = min(30 * (2 ** (fails - 1)), 900) self.retry[rel] = (fails, time.monotonic() + delay) log.warning("HTTP %d for %s, retry in %ds (attempt %d)", code, rel, delay, fails) time.sleep(self.cfg["post_delay"]) def do_baseline(cfg, db): count = 0 for rel, full in scan_files(cfg): try: st = os.stat(full) except OSError: continue mark(db, rel, "baseline", None, st.st_size, datetime.fromtimestamp(st.st_mtime, timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")) count += 1 log.info("baseline: marked %d existing file(s) as seen; nothing was posted", count) def sleep_interruptible(seconds): end = time.monotonic() + seconds while not STOP and time.monotonic() < end: time.sleep(min(1.0, end - time.monotonic())) def main(): parser = argparse.ArgumentParser(description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter) parser.add_argument("--baseline", action="store_true", help="mark existing raw/ files as seen without posting") parser.add_argument("--once", action="store_true", help="run one stabilised pass and exit") parser.add_argument("--dry-run", action="store_true", help="log what would be posted; post nothing, record nothing") parser.add_argument("--forget", metavar="REL_PATH", help="remove a path from seen-state so it is posted again") args = parser.parse_args() logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", stream=sys.stdout) cfg = get_config() problems = validate_config(cfg, need_url=not (args.baseline or args.forget or args.dry_run)) if problems: for p in problems: log.error(p) return 1 db = open_state(cfg["state_db"]) if args.forget: rel = unicodedata.normalize("NFC", args.forget) cur = db.execute("DELETE FROM files WHERE path = ?", (rel,)) db.commit() log.info("forgot %d row(s) for %s", cur.rowcount, rel) return 0 if args.baseline: do_baseline(cfg, db) return 0 def handle_signal(signum, _frame): global STOP STOP = True log.info("signal %d received, stopping", signum) signal.signal(signal.SIGTERM, handle_signal) signal.signal(signal.SIGINT, handle_signal) watcher = Watcher(cfg, db, dry_run=args.dry_run) log.info("watching %s (poll %.0fs, stable %.0fs, ext %s)", os.path.join(cfg["vault_root"], cfg["raw_dir"]), cfg["poll"], cfg["stable"], ",".join(cfg["extensions"])) if args.once: watcher.observe() # prime stability tracking sleep_interruptible(max(cfg["stable"], 1.0)) watcher.observe() # second observation (counts as the 'unchanged across scans' check) watcher.post_pass() return 0 while not STOP: try: watcher.post_pass() except Exception: # keep the daemon alive; launchd would restart us anyway log.exception("unexpected error in pass") sleep_interruptible(cfg["poll"]) return 0 if __name__ == "__main__": sys.exit(main())