# Video Analysis API v4.2
# Whisper transcription + emotion2vec voice emotion + HSEmotion facial emotions

import os
import base64
import math
import tempfile
import subprocess
import shutil
import threading
from concurrent.futures import ThreadPoolExecutor
from fastapi import FastAPI, UploadFile, File, Query
from fastapi.middleware.cors import CORSMiddleware
import torch
import service_auth
import numpy as np

print("=" * 50, flush=True)
print(f"[STARTUP] CUDA available: {torch.cuda.is_available()}", flush=True)
if torch.cuda.is_available():
    print(f"[STARTUP] GPU: {torch.cuda.get_device_name(0)}", flush=True)
print("=" * 50, flush=True)

# No interactive docs: the API is for the EchoDepth app only (service_auth.py)
app = FastAPI(title="EchoDepth Video Analysis API", docs_url=None, redoc_url=None, openapi_url=None)
# Pass check first, CORS added after so it wraps it: a refusal still reaches the browser readably
app.middleware("http")(service_auth.require_pass)
app.add_middleware(CORSMiddleware, allow_origins=service_auth.ALLOWED_ORIGINS, allow_credentials=False, allow_methods=["GET", "POST"], allow_headers=["*"])
print(f"[STARTUP] Service passes: {'enforced' if service_auth.enforcing() else 'OFF - set VIDEO_SERVICE_SECRET'}", flush=True)

# Lazy load models. Loads take the lock, so the start-up warm-up thread and a first request
# arriving meanwhile never load the same model twice.
_LOAD = threading.RLock()
# One analysis at a time. Requests are answered from a thread pool now (the endpoints are plain
# def), so /health and the job polls stay responsive during an analysis; this keeps two films from
# sharing the GPU and from mixing their per-request entries in MODEL_ERRORS.
_ANALYSE = threading.Lock()
# Entries in MODEL_ERRORS that describe one request rather than a model that failed to load.
# Cleared at the start of each analysis: before, one failed CEI pass was reported on every
# analysis after it until the service restarted.
_PER_REQUEST_ERRORS = ("cei", "stills", "proxy", "decode")


def _loaded_once(fn):
    """Run a model getter under the load lock."""
    def wrapper(*args, **kwargs):
        with _LOAD:
            return fn(*args, **kwargs)
    wrapper.__name__ = fn.__name__
    wrapper.__doc__ = fn.__doc__
    return wrapper


whisper_model = None
emotion2vec_model = None
face_emotion_model = None

# Why a model did not load, keyed by analyser. An optional model that fails to import used to be
# swallowed here and the request still answered 200 OK, so a service missing funasr returned a
# complete-looking analysis with the voice channel silently absent — and downstream, congruence
# defaulted rather than reporting itself unmeasured. Every response now says what actually ran.
MODEL_ERRORS: dict = {}

# emotion2vec labels (9 classes)
EMOTION2VEC_LABELS = ['angry', 'disgust', 'fearful', 'happy', 'neutral', 'other', 'sad', 'surprised', 'unknown']

# Normalize emotion labels from emotion2vec (handles Chinese/mixed labels)
EMOTION_LABEL_MAP = {
    '中立/neutral': 'neutral', '中立': 'neutral', 'neutral': 'neutral', 'neutra': 'neutral',
    '开心/happy': 'happy', '开心': 'happy', 'happy': 'happy',
    '生气/angry': 'angry', '生气': 'angry', 'angry': 'angry',
    '难过/sad': 'sad', '难过': 'sad', 'sad': 'sad',
    '恐惧/fear': 'fear', '恐惧': 'fear', 'fear': 'fear', 'fearful': 'fear',
    '惊讶/surprise': 'surprise', '惊讶': 'surprise', 'surprise': 'surprise', 'surprised': 'surprise',
    '厌恶/disgust': 'disgust', '厌恶': 'disgust', 'disgust': 'disgust',
    '其他/other': 'other', '其他': 'other', 'other': 'other',
    '<unk>': 'unknown', 'unknown': 'unknown', 'unk': 'unknown',
}

def normalize_emotion_label(label):
    """Normalize emotion label to clean English.

    emotion2vec_plus_large emits '吃惊/surprised', '恐惧/fearful' and '厌恶/disgusted', none of which
    were in the map, so three of its nine classes reached the app untranslated. Anything the map
    misses now falls back to the English half after the slash, mapped again.
    """
    label_lower = label.lower().strip()
    if label_lower in EMOTION_LABEL_MAP:
        return EMOTION_LABEL_MAP[label_lower]
    english = label_lower.split('/')[-1].strip()
    return EMOTION_LABEL_MAP.get(english, {'disgusted': 'disgust'}.get(english, english))

whisper_fast_model = None


@_loaded_once
def get_whisper_fast():
    """Whisper small, for short spoken answers people correct before sending (Reflections).

    Several times quicker than medium on this CPU. Leave cpu_threads at its default: os.cpu_count()
    reports the host's cores, not the Space's two, and oversubscribing them made it six times slower.
    The brand-name misses that ruled small out for
    ads do not matter here: every transcript is shown to its speaker to correct first.
    """
    global whisper_fast_model
    if whisper_fast_model is None:
        from faster_whisper import WhisperModel
        device = "cuda" if torch.cuda.is_available() else "cpu"
        compute = "float16" if device == "cuda" else "int8"
        print(f"[whisper] Loading small (fast profile) on {device}", flush=True)
        whisper_fast_model = WhisperModel("small", device=device, compute_type=compute)
    return whisper_fast_model


@_loaded_once
def get_whisper():
    global whisper_model
    if whisper_model is None:
        from faster_whisper import WhisperModel
        device = "cuda" if torch.cuda.is_available() else "cpu"
        compute = "float16" if device == "cuda" else "int8"
        # "medium" everywhere. "small" dropped a sung brand name at the end of a 30-second film
        # ("making smart moves with..." for "with Go.Compare") whatever it was prompted with;
        # medium with the brand as a hint hears it. The memorability words channel was fitted to
        # "small" transcripts of the training ads (2026-09-21); a medium re-run is the next cut.
        model_size = "medium"
        print(f"[whisper] Loading {model_size} on {device}", flush=True)
        whisper_model = WhisperModel(model_size, device=device, compute_type=compute)
    return whisper_model

@_loaded_once
def get_emotion2vec():
    """Load emotion2vec model for voice emotion recognition"""
    global emotion2vec_model
    if emotion2vec_model is None:
        try:
            from funasr import AutoModel
            print("[voice] Loading emotion2vec model...", flush=True)
            # Use HuggingFace model name (emotion2vec/) not ModelScope (iic/)
            emotion2vec_model = AutoModel(
                model="emotion2vec/emotion2vec_plus_large",
                hub="hf",
                device="cuda" if torch.cuda.is_available() else "cpu"
            )
            print("[voice] emotion2vec loaded successfully", flush=True)
        except Exception as e:
            print(f"[voice] Failed to load emotion2vec: {e}", flush=True)
            MODEL_ERRORS["voice_emotion"] = f"{type(e).__name__}: {e}"
            emotion2vec_model = "failed"
    return emotion2vec_model if emotion2vec_model != "failed" else None

@_loaded_once
def get_face_emotion():
    """Load HSEmotion ONNX model for facial emotion recognition"""
    global face_emotion_model
    if face_emotion_model is None:
        try:
            import onnxruntime as ort
            from hsemotion_onnx.facial_emotions import HSEmotionRecognizer, get_model_path
            print("[face] Loading HSEmotion ONNX...", flush=True)
            face_emotion_model = HSEmotionRecognizer(model_name='enet_b0_8_best_afew')
            # The library opens its session on the CPU provider only. With onnxruntime-gpu in the
            # image, reopen it on CUDA; the model and its outputs are the same, only where it runs.
            # Checked against torch too: onnxruntime-gpu lists the CUDA provider on a machine with
            # no GPU (the CPU fallback Space runs this same image).
            if torch.cuda.is_available() and "CUDAExecutionProvider" in ort.get_available_providers():
                face_emotion_model.ort_session = ort.InferenceSession(
                    get_model_path('enet_b0_8_best_afew'), providers=["CUDAExecutionProvider", "CPUExecutionProvider"])
            print(f"[face] HSEmotion ONNX loaded on {face_emotion_model.ort_session.get_providers()[0]}", flush=True)
        except Exception as e:
            import traceback
            print(f"[face] Failed to load HSEmotion ONNX: {e}", flush=True)
            print(f"[face] Traceback: {traceback.format_exc()}", flush=True)
            MODEL_ERRORS["face_emotion"] = f"{type(e).__name__}: {e}"
            face_emotion_model = "failed"
    return face_emotion_model if face_emotion_model != "failed" else None

