frigate/frigate/api/review.py

"""Review apis."""

import datetime
import logging
from functools import reduce
from pathlib import Path

import pandas as pd
from fastapi import APIRouter, Request
from fastapi.params import Depends
from fastapi.responses import JSONResponse
from peewee import Case, DoesNotExist, fn, operator
from playhouse.shortcuts import model_to_dict

from frigate.api.auth import (
    allow_any_authenticated,
    get_allowed_cameras_for_filter,
    get_current_user,
    require_camera_access,
    require_full_camera_access,
    require_role,
)
from frigate.api.defs.query.review_query_parameters import (
    ReviewActivityMotionQueryParams,
    ReviewQueryParams,
    ReviewSummaryQueryParams,
)
from frigate.api.defs.request.review_body import ReviewModifyMultipleBody
from frigate.api.defs.response.generic_response import GenericResponse
from frigate.api.defs.response.review_response import (
    ReviewActivityMotionResponse,
    ReviewSegmentResponse,
    ReviewSummaryResponse,
)
from frigate.api.defs.tags import Tags
from frigate.const import STREAM_TYPE_MAIN
from frigate.embeddings import EmbeddingsContext
from frigate.models import Recordings, ReviewSegment, UserReviewStatus
from frigate.review.types import SeverityEnum
from frigate.util.time import get_dst_transitions

logger = logging.getLogger(__name__)

router = APIRouter(tags=[Tags.review])


def get_label_clause(label: str, include_audio: bool = True):
    """Build a clause matching a label within a review segment's data.

    Verified objects are stored with a `-verified` suffix (eg. `person-verified`)
    so that variant is matched as well.
    """
    clause = (ReviewSegment.data["objects"].cast("text") % f'*"{label}"*') | (
        ReviewSegment.data["objects"].cast("text") % f'*"{label}-verified"*'
    )

    if include_audio:
        clause |= ReviewSegment.data["audio"].cast("text") % f'*"{label}"*'

    return clause


@router.get(
    "/review",
    response_model=list[ReviewSegmentResponse],
    dependencies=[Depends(allow_any_authenticated())],
)
async def review(
    params: ReviewQueryParams = Depends(),
    current_user: dict = Depends(get_current_user),
    allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
):
    if isinstance(current_user, JSONResponse):
        return current_user

    user_id = current_user["username"]

    cameras = params.cameras
    labels = params.labels
    zones = params.zones
    reviewed = params.reviewed
    limit = params.limit
    severity = params.severity
    before = params.before or datetime.datetime.now().timestamp()
    after = (
        params.after
        or (datetime.datetime.now() - datetime.timedelta(hours=24)).timestamp()
    )

    clauses = [
        (ReviewSegment.start_time < before)
        & ((ReviewSegment.end_time.is_null(True)) | (ReviewSegment.end_time > after))
    ]

    if cameras != "all":
        requested = set(cameras.split(","))
        filtered = requested.intersection(allowed_cameras)
        if not filtered:
            return JSONResponse(content=[])
        camera_list = list(filtered)
    else:
        camera_list = allowed_cameras
    clauses.append(ReviewSegment.camera << camera_list)

    if labels != "all":
        # use matching so segments with multiple labels
        # still match on a search where any label matches
        label_clauses = []
        filtered_labels = labels.split(",")

        for label in filtered_labels:
            label_clauses.append(get_label_clause(label))
        clauses.append(reduce(operator.or_, label_clauses))

    if zones != "all":
        # use matching so segments with multiple zones
        # still match on a search where any zone matches
        zone_clauses = []
        filtered_zones = zones.split(",")

        for zone in filtered_zones:
            zone_clauses.append(
                ReviewSegment.data["zones"].cast("text") % f'*"{zone}"*'
            )
        clauses.append(reduce(operator.or_, zone_clauses))

    if severity:
        clauses.append(ReviewSegment.severity == severity)

    # Join with UserReviewStatus to get per-user review status
    review_query = (
        ReviewSegment.select(
            ReviewSegment.id,
            ReviewSegment.camera,
            ReviewSegment.start_time,
            ReviewSegment.end_time,
            ReviewSegment.severity,
            ReviewSegment.thumb_path,
            ReviewSegment.data,
            fn.COALESCE(UserReviewStatus.has_been_reviewed, False).alias(
                "has_been_reviewed"
            ),
        )
        .left_outer_join(
            UserReviewStatus,
            on=(
                (ReviewSegment.id == UserReviewStatus.review_segment)
                & (UserReviewStatus.user_id == user_id)
            ),
        )
        .where(reduce(operator.and_, clauses))
    )

    # Filter unreviewed items without subquery
    if reviewed == 0:
        review_query = review_query.where(
            (UserReviewStatus.has_been_reviewed == False)
            | (UserReviewStatus.has_been_reviewed.is_null())
        )
    elif reviewed == 1:
        review_query = review_query.where(UserReviewStatus.has_been_reviewed == True)

    # Apply ordering and limit
    review_query = (
        review_query.order_by(ReviewSegment.severity.asc())
        .order_by(ReviewSegment.start_time.desc())
        .limit(limit)
        .dicts()
        .iterator()
    )

    return JSONResponse(content=[r for r in review_query])


