replica-omnisciente/scripts/vault-sync.py

#!/usr/bin/env python3
"""
vault-sync — Obsidian vault → GBrain ingestion daemon (MCP).

Watches a directory (synced from phone via Nextcloud/Remotely Save),
ingests markdown notes into GBrain via its MCP HTTP endpoint.

Runs as a systemd service on CT223 (pf-forja-do-conhecimento).

Config: /etc/vault-sync.json
  {
    "vault_dir": "/mnt/usb-pool/obsidian-vault",
    "gbrain_url": "http://127.0.0.1:18001/mcp",
    "gbrain_token": "gbrain_at_...",
    "poll_interval": 300,
    "min_file_age": 5,
    "state_db": "/var/lib/vault-sync/state.db"
  }
"""

import hashlib
import json
import os
import re
import signal
import sqlite3
import sys
import time
import urllib.request
import urllib.error
from datetime import datetime, timezone
from pathlib import Path

DEFAULT_CONFIG = {
    "vault_dir": "/mnt/usb-pool/obsidian-vault",
    "gbrain_url": "http://127.0.0.1:18001/mcp",
    "gbrain_token": "",
    "poll_interval": 300,
    "min_file_age": 5,
    "state_db": "/var/lib/vault-sync/state.db",
    "source_slug": "obsidian-vault",
}

WIKILINK_RE = re.compile(r"\[\[([^\]|]+)(?:\|[^\]]+)?\]\]")
FRONTMATTER_RE = re.compile(r"^---\n(.*?)\n---\n", re.DOTALL)
TAG_LINE_RE = re.compile(r"^tags?:\s*(.+)$", re.MULTILINE)
INLINE_TAG_RE = re.compile(r"(?:^|\s)#([a-zA-Z][a-zA-Z0-9_/\-]*)")


def mcp_call(url, token, tool, args=None):
    """Call a GBrain MCP tool via HTTP JSON-RPC."""
    params = {"name": tool, "arguments": args or {}}
    data = json.dumps({"jsonrpc": "2.0", "method": "tools/call", "params": params, "id": 1})
    req = urllib.request.Request(
        url,
        data=data.encode(),
        headers={
            "Content-Type": "application/json",
            "Accept": "application/json, text/event-stream",
            "Authorization": f"Bearer {token}",
        },
    )
    with urllib.request.urlopen(req, timeout=30) as resp:
        raw = resp.read().decode()
        for line in raw.splitlines():
            if line.startswith("data: "):
                return json.loads(line[6:])
        return json.loads(raw)


def slugify(text):
    """Convert a note title to a GBrain-safe slug."""
    s = text.lower().strip()
    s = re.sub(r"[^a-z0-9\s/-]", "", s)
    s = re.sub(r"[\s/]+", "-", s)
    s = re.sub(r"-+", "-", s).strip("-")
    return s[:80] or "untitled"


