from __future__ import annotations import hashlib import json import os import tempfile from pathlib import Path from typing import Any, Iterable from .paths import ENV_FILE def load_project_env() -> None: try: from dotenv import load_dotenv except ImportError: return if ENV_FILE.is_file(): load_dotenv(ENV_FILE, override=False) def atomic_write_text(path: Path, text: str) -> None: path.parent.mkdir(parents=True, exist_ok=True) fd, tmp = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) try: with os.fdopen(fd, "w", encoding="utf-8", newline="") as handle: handle.write(text) handle.flush() os.fsync(handle.fileno()) os.replace(tmp, path) finally: if os.path.exists(tmp): os.unlink(tmp) def atomic_write_json(path: Path, value: Any) -> None: atomic_write_text(path, json.dumps(value, ensure_ascii=False, indent=2) + "\n") def atomic_write_jsonl(path: Path, rows: Iterable[dict[str, Any]]) -> None: atomic_write_text(path, "".join(json.dumps(row, ensure_ascii=False) + "\n" for row in rows)) def load_json(path: Path, default: Any = None) -> Any: if not path.exists(): return default with path.open(encoding="utf-8") as handle: return json.load(handle) def read_jsonl(path: Path) -> list[dict[str, Any]]: if not path.exists(): return [] rows: list[dict[str, Any]] = [] with path.open(encoding="utf-8") as handle: for line_no, line in enumerate(handle, 1): if not line.strip(): continue value = json.loads(line) if not isinstance(value, dict): raise ValueError(f"{path}:{line_no}: expected a JSON object") rows.append(value) return rows def sha256_bytes(data: bytes) -> str: return hashlib.sha256(data).hexdigest() def sha256_file(path: Path) -> str: return sha256_bytes(path.read_bytes()) def sha256_text(text: str) -> str: return sha256_bytes(text.encode("utf-8")) def package_manifest(root: Path) -> list[dict[str, Any]]: return [ {"path": str(p.relative_to(root)), "sha256": sha256_file(p), "bytes": p.stat().st_size} for p in sorted(root.rglob("*")) if p.is_file() ] def package_hash(root: Path) -> str: payload = json.dumps(package_manifest(root), sort_keys=True, separators=(",", ":")) return sha256_text(payload)