44 lines
1.5 KiB
Python
44 lines
1.5 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
from dataclasses import asdict, dataclass, field
|
|
from pathlib import Path
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class OpenLaneCheckpoint:
|
|
max_pages: int
|
|
completed_pages: list[int] = field(default_factory=list)
|
|
failed_pages: list[int] = field(default_factory=list)
|
|
total_records: int = 0
|
|
last_saved_at: str | None = None
|
|
|
|
@property
|
|
def completed_set(self) -> set[int]:
|
|
return set(self.completed_pages)
|
|
|
|
|
|
class OpenLaneCheckpointStore:
|
|
def __init__(self, path: str | Path) -> None:
|
|
self.path = Path(path)
|
|
|
|
def load(self, max_pages: int) -> OpenLaneCheckpoint:
|
|
if not self.path.exists():
|
|
return OpenLaneCheckpoint(max_pages=max_pages)
|
|
data = json.loads(self.path.read_text(encoding="utf-8"))
|
|
return OpenLaneCheckpoint(
|
|
max_pages=int(data.get("max_pages") or max_pages),
|
|
completed_pages=sorted({int(page) for page in data.get("completed_pages", [])}),
|
|
failed_pages=sorted({int(page) for page in data.get("failed_pages", [])}),
|
|
total_records=int(data.get("total_records") or 0),
|
|
last_saved_at=data.get("last_saved_at"),
|
|
)
|
|
|
|
def save(self, checkpoint: OpenLaneCheckpoint) -> None:
|
|
self.path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = asdict(checkpoint)
|
|
self.path.write_text(
|
|
json.dumps(payload, ensure_ascii=False, indent=2, sort_keys=True),
|
|
encoding="utf-8",
|
|
)
|