46 lines
1.4 KiB
Python
46 lines
1.4 KiB
Python
import datetime
|
|
import json
|
|
import os
|
|
import threading
|
|
from pathlib import Path
|
|
from zoneinfo import ZoneInfo
|
|
|
|
from .navigator import RouteRequest
|
|
|
|
|
|
_MOSCOW = ZoneInfo("Europe/Moscow")
|
|
_LOG_PATH = Path("navigator_requests.json")
|
|
_LOCK = threading.RLock()
|
|
|
|
|
|
def append_navigator_request(request: RouteRequest) -> None:
|
|
"""Append a validated navigator request to the persistent JSON history."""
|
|
with _LOCK:
|
|
try:
|
|
with _LOG_PATH.open("r", encoding="utf-8") as file:
|
|
history = json.load(file)
|
|
except FileNotFoundError:
|
|
history = []
|
|
|
|
if not isinstance(history, list):
|
|
raise ValueError(f"{_LOG_PATH} must contain a JSON list")
|
|
|
|
history.append(
|
|
{
|
|
"time": datetime.datetime.now(_MOSCOW).isoformat(timespec="seconds"),
|
|
"start": {
|
|
"room": request.start.room,
|
|
"floor": request.start.floor,
|
|
},
|
|
"end": {
|
|
"room": request.end.room,
|
|
"floor": request.end.floor,
|
|
},
|
|
}
|
|
)
|
|
|
|
temporary_path = _LOG_PATH.with_suffix(".json.tmp")
|
|
with temporary_path.open("w", encoding="utf-8") as file:
|
|
json.dump(history, file, ensure_ascii=False, indent=2)
|
|
file.write("\n")
|
|
os.replace(temporary_path, _LOG_PATH)
|