app/services/analyzer.py

from __future__ import annotations

import hashlib
import json
import logging
import time
from typing import Protocol
from uuid import UUID, uuid4

from fastapi import HTTPException
from sqlalchemy.orm import Session

from ..config import settings
from ..db.models import Analysis, Feature
from ..schemas import (
    AnalysisResult,
    AnalysisSource,
    AnalyzeTextRequest,
    DetectedFeature,
    Lang,
    RiskLevel,
)
from .audio import temp_audio
from .features import extract
from .language import detect_language

log = logging.getLogger("fraud.analyzer")


class ClassifierLike(Protocol):
    def predict_proba(self, text: str) -> float: ...


def fuse(ml_proba: float, features: list[DetectedFeature]) -> float:
    """
    Disagreement-aware fusion of the ML probability and rule-based features.

    Base formula (kept for continuity with the spec):
        score = 0.6 * ml_proba + 0.4 * min(1, Σ feature_weights)

    Adjustments applied after the base score to be robust when one signal
    is missing or wrong:

    * **ML strong, no rule support** — cap below the HIGH threshold. A
      confident ML signal on text that triggers zero hand-crafted features
      is more likely overfit than genuine, so we degrade to MEDIUM.
    * **Strong rule signal, ML weak** — floor above the LOW threshold. The
      hand-crafted lexicons are precise; multiple matches mean the call /
      message looks like fraud even if ML disagrees (e.g. unusual wording
      the classifier hasn't seen).
    """
    rule_bonus = min(1.0, sum(f.weight for f in features))
    base = 0.6 * ml_proba + 0.4 * rule_bonus

    # Strong ML but no rule support — distrust slightly
    if ml_proba >= 0.7 and rule_bonus < 0.10:
        base = min(base, 0.55)
    # Strong rules but ML disagrees — force at least MEDIUM
    elif ml_proba < 0.30 and rule_bonus >= 0.50:
        base = max(base, 0.45)
    # Both agree strongly — confirm HIGH
    elif ml_proba >= 0.75 and rule_bonus >= 0.35:
        base = max(base, 0.75)

    return float(min(1.0, max(0.0, base)))


def to_risk_level(score: float, lo: float | None = None, hi: float | None = None) -> RiskLevel:
    lo = lo if lo is not None else settings.risk_low_threshold
    hi = hi if hi is not None else settings.risk_high_threshold
    if score < lo:
        return RiskLevel.LOW
    if score < hi:
        return RiskLevel.MEDIUM
    return RiskLevel.HIGH


def _preview(text: str, limit: int = 240) -> str:
    text = text.strip()
    return text if len(text) <= limit else text[: limit - 1] + "…"


def _persist(
    db: Session,
    *,
    analysis_id: UUID,
    device_id: UUID,
    source: AnalysisSource,
    language: Lang,
    risk_score: float,
    risk_level: RiskLevel,
    detected: list[DetectedFeature],
    transcript_preview: str | None,
    text_hash: str | None,
    duration_ms: int,
) -> Analysis:
    row = Analysis(
        id=str(analysis_id),
        device_id=str(device_id),
        source=source.value,
        language=language,
        risk_score=risk_score,
        risk_level=risk_level.value,
        duration_ms=duration_ms,
        transcript_preview=transcript_preview,
        text_hash=text_hash,
    )
    for feat in detected:
        row.features.append(
            Feature(
                type=feat.type.value,
                weight=feat.weight,
                evidence=json.dumps(feat.evidence, ensure_ascii=False),
            )
        )
    db.add(row)
    db.commit()
    db.refresh(row)
    return row


def _result_from_row(row: Analysis, detected: list[DetectedFeature], transcript: str | None) -> AnalysisResult:
    return AnalysisResult(
        id=UUID(row.id),
        created_at=row.created_at,
        source=AnalysisSource(row.source),
        language=row.language,
        risk_score=row.risk_score,
        risk_level=RiskLevel(row.risk_level),
        detected_features=detected,
        transcript=transcript,
        duration_ms=row.duration_ms,
    )


def analyze_text_sync(
    payload: AnalyzeTextRequest,
    device_id: UUID,
    db: Session,
    clf: ClassifierLike | None,
) -> AnalysisResult:
    started = time.perf_counter()
    text = payload.text
    lang = detect_language(text, payload.language_hint)
    detected = extract(text, lang)
    ml_proba = clf.predict_proba(text) if clf is not None else 0.0
    score = fuse(ml_proba, detected)
    level = to_risk_level(score)

    duration_ms = int((time.perf_counter() - started) * 1000)
    analysis_id = uuid4()
    text_hash = hashlib.sha256(text.encode("utf-8")).hexdigest()

    row = _persist(
        db,
        analysis_id=analysis_id,
        device_id=device_id,
        source=AnalysisSource.TEXT,
        language=lang,
        risk_score=score,
        risk_level=level,
        detected=detected,
        transcript_preview=_preview(text),
        text_hash=text_hash,
        duration_ms=duration_ms,
    )
    return _result_from_row(row, detected, transcript=text)


def analyze_audio_sync(
    audio_bytes: bytes,
    content_type: str | None,
    source: AnalysisSource,
    language_hint: Lang | None,
    device_id: UUID,
    db: Session,
    whisper,
    clf: ClassifierLike | None,
) -> AnalysisResult:
    from ..ml.whisper import transcribe

    started = time.perf_counter()
    suffix = _suffix_for_mime(content_type)

    with temp_audio(audio_bytes, suffix) as path:
        transcript, whisper_lang, _ = transcribe(whisper, path, language_hint)

    if not transcript or not transcript.strip():
        raise HTTPException(
            status_code=400,
            detail={"error": "EMPTY_AUDIO", "message": "Whisper returned no segments"},
        )

    lang: Lang = language_hint if language_hint in ("ru", "kk") else (
        whisper_lang if whisper_lang in ("ru", "kk") else detect_language(transcript)
    )
    detected = extract(transcript, lang)
    ml_proba = clf.predict_proba(transcript) if clf is not None else 0.0
    score = fuse(ml_proba, detected)
    level = to_risk_level(score)

    duration_ms = int((time.perf_counter() - started) * 1000)
    analysis_id = uuid4()
    text_hash = hashlib.sha256(transcript.encode("utf-8")).hexdigest()

    row = _persist(
        db,
        analysis_id=analysis_id,
        device_id=device_id,
        source=source,
        language=lang,
        risk_score=score,
        risk_level=level,
        detected=detected,
        transcript_preview=_preview(transcript),
        text_hash=text_hash,
        duration_ms=duration_ms,
    )
    return _result_from_row(row, detected, transcript=transcript)


_MIME_SUFFIX = {
    "audio/mpeg": ".mp3",
    "audio/mp3": ".mp3",
    "audio/wav": ".wav",
    "audio/x-wav": ".wav",
    "audio/mp4": ".m4a",
    "audio/m4a": ".m4a",
    "audio/x-m4a": ".m4a",
    "audio/ogg": ".ogg",
    "audio/webm": ".webm",
    "audio/aac": ".aac",
}


def _suffix_for_mime(mime: str | None) -> str:
    if not mime:
        return ".bin"
    return _MIME_SUFFIX.get(mime.lower(), ".bin")