ocr_reader = None

@_loaded_once
def get_ocr():
    """EasyOCR reader, loaded once. Optional like the emotion models: a service without it still
    answers, and says so, rather than pretending the film carried no on-screen copy."""
    global ocr_reader
    if ocr_reader is None:
        try:
            import easyocr
            print("[ocr] Loading EasyOCR...", flush=True)
            # On the GPU when there is one. It was pinned to the CPU when the Space had six
            # gigabytes of VRAM, where adding it beside Whisper, emotion2vec and HSEmotion killed
            # the service mid-request; the GPUs it runs on now have 24 GB or more, and OCR on full
            # frames was the slowest pass on the CPU.
            ocr_reader = easyocr.Reader(["en"], gpu=torch.cuda.is_available(), verbose=False)
            print(f"[ocr] EasyOCR loaded on {'cuda' if torch.cuda.is_available() else 'cpu'}", flush=True)
        except Exception as e:
            print(f"[ocr] Failed to load EasyOCR: {e}", flush=True)
            MODEL_ERRORS["screen_text"] = f"{type(e).__name__}: {e}"
            ocr_reader = "failed"
    return ocr_reader if ocr_reader != "failed" else None


class _TextPass:
    """Words on screen, at fps frames a second of the film. See analyze_screen_text."""

    def __init__(self, video_fps, fps=1.0, min_conf=0.4):
        self.reader = get_ocr()
        self.interval = max(1, int(video_fps / fps))
        self.video_fps = video_fps
        self.min_conf = min_conf
        self.out = []

    def wants(self, frame_num):
        return self.reader is not None and frame_num % self.interval == 0

    def add(self, frame, frame_num):
        t = frame_num / self.video_fps
        try:
            found = self.reader.readtext(frame)
        except Exception as e:
            print(f"[ocr] frame {t:.1f}s failed: {e}", flush=True)
            found = []
        words = [
            {"text": txt.strip(), "confidence": round(float(conf), 3)}
            for _box, txt, conf in found
            if conf >= self.min_conf and txt.strip()
        ]
        if words:
            self.out.append({"time": round(t, 2), "words": words,
                             "text": " ".join(w["text"] for w in words)})

    def finish(self):
        return self.out


def analyze_screen_text(video_path, fps=1.0, min_conf=0.4):
    """Read the words on screen, sampled through the film.

    An advert with no voiceover is not an advert with no message: the message is in the supers and
    the end frame. Reading only the audio channel and concluding "nothing was said" is how a
    perfectly clear piece of copy came back unscored.
    """
    return scan_frames(video_path, screen_text=True, faces=False, ocr_fps=fps, ocr_min_conf=min_conf)[0]


def brand_presence(text_timeline, transcript, duration, brand):
    """When the brand is actually present, and where.

    Ehrenberg-Bass: an advert nobody attributes is an advert that sells someone else's product.
    This is the measurable half of that — first appearance, share of the film it is on screen for,
    whether it is there in the opening seconds, and whether it closes the film.
    """
    if not brand or duration <= 0:
        return None
    # Comma-separated aliases, because a brand is rarely one string on screen: this campaign
    # carries WRU, Wales and Cymru, and matching only the legal name would have reported the
    # brand absent from a film that shows it four seconds in.
    aliases = [a.strip().lower() for a in brand.split(",") if a.strip()]
    if not aliases:
        return None
    tokens = []
    for a in aliases:
        parts = [t for t in a.replace("&", " ").split() if len(t) >= 3]
        tokens.extend(parts or [a])
    if not tokens:
        return None

    def _close(a, b, budget):
        """Levenshtein within a small budget — enough for one or two misread characters."""
        if abs(len(a) - len(b)) > budget:
            return False
        prev = list(range(len(b) + 1))
        for i, ca in enumerate(a, 1):
            cur = [i]
            for j, cb in enumerate(b, 1):
                cur.append(min(prev[j] + 1, cur[j - 1] + 1, prev[j - 1] + (ca != cb)))
            if min(cur) > budget:
                return False
            prev = cur
        return prev[-1] <= budget

    def hit(text):
        low = (text or "").lower()
        if any(t in low for t in tokens):
            return True
        # Brand names are set in display type, which is where OCR slips: this campaign's CYMRU
        # came back as GYMRU and GIMRU. An exact match would have reported the brand absent from
        # frames it is plainly on, which is the opposite of the finding.
        words = [w.strip(".,:;!?/()-") for w in low.split()]
        for t in tokens:
            budget = 1 if len(t) <= 5 else 2
            if any(_close(t, w, budget) for w in words if abs(len(w) - len(t)) <= budget):
                return True
        return False

    frames = [f for f in text_timeline if hit(f.get("text"))]
    sampled = len(text_timeline) or 1
    first = min((f["time"] for f in frames), default=None)
    last = max((f["time"] for f in frames), default=None)

    return {
        "brand": brand,
        "on_screen_frames": len(frames),
        "sampled_frames": len(text_timeline),
        "share_of_frames": round(100 * len(frames) / sampled, 1) if text_timeline else None,
        "first_seen_s": first,
        "last_seen_s": last,
        # The two placements that carry attribution: the hook, and the end frame nobody skips past.
        "in_opening_3s": bool(first is not None and first <= 3),
        "in_closing_3s": bool(last is not None and last >= duration - 3),
        "spoken": hit(transcript),
    }


text_emotion_model = None

# GoEmotions label -> (valence, arousal, dominance), 0-1. Same table the Stage Read pipeline uses,
# so a line of ad copy and a line of a speech are scored on one scale.
TEXT_VAD = {
    "admiration": (0.80, 0.50, 0.55), "amusement": (0.80, 0.60, 0.60), "anger": (0.15, 0.85, 0.75),
    "annoyance": (0.25, 0.60, 0.55), "approval": (0.75, 0.40, 0.60), "caring": (0.75, 0.40, 0.55),
    "confusion": (0.35, 0.50, 0.30), "curiosity": (0.65, 0.55, 0.50), "desire": (0.70, 0.65, 0.55),
    "disappointment": (0.20, 0.40, 0.30), "disapproval": (0.25, 0.50, 0.55), "disgust": (0.15, 0.60, 0.55),
    "embarrassment": (0.20, 0.55, 0.20), "excitement": (0.85, 0.85, 0.65), "fear": (0.15, 0.80, 0.20),
    "gratitude": (0.85, 0.45, 0.50), "grief": (0.10, 0.45, 0.20), "joy": (0.90, 0.70, 0.65),
    "love": (0.90, 0.60, 0.60), "nervousness": (0.25, 0.75, 0.25), "optimism": (0.80, 0.55, 0.65),
    "pride": (0.80, 0.60, 0.75), "realization": (0.55, 0.45, 0.50), "relief": (0.75, 0.30, 0.55),
    "remorse": (0.15, 0.40, 0.20), "sadness": (0.10, 0.30, 0.20), "surprise": (0.60, 0.80, 0.40),
    "neutral": (0.50, 0.30, 0.50),
}


@_loaded_once
def get_text_emotion():
    global text_emotion_model
    if text_emotion_model is None:
        try:
            from transformers import pipeline as hf_pipeline
            print("[text] Loading GoEmotions...", flush=True)
            text_emotion_model = hf_pipeline("text-classification",
                                             model="SamLowe/roberta-base-go_emotions",
                                             top_k=None, device=0 if torch.cuda.is_available() else -1)
            print("[text] GoEmotions loaded", flush=True)
        except Exception as e:
            print(f"[text] Failed to load GoEmotions: {e}", flush=True)
            MODEL_ERRORS["copy_emotion"] = f"{type(e).__name__}: {e}"
            text_emotion_model = "failed"
    return text_emotion_model if text_emotion_model != "failed" else None


