75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
""".manifest.json read/write for the vault.
|
|
|
|
The manifest records, per page, the source key, content hash, vault path and
|
|
update time — so batch ingest can skip already-sedimented threads (delta/dedup)
|
|
exactly like obsidian-wiki's native ``.manifest.json``.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import hashlib
|
|
import json
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
|
|
def content_hash(text: str) -> str:
|
|
"""Stable sha256 hex digest of ``text`` (used for dedup)."""
|
|
return hashlib.sha256((text or "").encode("utf-8")).hexdigest()
|
|
|
|
|
|
class Manifest:
|
|
"""JSON manifest stored at ``<vault>/.manifest.json``."""
|
|
|
|
def __init__(self, root: Path | str) -> None:
|
|
self.path = Path(root).resolve() / ".manifest.json"
|
|
|
|
def load(self) -> dict[str, Any]:
|
|
if not self.path.exists():
|
|
return {"version": 1, "updated": None, "pages": {}, "sources": {}}
|
|
try:
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
except (json.JSONDecodeError, OSError):
|
|
return {"version": 1, "updated": None, "pages": {}, "sources": {}}
|
|
data.setdefault("pages", {})
|
|
data.setdefault("sources", {})
|
|
return data
|
|
|
|
def save(self, data: dict[str, Any]) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
self.path.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
|
|
def record_page(
|
|
self,
|
|
*,
|
|
note_id: str,
|
|
vault_path: str,
|
|
source_key: str | None,
|
|
hash_value: str,
|
|
updated: str,
|
|
) -> None:
|
|
data = self.load()
|
|
data["pages"][note_id] = {
|
|
"vault_path": vault_path,
|
|
"source": source_key,
|
|
"hash": hash_value,
|
|
"updated": updated,
|
|
}
|
|
if source_key:
|
|
data["sources"][source_key] = {"note_id": note_id, "hash": hash_value, "updated": updated}
|
|
data["updated"] = updated
|
|
self.save(data)
|
|
|
|
def remove_page(self, note_id: str) -> None:
|
|
data = self.load()
|
|
page = data["pages"].pop(note_id, None)
|
|
if page and page.get("source"):
|
|
data["sources"].pop(page["source"], None)
|
|
self.save(data)
|
|
|
|
def has_source(self, source_key: str) -> bool:
|
|
return source_key in self.load().get("sources", {})
|
|
|
|
def get_source(self, source_key: str) -> dict[str, Any] | None:
|
|
return self.load().get("sources", {}).get(source_key)
|