Tracks version history for all regulatory documents.
| 406 | # --------------------------------------------------------------------------- |
| 407 | |
| 408 | class VersionRegistry: |
| 409 | """Tracks version history for all regulatory documents.""" |
| 410 | |
| 411 | def __init__(self, path: Path = VERSION_REGISTRY_PATH): |
| 412 | self.path = path |
| 413 | self.data = self._load() |
| 414 | |
| 415 | def _load(self) -> dict: |
| 416 | if self.path.exists(): |
| 417 | try: |
| 418 | with open(self.path, "r", encoding="utf-8") as f: |
| 419 | return json.load(f) |
| 420 | except Exception: |
| 421 | pass |
| 422 | return {"documents": {}, "last_updated": None} |
| 423 | |
| 424 | def save(self): |
| 425 | self.path.parent.mkdir(parents=True, exist_ok=True) |
| 426 | self.data["last_updated"] = datetime.now().isoformat() |
| 427 | with open(self.path, "w", encoding="utf-8") as f: |
| 428 | json.dump(self.data, f, indent=2, ensure_ascii=False) |
| 429 | |
| 430 | def get(self, doc_id: str) -> Optional[dict]: |
| 431 | return self.data["documents"].get(doc_id) |
| 432 | |
| 433 | def upsert(self, doc_id: str, current: dict): |
| 434 | existing = self.data["documents"].get(doc_id, {}) |
| 435 | history = existing.get("history", []) |
| 436 | if existing.get("current"): |
| 437 | history.append(existing["current"]) |
| 438 | self.data["documents"][doc_id] = {"current": current, "history": history} |
| 439 | |
| 440 | def find_outdated(self, days: int = 180) -> list: |
| 441 | cutoff = datetime.now() - timedelta(days=days) |
| 442 | result = [] |
| 443 | for doc_id, doc in self.data["documents"].items(): |
| 444 | lv = doc.get("current", {}).get("last_verified") |
| 445 | if lv: |
| 446 | try: |
| 447 | if datetime.fromisoformat(lv) < cutoff: |
| 448 | result.append(doc_id) |
| 449 | except Exception: |
| 450 | pass |
| 451 | return result |
| 452 | |
| 453 | |
| 454 | # --------------------------------------------------------------------------- |