def _normalise_case(text):
    """Set-in-caps copy to sentence case before scoring.

    The model reads capitals as shouting: the identical sentence scores 95 in capitals and 13 in
    sentence case. End frames are nearly always set in caps and a rewritten line arrives in
    sentence case, so without this every original beats every alternative on typesetting alone —
    which is the opposite of the comparison the report claims to be making.
    """
    letters = [c for c in text if c.isalpha()]
    if letters and sum(c.isupper() for c in letters) / len(letters) > 0.7:
        lowered = text.lower()
        return lowered[:1].upper() + lowered[1:]
    return text


def score_lines(lines):
    """Emotional content of each line of copy.

    `feel` is 1 minus the neutral score: how much emotional work the words are doing at all. It is
    the number that matters for advertising, because the failure mode is not being disliked, it is
    being ignored — and a line that reads as pure neutral has nothing for anyone to react to.
    """
    model = get_text_emotion()
    if not model or not lines:
        return []
    out = []
    for line in lines:
        text = (line or "").strip()
        if len(text) < 3:
            continue
        try:
            scores = {d["label"]: float(d["score"]) for d in model(_normalise_case(text)[:500])[0]}
        except Exception as e:
            print(f"[text] scoring failed: {e}", flush=True)
            continue
        total = sum(scores.values()) or 1e-9
        vad = [sum(TEXT_VAD.get(k, (0.5, 0.3, 0.5))[i] * v for k, v in scores.items()) / total for i in range(3)]
        top = sorted(((k, v) for k, v in scores.items() if k != "neutral"), key=lambda kv: -kv[1])[:3]
        out.append({
            "text": text,
            "feel": round(1 - scores.get("neutral", 0.0), 3),
            "valence": round(vad[0], 3),
            "arousal": round(vad[1], 3),
            "dominance": round(vad[2], 3),
            "top": [{"emotion": k, "score": round(v, 3)} for k, v in top],
        })
    return out


YUNET_MODEL = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stage_read", "models", "face_detection_yunet_2023mar.onnx")
# Frames wider than this are downscaled before detection: YuNet is a small network and 1080p buys
# nothing over 960 wide for faces that are in shot at all, at four times the cost.
YUNET_MAX_WIDTH = 960


_MEMORABILITY = None
try:
    import memorability as memorability_mod
except Exception as e:  # the service still starts without it; the pass then reports why
    memorability_mod = None
    MODEL_ERRORS["memorability"] = f"{type(e).__name__}: {e}"

@_loaded_once
def get_memorability():
    """CLIP, PerceptCLIP-Memorability and the fitted LAMBDA model, loaded on first use (about 3 GB of weights)."""
    global _MEMORABILITY
    if _MEMORABILITY is None:
        if memorability_mod is None:
            raise RuntimeError(MODEL_ERRORS.get("memorability", "memorability module missing"))
        yunet = os.path.join(os.path.dirname(os.path.abspath(__file__)), "stage_read", "models", "face_detection_yunet_2023mar.onnx")
        predictor = memorability_mod.Predictor()
        text_model = memorability_mod.TextModel() if predictor.needs_text() else None
        _MEMORABILITY = (memorability_mod.FrameModels(None, yunet), predictor, text_model)
    return _MEMORABILITY

def get_face_detector():
    """The detector behind the four-frames-a-second emotion pass.

    YuNet where its model is present (the Dockerfile fetches it for Stage Read), the Haar cascade
    otherwise. The cascade found no face on 43% of frames of a film that has a person in shot for
    most of it: it wants a frontal face of a decent size and loses profiles, small faces and motion
    blur, and the second-by-second chart came out as a comb. YuNet handles all three.

    This pass is not part of the Creative Effectiveness Index. The index's own two-a-second pass in
    cei_measures.py keeps the cascade, because the corpus was measured with it and must stay so.
    """
    import cv2
    if os.path.exists(YUNET_MODEL):
        try:
            det = cv2.FaceDetectorYN.create(YUNET_MODEL, "", (320, 320), score_threshold=0.6, nms_threshold=0.3, top_k=50)
            return ("yunet", det)
        except Exception as e:
            print(f"[face] YuNet unavailable ({e}); using the Haar cascade", flush=True)
    return ("haar", cv2.CascadeClassifier(cv2.data.haarcascades + 'haarcascade_frontalface_default.xml'))


def detect_faces(detector, frame, gray):
    """(x, y, w, h) boxes in the frame's own pixels, from whichever detector get_face_detector chose."""
    import cv2
    kind, det = detector
    if kind == "haar":
        return [tuple(int(v) for v in f) for f in det.detectMultiScale(gray, 1.1, 5, minSize=(60, 60))]
    h, w = frame.shape[:2]
    scale = min(1.0, YUNET_MAX_WIDTH / max(w, 1))
    small = cv2.resize(frame, (int(w * scale), int(h * scale)), interpolation=cv2.INTER_AREA) if scale < 1 else frame
    det.setInputSize((small.shape[1], small.shape[0]))
    _, faces = det.detect(small)
    out = []
    if faces is not None:
        for f in faces:
            x, y, fw, fh = (float(v) / scale for v in f[:4])
            if fw >= 24 and fh >= 24:
                out.append((max(0, int(x)), max(0, int(y)), int(fw), int(fh)))
    return out

def extract_audio(video_path, audio_path):
    """Returns None on success, or a message explaining the failure.

    The bare `except: return False` this replaced turned every cause — a corrupt upload, a missing
    ffmpeg, a process that could not start at all — into the same unactionable "Failed to extract
    audio" at the API boundary.
    """
    from ffmpeg_util import run_ffmpeg, FFmpegError
    try:
        run_ffmpeg(["-i", video_path, "-vn", "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", audio_path])
        return None
    except FFmpegError as e:
        print(f"[audio] {e}", flush=True)
        return str(e)
    except Exception as e:
        print(f"[audio] {type(e).__name__}: {e}", flush=True)
        return f"{type(e).__name__}: {e}"

def transcribe(audio_path, brand=None):
    """
    With a brand, its name goes to Whisper as the initial prompt, which biases recognition toward
    the term; a brand name that is sung, or unusual, is otherwise the word most likely to be lost.
    """
    model = get_whisper()
    prompt = f"{brand.split(',')[0].strip()}." if brand and brand.strip() else None
    segments, info = model.transcribe(audio_path, beam_size=5, language="en", initial_prompt=prompt)
    return [{"start": s.start, "end": s.end, "text": s.text.strip(), "confidence": round(math.exp(s.avg_logprob), 4)} for s in segments], info.language

def _voice_result(result):
    """emotion2vec's output for one clip, as the response carries it."""
    if result and len(result) > 0:
        scores = result[0].get("scores", [])
        labels = result[0].get("labels", EMOTION2VEC_LABELS)
        if scores:
            all_emotions = {normalize_emotion_label(labels[i]): round(float(scores[i]), 4) for i in range(len(scores))}
            top_idx = int(np.argmax(scores))
            return {
                "emotion": normalize_emotion_label(labels[top_idx]),
                "confidence": round(float(scores[top_idx]), 4),
                "all_emotions": all_emotions,
            }
    return None


def _voice_clip(model, y, sr):
    """One clip through emotion2vec, from memory.

    The clip used to be written to a temporary WAV and read back. The source is 16-bit PCM, so the
    samples are exact multiples of 1/32768 and that round trip changed nothing; passing the array
    skips a file write and read per segment. A funasr that refuses arrays gets the file instead.
    """
    import soundfile as sf
    try:
        return _voice_result(model.generate(y, fs=sr, granularity="utterance", extract_embedding=False))
    except Exception as e:
        print(f"[voice] array input refused ({type(e).__name__}); using a temporary file", flush=True)
    with tempfile.NamedTemporaryFile(suffix=".wav", delete=False) as f:
        sf.write(f.name, y, sr)
        temp_path = f.name
    try:
        return _voice_result(model.generate(temp_path, granularity="utterance", extract_embedding=False))
    finally:
        os.unlink(temp_path)


def _segment_samples(y, sr, start, end):
    """The samples librosa.load(offset=start, duration=end-start) returns, cut from the whole file."""
    i0 = int(start * sr)
    return y[i0:i0 + int((end - start) * sr)]


def analyze_voice_segments(audio_path, segments):
    """Voice emotion for each segment, reading the audio once.

    Each segment used to reload the file from its own offset and go through a temporary WAV. Same
    samples, same model, same output; one read instead of one per sentence.
    """
    import librosa
    model = get_emotion2vec()
    if not model or not segments:
        return [None] * len(segments)
    y_all, sr = librosa.load(audio_path, sr=16000)
    out = []
    for seg in segments:
        y = _segment_samples(y_all, sr, seg["start"], seg["end"])
        if len(y) < 1600:  # Too short
            out.append(None)
            continue
        try:
            out.append(_voice_clip(model, y, sr))
        except Exception as e:
            print(f"[voice] emotion2vec error: {e}", flush=True)
            out.append(None)
    return out


