123 lines
3.9 KiB
Python
123 lines
3.9 KiB
Python
#!/usr/bin/env python3
|
|
"""Import Markdown notes from a configured Obsidian vault into Second-Brain."""
|
|
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
BRAIN_DIR = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(BRAIN_DIR))
|
|
|
|
from src.engram import Engram, Grounding
|
|
from src.store import EngramStore
|
|
|
|
|
|
DATA_DIR = BRAIN_DIR / "data"
|
|
CONFIG_PATH = DATA_DIR / "obsidian_config.json"
|
|
STATE_PATH = DATA_DIR / "obsidian_ingest_state.json"
|
|
|
|
|
|
def load_json(path: Path, default: dict) -> dict:
|
|
if not path.exists():
|
|
return default
|
|
try:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
except Exception:
|
|
return default
|
|
|
|
|
|
def write_json(path: Path, data: dict) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
|
|
def sha256_text(text: str) -> str:
|
|
return hashlib.sha256(text.encode("utf-8")).hexdigest()
|
|
|
|
|
|
def is_excluded(relative: str, patterns: list[str]) -> bool:
|
|
p = Path(relative)
|
|
for pattern in patterns:
|
|
if p.match(pattern) or relative.startswith(pattern.rstrip("*")):
|
|
return True
|
|
return False
|
|
|
|
|
|
def main() -> int:
|
|
cfg = load_json(CONFIG_PATH, {})
|
|
if not cfg.get("enabled", {}).get("ingest", False):
|
|
print(json.dumps({"success": True, "skipped": "ingest disabled"}))
|
|
return 0
|
|
|
|
vault = Path(cfg.get("vault_path") or "")
|
|
if not vault.exists() or not vault.is_dir():
|
|
print(json.dumps({"success": True, "skipped": "vault_path missing or invalid", "vault_path": str(vault)}))
|
|
return 0
|
|
if cfg.get("require_obsidian_dir", True) and not (vault / ".obsidian").is_dir():
|
|
print(json.dumps({"success": True, "skipped": "vault is missing .obsidian", "vault_path": str(vault)}))
|
|
return 0
|
|
|
|
ingest_cfg = cfg.get("ingest", {})
|
|
include_glob = ingest_cfg.get("include_glob", "**/*.md")
|
|
exclude_globs = ingest_cfg.get("exclude_globs", [".obsidian/**", ".trash/**", "SecondBrain/**"])
|
|
min_chars = int(ingest_cfg.get("min_chars", 20))
|
|
max_files = int(ingest_cfg.get("max_files_per_run", 2000))
|
|
max_content_chars = int(ingest_cfg.get("max_content_chars", 4000))
|
|
state_max_seen = int(ingest_cfg.get("state_max_seen", 5000))
|
|
|
|
state = load_json(STATE_PATH, {"version": 2, "files": {}})
|
|
seen = dict(state.get("files", {}))
|
|
store = EngramStore(os.environ.get("BRAIN_DB") or str(DATA_DIR / "brain.sqlite"))
|
|
|
|
imported = 0
|
|
unchanged = 0
|
|
skipped = 0
|
|
processed = 0
|
|
for path in sorted(vault.glob(include_glob)):
|
|
if processed >= max_files:
|
|
break
|
|
if not path.is_file():
|
|
continue
|
|
rel = path.relative_to(vault).as_posix()
|
|
if is_excluded(rel, exclude_globs):
|
|
skipped += 1
|
|
continue
|
|
processed += 1
|
|
try:
|
|
text = path.read_text(encoding="utf-8", errors="replace").strip()
|
|
except Exception:
|
|
skipped += 1
|
|
continue
|
|
if len(text) < min_chars:
|
|
skipped += 1
|
|
continue
|
|
digest = sha256_text(text)
|
|
if seen.get(rel) == digest:
|
|
unchanged += 1
|
|
continue
|
|
|
|
content = text[:max_content_chars]
|
|
eg = Engram.create(
|
|
content=content,
|
|
source=f"obsidian:{rel}",
|
|
tags=["obsidian", "imported"],
|
|
grounding=Grounding.SOURCED,
|
|
)
|
|
eg.metadata["obsidian_path"] = rel
|
|
eg.metadata["obsidian_hash"] = digest
|
|
store.save(eg)
|
|
seen[rel] = digest
|
|
imported += 1
|
|
|
|
if len(seen) > state_max_seen:
|
|
seen = dict(list(seen.items())[-state_max_seen:])
|
|
write_json(STATE_PATH, {"version": 2, "files": seen})
|
|
print(json.dumps({"success": True, "imported": imported, "unchanged": unchanged, "skipped": skipped, "processed": processed}))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|