89 lines
2.4 KiB
Python
89 lines
2.4 KiB
Python
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)
|