def analyze_voice_emotion(audio_path, start, end):
    """Voice emotion for one stretch of the audio (the transcription jobs read it one at a time)."""
    return analyze_voice_segments(audio_path, [{"start": start, "end": end}])[0]

# HSEmotion-ONNX labels (correct order: Anger, Contempt, Disgust, Fear, Happiness, Neutral, Sadness, Surprise)
FACE_EMOTION_LABELS = ['angry', 'contempt', 'disgust', 'fear', 'happy', 'neutral', 'sad', 'surprise']
# Face crops sent to HSEmotion in one call. One call per face was most of the cost of the pass on a
# GPU; the scores are the model's own either way (a batch is the same per-image forward pass).
FACE_BATCH = 32


class _FacePass:
    """Faces and their emotions, at fps frames a second of the film. See analyze_faces."""

    def __init__(self, video_fps, fps=0.5, use_emotions=True):
        self.detector = get_face_detector()
        self.emotion_model = get_face_emotion() if use_emotions else None
        self.interval = max(1, int(video_fps / fps))
        self.video_fps = video_fps
        self.timeline = []
        self.pending = []  # (timeline index, face index, crop) awaiting HSEmotion

    def wants(self, frame_num):
        return frame_num % self.interval == 0

    def add(self, frame, frame_num):
        import cv2
        time = frame_num / self.video_fps
        gray = cv2.cvtColor(frame, cv2.COLOR_BGR2GRAY)
        faces = detect_faces(self.detector, frame, gray)

        emotions = {}
        # Every face's box and reading, in the frame's own pixels, so the player can draw the
        # measurement on the person it was taken from.
        per_face = [{"box": [int(x), int(y), int(w), int(h)], "emotions": {}} for (x, y, w, h) in faces]
        self.timeline.append({"time": time, "faces": len(faces), "emotions": emotions,
                              "size": [int(frame.shape[1]), int(frame.shape[0])], "per_face": per_face})
        if len(faces) > 0 and self.emotion_model:
            for fi, (x, y, w, h) in enumerate(faces):
                # Expand face region slightly for better detection
                pad = int(w * 0.1)
                x1 = max(0, x - pad)
                y1 = max(0, y - pad)
                x2 = min(frame.shape[1], x + w + pad)
                y2 = min(frame.shape[0], y + h + pad)
                crop = frame[y1:y2, x1:x2]
                if crop.size:
                    self.pending.append((len(self.timeline) - 1, fi, crop.copy()))
            if len(self.pending) >= FACE_BATCH:
                self._flush()
        elif len(faces) > 0:
            # Fallback: basic heuristic if HSEmotion not available
            x, y, w, h = faces[0]
            face_roi = gray[y:y+h, x:x+w]
            brightness = face_roi.mean() / 255
            emotions["neutral"] = round(0.5 + brightness * 0.2, 3)

    def _flush(self):
        if not self.pending:
            return
        batch, self.pending = self.pending, []
        try:
            _, scores = self.emotion_model.predict_multi_emotions([c for _, _, c in batch], logits=False)
        except Exception as e:
            print(f"[face] HSEmotion batch error ({e}); scoring faces one at a time", flush=True)
            scores = []
            for _, _, crop in batch:
                try:
                    _, sc = self.emotion_model.predict_emotions(crop, logits=False)
                    scores.append(sc[0] if len(sc.shape) > 1 else sc)
                except Exception as e2:
                    print(f"[face] HSEmotion error for face: {e2}", flush=True)
                    scores.append(None)
        for (ti, fi, _), sc in zip(batch, scores):
            if sc is None or len(sc) == 0:
                continue
            face_emo = {FACE_EMOTION_LABELS[i]: float(sc[i]) for i in range(min(len(sc), len(FACE_EMOTION_LABELS)))}
            self.timeline[ti]["per_face"][fi]["emotions"] = {k: round(v, 4) for k, v in face_emo.items()}
            self.timeline[ti].setdefault("_raw", []).append(face_emo)

    def finish(self):
        if self.emotion_model:
            self._flush()
        # Aggregate: average emotions across all faces in each frame
        for entry in self.timeline:
            raw = entry.pop("_raw", None)
            if raw:
                for label in FACE_EMOTION_LABELS:
                    entry["emotions"][label] = round(sum(f.get(label, 0) for f in raw) / len(raw), 4)
        return self.timeline


def scan_frames(video_path, screen_text=True, faces=True, face_fps=4, use_emotions=True, ocr_fps=1.0, ocr_min_conf=0.4):
    """The on-screen text pass and the face pass from one read of the film.

    Each pass used to open the file and decode every frame itself. They read the same frames they
    always did (each keeps its own sampling interval), from a single decode. Returns
    (text_timeline, face_timeline).
    """
    import cv2
    cap = cv2.VideoCapture(video_path)
    if not cap.isOpened():
        return [], []
    video_fps = cap.get(cv2.CAP_PROP_FPS) or 30
    text_pass = _TextPass(video_fps, ocr_fps, ocr_min_conf) if screen_text else None
    face_pass = _FacePass(video_fps, face_fps, use_emotions) if faces else None
    frame_num = 0
    try:
        while True:
            ret, frame = cap.read()
            if not ret:
                break
            if text_pass and text_pass.wants(frame_num):
                text_pass.add(frame, frame_num)
            if face_pass and face_pass.wants(frame_num):
                face_pass.add(frame, frame_num)
            frame_num += 1
    finally:
        cap.release()
    return (text_pass.finish() if text_pass else []), (face_pass.finish() if face_pass else [])


def analyze_faces(video_path, fps=0.5, use_emotions=True):
    """Analyze faces in video frames with optional HSEmotion"""
    return scan_frames(video_path, screen_text=False, faces=True, face_fps=fps, use_emotions=use_emotions)[1]

def video_duration(video_path):
    """Running time from the container."""
    import cv2
    cap = cv2.VideoCapture(str(video_path))
    try:
        fps = cap.get(cv2.CAP_PROP_FPS) or 0
        frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT) or 0)
        return round(frames / fps, 3) if fps > 0 and frames > 0 else None
    finally:
        cap.release()


def cut_stills(video_path, duration, width=320, max_frames=240):
    """One small JPEG per second of the film, as data URLs.

    The report puts a frame beside every moment it names, and it cuts those frames in the browser
    from the file it already has open. That fails outright on a master the browser cannot decode -
    ProRes, which is what an agency has to hand, or HEVC - and the moments print as "no still". This
    strip is the fallback: the service has the film decoded already, so it costs one seek per
    second. At 320 wide and quality 72 a frame is about 12 KB, so a 30-second film adds about
    500 KB to the response. Sampled mid-second (0.5, 1.5, ...) so a black slate at 0 is never the
    frame chosen. Longer than four minutes, the step widens so the strip stays under max_frames.
    """
    import cv2  # imported per function throughout this module, so a missing wheel fails one pass rather than the service
    if not duration or duration <= 0:
        return []
    step = max(1.0, float(math.ceil(duration / max_frames)))
    cap = cv2.VideoCapture(video_path)
    out = []
    try:
        t = 0.5
        while t < duration:
            cap.set(cv2.CAP_PROP_POS_MSEC, t * 1000.0)
            ok, frame = cap.read()
            if ok and frame is not None and frame.size:
                h, w = frame.shape[:2]
                scale = width / max(w, 1)
                small = cv2.resize(frame, (width, max(1, int(round(h * scale)))), interpolation=cv2.INTER_AREA)
                ok2, buf = cv2.imencode(".jpg", small, [cv2.IMWRITE_JPEG_QUALITY, 72])
                if ok2:
                    out.append({"time": round(t, 2),
                                "jpeg": "data:image/jpeg;base64," + base64.b64encode(buf.tobytes()).decode("ascii")})
            t += step
    finally:
        cap.release()
    return out


