weighing-controller/orange_pi/outbox.py

"""
outbox.py — надёжная доставка взвешиваний на сервер (store-and-forward).

При обрыве сети OPi↔сервер события не теряются: кладём JPEG на диск + метаданные
в SQLite, и досылаем при восстановлении связи. Сервер дедупит по event_id (upsert).
Семантика: at-least-once (повторная отправка безопасна).

Зависимости: только stdlib (sqlite3, urllib, json).
"""

import os
import time
import base64
import sqlite3
import threading
import logging
import json
import urllib.request
import urllib.error

log = logging.getLogger("outbox")


class CaptureOutbox:
    def __init__(self, server_url, device_id,
                 db_path="/root/scales_outbox/outbox.db",
                 photo_dir="/root/scales_outbox/photos",
                 timeout=15, max_attempts=20):
        self.url = server_url.rstrip("/") + "/api/capture"
        self.device_id = device_id
        self.timeout = timeout
        self.max_attempts = max_attempts
        self.photo_dir = photo_dir

        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        os.makedirs(photo_dir, exist_ok=True)

        self._lock = threading.Lock()        # защита БД (heartbeat + main)
        self._flush_lock = threading.Lock()  # одна flush за раз
        self._db = sqlite3.connect(db_path, check_same_thread=False)
        self._db.execute("PRAGMA journal_mode=WAL")
        self._db.execute("""
            CREATE TABLE IF NOT EXISTS outbox (
                event_id    TEXT PRIMARY KEY,
                weight      REAL,
                weight_raw  TEXT,
                captured_at TEXT,
                photo_path  TEXT NOT NULL,
                created_at  TEXT NOT NULL,
                attempts    INTEGER NOT NULL DEFAULT 0,
                status      TEXT NOT NULL DEFAULT 'pending'   -- pending | failed
            )""")
        self._db.commit()
        self._reconcile()

    def _reconcile(self):
        """После рестарта убрать осиротевшие JPEG."""
        with self._lock:
            known = {r[0] for r in self._db.execute("SELECT photo_path FROM outbox")}
        for name in os.listdir(self.photo_dir):
            p = os.path.join(self.photo_dir, name)
            if p not in known:
                try:
                    os.remove(p)
                except OSError:
                    pass

    # ── публичный API ─────────────────────────────────────────────
    def enqueue(self, event_id, weight, weight_raw, captured_at, jpeg_bytes):
        """Положить взвешивание в очередь. JPEG на диск, метаданные в БД."""
        photo_path = os.path.join(self.photo_dir, f"{event_id}.jpg")
        with open(photo_path, "wb") as f:
            f.write(jpeg_bytes)
        with self._lock:
            self._db.execute(
                "INSERT OR REPLACE INTO outbox "
                "(event_id, weight, weight_raw, captured_at, photo_path, created_at, attempts, status) "
                "VALUES (?,?,?,?,?,?,0,'pending')",
                (event_id, weight, weight_raw, captured_at, photo_path,
                 time.strftime("%Y-%m-%dT%H:%M:%SZ", time.gmtime())))
            self._db.commit()
        log.info("enqueued %s (%.1f kg)", event_id, weight)

    def pending_count(self):
        """Количество событий в очереди."""
        with self._lock:
            return self._db.execute(
                "SELECT COUNT(*) FROM outbox WHERE status='pending'").fetchone()[0]

    def flush(self):
        """Досылает pending-события (старые первыми). Вернёт (отправлено, осталось)."""
        if not self._flush_lock.acquire(blocking=False):
            return (0, self.pending_count())
        sent = 0
        try:
            while True:
                with self._lock:
                    row = self._db.execute(
                        "SELECT event_id, weight, weight_raw, captured_at, photo_path, attempts "
                        "FROM outbox WHERE status='pending' ORDER BY rowid LIMIT 1").fetchone()
                if row is None:
                    break
                event_id, weight, weight_raw, captured_at, photo_path, attempts = row
                try:
                    with open(photo_path, "rb") as f:
                        b64 = base64.b64encode(f.read()).decode()
                except OSError:
                    self._drop(event_id)
                    continue

                payload = json.dumps({
                    "event_id": event_id,
                    "weight": weight,
                    "weight_raw": weight_raw,
                    "device_id": self.device_id,
                    "captured_at": captured_at,
                    "photo_b64": b64,
                    "media_type": "image/jpeg",
                }).encode()

                try:
                    req = urllib.request.Request(
                        self.url,
                        data=payload,
                        headers={"Content-Type": "application/json"})
                    with urllib.request.urlopen(req, timeout=self.timeout) as r:
                        resp = r.read().decode()
                        if _ok_response(resp):
                            self._delete(event_id)
                            sent += 1
                            log.info("delivered %s", event_id)
                        else:
                            log.warning("server rejected %s: %s", event_id, resp[:100])
                            self._park(event_id)
                except urllib.error.HTTPError as e:
                    if 400 <= e.code < 500:
                        log.error("rejected %s: %d", event_id, e.code)
                        self._park(event_id)
                    else:
                        self._bump(event_id, attempts)
                        break
                except (urllib.error.URLError, OSError, TimeoutError) as e:
                    log.warning("network unavailable (%s), pending=%d",
                                type(e).__name__, self.pending_count())
                    break
        finally:
            self._flush_lock.release()
        return (sent, self.pending_count())

    # ── внутреннее ────────────────────────────────────────────────
    def _delete(self, event_id):
        with self._lock:
            row = self._db.execute("SELECT photo_path FROM outbox WHERE event_id=?",
                                   (event_id,)).fetchone()
            self._db.execute("DELETE FROM outbox WHERE event_id=?", (event_id,))
            self._db.commit()
        if row:
            try:
                os.remove(row[0])
            except OSError:
                pass

    def _drop(self, event_id):
        """Удалить событие (фото потеряно)."""
        with self._lock:
            self._db.execute("DELETE FROM outbox WHERE event_id=?", (event_id,))
            self._db.commit()

    def _park(self, event_id):
        """Паркировать событие (не отправлять больше)."""
        with self._lock:
            self._db.execute("UPDATE outbox SET status='failed' WHERE event_id=?", (event_id,))
            self._db.commit()

    def _bump(self, event_id, attempts):
        """Увеличить счётчик попыток, паркировать если превышен лимит."""
        status = "failed" if attempts + 1 >= self.max_attempts else "pending"
        with self._lock:
            self._db.execute("UPDATE outbox SET attempts=attempts+1, status=? WHERE event_id=?",
                             (status, event_id))
            self._db.commit()


def _ok_response(text):
    """Проверить, что в ответе есть 'ok': true."""
    try:
        return bool(json.loads(text).get("ok"))
    except Exception:
        return False