364 lines
15 KiB
Python
364 lines
15 KiB
Python
#!/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())
|