def decodable_copy(video_path):
    """The path the frame passes should read: the upload itself, unless OpenCV cannot decode it.

    OpenCV's Linux wheel ships an ffmpeg with no software AV1 decoder, and YouTube serves a third
    of its downloads as AV1. Every frame pass then returned nothing at all, quietly - "Your
    platform doesn't support hardware accelerated AV1 decoding" in the log, and a film reported as
    measured with no faces in it. Probed by reading one frame. On failure the system ffmpeg, which
    does decode AV1, writes an H.264 copy for the frame passes to read; audio still comes from the
    original, and the copy is the caller's to delete. Returns (path_to_read, copy_to_delete).
    """
    import cv2
    cap = cv2.VideoCapture(video_path)
    try:
        ok, frame = cap.read()
    finally:
        cap.release()
    if ok and frame is not None and frame.size:
        return video_path, None
    from ffmpeg_util import run_ffmpeg, FFmpegError
    copy = video_path + ".h264.mp4"
    print("[process] OpenCV could not decode a frame; re-encoding to H.264 for the frame passes", flush=True)
    try:
        # crf 16 is visually lossless for a face pass that downsamples anyway; veryfast because the
        # point is to get frames at all, not to archive them.
        encode_h264(["-i", video_path, "-an"], 16, ["-pix_fmt", "yuv420p", copy])
    except FFmpegError as e:
        print(f"[process] re-encode failed: {e}", flush=True)
        MODEL_ERRORS["decode"] = f"FFmpegError: {e}"
        return video_path, None
    return copy, copy


_NVENC = None


def nvenc_available():
    """Whether ffmpeg can encode H.264 on the GPU here: listed, and a one-frame test encode works.

    Checked once. The encoder needs the driver's video capability in the container
    (NVIDIA_DRIVER_CAPABILITIES includes video, set in the Dockerfile); without it ffmpeg lists
    h264_nvenc and then fails to open it, which is why this tries an encode rather than trusting the list.
    """
    global _NVENC
    if _NVENC is None:
        _NVENC = False
        if torch.cuda.is_available():
            try:
                from ffmpeg_util import run_ffmpeg
                run_ffmpeg(["-f", "lavfi", "-i", "color=c=black:s=256x256:d=0.1", "-frames:v", "1",
                            "-c:v", "h264_nvenc", "-f", "null", "-"], timeout=60)
                _NVENC = True
            except Exception as e:
                print(f"[encode] GPU H.264 encoder unavailable, using libx264: {e}", flush=True)
        print(f"[encode] H.264 encoder: {'h264_nvenc' if _NVENC else 'libx264'}", flush=True)
    return _NVENC


def h264_args(crf, preset="veryfast"):
    """Encoder arguments for an H.264 copy at about the quality of libx264 at this crf."""
    if nvenc_available():
        return ["-c:v", "h264_nvenc", "-preset", "p4", "-rc", "vbr", "-cq", str(crf), "-b:v", "0"]
    return ["-c:v", "libx264", "-preset", preset, "-crf", str(crf)]


def encode_h264(args_before, crf, args_after, timeout=None):
    """run_ffmpeg with an H.264 video encoder, on the GPU when it can; falls back to libx264 if the GPU encode fails."""
    from ffmpeg_util import run_ffmpeg
    try:
        return run_ffmpeg(args_before + h264_args(crf) + args_after, **({"timeout": timeout} if timeout else {}))
    except Exception:
        if not nvenc_available():
            raise
        print("[encode] GPU encode failed; retrying with libx264", flush=True)
        return run_ffmpeg(args_before + ["-c:v", "libx264", "-preset", "veryfast", "-crf", str(crf)] + args_after,
                          **({"timeout": timeout} if timeout else {}))


PLAYABLE_CODECS = {"h264", "vp8", "vp9", "av1"}
# A proxy bigger than this is left out of the response rather than making it unwieldy.
PROXY_MAX_BYTES = 40 * 1024 * 1024


def probe_codec(video_path):
    """codec_name and pix_fmt of the first video stream, from the system ffprobe; (None, None) if it cannot say."""
    try:
        from ffmpeg_util import ffmpeg_exe
        exe = shutil.which("ffprobe") or os.path.join(os.path.dirname(ffmpeg_exe()), "ffprobe")
        r = subprocess.run([exe, "-v", "error", "-select_streams", "v:0", "-show_entries", "stream=codec_name,pix_fmt",
                            "-of", "csv=p=0", video_path], capture_output=True, text=True, timeout=60)
        parts = (r.stdout.strip().splitlines() or [""])[0].split(",")
        return (parts[0] or None, parts[1] if len(parts) > 1 and parts[1] else None)
    except Exception:
        return (None, None)


def make_proxy(video_path, force=False):
    """A browser-playable copy of a master the browser cannot decode, or None when the upload already is one.

    Agencies send ProRes. A browser plays its audio track over a black frame, the thumbnail comes out
    black, and the report cannot cut a still from it. This writes a 720p H.264 with AAC audio, the
    one format every browser plays, and hands it back base64 in the response so the app can play it,
    cut frames from it and store it as the playable asset beside the original. A master that is
    already H.264, VP8, VP9 or AV1 needs none of this and gets None.
    """
    from ffmpeg_util import run_ffmpeg
    codec, pix = probe_codec(video_path)
    if not force and codec in PLAYABLE_CODECS and (pix or "yuv420p") == "yuv420p":
        return None
    out = video_path + ".proxy.mp4"
    print(f"[proxy] {codec or 'unknown'} {pix or ''} is not browser-playable; writing a 720p H.264 proxy", flush=True)
    # Video and audio only, and no timecode: a ProRes master carries a timecode, and the mp4 muxer
    # writes it back as a data track that shows up as an unknown stream some browsers refuse the
    # whole file for. Mapping the streams is not enough, because the muxer takes it from metadata.
    encode_h264(["-i", video_path, "-map", "0:v:0", "-map", "0:a?", "-dn", "-sn", "-map_metadata", "-1",
                 "-write_tmcd", "0", "-vf", "scale=-2:min(720\\,ih)"], 23,
                ["-pix_fmt", "yuv420p", "-c:a", "aac", "-b:a", "128k", "-movflags", "+faststart", out],
                timeout=900)
    try:
        size = os.path.getsize(out)
        if size > PROXY_MAX_BYTES:
            MODEL_ERRORS["proxy"] = f"proxy is {size // 1_000_000} MB, over the {PROXY_MAX_BYTES // 1_000_000} MB the response carries"
            return None
        with open(out, "rb") as fh:
            data = base64.b64encode(fh.read()).decode("ascii")
    finally:
        if os.path.exists(out):
            os.unlink(out)
    print(f"[proxy] {size // 1000} KB", flush=True)
    return {"mp4_base64": data, "bytes": size, "source_codec": f"{codec or 'unknown'}{(' ' + pix) if pix else ''}"}


def process_video(video_path, emotions=True, vocal_emotion=True, screen_text=True, brand=None, cei=False, stills=False, proxy=False, memorability=False):
    with _ANALYSE:
        for key in _PER_REQUEST_ERRORS:
            MODEL_ERRORS.pop(key, None)
        if memorability_mod is not None:
            MODEL_ERRORS.pop("memorability", None)
        return _process_video(video_path, emotions, vocal_emotion, screen_text, brand, cei, stills, proxy, memorability)


