multica deployment

hermes/obsidian-file-watcher/wiki-llm-watcher.py

@@ -0,0 +1,363 @@
+#!/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())