res_levis_API/logica_reslevis/tracker.py

import json
from typing import List, Dict, Any, Optional

from fastapi.encoders import jsonable_encoder

from schemas.reslevis import TrackerItem
from .config import TRACKER_JSON_PATH
from .gateway import _LockedFile, _atomic_write, _norm_str, _index_by_id


class TrackerJsonRepository:
    def __init__(self, json_path: str = TRACKER_JSON_PATH):
        self.path = json_path

    def _read_all(self) -> List[Dict[str, Any]]:
        with _LockedFile(self.path, "r") as fp:
            try:
                fp.seek(0)
                data = fp.read().strip()
                return json.loads(data) if data else []
            except json.JSONDecodeError:
                return []

    def _write_all(self, rows: List[Dict[str, Any]]) -> None:
        payload = json.dumps(rows, ensure_ascii=False, indent=2)
        _atomic_write(self.path, payload)

    def list(self, search_string: Optional[str] = None) -> List[Dict[str, Any]]:
        rows = self._read_all()
        if search_string:
            s = search_string.lower()
            rows = [
                r
                for r in rows
                if any(
                    isinstance(r.get(k), str) and s in (r.get(k) or "").lower()
                    for k in ("name", "mac", "model", "status", "position", "notes")
                )
            ]
        return rows

    def add(self, item: TrackerItem) -> None:
        rows = self._read_all()
        obj = jsonable_encoder(item)
        obj_id = _norm_str(obj.get("id"))
        mac = _norm_str(obj.get("mac"))

        if any(_norm_str(r.get("id")) == obj_id for r in rows):
            raise ValueError(f"Tracker con id '{obj_id}' già presente")

        if mac and any(_norm_str(r.get("mac")) == mac for r in rows):
            raise ValueError(f"Tracker con mac '{mac}' già presente")

        rows.append(obj)
        self._write_all(rows)

    def update(self, item: TrackerItem) -> None:
        rows = self._read_all()
        obj = jsonable_encoder(item)
        obj_id = _norm_str(obj.get("id"))
        mac = _norm_str(obj.get("mac"))

        idx = _index_by_id(rows, obj_id)
        if idx is None:
            raise ValueError(f"Tracker con id '{obj_id}' non trovato")

        if mac and any(
            _norm_str(r.get("mac")) == mac and _norm_str(r.get("id")) != obj_id
            for r in rows
        ):
            raise ValueError(f"Tracker con mac '{mac}' già presente")

        rows[idx] = obj
        self._write_all(rows)

    def remove(self, tracker_id: str) -> None:
        rows = self._read_all()
        idx = _index_by_id(rows, tracker_id)
        if idx is None:
            raise ValueError(f"Tracker con id '{tracker_id}' non trovato")

        del rows[idx]
        self._write_all(rows)