def _process_video(video_path, emotions, vocal_emotion, screen_text, brand, cei, stills, proxy, memorability):
    audio_path = video_path + ".wav"

    audio_error = extract_audio(video_path, audio_path)
    if audio_error:
        return {"success": False, "error": f"Failed to extract audio. {audio_error}"}

    # See decodable_copy. Everything that reads frames reads frames_path; duration and audio come
    # from the upload itself.
    frames_path, frames_copy = decodable_copy(video_path)

    try:
        # From the container, read once. See below for why not from the transcript.
        container_duration = video_duration(video_path)

        # The passes run side by side rather than one after another. The sound (Whisper, then
        # emotion2vec per segment) needs nothing from the pictures; the text-and-faces read, the
        # Creative Effectiveness pass and memorability each read the film themselves. Only
        # memorability waits, for the transcript its words channel scores. Each pass is the same
        # code on the same input as before, so a film measures the same; it just takes the time of
        # the longest branch rather than the sum of all of them.
        def sound():
            print("[process] Transcribing...", flush=True)
            segs, lang = transcribe(audio_path, brand)
            voices = []
            if vocal_emotion and segs:
                print("[process] Analyzing voice emotions (emotion2vec)...", flush=True)
                voices = analyze_voice_segments(audio_path, segs)
            return segs, lang, voices

        def pictures():
            if screen_text:
                print("[process] Reading on-screen text...", flush=True)
            print(f"[process] Analyzing faces (emotions={emotions})...", flush=True)
            return scan_frames(frames_path, screen_text=screen_text, faces=True, face_fps=4, use_emotions=emotions)

        def effectiveness():
            # Creative Effectiveness Index inputs. A separate pass on purpose: it samples at 2fps
            # through DeepFace because the 138-video corpus the index compares against was scored
            # that way, and a measure taken with this service's 4fps HSEmotion pass would not sit
            # on the same scale as the thresholds.
            try:
                import cei_measures
                # DeepFace pulls in TensorFlow on first use. Imported under the load lock, as every
                # model is: two threads importing torch, transformers and TensorFlow at once is what
                # left "KeyError: 'torch.distributed.fsdp'" and dropped whole passes.
                with _LOAD:
                    cei_measures._deepface()
                print("[process] Measuring for the Creative Effectiveness Index...", flush=True)
                return cei_measures.measure(frames_path, audio_path)
            except Exception as e:
                import traceback
                print(f"[cei] failed: {e}", flush=True)
                print(traceback.format_exc(), flush=True)
                MODEL_ERRORS["cei"] = f"{type(e).__name__}: {e}"
                return None

        def recall(sound_future):
            # Memorability: predicted brand recall a day or more after viewing, from the pictures
            # and the cutting alone. A model fitted to LAMBDA, 2,183 real ads whose viewers were
            # asked 24 to 72 hours later which brands they remembered (see memorability.py). Its
            # own pass because it reads the film at 1fps through CLIP, which none of the others do.
            try:
                segs = sound_future.result()[0]
                print("[process] Reading memorability...", flush=True)
                models, predictor, text_model = get_memorability()
                out = memorability_mod.read_memorability(
                    frames_path, models, predictor, container_duration,
                    transcript=" ".join(seg.get("text", "") for seg in segs) if segs else "",
                    brand=(brand or "").split(",")[0].strip() or None, text_model=text_model)
                if out.get("error"):
                    MODEL_ERRORS["memorability"] = out["error"]
                return out
            except Exception as e:
                import traceback
                print(f"[memorability] failed: {e}", flush=True)
                print(traceback.format_exc(), flush=True)
                MODEL_ERRORS["memorability"] = f"{type(e).__name__}: {e}"
                return None

        def playable():
            # A playable copy for the browser; see make_proxy. Never fatal. Alongside the rest
            # because a ProRes master's re-encode is minutes on its own.
            try:
                return make_proxy(video_path)
            except Exception as e:
                print(f"[proxy] failed: {e}", flush=True)
                MODEL_ERRORS["proxy"] = f"{type(e).__name__}: {e}"
                return None

        with ThreadPoolExecutor(max_workers=5, thread_name_prefix="analyse") as pool:
            sound_f = pool.submit(sound)
            pictures_f = pool.submit(pictures)
            cei_f = pool.submit(effectiveness) if cei else None
            recall_f = pool.submit(recall, sound_f) if memorability else None
            proxy_f = pool.submit(playable) if proxy else None
            segments, language, voices = sound_f.result()
            text_timeline, face_timeline = pictures_f.result()
            cei_metrics = cei_f.result() if cei_f else None
            memorability_out = recall_f.result() if recall_f else None
            proxy_out = proxy_f.result() if proxy_f else None

        # A film with no speech used to return before any of the frame passes: a music-only
        # advert, which is most of the effectiveness corpus, came back as a 0-second film with
        # nothing measured and no error. Every pass tolerates an empty transcript, so a silent
        # film goes through like any other and its voice measures are simply absent.
        if not segments:
            print("[process] No speech; measured the pictures anyway", flush=True)

        # The emotional content of the copy, scored once per distinct line.
        copy_emotion = []
        if screen_text and text_timeline:
            seen, distinct = set(), []
            for f in text_timeline:
                t = (f.get("text") or "").strip()
                key = t.lower()
                if t and key not in seen:
                    seen.add(key)
                    distinct.append(t)
            print(f"[process] Scoring {len(distinct)} lines of on-screen copy...", flush=True)
            copy_emotion = score_lines(distinct)

        timeline = []
        for i, seg in enumerate(segments):
            entry = {
                "start_time": seg["start"],
                "end_time": seg["end"],
                "text": seg["text"],
            }
            # Voice emotion (emotion2vec)
            voice_emo = voices[i] if i < len(voices) else None
            if voice_emo:
                entry["voice_emotion"] = voice_emo

            # Find closest face frame
            seg_mid = (seg["start"] + seg["end"]) / 2
            closest = min(face_timeline, key=lambda f: abs(f["time"] - seg_mid), default=None)
            if closest and abs(closest["time"] - seg_mid) < 2:
                entry["faces_detected"] = closest["faces"]
                if closest["emotions"]:
                    entry["facial_emotions"] = closest["emotions"]
            timeline.append(entry)

        # From the container, not from the last speech segment. Deriving it from the transcript
        # made a 20-second advert report as 16 seconds whenever nobody spoke near the end, and
        # every percentage-of-runtime figure moved with it between runs on the same file.
        duration = container_duration or (max(t["end_time"] for t in timeline) if timeline else 0)
        transcript = " ".join(t["text"] for t in timeline)

        features_enabled = ["transcription"]
        if vocal_emotion:
            features_enabled.append("vocal_emotion")
        if emotions:
            features_enabled.append("facial_emotions")

        # Stills for the report's moment cards; see cut_stills. Never fatal: a film with no
        # pictures against its moments is still a measured film.
        stills_out = None
        if stills:
            try:
                print("[process] Cutting stills...", flush=True)
                stills_out = cut_stills(frames_path, duration)
                print(f"[process] {len(stills_out)} stills", flush=True)
            except Exception as e:
                print(f"[stills] failed: {e}", flush=True)
                MODEL_ERRORS["stills"] = f"{type(e).__name__}: {e}"

        return {
            "success": True,
            "timeline": timeline,
            # Facial emotion at the sampling rate, not collapsed onto speech segments. The timeline
            # above has one entry per thing that was said, so an advert with music and no voiceover
            # produced a single data point from sixty-odd measured frames — no curve, no peak, and
            # a report with nothing in it. The visual channel is the whole content of that advert.
            "face_timeline": face_timeline,
            "text_timeline": text_timeline,
            "screen_text_full": " ".join(f["text"] for f in text_timeline),
            "copy_emotion": copy_emotion,
            "cei": cei_metrics,
            "memorability": memorability_out,
            "stills": stills_out,
            "proxy": proxy_out,
            "brand_presence": brand_presence(text_timeline, transcript, duration, brand),
            "transcript_full": transcript,
            "duration": duration,
            "metadata": {
                "language": language,
                "segments": len(timeline),
                # What was asked for. `analysers` below is what actually ran — the two differ
                # whenever an optional model is missing, and only the second one is a fact.
                "features_enabled": features_enabled,
                "analysers": {
                    "transcription": {"ran": True},
                    "voice_emotion": {
                        "ran": vocal_emotion and "voice_emotion" not in MODEL_ERRORS,
                        "requested": bool(vocal_emotion),
                        "error": MODEL_ERRORS.get("voice_emotion"),
                    },
                    "face_emotion": {
                        "ran": emotions and "face_emotion" not in MODEL_ERRORS,
                        "requested": bool(emotions),
                        "error": MODEL_ERRORS.get("face_emotion"),
                    },
                    "screen_text": {
                        "ran": screen_text and "screen_text" not in MODEL_ERRORS,
                        "requested": bool(screen_text),
                        "error": MODEL_ERRORS.get("screen_text"),
                    },
                    "copy_emotion": {
                        "ran": bool(copy_emotion),
                        "requested": bool(screen_text),
                        "error": MODEL_ERRORS.get("copy_emotion"),
                    },
                    "cei": {
                        "ran": cei_metrics is not None,
                        "requested": bool(cei),
                        "error": MODEL_ERRORS.get("cei"),
                    },
                    "memorability": {
                        "ran": memorability_out is not None and "memorability" not in MODEL_ERRORS,
                        "requested": bool(memorability),
                        "error": MODEL_ERRORS.get("memorability"),
                    },
                },
                "degraded": sorted(MODEL_ERRORS.keys()),
            }
        }
    finally:
        if os.path.exists(audio_path):
            os.unlink(audio_path)
        if frames_copy and os.path.exists(frames_copy):
            os.unlink(frames_copy)

