#!/usr/bin/env python3 """Persistent collector settings and MAC→name map.""" from __future__ import annotations import json import logging from pathlib import Path from typing import Dict from config import ( DEFAULT_REPORT_INTERVAL_MIN, names_path, settings_path, ) log = logging.getLogger("collector.settings") def normalize_addr(addr: str) -> str: return (addr or "").strip().upper() def load_report_interval_min(path: Path | None = None) -> int: p = path or settings_path() if not p.exists(): return DEFAULT_REPORT_INTERVAL_MIN try: raw = json.loads(p.read_text(encoding="utf-8")) v = int(raw.get("report_interval_min", DEFAULT_REPORT_INTERVAL_MIN)) return max(1, min(65535, v)) except (OSError, ValueError, TypeError) as e: log.warning("settings load failed: %s", e) return DEFAULT_REPORT_INTERVAL_MIN def save_report_interval_min(minutes: int, path: Path | None = None) -> int: p = path or settings_path() v = max(1, min(65535, int(minutes))) data = {"report_interval_min": v} tmp = Path(str(p) + ".tmp") try: p.parent.mkdir(parents=True, exist_ok=True) tmp.write_text(json.dumps(data, indent=2), encoding="utf-8") tmp.replace(p) except OSError as e: log.error("settings save failed: %s", e) return v def load_names(path: Path | None = None) -> Dict[str, str]: p = path or names_path() if not p.exists(): return {} try: raw = json.loads(p.read_text(encoding="utf-8")) except (OSError, ValueError) as e: log.warning("names load failed: %s", e) return {} if not isinstance(raw, dict): return {} out: Dict[str, str] = {} for k, v in raw.items(): addr = normalize_addr(str(k)) name = str(v).strip() if addr and name: out[addr] = name return out def save_names(names: Dict[str, str], path: Path | None = None) -> Dict[str, str]: p = path or names_path() cleaned: Dict[str, str] = {} for k, v in (names or {}).items(): addr = normalize_addr(str(k)) name = str(v).strip() if addr and name: cleaned[addr] = name tmp = Path(str(p) + ".tmp") try: p.parent.mkdir(parents=True, exist_ok=True) tmp.write_text(json.dumps(cleaned, indent=2, ensure_ascii=False), encoding="utf-8") tmp.replace(p) except OSError as e: log.error("names save failed: %s", e) return cleaned