class VaultSync:
    def __init__(self, config_path="/etc/vault-sync.json"):
        self.config = dict(DEFAULT_CONFIG)
        try:
            with open(config_path) as f:
                self.config.update(json.load(f))
        except (OSError, json.JSONDecodeError):
            pass

        self.vault_dir = Path(self.config["vault_dir"])
        self.gbrain_url = self.config["gbrain_url"].rstrip("/")
        self.gbrain_token = self.config["gbrain_token"]
        self.poll_interval = self.config["poll_interval"]
        self.min_file_age = self.config["min_file_age"]
        self.source_slug = self.config["source_slug"]

        os.makedirs(os.path.dirname(self.config["state_db"]), exist_ok=True)
        self.db = sqlite3.connect(self.config["state_db"])
        self.db.execute("""
            CREATE TABLE IF NOT EXISTS ingested (
                path TEXT PRIMARY KEY,
                mtime REAL,
                content_hash TEXT,
                gbrain_slug TEXT,
                ingested_at TEXT
            )
        """)
        self.db.commit()

        self.running = True
        signal.signal(signal.SIGTERM, self._shutdown)
        signal.signal(signal.SIGINT, self._shutdown)

    def _shutdown(self, *_):
        print("Shutting down...", flush=True)
        self.running = False

    def run(self):
        print(f"vault-sync: {self.vault_dir} → {self.gbrain_url}", flush=True)
        print(f"  poll: {self.poll_interval}s, min_age: {self.min_file_age}s", flush=True)

        while self.running:
            try:
                self.sync_once()
            except Exception as e:
                print(f"Sync error: {e}", file=sys.stderr, flush=True)
            for _ in range(int(self.poll_interval)):
                if not self.running:
                    break
                time.sleep(1)

        self.db.close()
        print("vault-sync stopped.", flush=True)

    def sync_once(self):
        if not self.vault_dir.exists():
            print(f"  vault dir not found: {self.vault_dir}", flush=True)
            return

        notes = sorted(self.vault_dir.rglob("*.md"))
        total = len(notes)
        new = changed = skipped = errors = 0

        for note_path in notes:
            if not self.running:
                break
            rel = note_path.relative_to(self.vault_dir)
            stat = note_path.stat()
            age = time.time() - stat.st_mtime

            if age < self.min_file_age:
                skipped += 1
                continue

            row = self.db.execute(
                "SELECT mtime FROM ingested WHERE path = ?", (str(rel),)
            ).fetchone()
            if row and row[0] == stat.st_mtime:
                skipped += 1
                continue

            try:
                result = self.ingest_note(note_path, rel)
                if result == "new":
                    new += 1
                elif result == "updated":
                    changed += 1
                else:
                    skipped += 1
            except Exception as e:
                errors += 1
                print(f"  ERROR {rel}: {e}", file=sys.stderr, flush=True)

        if new or changed or errors:
            print(
                f"  {total} notes: {new} new, {changed} updated, "
                f"{skipped} skipped, {errors} errors",
                flush=True,
            )

    def ingest_note(self, note_path, rel):
        content = note_path.read_text(encoding="utf-8", errors="replace")
        if not content.strip():
            return "skip"

        content_hash = hashlib.sha1(content.encode()).hexdigest()[:16]
        stat = note_path.stat()

        # Parse frontmatter
        frontmatter = {}
        body = content
        fm = FRONTMATTER_RE.match(content)
        if fm:
            body = content[fm.end():]
            for line in fm.group(1).splitlines():
                if ":" in line:
                    k, v = line.split(":", 1)
                    frontmatter[k.strip().lower()] = v.strip()

        # Title: frontmatter title > first heading > filename
        title = frontmatter.get("title", "")
        if not title:
            heading = re.search(r"^#\s+(.+)$", body, re.MULTILINE)
            title = heading.group(1).strip() if heading else note_path.stem

        # Wikilinks
        wikilinks = list(set(WIKILINK_RE.findall(body)))

        # Tags: frontmatter tags + inline #tags
        tags = set()
        tag_match = TAG_LINE_RE.search(content)
        if tag_match:
            raw = tag_match.group(1)
            # Handle YAML arrays: [tag1, tag2] or plain: tag1, tag2
            if raw.startswith("[") and raw.endswith("]"):
                raw = raw[1:-1]
            for t in raw.split(","):
                t = t.strip().strip("'\"[]").lstrip("#")
                if t:
                    tags.add(t)
        for t in INLINE_TAG_RE.findall(body):
            tags.add(t)
        tags = sorted(tags - {""})

        # Slug for GBrain page
        slug = slugify(str(rel).replace(".md", ""))

        # Check if this is new or update
        existing = self.db.execute(
            "SELECT gbrain_slug FROM ingested WHERE path = ?", (str(rel),)
        ).fetchone()
        is_update = existing is not None

        # Build GBrain page content with frontmatter
        gbrain_frontmatter = {
            "type": "note",
            "source": self.source_slug,
            "title": title,
            "original_path": str(rel),
            "content_hash": content_hash,
            "file_mtime": stat.st_mtime,
            "synced_at": datetime.now(timezone.utc).isoformat(),
            **{k: v for k, v in frontmatter.items() if k not in ("title",)},
        }
        fm_lines = ["---"]
        for k, v in gbrain_frontmatter.items():
            if isinstance(v, str) and (":" in v or '"' in v):
                v = json.dumps(v)
            elif isinstance(v, (int, float)):
                v = str(v)
            elif isinstance(v, list):
                v = json.dumps(v)
            fm_lines.append(f"{k}: {v}")
        fm_lines.append("---\n")

        page_content = "\n".join(fm_lines) + body

        # PUT page to GBrain
        try:
            result = mcp_call(self.gbrain_url, self.gbrain_token, "put_page", {
                "slug": slug,
                "content": page_content,
            })
            gbrain_slug = slug
        except Exception as e:
            raise RuntimeError(f"put_page failed: {e}")

        # Add tags
        for tag in tags[:20]:
            try:
                mcp_call(self.gbrain_url, self.gbrain_token, "add_tag", {
                    "slug": gbrain_slug,
                    "tag": tag,
                })
            except Exception:
                pass  # tag errors are non-fatal

        # Add wikilinks as graph links
        for link_target in wikilinks[:50]:
            target_slug = slugify(link_target)
            try:
                mcp_call(self.gbrain_url, self.gbrain_token, "add_link", {
                    "from": gbrain_slug,
                    "to": target_slug,
                    "link_type": "wikilink",
                })
            except Exception:
                pass  # link errors non-fatal (target may not exist yet)

        # Log the ingest
        try:
            mcp_call(self.gbrain_url, self.gbrain_token, "log_ingest", {
                "source": self.source_slug,
                "external_id": str(rel),
                "action": "update" if is_update else "create",
                "title": title,
            })
        except Exception:
            pass

        now = datetime.now(timezone.utc).isoformat()
        self.db.execute(
            """INSERT OR REPLACE INTO ingested (path, mtime, content_hash, gbrain_slug, ingested_at)
               VALUES (?, ?, ?, ?, ?)""",
            (str(rel), stat.st_mtime, content_hash, gbrain_slug, now),
        )
        self.db.commit()

        return "updated" if is_update else "new"


def main():
    config_path = os.environ.get("VAULT_SYNC_CONFIG", "/etc/vault-sync.json")
    sync = VaultSync(config_path)
    sync.run()


if __name__ == "__main__":
    main()