@app.get("/")
def root():
    return {"service": "EchoDepth Video Analysis", "version": "4.2-emotion2vec", "gpu": torch.cuda.is_available()}

@app.get("/health")
def health():
    return {"status": "ok", "gpu": torch.cuda.is_available(), "auth": "enforced" if service_auth.enforcing() else "off"}

@app.get("/models")
def models_status():
    """Check which models are loaded/available"""
    import traceback
    status = {
        "whisper": "not_loaded",
        "hsemotion": "not_loaded",
        "emotion2vec": "not_loaded",
        "errors": [],
        "gpu": torch.cuda.is_available(),
        "gpu_name": torch.cuda.get_device_name(0) if torch.cuda.is_available() else None
    }

    # Check whisper
    try:
        model = get_whisper()
        status["whisper"] = "loaded" if model else "failed"
    except Exception as e:
        status["whisper"] = "error"
        status["errors"].append(f"whisper: {str(e)}")

    # Check HSEmotion
    try:
        model = get_face_emotion()
        status["hsemotion"] = "loaded" if model else "failed"
        if not model:
            status["errors"].append("hsemotion: model returned None (check logs)")
    except Exception as e:
        status["hsemotion"] = "error"
        status["errors"].append(f"hsemotion: {str(e)}\n{traceback.format_exc()}")

    # Check emotion2vec
    try:
        model = get_emotion2vec()
        status["emotion2vec"] = "loaded" if model else "failed"
    except Exception as e:
        status["emotion2vec"] = "error"
        status["errors"].append(f"emotion2vec: {str(e)}")

    # Where each pass runs: the check that the GPU is being used, not just present
    face = face_emotion_model if face_emotion_model not in (None, "failed") else None
    status["devices"] = {
        "whisper": "cuda" if torch.cuda.is_available() else "cpu",
        "face_emotion": face.ort_session.get_providers()[0] if face else None,
        "screen_text": ("cuda" if torch.cuda.is_available() else "cpu") if ocr_reader not in (None, "failed") else None,
        "h264_encoder": "h264_nvenc" if nvenc_available() else "libx264",
        "cei_deepface": "cpu (deliberately; see cei_measures._deepface)",
    }
    return status

@app.post("/score-text")
def score_text(payload: dict):
    """Score arbitrary lines of copy on the same scale as the copy read off the film.

    Exists so a rewritten call to action is measured by the model that judged the original, rather
    than against a second opinion that cannot be compared with the first.
    """
    lines = payload.get("lines") or []
    if not isinstance(lines, list):
        return {"success": False, "error": "lines must be an array of strings"}
    scored = score_lines([str(l) for l in lines[:40]])
    return {"success": True, "scored": scored, "error": MODEL_ERRORS.get("copy_emotion")}


@app.post("/memorability-words")
def memorability_words(payload: dict):
    """
    Score different words against the same pictures. `channels` is the per-channel prediction the
    film's own read returned; only the words channel is recomputed from `transcript`. For scoring a
    rewritten script against the film as shot.
    """
    transcript = (payload.get("transcript") or "").strip()
    channels = payload.get("channels") or {}
    if not channels:
        return {"success": False, "error": "channels from the film's memorability read are required"}
    try:
        _, predictor, text_model = get_memorability()
        if text_model is None or "words" not in predictor.channels:
            return {"success": False, "error": "this model has no words channel"}
        return {"success": True, **memorability_mod.score_words(predictor, text_model, transcript, payload.get("brand"), channels)}
    except Exception as e:
        return {"success": False, "error": f"{type(e).__name__}: {e}"}

@app.post("/proxy")
def proxy_only(file: UploadFile = File(...)):
    """A playable 720p H.264 copy of an upload, and nothing else.

    For attaching a video to an analysis that was saved without one: the film was measured, the
    stored copy failed or was never made, and re-running eleven minutes of analysis to get a
    player back is the wrong price. Always transcodes, so the result is small and plays anywhere
    whatever came in.
    """
    print(f"[API] Proxy only: {file.filename}", flush=True)
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp:
        shutil.copyfileobj(file.file, tmp)
        video_path = tmp.name
    try:
        MODEL_ERRORS.pop("proxy", None)
        out = make_proxy(video_path, force=True)
        return {"success": out is not None, "proxy": out, "error": MODEL_ERRORS.get("proxy")}
    except Exception as e:
        return {"success": False, "proxy": None, "error": f"{type(e).__name__}: {e}"}
    finally:
        if os.path.exists(video_path):
            os.unlink(video_path)


# A plain def, like the other upload routes: FastAPI runs it on its thread pool. As an async def it
# ran the whole analysis on the event loop, so /health, the transcription and Stage Read polls and
# every other request waited until the film was done.
@app.post("/analyze")
def analyze(
    file: UploadFile = File(...),
    emotions: bool = Query(True, description="Enable facial emotion detection (HSEmotion)"),
    vocal_emotion: bool = Query(True, description="Enable voice emotion detection (emotion2vec)"),
    screen_text: bool = Query(True, description="Read on-screen copy (EasyOCR)"),
    brand: str = Query(None, description="Brand name to look for on screen and in speech"),
    cei: bool = Query(False, description="Measure for the Creative Effectiveness Index (DeepFace, 2fps)"),
    stills: bool = Query(False, description="Return one 320px JPEG per second, for reports on a master the browser cannot decode"),
    proxy: bool = Query(False, description="Return a 720p H.264 copy when the upload is not browser-playable (ProRes, HEVC)"),
    memorability: bool = Query(False, description="Predict brand recall a day later from the pictures (CLIP at 1fps, model fitted to LAMBDA)")
):
    print(f"[API] Received: {file.filename} (emotions={emotions}, vocal_emotion={vocal_emotion}, screen_text={screen_text}, brand={brand}, cei={cei}, stills={stills}, proxy={proxy}, memorability={memorability})", flush=True)
    with tempfile.NamedTemporaryFile(delete=False, suffix=".mp4") as tmp:
        shutil.copyfileobj(file.file, tmp)
        video_path = tmp.name
    try:
        return process_video(video_path, emotions=emotions, vocal_emotion=vocal_emotion, screen_text=screen_text, brand=brand, cei=cei, stills=stills, proxy=proxy, memorability=memorability)
    finally:
        if os.path.exists(video_path):
            os.unlink(video_path)

# ------------------------------------------------------------------------------------------------
# Transcription jobs: Voice Analysis, Practice and script audio.
#
# These used to go to AssemblyAI, which trains on customer audio unless opted out and keeps
# transcripts until they are deleted. They now run here, behind the app's signed pass: Whisper for
# the words, emotion2vec for the voice, GoEmotions for what the words mean.
#
# A job, not a request: on CPU hardware Whisper medium runs at around real time, so a few minutes
# of audio outlasts any sensible HTTP timeout. The caller posts the file and polls.
#
# RETENTION. The upload and its audio are deleted as soon as the job finishes, pass or fail. The
# result is held in memory for TRANSCRIBE_TTL_S so the caller can collect it, then dropped. Nothing
# is written anywhere else.
# ------------------------------------------------------------------------------------------------
import re as _re
import threading as _threading
import time as _time
import uuid as _uuid
from fastapi import HTTPException as _HTTPException

TRANSCRIBE_JOBS: dict = {}
TRANSCRIBE_TTL_S = 3600
_TRANSCRIBE_RUN = _threading.Lock()  # one at a time: the CPU is shared with /analyze
_JOB_ID = _re.compile(r"^[0-9a-f]{32}$")


def _text_sentiment(score):
    """AssemblyAI-style POSITIVE / NEGATIVE / NEUTRAL from GoEmotions, with a strength 0-1.

    Valence is the emotion-weighted pleasantness (0.5 is neither); feel is how much emotion the
    words carry at all. A line with little feeling is NEUTRAL whatever its lean.
    """
    if not score:
        return "NEUTRAL", 0.0
    valence, feel = score["valence"], score["feel"]
    lean = (valence - 0.5) * 2  # -1 .. 1
    if feel < 0.25 or abs(lean) < 0.15:
        return "NEUTRAL", round(1 - feel, 3)
    return ("POSITIVE" if lean > 0 else "NEGATIVE"), round(min(1.0, abs(lean) * 0.5 + feel * 0.5), 3)


