spb/air Public MIT

1# =============================================================================2# Projet : AIR — Accounting Intermediate Representation3# Auteur : Simon-Pierre Boucher4# Contact : contact@spboucher.ai5# Fichier : workspace.py6# Description : AIR home — a managed local data directory: ledger, archived documents, exports.7# =============================================================================8"""The AIR home: managed local data for standalone users.910Users without any third-party system keep everything in one directory:1112    <home>/13    ├── meta.json         # company name, default policy set, created date14    ├── ledger.jsonl      # the hash-chained book of record15    ├── documents/        # every AIR document ever posted (content-addressed)16    └── exports/          # generated CSV/JSON exports and reports1718Documents are archived content-addressed (sha256 prefix + original name), so19the exact input of every posting can be replayed later — pairing with the20ledger's idempotency keys and the compiler's determinism, the entire book of21record is reproducible from this directory alone.22"""23from __future__ import annotations2425import hashlib26import json27from dataclasses import dataclass28from pathlib import Path2930from kernel.ledger import Ledger3132META_FILE = "meta.json"333435class WorkspaceError(RuntimeError):36    pass373839@dataclass40class Workspace:41    root: Path4243    # -- lifecycle -----------------------------------------------------------44    @classmethod45    def init(46        cls,47        root: str | Path,48        *,49        name: str = "my-books",50        policies: str | None = None,51        created: str | None = None,52    ) -> "Workspace":53        root = Path(root)54        if (root / META_FILE).exists():55            raise WorkspaceError(f"workspace already initialized: {root}")56        (root / "documents").mkdir(parents=True, exist_ok=True)57        (root / "exports").mkdir(parents=True, exist_ok=True)58        meta = {59            "_author": "Simon-Pierre Boucher <contact@spboucher.ai>",60            "air_home_version": "1",61            "name": name,62            "policies": policies,63            "created": created,64        }65        (root / META_FILE).write_text(json.dumps(meta, indent=2), encoding="utf-8")66        (root / "ledger.jsonl").touch()67        return cls(root=root)6869    @classmethod70    def open(cls, root: str | Path) -> "Workspace":71        root = Path(root)72        if not (root / META_FILE).exists():73            raise WorkspaceError(74                f"no AIR home at {root} — run: air init --home {root} "75                "--policies <policy-set.yaml>"76            )77        return cls(root=root)7879    # -- accessors -------------------------------------------------------------80    @property81    def meta(self) -> dict:82        return json.loads((self.root / META_FILE).read_text(encoding="utf-8"))8384    @property85    def ledger_path(self) -> Path:86        return self.root / "ledger.jsonl"8788    @property89    def exports_dir(self) -> Path:90        return self.root / "exports"9192    def ledger(self) -> Ledger:93        return Ledger(path=self.ledger_path)9495    def default_policies(self) -> str | None:96        return self.meta.get("policies")9798    # -- data management -----------------------------------------------------------99    def archive_document(self, path: str | Path) -> Path:100        """Store a posted AIR document content-addressed; idempotent."""101        path = Path(path)102        content = path.read_bytes()103        digest = hashlib.sha256(content).hexdigest()[:12]104        target = self.root / "documents" / f"{digest}_{path.name}"105        if not target.exists():106            target.write_bytes(content)107        return target108109    def save_export(self, filename: str, content: str) -> Path:110        target = self.exports_dir / filename111        target.write_text(content, encoding="utf-8")112        return target113114    def status(self) -> dict:115        ledger = self.ledger()116        documents = sorted((self.root / "documents").glob("*"))117        exports = sorted(self.exports_dir.glob("*"))118        return {119            "home": str(self.root),120            "name": self.meta.get("name"),121            "policies": self.meta.get("policies"),122            "entries": len(ledger.entries),123            "chain_valid": ledger.verify_chain(),124            "documents_archived": len(documents),125            "exports": len(exports),126        }127