#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""Scheme 3 watcher for tagfeed.
|
|
This watcher is event-driven:
|
- it listens to vault file changes
|
- it ignores generated outputs that would cause self-trigger loops
|
- it reruns sync_excalidraw_tagfeed.py with debounce
|
"""
|
|
import os
|
import queue
|
import re
|
import subprocess
|
import sys
|
import threading
|
import time
|
import traceback
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Optional
|
|
from watchdog.events import FileSystemEventHandler
|
from watchdog.observers import Observer
|
|
from tagfeed import DIARY_BACKFILL_DAYS, update_diaries
|
|
|
VAULT = Path(os.environ.get("EXCALIDRAW_TAGFEED_VAULT", "/Users/ar/Downloads/Syn/Ob/Ob.temp")).expanduser()
|
SYNC_SCRIPT = VAULT / "X2.Archived/scripts/sync_excalidraw_tagfeed.py"
|
LOG_PATH = Path(os.environ.get("EXCALIDRAW_TAGFEED_LOG", "/Users/ar/.excalidraw-tagfeed-watch.log"))
|
DEBOUNCE_SECONDS = float(os.environ.get("EXCALIDRAW_TAGFEED_DEBOUNCE", "1.5"))
|
OUTPUT_REL = Path("00index/00index.md")
|
GENERATED_DIR = Path("00index/tagfeed-md")
|
SKIP_DIRS = {
|
".obsidian",
|
".git",
|
".trash",
|
".hg",
|
".svn",
|
".agents",
|
".claude",
|
".copilot",
|
".smart-env",
|
".workbuddy",
|
"X2.Archived",
|
"X.Attachment",
|
}
|
|
|
def log(message: str) -> None:
|
stamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
line = f"[{stamp}] {message}"
|
print(line, flush=True)
|
try:
|
with LOG_PATH.open("a", encoding="utf-8") as fh:
|
fh.write(line + "\n")
|
except OSError:
|
pass
|
|
|
def is_relevant(path: Path) -> bool:
|
try:
|
rel = path.relative_to(VAULT)
|
except ValueError:
|
return False
|
if rel == OUTPUT_REL:
|
return False
|
if rel == GENERATED_DIR or GENERATED_DIR in rel.parents:
|
return False
|
parts = rel.parts
|
if parts and parts[0] in SKIP_DIRS:
|
return False
|
if path.name.startswith("."):
|
return False
|
if path.suffix.lower() not in {".md", ".markdown", ".excalidraw", ".canvas"}:
|
return False
|
return True
|
|
|
def run_sync() -> str:
|
result = subprocess.run(
|
[sys.executable, str(SYNC_SCRIPT), str(VAULT)],
|
capture_output=True,
|
text=True,
|
timeout=180,
|
)
|
output = (result.stdout or "").strip()
|
if result.returncode != 0:
|
err = (result.stderr or output).strip()
|
raise RuntimeError(err[-1000:] or f"sync exited {result.returncode}")
|
if output:
|
log(output)
|
match = re.search(r"Excalidraw-Tagfeed:\s+(\w+)\s+", output)
|
return match.group(1) if match else "unknown"
|
|
|
def run_daily_activity_sync(known_written) -> int:
|
diary_dir = VAULT / "X0.Diary"
|
if not diary_dir.is_dir():
|
return 0
|
return update_diaries(str(VAULT), DIARY_BACKFILL_DAYS, log=log, known_written=known_written)
|
|
|
class DebouncedSync:
|
def __init__(self, debounce_seconds: float) -> None:
|
self.debounce_seconds = debounce_seconds
|
self._lock = threading.Lock()
|
self._timer: Optional[threading.Timer] = None
|
self._running = False
|
self._pending = False
|
self.known_written = {}
|
|
def trigger(self, reason: str) -> None:
|
log(f"event: {reason}")
|
with self._lock:
|
if self._running:
|
self._pending = True
|
return
|
if self._timer is not None:
|
self._timer.cancel()
|
self._timer = threading.Timer(self.debounce_seconds, self._fire)
|
self._timer.daemon = True
|
self._timer.start()
|
|
def _fire(self) -> None:
|
with self._lock:
|
self._timer = None
|
self._running = True
|
try:
|
status = run_sync()
|
if status == "missingframe":
|
log("Frame0-all unavailable; waiting for next event")
|
except Exception:
|
log("sync failed:\n" + traceback.format_exc().rstrip())
|
try:
|
changed = run_daily_activity_sync(self.known_written)
|
if changed:
|
log(f"Daily activity: updated {changed} diaries")
|
except Exception:
|
log("daily activity sync failed:\n" + traceback.format_exc().rstrip())
|
finally:
|
with self._lock:
|
self._running = False
|
pending = self._pending
|
self._pending = False
|
if pending:
|
self.trigger("coalesced")
|
|
|
class VaultEventHandler(FileSystemEventHandler):
|
def __init__(self, scheduler: DebouncedSync) -> None:
|
self.scheduler = scheduler
|
|
def on_any_event(self, event) -> None:
|
if getattr(event, "is_directory", False):
|
return
|
src = Path(getattr(event, "src_path", "") or "")
|
dest = Path(getattr(event, "dest_path", "") or "")
|
if is_relevant(src) or is_relevant(dest):
|
self.scheduler.trigger(f"{event.event_type}: {src or dest}")
|
|
|
def main() -> None:
|
if not VAULT.is_dir():
|
raise SystemExit(f"vault not found: {VAULT}")
|
if not SYNC_SCRIPT.is_file():
|
raise SystemExit(f"sync script not found: {SYNC_SCRIPT}")
|
|
log(f"scheme3 watch start vault={VAULT} mode=event debounce={DEBOUNCE_SECONDS:g}s")
|
scheduler = DebouncedSync(DEBOUNCE_SECONDS)
|
handler = VaultEventHandler(scheduler)
|
observer = Observer()
|
observer.schedule(handler, str(VAULT), recursive=True)
|
observer.start()
|
scheduler.trigger("startup")
|
|
try:
|
while True:
|
time.sleep(1)
|
except KeyboardInterrupt:
|
log("watch stopped by user")
|
finally:
|
observer.stop()
|
observer.join(timeout=5)
|
|
|
if __name__ == "__main__":
|
main()
|
