import json
|
import logging
|
import time
|
from pathlib import Path
|
from threading import Lock
|
from typing import Any, Literal
|
|
from app.config import settings
|
|
logger = logging.getLogger(__name__)
|
|
InitSource = Literal["auto", "manual"]
|
|
|
class InitRegistry:
|
def __init__(self, storage_path: Path | None = None) -> None:
|
self.storage_path = storage_path or settings.resolved_initialized_devices_file
|
self._lock = Lock()
|
self._data = self._load()
|
|
def _load(self) -> dict[str, Any]:
|
if self.storage_path.exists():
|
with self.storage_path.open(encoding="utf-8") as f:
|
return json.load(f)
|
return {"devices": {}}
|
|
def _save(self) -> None:
|
self.storage_path.parent.mkdir(parents=True, exist_ok=True)
|
with self.storage_path.open("w", encoding="utf-8") as f:
|
json.dump(self._data, f, ensure_ascii=False, indent=2)
|
|
def is_initialized(self, device_id: str) -> bool:
|
with self._lock:
|
return device_id in self._data["devices"]
|
|
def try_claim(
|
self,
|
device_id: str,
|
source: InitSource,
|
mode: str | None = None,
|
timestamp: int | None = None,
|
) -> bool:
|
with self._lock:
|
if device_id in self._data["devices"]:
|
return False
|
self._data["devices"][device_id] = {
|
"initialized_at": timestamp or int(time.time()),
|
"source": source,
|
"mode": mode,
|
}
|
self._save()
|
return True
|
|
def mark_initialized(
|
self,
|
device_id: str,
|
source: InitSource,
|
mode: str | None = None,
|
timestamp: int | None = None,
|
) -> None:
|
with self._lock:
|
self._data["devices"][device_id] = {
|
"initialized_at": timestamp or int(time.time()),
|
"source": source,
|
"mode": mode,
|
}
|
self._save()
|
|
def remove(self, device_id: str) -> bool:
|
with self._lock:
|
if device_id not in self._data["devices"]:
|
return False
|
del self._data["devices"][device_id]
|
self._save()
|
return True
|
|
def list_initialized(self) -> dict[str, Any]:
|
with self._lock:
|
return dict(self._data["devices"])
|
|
def get_record(self, device_id: str) -> dict[str, Any] | None:
|
with self._lock:
|
record = self._data["devices"].get(device_id)
|
return dict(record) if record else None
|
|
def update_record(self, device_id: str, **fields: Any) -> bool:
|
with self._lock:
|
if device_id not in self._data["devices"]:
|
return False
|
self._data["devices"][device_id].update(fields)
|
self._save()
|
return True
|
