#!/usr/bin/env python3
|
# -*- coding: utf-8 -*-
|
"""Standalone watcher for ob02 daily activity.
|
|
This watcher is deliberately separated from tagfeed. It listens to source file
|
changes and updates only today's daily-activity block.
|
"""
|
|
from datetime import datetime
|
from pathlib import Path
|
from typing import Optional
|
import os
|
import threading
|
import time
|
import traceback
|
|
from watchdog.events import FileSystemEventHandler
|
from watchdog.observers import Observer
|
|
from daily_activity import DEFAULT_VAULT, TRACKED_SUFFIXES, should_skip_diary_path, update_diaries
|
|
|
VAULT = Path(os.environ.get("DAILY_ACTIVITY_VAULT", DEFAULT_VAULT)).expanduser()
|
LOG_PATH = Path(os.environ.get("DAILY_ACTIVITY_WATCH_LOG", "/Users/ar/.daily-activity-watch.log"))
|
DEBOUNCE_SECONDS = float(os.environ.get("DAILY_ACTIVITY_DEBOUNCE", "1.5"))
|
AUTO_DIARY_BACKFILL_DAYS = 0
|
|
|
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:
|
if not path:
|
return False
|
try:
|
path.relative_to(VAULT)
|
except ValueError:
|
return False
|
if path.name == ".DS_Store" or path.name.startswith("."):
|
return False
|
if should_skip_diary_path(str(path), str(VAULT)):
|
return False
|
return path.name.lower().endswith(TRACKED_SUFFIXES)
|
|
|
class DebouncedDailyActivity:
|
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:
|
changed = update_diaries(
|
str(VAULT),
|
AUTO_DIARY_BACKFILL_DAYS,
|
log=log,
|
known_written=self.known_written,
|
)
|
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: DebouncedDailyActivity) -> 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}")
|
|
log(f"daily activity watch start vault={VAULT} mode=event debounce={DEBOUNCE_SECONDS:g}s")
|
scheduler = DebouncedDailyActivity(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("daily activity watch stopped by user")
|
finally:
|
observer.stop()
|
observer.join(timeout=5)
|
|
|
if __name__ == "__main__":
|
main()
|