@router.get(
    "/review_ids",
    response_model=list[ReviewSegmentResponse],
    dependencies=[Depends(allow_any_authenticated())],
)
async def review_ids(request: Request, ids: str):
    ids = ids.split(",")

    if not ids:
        return JSONResponse(
            content=({"success": False, "message": "Valid list of ids must be sent"}),
            status_code=400,
        )

    try:
        reviews = list(
            ReviewSegment.select().where(ReviewSegment.id << ids).dicts().iterator()
        )
    except Exception:
        return JSONResponse(
            content=({"success": False, "message": "Review segments not found"}),
            status_code=400,
        )

    found_ids = {r["id"] for r in reviews}
    for review_id in ids:
        if review_id not in found_ids:
            return JSONResponse(
                content=(
                    {"success": False, "message": f"Review {review_id} not found"}
                ),
                status_code=404,
            )

    for review in reviews:
        await require_camera_access(review["camera"], request=request)

    return JSONResponse(reviews)


@router.get(
    "/review/summary",
    response_model=ReviewSummaryResponse,
    dependencies=[Depends(allow_any_authenticated())],
)
async def review_summary(
    params: ReviewSummaryQueryParams = Depends(),
    current_user: dict = Depends(get_current_user),
    allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
):
    if isinstance(current_user, JSONResponse):
        return current_user

    user_id = current_user["username"]

    day_ago = (datetime.datetime.now() - datetime.timedelta(hours=24)).timestamp()

    cameras = params.cameras
    labels = params.labels
    zones = params.zones

    clauses = [(ReviewSegment.start_time > day_ago)]

    if cameras != "all":
        requested = set(cameras.split(","))
        filtered = requested.intersection(allowed_cameras)
        if not filtered:
            return JSONResponse(content={})
        camera_list = list(filtered)
    else:
        camera_list = allowed_cameras
    clauses.append(ReviewSegment.camera << camera_list)

    if labels != "all":
        # use matching so segments with multiple labels
        # still match on a search where any label matches
        label_clauses = []
        filtered_labels = labels.split(",")

        for label in filtered_labels:
            label_clauses.append(get_label_clause(label))
        clauses.append(reduce(operator.or_, label_clauses))
    if zones != "all":
        # use matching so segments with multiple zones
        # still match on a search where any zone matches
        zone_clauses = []
        filtered_zones = zones.split(",")

        for zone in filtered_zones:
            zone_clauses.append(
                ReviewSegment.data["zones"].cast("text") % f'*"{zone}"*'
            )
        clauses.append(reduce(operator.or_, zone_clauses))

    last_24_query = (
        ReviewSegment.select(
            fn.SUM(
                Case(
                    None,
                    [
                        (
                            (ReviewSegment.severity == SeverityEnum.alert)
                            & (UserReviewStatus.has_been_reviewed == True),
                            1,
                        )
                    ],
                    0,
                )
            ).alias("reviewed_alert"),
            fn.SUM(
                Case(
                    None,
                    [
                        (
                            (ReviewSegment.severity == SeverityEnum.detection)
                            & (UserReviewStatus.has_been_reviewed == True),
                            1,
                        )
                    ],
                    0,
                )
            ).alias("reviewed_detection"),
            fn.SUM(
                Case(
                    None,
                    [
                        (
                            (ReviewSegment.severity == SeverityEnum.alert),
                            1,
                        )
                    ],
                    0,
                )
            ).alias("total_alert"),
            fn.SUM(
                Case(
                    None,
                    [
                        (
                            (ReviewSegment.severity == SeverityEnum.detection),
                            1,
                        )
                    ],
                    0,
                )
            ).alias("total_detection"),
        )
        .left_outer_join(
            UserReviewStatus,
            on=(
                (ReviewSegment.id == UserReviewStatus.review_segment)
                & (UserReviewStatus.user_id == user_id)
            ),
        )
        .where(reduce(operator.and_, clauses))
        .dicts()
        .get()
    )

    clauses = []

    if cameras != "all":
        requested = set(cameras.split(","))
        filtered = requested.intersection(allowed_cameras)
        if not filtered:
            return JSONResponse(content={})
        camera_list = list(filtered)
    else:
        camera_list = allowed_cameras
    clauses.append(ReviewSegment.camera << camera_list)

    if labels != "all":
        # use matching so segments with multiple labels
        # still match on a search where any label matches
        label_clauses = []
        filtered_labels = labels.split(",")

        for label in filtered_labels:
            label_clauses.append(get_label_clause(label, include_audio=False))

        clauses.append(reduce(operator.or_, label_clauses))

    # Find the time range of available data
    time_range_query = (
        ReviewSegment.select(
            fn.MIN(ReviewSegment.start_time).alias("min_time"),
            fn.MAX(ReviewSegment.start_time).alias("max_time"),
        )
        .where(reduce(operator.and_, clauses) if clauses else True)
        .dicts()
        .get()
    )

    min_time = time_range_query.get("min_time")
    max_time = time_range_query.get("max_time")

    data = {
        "last24Hours": last_24_query,
    }

    # If no data, return early
    if min_time is None or max_time is None:
        return JSONResponse(content=data)

    # Get DST transition periods
    dst_periods = get_dst_transitions(params.timezone, min_time, max_time)

    day_in_seconds = 60 * 60 * 24

    # Query each DST period separately with the correct offset
    for period_start, period_end, period_offset in dst_periods:
        # Calculate hour/minute modifiers for this period
        hours_offset = int(period_offset / 60 / 60)
        minutes_offset = int(period_offset / 60 - hours_offset * 60)
        period_hour_modifier = f"{hours_offset} hour"
        period_minute_modifier = f"{minutes_offset} minute"

        # Build clauses including time range for this period
        period_clauses = clauses.copy()
        period_clauses.append(
            (ReviewSegment.start_time >= period_start)
            & (ReviewSegment.start_time <= period_end)
        )

        period_query = (
            ReviewSegment.select(
                fn.strftime(
                    "%Y-%m-%d",
                    fn.datetime(
                        ReviewSegment.start_time,
                        "unixepoch",
                        period_hour_modifier,
                        period_minute_modifier,
                    ),
                ).alias("day"),
                fn.SUM(
                    Case(
                        None,
                        [
                            (
                                (ReviewSegment.severity == SeverityEnum.alert)
                                & (UserReviewStatus.has_been_reviewed == True),
                                1,
                            )
                        ],
                        0,
                    )
                ).alias("reviewed_alert"),
                fn.SUM(
                    Case(
                        None,
                        [
                            (
                                (ReviewSegment.severity == SeverityEnum.detection)
                                & (UserReviewStatus.has_been_reviewed == True),
                                1,
                            )
                        ],
                        0,
                    )
                ).alias("reviewed_detection"),
                fn.SUM(
                    Case(
                        None,
                        [
                            (
                                (ReviewSegment.severity == SeverityEnum.alert),
                                1,
                            )
                        ],
                        0,
                    )
                ).alias("total_alert"),
                fn.SUM(
                    Case(
                        None,
                        [
                            (
                                (ReviewSegment.severity == SeverityEnum.detection),
                                1,
                            )
                        ],
                        0,
                    )
                ).alias("total_detection"),
            )
            .left_outer_join(
                UserReviewStatus,
                on=(
                    (ReviewSegment.id == UserReviewStatus.review_segment)
                    & (UserReviewStatus.user_id == user_id)
                ),
            )
            .where(reduce(operator.and_, period_clauses))
            .group_by(
                (ReviewSegment.start_time + period_offset).cast("int") / day_in_seconds
            )
            .order_by(ReviewSegment.start_time.desc())
        )

        # Merge results from this period
        for e in period_query.dicts().iterator():
            day_key = e["day"]
            if day_key in data:
                # Merge counts if day already exists (edge case at DST boundary)
                data[day_key]["reviewed_alert"] += e["reviewed_alert"] or 0
                data[day_key]["reviewed_detection"] += e["reviewed_detection"] or 0
                data[day_key]["total_alert"] += e["total_alert"] or 0
                data[day_key]["total_detection"] += e["total_detection"] or 0
            else:
                data[day_key] = e

    return JSONResponse(content=data)


