86 lines
2.5 KiB
Python
86 lines
2.5 KiB
Python
#!/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
|