def _sentences(audio_path):
    """Whisper's words regrouped into sentences, each with its own start and end.

    Whisper segments by pauses, so one segment can hold a compliment and a complaint, which then
    cancel out to NEUTRAL. The voice page charts sentiment sentence by sentence, as AssemblyAI
    returned it, so the words are split at sentence ends (or a long pause) instead.
    """
    model = get_whisper()
    # vad_filter skips the silent stretches of a recording rather than decoding them. Spoken audio
    # only: /analyze keeps it off, because a sung brand line is the word most likely to be lost and
    # the voice activity detector is built for speech.
    segments, info = model.transcribe(audio_path, beam_size=5, language="en", word_timestamps=True, vad_filter=True)
    out, words, confs = [], [], []

    def flush():
        if words:
            text = "".join(w.word for w in words).strip()
            if text:
                out.append({"start": words[0].start, "end": words[-1].end, "text": text,
                            "confidence": round(sum(confs) / len(confs), 4) if confs else None})
        words.clear(); confs.clear()

    for seg in segments:
        for w in (seg.words or []):
            if words and w.start - words[-1].end > 1.5:
                flush()
            words.append(w)
            confs.append(float(w.probability))
            if w.word.strip().endswith((".", "?", "!")):
                flush()
    flush()
    return out, info.language


def _sweep_transcribe_jobs():
    cutoff = _time.time() - TRANSCRIBE_TTL_S
    for job_id in [k for k, j in TRANSCRIBE_JOBS.items() if j.get("finished_at") and j["finished_at"] < cutoff]:
        TRANSCRIBE_JOBS.pop(job_id, None)


def _fast_segments(audio_path):
    """Whisper small, greedy, silence skipped, no word timings: the fast profile's words."""
    segments, info = get_whisper_fast().transcribe(
        audio_path, beam_size=1, language="en", vad_filter=True, condition_on_previous_text=False,
    )
    return [{"start": s.start, "end": s.end, "text": s.text.strip(), "confidence": None}
            for s in segments if s.text.strip()], info.language


def _run_transcription(job_id, src_path, vocal, fast=False):
    """fast: the Reflections profile. Whisper small and greedy, no per-sentence text sentiment,
    and one voice-emotion read over the whole answer instead of one per sentence."""
    job = TRANSCRIBE_JOBS[job_id]
    audio_path = src_path + ".wav"
    try:
        with _TRANSCRIBE_RUN:
            job["status"] = "running"
            err = extract_audio(src_path, audio_path)
            if err:
                raise RuntimeError(f"Could not read the audio: {err}")
            import soundfile as sf
            duration = float(sf.info(audio_path).duration)
            if fast:
                # The words are what the speaker waits for, so the job is done as soon as they are;
                # the voice read follows in the same job, and voice_pending says when it has landed.
                segments, language = _fast_segments(audio_path)
                out = [{
                    "start": round(seg["start"], 3), "end": round(seg["end"], 3), "text": seg["text"],
                    "sentiment": "NEUTRAL", "sentiment_confidence": 0.0, "asr_confidence": None,
                    "voice_emotion": None,
                } for seg in segments]
                result = {
                    "text": " ".join(s["text"] for s in out).strip(),
                    "language": language,
                    "duration": round(duration, 3),
                    "confidence": None,
                    "segments": out,
                    "engine": "whisper-small-fast+emotion2vec",
                    "voice_pending": bool(vocal and segments),
                }
                job.update(status="done", result=result)
                if result["voice_pending"]:
                    try:
                        voice = analyze_voice_emotion(audio_path, 0, duration)
                        for seg in out:
                            seg["voice_emotion"] = voice
                    finally:
                        result["voice_pending"] = False
                return
            segments, language = _sentences(audio_path)
            out = []
            for seg in segments:
                scored = score_lines([seg["text"]])
                sentiment, strength = _text_sentiment(scored[0] if scored else None)
                voice = analyze_voice_emotion(audio_path, seg["start"], seg["end"]) if vocal else None
                out.append({
                    "start": round(seg["start"], 3), "end": round(seg["end"], 3), "text": seg["text"],
                    "sentiment": sentiment, "sentiment_confidence": strength,
                    "asr_confidence": seg.get("confidence"),
                    "voice_emotion": voice,
                })
            confs = [s["asr_confidence"] for s in out if s["asr_confidence"] is not None]
            job.update(status="done", result={
                "text": " ".join(s["text"] for s in out).strip(),
                "language": language,
                "duration": round(duration, 3),
                "confidence": round(sum(confs) / len(confs), 4) if confs else None,
                "segments": out,
                "engine": "whisper-medium+emotion2vec+goemotions",
            })
    except Exception as e:
        print(f"[transcribe] job failed: {type(e).__name__}: {e}", flush=True)
        job.update(status="error", error=f"{e}")
    finally:
        job["finished_at"] = _time.time()
        for p in (src_path, audio_path):
            try:
                if os.path.exists(p):
                    os.unlink(p)
            except OSError:
                pass


def _warm_fast_profile():
    """Load the models as the Space starts, so the first job after a wake does not wait for them.

    The fast transcription profile first, because a speaker is waiting on it; then everything
    /analyze uses. The weights are in the image (prefetch_models.py), so this is loading, not
    downloading. In the background: the service answers meanwhile, and the load lock means a
    request that arrives first simply waits for the model it needs.
    """
    try:
        get_whisper_fast()
        get_emotion2vec()
        print("[whisper] fast profile ready", flush=True)
    except Exception as e:
        print(f"[whisper] fast profile warm-up failed: {e}", flush=True)
    for name, load in (("whisper medium", get_whisper), ("HSEmotion", get_face_emotion), ("EasyOCR", get_ocr),
                       ("GoEmotions", get_text_emotion), ("memorability", get_memorability)):
        try:
            load()
            print(f"[warm] {name} ready", flush=True)
        except Exception as e:
            print(f"[warm] {name} failed: {e}", flush=True)
    try:
        import cei_measures
        with _LOAD:
            cei_measures.warm()
        print("[warm] DeepFace ready", flush=True)
    except Exception as e:
        print(f"[warm] DeepFace failed: {e}", flush=True)
    nvenc_available()


if os.environ.get("ECHODEPTH_NO_WARMUP") != "1":
    _threading.Thread(target=_warm_fast_profile, daemon=True).start()


@app.post("/transcribe")
def transcribe_submit(file: UploadFile = File(...), vocal_emotion: bool = Query(True),
                            profile: str = Query("full", description="full, or fast for short answers the speaker corrects")):
    """Start a transcription job. Returns {job_id}; poll GET /transcribe/{job_id}."""
    _sweep_transcribe_jobs()
    suffix = os.path.splitext(file.filename or "")[1] or ".bin"
    with tempfile.NamedTemporaryFile(delete=False, suffix=suffix) as tmp:
        shutil.copyfileobj(file.file, tmp)
        src_path = tmp.name
    job_id = _uuid.uuid4().hex
    TRANSCRIBE_JOBS[job_id] = {"status": "queued", "created_at": _time.time()}
    _threading.Thread(target=_run_transcription, args=(job_id, src_path, vocal_emotion, profile == "fast"), daemon=True).start()
    return {"job_id": job_id, "status": "queued"}


@app.get("/transcribe/{job_id}")
def transcribe_status(job_id: str):
    if not _JOB_ID.match(job_id or ""):
        raise _HTTPException(400, "bad job id")
    job = TRANSCRIBE_JOBS.get(job_id)
    if not job:
        raise _HTTPException(404, "unknown or expired job")
    out = {"status": job["status"]}
    if job["status"] == "done":
        out["result"] = job["result"]
    if job["status"] == "error":
        out["error"] = job.get("error")
    return out


# Stage Read (multi-speaker stage measurement) — see docs/stage-read-spec.md. Optional: the service
# still starts if its extra dependencies are missing.
try:
    from stage_read.api import router as stage_read_router
    app.include_router(stage_read_router)
    print("[STARTUP] Stage Read router mounted", flush=True)
except Exception as _e:
    print(f"[STARTUP] Stage Read unavailable: {_e}", flush=True)

if __name__ == "__main__":
    import uvicorn
    uvicorn.run(app, host="0.0.0.0", port=7860)