frigate-notify/app/routes/known_objects.py

"""Known objects API endpoints for frigate-notify v2."""

import logging
from typing import Optional

from fastapi import APIRouter, HTTPException, Query
from fastapi.responses import Response

from database import get_connection
from models import KnownObjectResponse, KnownObjectUpdate

logger = logging.getLogger(__name__)

router = APIRouter(prefix="/api/known-objects", tags=["known-objects"])


@router.get("")
async def list_known_objects(
    type: Optional[str] = Query(default=None, pattern="^(person|vehicle|pet)$"),
):
    """List all known objects with embedding_count. Optional type filter."""
    conn = get_connection()

    query = """
        SELECT
            ko.id, ko.name, ko.type, ko.action,
            ko.created_at, ko.updated_at,
            COUNT(e.id) AS embedding_count
        FROM known_objects ko
        LEFT JOIN embeddings e ON ko.id = e.known_object_id
    """
    params: list = []

    if type:
        query += " WHERE ko.type = ?"
        params.append(type)

    query += " GROUP BY ko.id ORDER BY ko.name"

    cursor = conn.execute(query, params)
    rows = cursor.fetchall()

    return [
        {
            "id": row["id"],
            "name": row["name"],
            "type": row["type"],
            "action": row["action"],
            "created_at": row["created_at"],
            "updated_at": row["updated_at"],
            "embedding_count": row["embedding_count"],
        }
        for row in rows
    ]


@router.get("/{object_id}")
async def get_known_object(object_id: int):
    """Get a single known object with list of its embeddings (id, created_at only)."""
    conn = get_connection()

    cursor = conn.execute(
        "SELECT id, name, type, action, created_at, updated_at FROM known_objects WHERE id = ?",
        (object_id,),
    )
    row = cursor.fetchone()
    if not row:
        raise HTTPException(status_code=404, detail="Known object not found")

    # Get embeddings list
    emb_cursor = conn.execute(
        "SELECT id, created_at FROM embeddings WHERE known_object_id = ? ORDER BY created_at",
        (object_id,),
    )
    embeddings = [
        {"id": e["id"], "created_at": e["created_at"]} for e in emb_cursor.fetchall()
    ]

    return {
        "id": row["id"],
        "name": row["name"],
        "type": row["type"],
        "action": row["action"],
        "created_at": row["created_at"],
        "updated_at": row["updated_at"],
        "embeddings": embeddings,
    }


@router.get("/{object_id}/thumbnails/{embedding_id}")
async def get_embedding_thumbnail(object_id: int, embedding_id: int):
    """Return raw JPEG thumbnail from embeddings table."""
    conn = get_connection()

    cursor = conn.execute(
        "SELECT thumbnail FROM embeddings WHERE id = ? AND known_object_id = ?",
        (embedding_id, object_id),
    )
    row = cursor.fetchone()
    if not row:
        raise HTTPException(status_code=404, detail="Embedding not found")
    if row["thumbnail"] is None:
        raise HTTPException(status_code=404, detail="No thumbnail available")

    return Response(content=row["thumbnail"], media_type="image/jpeg")


@router.put("/{object_id}")
async def update_known_object(object_id: int, body: KnownObjectUpdate):
    """Update name/type/action for a known object. Sets updated_at."""
    conn = get_connection()

    cursor = conn.execute(
        "SELECT id FROM known_objects WHERE id = ?",
        (object_id,),
    )
    if not cursor.fetchone():
        raise HTTPException(status_code=404, detail="Known object not found")

    updates = []
    params: list = []

    if body.name is not None:
        updates.append("name = ?")
        params.append(body.name)
    if body.type is not None:
        updates.append("type = ?")
        params.append(body.type)
    if body.action is not None:
        updates.append("action = ?")
        params.append(body.action)

    if not updates:
        raise HTTPException(status_code=400, detail="No fields to update")

    updates.append("updated_at = CURRENT_TIMESTAMP")
    params.append(object_id)

    conn.execute(
        f"UPDATE known_objects SET {', '.join(updates)} WHERE id = ?",
        params,
    )
    conn.commit()

    # Return the updated object
    cursor = conn.execute(
        "SELECT id, name, type, action, created_at, updated_at FROM known_objects WHERE id = ?",
        (object_id,),
    )
    row = cursor.fetchone()

    return KnownObjectResponse(
        id=row["id"],
        name=row["name"],
        type=row["type"],
        action=row["action"],
        created_at=row["created_at"],
        updated_at=row["updated_at"],
    )


@router.delete("/{object_id}")
async def delete_known_object(object_id: int):
    """Delete a known object. Embeddings cascade delete automatically."""
    conn = get_connection()

    cursor = conn.execute(
        "SELECT id FROM known_objects WHERE id = ?",
        (object_id,),
    )
    if not cursor.fetchone():
        raise HTTPException(status_code=404, detail="Known object not found")

    conn.execute("DELETE FROM known_objects WHERE id = ?", (object_id,))
    conn.commit()

    return {"status": "ok"}