app/service.py

"""Notice assessment orchestration."""

from __future__ import annotations

import json
import re
import time
from typing import Any

from app import model_endpoint
from app.config import EXAMPLE_CACHE_PATH
from app.schema import normalize_assessment
from app.trace import queue_trace


def load_example_cache() -> dict[str, dict[str, Any]]:
    try:
        document = json.loads(EXAMPLE_CACHE_PATH.read_text(encoding="utf-8"))
        examples = document["examples"]
    except (OSError, KeyError, TypeError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"Invalid example cache: {exc}") from exc
    if not isinstance(examples, dict):
        raise RuntimeError("Invalid example cache: examples must be an object.")
    return {
        str(example_id): normalize_assessment(assessment)
        for example_id, assessment in examples.items()
    }


EXAMPLE_ASSESSMENTS = load_example_cache()


def parse_model_json(
    content: str,
    telemetry: dict[str, Any] | None = None,
) -> dict[str, Any]:
    """Parse model JSON; retained as a testable public helper."""
    telemetry = telemetry if telemetry is not None else {}
    candidate = content.strip()
    if candidate.startswith("```"):
        candidate = re.sub(r"^```(?:json)?\s*", "", candidate, flags=re.I)
        candidate = re.sub(r"\s*```$", "", candidate)
    parse_started = time.perf_counter()
    try:
        value = json.loads(candidate)
    except json.JSONDecodeError:
        match = re.search(r"\{.*\}", candidate, re.S)
        if not match:
            raise ValueError("Model did not return JSON.") from None
        value = json.loads(match.group(0))
    telemetry["parse_ms"] = (time.perf_counter() - parse_started) * 1000
    telemetry["parse_completed"] = True
    normalize_started = time.perf_counter()
    try:
        result = normalize_assessment(value)
    finally:
        telemetry["normalize_ms"] = (
            time.perf_counter() - normalize_started
        ) * 1000
    telemetry["normalize_completed"] = True
    return result


def sanitize_model_guidance(assessment: dict[str, Any]) -> dict[str, Any]:
    """Replace unsafe or invented verification advice."""
    replacements = {
        "social media": (
            "Use contact details from an independently located official website, "
            "app, card, or statement."
        ),
        "national anti-fraud centre": (
            "Use the relevant service's official reporting channel if needed."
        ),
        "national cyber security centre": (
            "Use the relevant service's official reporting channel if needed."
        ),
    }
    sanitized_steps: list[str] = []
    for item in assessment["safe_next_steps"]:
        lowered = item.lower()
        named_reporting_body = "report" in lowered and any(
            phrase in lowered
            for phrase in (
                "authority",
                "centre",
                "center",
                "cybercrime unit",
                "cyber security",
                "anti-fraud",
            )
        )
        replacement = (
            "Use the relevant service's official reporting channel if needed."
            if named_reporting_body
            else next(
                (value for phrase, value in replacements.items() if phrase in lowered),
                item,
            )
        )
        if replacement not in sanitized_steps:
            sanitized_steps.append(replacement)
    assessment["safe_next_steps"] = sanitized_steps
    return assessment


def analyze_notice(
    text: str = "",
    image_data_url: str = "",
    example_id: str = "",
    save_trace: bool = True,
    output_language: str = "en",
) -> dict[str, Any]:
    """Analyze supplied input with the local model and optionally queue a trace."""
    text = (text or "").strip()
    image_data_url = image_data_url or ""
    example_id = (example_id or "").strip()
    output_language = "ur" if output_language == "ur" else "en"

    def notice_image_warning(status: dict[str, Any]) -> dict[str, Any]:
        return finish(
            {
                "ok": False,
                "warning": True,
                "error": (
                    "This image does not contain readable notice text. "
                    "Upload a clear screenshot of the full notice or message."
                ),
                "error_code": "noticeImageRequiredWarning",
                "status": status,
            }
        )

    def finish(response: dict[str, Any]) -> dict[str, Any]:
        if save_trace:
            try:
                trace_id, queued = queue_trace(
                    text=text,
                    image_data_url=image_data_url,
                    example_id=example_id,
                    assessment=response.get("assessment"),
                )
                response["trace"] = {"trace_id": trace_id, "status": queued}
            except Exception:
                response["trace"] = {"trace_id": "", "status": "failed"}
        else:
            response["trace"] = {"trace_id": "", "status": "disabled"}
        return response

    valid_example = example_id in EXAMPLE_ASSESSMENTS
    if not text and not image_data_url and not valid_example:
        return finish(
            {
                "ok": False,
                "error": "Paste a message or upload a screenshot to continue.",
                "status": model_endpoint.model_status(),
            }
        )

    if valid_example:
        return finish(
            {
                "ok": True,
                "assessment": dict(EXAMPLE_ASSESSMENTS[example_id]),
                "status": model_endpoint.model_status(),
                "source": "cached_local_example",
            }
        )

    status = model_endpoint.model_status()
    if not status["connected"]:
        return finish(
            {
                "ok": False,
                "error": (
                    "The model runtime is not ready. Install the configured "
                    "backend and check the model configuration."
                ),
                "error_code": "modelConfigurationError",
                "status": status,
            }
        )

    try:
        result = model_endpoint.call_model(text, image_data_url, output_language)
        return finish(
            {
                "ok": True,
                "assessment": sanitize_model_guidance(result),
                "status": status,
                "source": "local_model",
            }
        )
    except model_endpoint.NoticeImageInputError:
        return notice_image_warning(status)
    except model_endpoint.ModelRuntimeError as exc:
        if image_data_url and "Nemotron-Parse" in str(exc):
            message = str(exc)
            error_code = "ocrUnavailableError"
        else:
            message = "The local model is unavailable or could not be loaded."
            error_code = "modelUnavailableError"
    except (RuntimeError, ValueError) as exc:
        exc_text = str(exc)
        if "ZeroGPU quota" in exc_text or "exceeded your ZeroGPU" in exc_text:
            message = "GPU quota exceeded. Please try again later or authenticate with a Hugging Face token for more quota."
            error_code = "gpuQuotaError"
        elif image_data_url and not text:
            return notice_image_warning(status)
        else:
            message = "The local model returned an invalid response. Please try again."
            error_code = "modelInvalidError"

    return finish(
        {
            "ok": False,
            "error": message,
            "error_code": error_code,
            "status": {**status, "connected": False, "label": "Local model unavailable"},
        }
    )