"""Door-sensor-triggered appliance cameras — the "which fridge is it in" half of
pantry-vision, from docs/fridge-item-location.md.

A Zigbee contact sensor on each cold appliance fires an HA automation, which POSTs
`/doorway-event` here. This module then pulls a snapshot (or a short burst) from that
appliance's camera via Frigate, runs it through the same vision identification the
kitchen display uses, and records what it saw as a **hint**.

WHY THIS IS A HINT AND NOT A FACT
---------------------------------
Read `docs/fridge-item-location.md` before changing anything here; the short version:

- **A camera at the door can only ever see the doorway.** It does not know whether the
  item was going in or coming out, it misses when two things are carried at once, and
  it sees nothing at all when an arm is in the way. Anything that presented its output
  as the truth about where food is would be wrong several times a week, silently.
- So a hint is written with a timestamp, an appliance, and a confidence, and it is
  shown as "last seen going past the loggia fridge, Tue 18:42" — a sentence a person
  can evaluate. **Nothing here writes stock.** Moving an item between appliances is
  `/transfer`, which a person taps, exactly like every other write in this service.
- Which means a wrong hint costs a glance. That is the whole design budget.

WHY A SEPARATE SQLITE FILE, WHEN "GROCY IS THE SYSTEM OF RECORD"
----------------------------------------------------------------
Grocy owns *stock*: what exists, how much, until when, and — via locations — where it
is meant to be. It has nowhere to put "a camera thinks it saw something like this go
past that door 40 seconds ago, and might be wrong". That is not inventory, it is
observation, and it has a retention life measured in days. Keeping it out of Grocy is
what stops a guess from ever being mistaken for stock — the same separation
digest-engine draws between its archive and its output.

The file is best-effort in the same way `archive.py` is over in digest-engine: a
corrupt or unwritable database logs a warning and the identification still happens.
No hint is worth failing a request over.
"""

from __future__ import annotations

import logging
import os
import sqlite3
import time
import urllib.error
import urllib.request
from datetime import datetime, timedelta, timezone
from pathlib import Path

LOG = logging.getLogger("pantry-vision.doorway")

DB_PATH = Path(os.environ.get("PANTRY_HINTS_DB_PATH", "/data/pantry-hints.db"))

# Same Frigate this project's `chores` already pulls snapshots from, same env name.
# Unset means the door-event endpoint accepts the event and records the opening
# without a picture — which is still worth having, see the module docstring.
FRIGATE_URL = os.environ.get("FRIGATE_URL", "").rstrip("/")
FRIGATE_TIMEOUT = float(os.environ.get("FRIGATE_TIMEOUT", "15"))

# How many frames to take per door event, and how far apart. An item crosses the
# doorway in about a second, so one frame is a coin toss and ten is a queue at the
# vision model. Three is a compromise, and the burst stops early on the first frame
# that actually identifies something.
BURST_FRAMES = int(os.environ.get("PANTRY_DOORWAY_BURST", "3"))
BURST_INTERVAL = float(os.environ.get("PANTRY_DOORWAY_BURST_INTERVAL", "0.7"))

# Hints are worthless once they are old — "last seen a month ago" tells you nothing a
# person didn't already know. Pruned on write, so the file cannot grow without bound.
HINT_RETENTION_DAYS = int(os.environ.get("PANTRY_HINT_RETENTION_DAYS", "30"))


def _now() -> str:
    return datetime.now(timezone.utc).isoformat().replace("+00:00", "Z")


