import base64 import json import sqlite3 from edge.events import Event, EventType, Severity class OfflineQueue: def __init__(self, db_path: str, client): self.client = client self.db = sqlite3.connect(db_path) self.db.execute( "CREATE TABLE IF NOT EXISTS queue (" "id INTEGER PRIMARY KEY AUTOINCREMENT, payload TEXT, snaps TEXT)" ) self.db.commit() def _rehydrate(self, payload: dict, snaps: list[bytes]) -> tuple[Event, list[bytes]]: e = Event(EventType(payload["event_type"]), Severity(payload["severity"]), payload["camera_id"], payload["track_id"], payload["zone_id"], payload["occurred_at"], payload["duration_s"], payload["metrics"], tuple(payload["spatial"]) if payload.get("spatial") else None) return e, snaps def enqueue(self, event: Event, snapshots: list[bytes]) -> None: payload = event.to_payload("") # device_id는 client가 다시 채움 payload["event_type"] = event.event_type.value payload["severity"] = event.severity.value payload["spatial"] = list(event.spatial) if event.spatial else None snaps = [base64.b64encode(s).decode() for s in snapshots] self.db.execute("INSERT INTO queue (payload, snaps) VALUES (?, ?)", (json.dumps(payload), json.dumps(snaps))) self.db.commit() def try_send(self, event: Event, snapshots: list[bytes]) -> bool: if self.client.send(event, snapshots): return True self.enqueue(event, snapshots) return False def flush(self) -> int: rows = self.db.execute("SELECT id, payload, snaps FROM queue ORDER BY id").fetchall() sent = 0 for rid, payload_json, snaps_json in rows: payload = json.loads(payload_json) snaps = [base64.b64decode(s) for s in json.loads(snaps_json)] event, snapshots = self._rehydrate(payload, snaps) if self.client.send(event, snapshots): self.db.execute("DELETE FROM queue WHERE id = ?", (rid,)) self.db.commit() sent += 1 else: break # 여전히 오프라인 — 순서 보존 return sent def pending(self) -> int: return self.db.execute("SELECT COUNT(*) FROM queue").fetchone()[0]