100 lines
2.6 KiB
Python
100 lines
2.6 KiB
Python
#!/usr/bin/env python3
|
|
"""UI bridge files: status JSON, log jsonl, command file."""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import logging
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from config import UI_LOG_LIMIT, cmd_path, history_path, status_path, ui_log_path
|
|
|
|
log = logging.getLogger("collector.ui")
|
|
|
|
|
|
def write_status(status: dict) -> None:
|
|
path = status_path()
|
|
tmp = Path(str(path) + ".tmp")
|
|
try:
|
|
tmp.write_text(json.dumps(status, ensure_ascii=False), encoding="utf-8")
|
|
tmp.replace(path)
|
|
except OSError as e:
|
|
log.error("status write failed: %s", e)
|
|
|
|
|
|
def write_history(data: dict) -> None:
|
|
path = history_path()
|
|
tmp = Path(str(path) + ".tmp")
|
|
try:
|
|
tmp.write_text(json.dumps(data, ensure_ascii=False), encoding="utf-8")
|
|
tmp.replace(path)
|
|
except OSError as e:
|
|
log.error("history write failed: %s", e)
|
|
|
|
|
|
def read_history() -> dict:
|
|
path = history_path()
|
|
if not path.exists():
|
|
return {}
|
|
try:
|
|
raw = json.loads(path.read_text(encoding="utf-8"))
|
|
except (OSError, ValueError):
|
|
return {}
|
|
return raw if isinstance(raw, dict) else {}
|
|
|
|
|
|
def append_ui_log(level: str, message: str, **extra: Any) -> None:
|
|
path = ui_log_path()
|
|
entry = {
|
|
"ts": time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime()),
|
|
"level": level,
|
|
"msg": message,
|
|
}
|
|
entry.update(extra)
|
|
try:
|
|
with path.open("a", encoding="utf-8") as f:
|
|
f.write(json.dumps(entry, ensure_ascii=False) + "\n")
|
|
_truncate_log(path, UI_LOG_LIMIT)
|
|
except OSError as e:
|
|
log.error("ui log write failed: %s", e)
|
|
|
|
|
|
def _truncate_log(path: Path, limit: int) -> None:
|
|
try:
|
|
lines = path.read_text(encoding="utf-8").splitlines()
|
|
except OSError:
|
|
return
|
|
if len(lines) <= limit:
|
|
return
|
|
try:
|
|
path.write_text("\n".join(lines[-limit:]) + "\n", encoding="utf-8")
|
|
except OSError:
|
|
pass
|
|
|
|
|
|
def read_cmd() -> Optional[Dict[str, Any]]:
|
|
path = cmd_path()
|
|
if not path.exists():
|
|
return None
|
|
try:
|
|
raw = path.read_text(encoding="utf-8").strip()
|
|
path.unlink(missing_ok=True)
|
|
except OSError as e:
|
|
log.error("cmd read failed: %s", e)
|
|
return None
|
|
if not raw:
|
|
return {"op": "advertise"}
|
|
try:
|
|
return json.loads(raw)
|
|
except ValueError:
|
|
return {"op": "advertise"}
|
|
|
|
|
|
def write_advertise_cmd() -> None:
|
|
"""Helper for external callers (tests)."""
|
|
try:
|
|
cmd_path().write_text(json.dumps({"op": "advertise"}), encoding="utf-8")
|
|
except OSError as e:
|
|
log.error("cmd write failed: %s", e)
|