def appliances() -> dict[str, dict]:
    """PANTRY_DOOR_APPLIANCES: "id:Grocy location name:frigate_camera, ..."

    The Grocy location name is the same string `PANTRY_LOCATION_*` uses, because the
    point of the whole feature is to answer *which appliance*, and "which appliance"
    has to mean the same thing here as it does on the confirm screen. The camera is
    optional: an appliance with a door sensor and no camera still records openings,
    which is the half of this feature that pays for itself.
    """
    raw = os.environ.get("PANTRY_DOOR_APPLIANCES", "").strip()
    result: dict[str, dict] = {}
    for entry in raw.split(","):
        entry = entry.strip()
        if not entry:
            continue
        parts = [p.strip() for p in entry.split(":")]
        if len(parts) < 2 or not parts[0] or not parts[1]:
            LOG.warning("pantry-vision: ignoring malformed PANTRY_DOOR_APPLIANCES entry %r", entry)
            continue
        result[parts[0]] = {
            "id": parts[0],
            "location_name": parts[1],
            "camera": parts[2] if len(parts) > 2 and parts[2] else "",
        }
    return result


# --- the hint store ---------------------------------------------------------------
def _db() -> sqlite3.Connection | None:
    try:
        DB_PATH.parent.mkdir(parents=True, exist_ok=True)
        conn = sqlite3.connect(DB_PATH, timeout=5)
        conn.row_factory = sqlite3.Row
        conn.execute(
            """
            CREATE TABLE IF NOT EXISTS hints (
                id INTEGER PRIMARY KEY,
                appliance TEXT NOT NULL,
                location_name TEXT NOT NULL,
                door_state TEXT NOT NULL,
                product_id INTEGER,
                kind TEXT,
                name TEXT,
                confidence TEXT,
                identified INTEGER NOT NULL DEFAULT 0,
                created_at TEXT NOT NULL
            )
            """
        )
        conn.execute("CREATE INDEX IF NOT EXISTS hints_by_product ON hints (product_id, created_at)")
        return conn
    except (sqlite3.Error, OSError):
        # OSError as well as sqlite3.Error, and this is not belt-and-braces: mkdir on
        # an unmounted or read-only /data raises PermissionError, which is an OSError
        # and not a database error at all. Without it, a deployment that forgot the
        # bind mount would take down /inventory — the one screen that has nothing to
        # do with hints — instead of quietly having no sightings to show.
        LOG.warning("pantry-vision: hint database unusable at %s", DB_PATH, exc_info=True)
        return None


def record_hint(appliance: dict, door_state: str, proposal: dict | None, product_id: int | None) -> None:
    conn = _db()
    if conn is None:
        return
    try:
        with conn:
            conn.execute(
                "INSERT INTO hints (appliance, location_name, door_state, product_id, kind, name, "
                "confidence, identified, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)",
                (
                    appliance["id"],
                    appliance["location_name"],
                    door_state,
                    product_id,
                    (proposal or {}).get("kind") or "",
                    (proposal or {}).get("name") or "",
                    (proposal or {}).get("confidence") or "",
                    1 if proposal else 0,
                    _now(),
                ),
            )
            cutoff = (datetime.now(timezone.utc) - timedelta(days=HINT_RETENTION_DAYS)).isoformat()
            conn.execute("DELETE FROM hints WHERE created_at < ?", (cutoff.replace("+00:00", "Z"),))
    except sqlite3.Error:
        LOG.warning("pantry-vision: could not write a doorway hint", exc_info=True)
    finally:
        conn.close()


def recent_hints(limit: int = 50) -> list[dict]:
    conn = _db()
    if conn is None:
        return []
    try:
        rows = conn.execute(
            "SELECT appliance, location_name, door_state, product_id, kind, name, confidence, "
            "identified, created_at FROM hints ORDER BY created_at DESC LIMIT ?",
            (max(1, min(500, limit)),),
        ).fetchall()
        return [dict(row) for row in rows]
    except sqlite3.Error:
        LOG.warning("pantry-vision: could not read doorway hints", exc_info=True)
        return []
    finally:
        conn.close()