@router.post(
    "/reviews/viewed",
    response_model=GenericResponse,
    dependencies=[Depends(allow_any_authenticated())],
)
async def set_multiple_reviewed(
    request: Request,
    body: ReviewModifyMultipleBody,
    current_user: dict = Depends(get_current_user),
):
    if isinstance(current_user, JSONResponse):
        return current_user

    user_id = current_user["username"]

    reviews = list(
        ReviewSegment.select(ReviewSegment.id, ReviewSegment.camera).where(
            ReviewSegment.id << body.ids
        )
    )

    for review in reviews:
        await require_camera_access(review.camera, request=request)

    found_ids = [r.id for r in reviews]

    if found_ids:
        existing_statuses = list(
            UserReviewStatus.select().where(
                (UserReviewStatus.user_id == user_id)
                & (UserReviewStatus.review_segment << found_ids)
            )
        )

        status_by_review = {s.review_segment_id: s for s in existing_statuses}

        to_update = []
        to_create = []

        for review_id in found_ids:
            if review_id in status_by_review:
                status = status_by_review[review_id]
                if status.has_been_reviewed != body.reviewed:
                    status.has_been_reviewed = body.reviewed
                    to_update.append(status)
            else:
                to_create.append(
                    {
                        "user_id": user_id,
                        "review_segment_id": review_id,
                        "has_been_reviewed": body.reviewed,
                    }
                )

        if to_update:
            UserReviewStatus.bulk_update(
                to_update, fields=[UserReviewStatus.has_been_reviewed], batch_size=100
            )

        if to_create:
            UserReviewStatus.insert_many(to_create).on_conflict_ignore().execute()

    return JSONResponse(
        content=(
            {
                "success": True,
                "message": f"Marked multiple items as {'reviewed' if body.reviewed else 'unreviewed'}",
            }
        ),
        status_code=200,
    )


