jensandClaude Sonnet 4.6
b57e2a9cc2
Replace load_today_records with load_last_24h_records which computes a rolling 24 h cutoff, globs log files for both the current and previous calendar day, and filters records by timestamp — so clients always receive a full 24 h window regardless of when midnight falls. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
63 lines
2.2 KiB
Python
63 lines
2.2 KiB
Python
import json
|
|
import logging
|
|
from datetime import datetime, timedelta, timezone
|
|
from pathlib import Path
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
|
|
def auto_out_path(logs_dir: Path, vin: str) -> Path:
|
|
stamp = datetime.now(timezone.utc).strftime("%Y_%m_%d_%H_%M_%S")
|
|
return logs_dir / f"{stamp}_{vin}.json"
|
|
|
|
|
|
def load_store(path: Path, vin: str, interval: int) -> dict:
|
|
if path.exists():
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
if isinstance(data, dict) and "records" in data:
|
|
return data
|
|
except json.JSONDecodeError:
|
|
log.warning("Output file is corrupt — starting fresh")
|
|
return {
|
|
"meta": {
|
|
"vin": vin,
|
|
"created_at": datetime.now(timezone.utc).isoformat(),
|
|
"interval_seconds": interval,
|
|
},
|
|
"records": [],
|
|
}
|
|
|
|
|
|
def save_store(path: Path, store: dict, max_records: int | None) -> None:
|
|
if max_records and len(store["records"]) > max_records:
|
|
store["records"] = store["records"][-max_records:]
|
|
path.write_text(json.dumps(store, indent=2, default=str))
|
|
|
|
|
|
def load_last_24h_records(logs_dir: Path, vin: str) -> list:
|
|
"""Load records from the last 24 hours across log files (may span midnight)."""
|
|
now = datetime.now(timezone.utc)
|
|
cutoff = now - timedelta(hours=24)
|
|
cutoff_str = cutoff.isoformat()
|
|
|
|
# Log files are named YYYY_MM_DD_…_VIN.json; the 24 h window can span two calendar days.
|
|
prefixes = {
|
|
cutoff.strftime("%Y_%m_%d_"),
|
|
now.strftime("%Y_%m_%d_"),
|
|
}
|
|
|
|
records = []
|
|
for prefix in sorted(prefixes):
|
|
for path in sorted(logs_dir.glob(f"{prefix}*_{vin}.json")):
|
|
try:
|
|
data = json.loads(path.read_text())
|
|
if isinstance(data, dict) and "records" in data:
|
|
kept = [r for r in data["records"] if r.get("ts", "") >= cutoff_str]
|
|
records.extend(kept)
|
|
log.debug("Loaded %d/%d records from %s", len(kept), len(data["records"]), path.name)
|
|
except (json.JSONDecodeError, OSError):
|
|
log.warning("Could not load %s", path)
|
|
|
|
records.sort(key=lambda r: r.get("ts", ""))
|
|
return records |