def last_seen_by_product() -> dict[int, dict]:
    """{product_id: the most recent identified hint for it}.

    Only identified hints: a door opening with nothing recognised in it says something
    about the door, not about any particular jar, and attaching it to an item would be
    inventing the very link this module refuses to invent.
    """
    conn = _db()
    if conn is None:
        return {}
    try:
        # SQLite's documented bare-column rule: with MAX() in the select list, the
        # other columns come from the row that matched it. That is what makes this one
        # query instead of one per product.
        rows = conn.execute(
            "SELECT product_id, appliance, location_name, confidence, MAX(created_at) AS created_at "
            "FROM hints WHERE product_id IS NOT NULL AND identified = 1 GROUP BY product_id"
        ).fetchall()
        return {int(row["product_id"]): dict(row) for row in rows}
    except sqlite3.Error:
        LOG.warning("pantry-vision: could not read last-seen hints", exc_info=True)
        return {}
    finally:
        conn.close()


# --- the camera -------------------------------------------------------------------
def _snapshot(camera: str) -> bytes | None:
    """One frame from Frigate, or None. Same endpoint shape `chores/check.py` uses.

    VERIFY against a real Frigate: `/api/<camera>/latest.jpg` is its documented
    always-available snapshot path, but nothing in this project has called it yet.
    """
    if not (FRIGATE_URL and camera):
        return None
    try:
        with urllib.request.urlopen(f"{FRIGATE_URL}/api/{camera}/latest.jpg", timeout=FRIGATE_TIMEOUT) as resp:
            return resp.read()
    except (urllib.error.URLError, urllib.error.HTTPError, TimeoutError, OSError):
        LOG.warning("pantry-vision: could not fetch a snapshot from camera %r", camera, exc_info=True)
        return None


def handle_event(appliance: dict, door_state: str, identify, match) -> dict:
    """Take a burst, identify the first frame that shows something, record the hint.

    Runs off the request thread (see server.py) because a burst is up to three vision
    calls and Home Assistant's `rest_command` should not be sitting on a socket for
    however long the LLM host takes. `identify` and `match` are passed in rather than
    imported so this module never has to know about Grocy — it deals in doors,
    cameras and hints.
    """
    camera = appliance.get("camera") or ""
    if not camera:
        record_hint(appliance, door_state, None, None)
        LOG.info("pantry-vision: %s %s (no camera configured)", appliance["id"], door_state)
        return {"appliance": appliance["id"], "identified": False, "reason": "no camera configured"}

    proposal = None
    for attempt in range(max(1, BURST_FRAMES)):
        if attempt:
            time.sleep(BURST_INTERVAL)
        image = _snapshot(camera)
        if image is None:
            break
        candidate = identify(image)
        # `present: false` is the model saying the frame holds no grocery item — an
        # arm, a closed door, an empty kitchen. That is the expected answer for most
        # frames of most openings, and it is not a failure.
        if candidate.get("present") and not candidate.get("degraded"):
            proposal = candidate
            break

    if proposal is None:
        record_hint(appliance, door_state, None, None)
        LOG.info("pantry-vision: %s %s — nothing recognised in %d frame(s)", appliance["id"], door_state, BURST_FRAMES)
        return {"appliance": appliance["id"], "identified": False, "reason": "nothing recognised"}

    # An exact stock match makes the hint attachable to a row the household can act
    # on. Without one it is still worth recording — "something like eggs went past the
    # loggia freezer" is a useful thing to have seen, and it is exactly what a person
    # would search for after failing to find eggs.
    product_id = None
    try:
        matches = match(proposal.get("kind", ""), proposal.get("name", ""))
        exact = [m for m in matches if m.get("exact")]
        if exact:
            product_id = int(exact[0]["product_id"])
        elif len(matches) == 1:
            product_id = int(matches[0]["product_id"])
    except Exception:
        LOG.warning("pantry-vision: could not match a doorway identification to stock", exc_info=True)

    record_hint(appliance, door_state, proposal, product_id)
    LOG.info(
        "pantry-vision: %s %s — saw %r (confidence %s, stock row %s)",
        appliance["id"], door_state, proposal.get("name"), proposal.get("confidence"), product_id,
    )
    return {
        "appliance": appliance["id"],
        "identified": True,
        "name": proposal.get("name"),
        "confidence": proposal.get("confidence"),
        "product_id": product_id,
    }
