| 71 | |
| 72 | |
| 73 | class FileSpendControlStorage(SpendControlStorage): |
| 74 | def __init__(self, path: Path | None = None) -> None: |
| 75 | self._path = path or (_DATA_DIR / "spending.json") |
| 76 | |
| 77 | def load(self) -> dict[str, Any] | None: |
| 78 | try: |
| 79 | if self._path.exists(): |
| 80 | raw = json.loads(self._path.read_text()) |
| 81 | return self._validate(raw) |
| 82 | except Exception: |
| 83 | pass |
| 84 | return None |
| 85 | |
| 86 | def save(self, data: dict[str, Any]) -> None: |
| 87 | try: |
| 88 | self._path.parent.mkdir(parents=True, exist_ok=True, mode=0o700) |
| 89 | self._path.write_text(json.dumps(data, indent=2, default=str)) |
| 90 | self._path.chmod(0o600) |
| 91 | except Exception as exc: |
| 92 | import sys |
| 93 | print(f"[UncommonRoute] Failed to save spending data: {exc}", file=sys.stderr) |
| 94 | |
| 95 | @staticmethod |
| 96 | def _validate(raw: dict) -> dict[str, Any]: |
| 97 | limits: dict[str, float] = {} |
| 98 | for key in ("per_request", "hourly", "daily", "session"): |
| 99 | val = raw.get("limits", {}).get(key) |
| 100 | if isinstance(val, (int, float)) and val > 0: |
| 101 | limits[key] = float(val) |
| 102 | history: list[dict] = [] |
| 103 | for r in raw.get("history", []): |
| 104 | if isinstance(r, dict) and isinstance(r.get("timestamp"), (int, float)) and isinstance(r.get("amount"), (int, float)): |
| 105 | history.append({ |
| 106 | "timestamp": r["timestamp"], |
| 107 | "amount": r["amount"], |
| 108 | "model": r.get("model"), |
| 109 | "action": r.get("action"), |
| 110 | }) |
| 111 | return {"limits": limits, "history": history} |
| 112 | |
| 113 | |
| 114 | class InMemorySpendControlStorage(SpendControlStorage): |