Add Obsidian ingest and export tasks
This commit is contained in:
109
cron_tasks/export_obsidian.py
Normal file
109
cron_tasks/export_obsidian.py
Normal file
@@ -0,0 +1,109 @@
|
||||
#!/usr/bin/env python3
|
||||
"""Export Second-Brain engrams as Markdown notes into a configured Obsidian vault."""
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from pathlib import Path
|
||||
|
||||
BRAIN_DIR = Path(__file__).resolve().parents[1]
|
||||
sys.path.insert(0, str(BRAIN_DIR))
|
||||
|
||||
from src.store import EngramStore
|
||||
|
||||
|
||||
DATA_DIR = BRAIN_DIR / "data"
|
||||
CONFIG_PATH = DATA_DIR / "obsidian_config.json"
|
||||
DEFAULT_STATE_PATH = DATA_DIR / "obsidian_export_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 slugify(value: str) -> str:
|
||||
slug = re.sub(r"[^A-Za-z0-9._-]+", "-", value).strip("-._")
|
||||
return slug[:80] or "engram"
|
||||
|
||||
|
||||
def render_markdown(eg) -> str:
|
||||
tags = eg.metadata.get("tags", []) or []
|
||||
tag_line = " ".join(f"#{slugify(str(t))}" for t in tags if str(t).strip())
|
||||
return (
|
||||
"---\n"
|
||||
f"id: {eg.id}\n"
|
||||
f"source: {json.dumps(eg.metadata.get('source', ''), ensure_ascii=False)}\n"
|
||||
f"created: {eg.metadata.get('created', '')}\n"
|
||||
f"modified: {eg.metadata.get('modified', '')}\n"
|
||||
f"confidence: {eg.compute_confidence():.3f}\n"
|
||||
f"tags: {json.dumps(tags, ensure_ascii=False)}\n"
|
||||
"---\n\n"
|
||||
f"{tag_line}\n\n"
|
||||
f"{eg.content.strip()}\n"
|
||||
)
|
||||
|
||||
|
||||
def main() -> int:
|
||||
cfg = load_json(CONFIG_PATH, {})
|
||||
if not cfg.get("enabled", {}).get("export", False):
|
||||
print(json.dumps({"success": True, "skipped": "export 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
|
||||
|
||||
export_cfg = cfg.get("export", {})
|
||||
subdir = export_cfg.get("subdir", "SecondBrain")
|
||||
max_per_run = int(export_cfg.get("max_per_run", 2000))
|
||||
state_path_raw = export_cfg.get("state_path")
|
||||
state_path = (BRAIN_DIR.parent / state_path_raw).resolve() if state_path_raw else DEFAULT_STATE_PATH
|
||||
state = load_json(state_path, {"version": 1, "hash_by_id": {}})
|
||||
hashes = dict(state.get("hash_by_id", {}))
|
||||
|
||||
out_dir = vault / subdir
|
||||
out_dir.mkdir(parents=True, exist_ok=True)
|
||||
|
||||
store = EngramStore(os.environ.get("BRAIN_DB") or str(DATA_DIR / "brain.sqlite"))
|
||||
exported = 0
|
||||
unchanged = 0
|
||||
for eg in store.get_all(limit=max_per_run):
|
||||
text = render_markdown(eg)
|
||||
digest = sha256_text(text)
|
||||
eg_id = str(eg.id)
|
||||
if hashes.get(eg_id) == digest:
|
||||
unchanged += 1
|
||||
continue
|
||||
title = slugify(eg.content.splitlines()[0][:60] if eg.content else eg_id)
|
||||
path = out_dir / f"{title}-{eg_id[:8]}.md"
|
||||
path.write_text(text, encoding="utf-8")
|
||||
hashes[eg_id] = digest
|
||||
exported += 1
|
||||
|
||||
write_json(state_path, {"version": 1, "hash_by_id": hashes})
|
||||
print(json.dumps({"success": True, "exported": exported, "unchanged": unchanged, "target": str(out_dir)}))
|
||||
return 0
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
raise SystemExit(main())
|
||||
122
cron_tasks/ingest_obsidian.py
Normal file
122
cron_tasks/ingest_obsidian.py
Normal file
@@ -0,0 +1,122 @@
|
||||
#!/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())
|
||||
Reference in New Issue
Block a user