@router.post(
    "/reviews/delete",
    response_model=GenericResponse,
    dependencies=[Depends(require_role(["admin"]))],
)
def delete_reviews(body: ReviewModifyMultipleBody):
    list_of_ids = body.ids
    reviews = (
        ReviewSegment.select(
            ReviewSegment.camera,
            ReviewSegment.start_time,
            ReviewSegment.end_time,
        )
        .where(ReviewSegment.id << list_of_ids)
        .dicts()
        .iterator()
    )
    recording_ids = []

    for review in reviews:
        start_time = review["start_time"]
        end_time = review["end_time"]
        camera_name = review["camera"]
        recordings = (
            Recordings.select(Recordings.id, Recordings.path)
            .where(
                Recordings.start_time.between(start_time, end_time)
                | Recordings.end_time.between(start_time, end_time)
                | (
                    (start_time > Recordings.start_time)
                    & (end_time < Recordings.end_time)
                )
            )
            .where(Recordings.camera == camera_name)
            .dicts()
            .iterator()
        )

        for recording in recordings:
            Path(recording["path"]).unlink(missing_ok=True)
            recording_ids.append(recording["id"])

    # delete recordings and review segments
    Recordings.delete().where(Recordings.id << recording_ids).execute()
    ReviewSegment.delete().where(ReviewSegment.id << list_of_ids).execute()
    UserReviewStatus.delete().where(
        UserReviewStatus.review_segment << list_of_ids
    ).execute()

    return JSONResponse(
        content=({"success": True, "message": "Deleted review items."}), status_code=200
    )


@router.get(
    "/review/activity/motion",
    response_model=list[ReviewActivityMotionResponse],
    dependencies=[Depends(allow_any_authenticated())],
)
def motion_activity(
    params: ReviewActivityMotionQueryParams = Depends(),
    allowed_cameras: list[str] = Depends(get_allowed_cameras_for_filter),
):
    """Get motion and audio activity."""
    cameras = params.cameras
    before = params.before or datetime.datetime.now().timestamp()
    after = (
        params.after
        or (datetime.datetime.now() - datetime.timedelta(hours=1)).timestamp()
    )
    # get scale in seconds
    scale = params.scale

    clauses = [(Recordings.start_time > after) & (Recordings.end_time < before)]
    clauses.append(Recordings.motion > 0)
    # sub rows duplicate the camera's motion stats, so only count main rows
    clauses.append(Recordings.stream_type == STREAM_TYPE_MAIN)

    if cameras != "all":
        requested = set(cameras.split(","))
        filtered = requested.intersection(allowed_cameras)
        if not filtered:
            return JSONResponse(content=[])
        camera_list = list(filtered)
    else:
        camera_list = list(allowed_cameras)

    clauses.append(Recordings.camera << camera_list)

    data: list[Recordings] = (
        Recordings.select(
            Recordings.camera,
            Recordings.start_time,
            Recordings.motion,
        )
        .where(reduce(operator.and_, clauses))
        .order_by(Recordings.start_time.asc())
        .dicts()
        .iterator()
    )

    # resample data using pandas to get activity on scaled basis
    df = pd.DataFrame(data, columns=["start_time", "motion", "camera"])

    if df.empty:
        logger.warning("No motion data found for the requested time range")
        return JSONResponse(content=[])

    df = df.astype(dtype={"motion": "float32"})

    # set date as datetime index
    df["start_time"] = pd.to_datetime(df["start_time"], unit="s")
    df.set_index(["start_time"], inplace=True)

    # normalize data
    motion = df["motion"].resample(f"{scale}s").max().fillna(0.0).to_frame()

    if len(camera_list) == 1:
        cameras = df["camera"].resample(f"{scale}s").first().fillna("")
    else:
        cameras = df["camera"].resample(f"{scale}s").agg(lambda x: ",".join(set(x)))
    df = motion.join(cameras)

    length = df.shape[0]
    chunk = int(60 * (60 / scale))

    for i in range(0, length, chunk):
        part = df.iloc[i : i + chunk]
        min_val, max_val = part["motion"].min(), part["motion"].max()
        if min_val != max_val:
            df.iloc[i : i + chunk, 0] = (
                part["motion"].sub(min_val).div(max_val - min_val).mul(100).fillna(0)
            )
        else:
            df.iloc[i : i + chunk, 0] = 0.0

    # Drop resample gap-fill buckets. The resample above emits a row for every
    # {scale}s bucket spanning the range, and buckets with no recording get a
    # motion of 0 (from fillna) and an empty camera (from joining an empty set).
    df = df[df["camera"] != ""]

    # change types for output
    df.index = df.index.as_unit("s").astype(int)
    normalized = df.reset_index().to_dict("records")
    return JSONResponse(content=normalized)


@router.get(
    "/review/event/{event_id}",
    response_model=ReviewSegmentResponse,
    dependencies=[Depends(allow_any_authenticated())],
)
async def get_review_from_event(request: Request, event_id: str):
    try:
        review = ReviewSegment.get(
            ReviewSegment.data["detections"].cast("text") % f'*"{event_id}"*'
        )
        await require_camera_access(review.camera, request=request)
        return JSONResponse(model_to_dict(review))
    except DoesNotExist:
        return JSONResponse(
            content={"success": False, "message": "Review item not found"},
            status_code=404,
        )


@router.get(
    "/review/{review_id}",
    response_model=ReviewSegmentResponse,
    dependencies=[Depends(allow_any_authenticated())],
)
async def get_review(request: Request, review_id: str):
    try:
        review = ReviewSegment.get(ReviewSegment.id == review_id)
        await require_camera_access(review.camera, request=request)
        return JSONResponse(content=model_to_dict(review))
    except DoesNotExist:
        return JSONResponse(
            content={"success": False, "message": "Review item not found"},
            status_code=404,
        )


@router.put(
    "/review/{review_id}/regenerate_description",
    response_model=GenericResponse,
    dependencies=[Depends(require_role(["admin"]))],
    summary="Generate a review item description",
    description="""Re-runs a review item through the GenAI descriptions process.
    Frames are always taken from recordings, and both alerts and detections are
    accepted regardless of the camera's GenAI alerts/detections toggles.
    """,
)
async def regenerate_review_description(request: Request, review_id: str):
    try:
        review: ReviewSegment = ReviewSegment.get(ReviewSegment.id == review_id)
    except DoesNotExist:
        return JSONResponse(
            content={
                "success": False,
                "message": "Review " + review_id + " not found",
            },
            status_code=404,
        )

    await require_camera_access(review.camera, request=request)

    if review.end_time is None:
        return JSONResponse(
            content={
                "success": False,
                "message": "Review " + review_id + " has not ended yet",
            },
            status_code=400,
        )

    camera_config = request.app.frigate_config.cameras.get(review.camera)

    if camera_config is None or not camera_config.review.genai.enabled:
        return JSONResponse(
            content={
                "success": False,
                "message": "GenAI descriptions must be enabled for this camera",
            },
            status_code=400,
        )

    if request.app.genai_manager.description_client is None:
        return JSONResponse(
            content={
                "success": False,
                "message": "A GenAI provider with the descriptions role must be configured",
            },
            status_code=400,
        )

    context: EmbeddingsContext = request.app.embeddings
    context.regenerate_review_description(review_id)

    return JSONResponse(
        content={
            "success": True,
            "message": "Review "
            + review_id
            + " description generation has been requested",
        },
        status_code=202,
    )


@router.delete(
    "/review/{review_id}/viewed",
    response_model=GenericResponse,
    dependencies=[Depends(allow_any_authenticated())],
)
async def set_not_reviewed(
    request: Request,
    review_id: str,
    current_user: dict = Depends(get_current_user),
):
    if isinstance(current_user, JSONResponse):
        return current_user

    user_id = current_user["username"]

    try:
        review: ReviewSegment = ReviewSegment.get(ReviewSegment.id == review_id)
    except DoesNotExist:
        return JSONResponse(
            content=(
                {"success": False, "message": "Review " + review_id + " not found"}
            ),
            status_code=404,
        )

    await require_camera_access(review.camera, request=request)

    try:
        user_review = UserReviewStatus.get(
            UserReviewStatus.user_id == user_id,
            UserReviewStatus.review_segment == review,
        )
        # we could update here instead of delete if we need
        user_review.delete_instance()
    except DoesNotExist:
        pass  # Already effectively "not reviewed"

    return JSONResponse(
        content=({"success": True, "message": f"Set Review {review_id} as not viewed"}),
        status_code=200,
    )


# Intentionally not camera scoped, as the summary correlates each flagged event
# with overlapping activity on other cameras. Restricted to callers who can
# already see every camera, so the unscoped query discloses nothing.
@router.post(
    "/review/summarize/start/{start_ts}/end/{end_ts}",
    dependencies=[Depends(require_full_camera_access)],
    description="Use GenAI to summarize review items over a period of time.",
)
def generate_review_summary(request: Request, start_ts: float, end_ts: float):
    if not request.app.genai_manager.description_client:
        return JSONResponse(
            content=(
                {
                    "success": False,
                    "message": "GenAI must be configured to use this feature.",
                }
            ),
            status_code=400,
        )

    context: EmbeddingsContext = request.app.embeddings
    summary = context.generate_review_summary(start_ts, end_ts)

    if summary:
        return JSONResponse(
            content=({"success": True, "summary": summary}), status_code=200
        )
    else:
        return JSONResponse(
            content=({"success": False, "message": "Failed to create summary."}),
            status_code=500,
        )