ai_analysis.py

"""Gemini steps: listen (per chunk), analyse (per file), synthesise (across files).

Plus the local checks that keep the AI output honest:
* timestamps are parsed, re-based to the full recording, clamped and, if the
  model overshoots the clip length, rescaled;
* every quote the model cites is looked up in the transcript and marked
  exact / close / not found, and its time is taken from the matching line;
* speech metrics (words per minute, talk share, turns, fillers) are computed
  from the transcript, not asked of the model.
"""

from __future__ import annotations

import copy
import difflib
import json
import re
import threading
from collections import OrderedDict
from dataclasses import asdict, dataclass, field
from typing import Callable

import numpy as np

import config
from audio_io import encode_opus
from gemini_service import Gemini, GeminiCallError, audio_part, text_part
from schemas import ChunkListening, CrossSynthesis, FileAnalysis

# --- time helpers ------------------------------------------------------------


def fmt_time(seconds: float | None) -> str:
    if seconds is None or not np.isfinite(seconds):
        return "--:--"
    s = int(round(max(0.0, seconds)))
    h, rem = divmod(s, 3600)
    m, sec = divmod(rem, 60)
    return f"{h}:{m:02d}:{sec:02d}" if h else f"{m:02d}:{sec:02d}"


_TS = re.compile(r"^\s*\[?\s*(?:(\d+):)?(\d{1,3}):(\d{1,2}(?:[.,]\d+)?)\s*\]?\s*$")
_TS_WORDS = re.compile(r"^\s*(?:(\d+)\s*h)?\s*(?:(\d+)\s*m(?:in)?)?\s*(?:(\d+(?:\.\d+)?)\s*s(?:ec)?)?\s*$", re.I)


def parse_time(value: str | float | int | None) -> float | None:
    """'04:31', '1:02:10', '4:31.5', '271', '4m31s' -> seconds."""
    if value is None:
        return None
    if isinstance(value, (int, float)):
        return float(value) if np.isfinite(value) and value >= 0 else None
    text = str(value).strip()
    if not text:
        return None
    m = _TS.match(text)
    if m:
        h, mnt, sec = m.groups()
        return int(h or 0) * 3600 + int(mnt) * 60 + float(sec.replace(",", "."))
    try:
        v = float(text)
        return v if v >= 0 else None
    except ValueError:
        pass
    m = _TS_WORDS.match(text)
    if m and any(m.groups()):
        h, mnt, sec = m.groups()
        return int(h or 0) * 3600 + int(mnt or 0) * 60 + float(sec or 0)
    return None


# --- data containers -----------------------------------------------------------


@dataclass
class TranscriptLine:
    start: float
    end: float
    speaker: str
    text: str
    chunk: int


@dataclass
class Listening:
    lines: list[TranscriptLine] = field(default_factory=list)
    speakers: dict = field(default_factory=dict)          # label -> description
    language: str = ""
    has_speech: bool = False
    sound_events: list = field(default_factory=list)      # [[t, note]]
    vocal_tone: list = field(default_factory=list)        # [[t, note]]
    descriptions: list = field(default_factory=list)      # [[start, end, text]]
    quality_notes: list = field(default_factory=list)     # [[start, end, text]]
    chunk_errors: list = field(default_factory=list)      # [str]
    notes: list = field(default_factory=list)             # timestamp repairs etc.
    chunks: list = field(default_factory=list)            # [[start, end]]

    def to_dict(self) -> dict:
        return asdict(self)

    def transcript_text(self, prefix: str = "") -> str:
        return "\n".join(f"[{prefix}{fmt_time(x.start)}] {x.speaker}: {x.text}" for x in self.lines)


_CACHE: "OrderedDict[tuple, Listening]" = OrderedDict()
_CACHE_LOCK = threading.Lock()
_CACHE_MAX = 64


# --- 1. listening --------------------------------------------------------------

LISTEN_SYSTEM = (
    "You are a meticulous audio analyst and professional transcriptionist. You receive one clip "
    "of a longer recording and report exactly what is audible in it. Never invent or complete "
    "words you cannot hear; write [inaudible] instead. Keep speech in its original language (do "
    "not translate). Every time you give is measured from the start of THIS clip (00:00 is its "
    "first moment) and can never exceed the clip length."
)


def _listen_prompt(name: str, idx: int, total: int, start: float, end: float,
                   roster: dict, context: str) -> str:
    parts = [
        f'Recording "{name}" - clip {idx + 1} of {total}, covering {fmt_time(start)} to '
        f"{fmt_time(end)} of the full recording. This clip is {fmt_time(end - start)} long "
        f"({end - start:.0f} seconds).",
    ]
    if context.strip():
        parts.append("Background from the user (may help with names and technical terms):\n" + context.strip()[:2000])
    if roster:
        lines = "\n".join(f"- {k}: {v}" for k, v in list(roster.items())[:12])
        parts.append(
            "Speakers already identified in earlier clips. Reuse exactly these labels for the same "
            "voices, and add new labels only for new voices:\n" + lines
        )
    parts.append(
        "Tasks:\n"
        "1. Transcribe all speech verbatim, keeping fillers (um, uh), false starts and repetitions. "
        "Start a new segment whenever the speaker changes and split long turns into segments of at "
        "most about 30 seconds.\n"
        "2. Label speakers consistently. Use a role label when the role is clear (Interviewer, "
        "Participant, Host, Guest, Caller, Teacher, Student, Presenter...), otherwise Speaker 1, "
        "Speaker 2... Do not use people's real names as labels.\n"
        "3. List non-speech sounds with their times (laughter, music, applause, phones, background "
        "noise, long silences, interruptions).\n"
        "4. List moments where the WAY someone speaks changes (hesitant, emotional, animated, "
        "frustrated, laughing, whispering...).\n"
        "5. Describe the clip overall: setting and recording environment; for music give genre, "
        "instruments, mood and tempo feel.\n"
        "6. Comment briefly on the audio quality (noise, echo, distortion, level, overlapping speech).\n"
        "If there is no speech, set has_speech to false and return an empty segments list."
    )
    return "\n\n".join(parts)


def _normalise_times(values: list[float | None], clip_len: float, clip_start: float) -> tuple[list[float | None], str]:
    """Map model times onto the full recording.

    Handles times given relative to the full recording instead of the clip,
    and a model clock that runs past the clip end (rescaled to fit)."""
    known = [v for v in values if v is not None]
    note = ""
    if not known:
        return values, note
    hi = max(known)
    if clip_start > 0 and min(known) >= clip_start - 5 and hi > clip_len + 5 and hi <= clip_start + clip_len + 30:
        values = [None if v is None else v - clip_start for v in values]
        known = [v for v in values if v is not None]
        hi = max(known)
        note = "times were given relative to the full recording and were re-based"
    if hi > clip_len * 1.02 + 2:
        scale = clip_len / hi
        values = [None if v is None else v * scale for v in values]
        note = (note + "; " if note else "") + f"times ran {hi - clip_len:.0f} s past the clip end and were rescaled by {scale:.3f}"
    out = [None if v is None else clip_start + min(max(v, 0.0), clip_len) for v in values]
    return out, note


def listen(gem: Gemini, name: str, sha256: str, samples: np.ndarray, chunks: list[tuple[float, float]],
           context: str = "", progress: Callable[[str, int], None] | None = None) -> Listening:
    """Transcribe and describe the recording clip by clip.

    progress(message, chunks_completed_by_this_call) is called before (0) and
    after (1) each clip."""
    key = (sha256, tuple(chunks), tuple(gem.models), context.strip())
    with _CACHE_LOCK:
        if key in _CACHE:
            _CACHE.move_to_end(key)
            return copy.deepcopy(_CACHE[key])

    result = Listening(chunks=[list(c) for c in chunks])
    roster: dict[str, str] = {}
    languages: list[str] = []
    sr = config.ANALYSIS_SR
    for i, (a, b) in enumerate(chunks):
        if progress:
            progress(f"listening to {fmt_time(a)}-{fmt_time(b)} (part {i + 1}/{len(chunks)})", 0)
        audio = encode_opus(samples[int(a * sr): int(b * sr)])
        prompt = _listen_prompt(name, i, len(chunks), a, b, roster, context)
        try:
            r = gem.generate([audio_part(audio), text_part(prompt)], ChunkListening, LISTEN_SYSTEM,
                             label=f"{name} {fmt_time(a)}-{fmt_time(b)}")
        except GeminiCallError as exc:
            result.chunk_errors.append(f"{fmt_time(a)}-{fmt_time(b)}: {exc}")
            if progress:
                progress(f"part {i + 1}/{len(chunks)} failed", 1)
            continue
        if progress:
            progress(f"heard {fmt_time(a)}-{fmt_time(b)} (part {i + 1}/{len(chunks)})", 1)

        clip_len = b - a
        seg_times = []
        for s in r.segments:
            seg_times += [parse_time(s.start), parse_time(s.end)]
        ev_times = [parse_time(e.time) for e in r.sound_events]
        tone_times = [parse_time(e.time) for e in r.vocal_tone]
        # One mapping for the whole clip so segments and notes stay consistent.
        mapped, note = _normalise_times(seg_times + ev_times + tone_times, clip_len, a)
        if note:
            result.notes.append(f"{fmt_time(a)}-{fmt_time(b)}: {note}")
        seg_mapped = mapped[: len(seg_times)]
        ev_mapped = mapped[len(seg_times): len(seg_times) + len(ev_times)]
        tone_mapped = mapped[len(seg_times) + len(ev_times):]

        prev_end = a
        for j, s in enumerate(r.segments):
            text = " ".join(s.text.split())
            if not text:
                continue
            st, en = seg_mapped[2 * j], seg_mapped[2 * j + 1]
            st = prev_end if st is None else max(st, a)
            en = st if en is None else en
            en = max(en, st)
            result.lines.append(TranscriptLine(round(st, 2), round(min(en, b), 2), s.speaker.strip() or "Speaker", text, i))
            prev_end = en
        for t, e in zip(ev_mapped, r.sound_events):
            if e.note.strip():
                result.sound_events.append([round(t if t is not None else a, 2), e.note.strip()])
        for t, e in zip(tone_mapped, r.vocal_tone):
            if e.note.strip():
                result.vocal_tone.append([round(t if t is not None else a, 2), e.note.strip()])
        for sp in r.speakers:
            if sp.label.strip() and sp.label.strip() not in roster:
                roster[sp.label.strip()] = sp.description.strip()
        if r.has_speech:
            result.has_speech = True
        if r.language and r.language.lower() not in ("none", "n/a", "unknown", ""):
            languages.append(r.language.strip())
        if r.audio_description.strip():
            result.descriptions.append([a, b, r.audio_description.strip()])
        if r.audio_quality_notes.strip():
            result.quality_notes.append([a, b, r.audio_quality_notes.strip()])

    result.lines.sort(key=lambda x: (x.start, x.chunk))
    used = {x.speaker for x in result.lines}
    result.speakers = {k: v for k, v in roster.items() if k in used or not used}
    for spk in used - set(result.speakers):
        result.speakers[spk] = ""
    if languages:
        result.language = language_name(max(set(languages), key=languages.count))
    result.has_speech = result.has_speech or bool(result.lines)
    if len(result.chunk_errors) < len(chunks):          # cache only useful results
        with _CACHE_LOCK:
            _CACHE[key] = copy.deepcopy(result)
            while len(_CACHE) > _CACHE_MAX:
                _CACHE.popitem(last=False)
    return result


_LANGS = {"en": "English", "es": "Spanish", "fr": "French", "de": "German", "it": "Italian", "pt": "Portuguese",
          "nl": "Dutch", "hi": "Hindi", "zh": "Chinese", "ja": "Japanese", "ko": "Korean", "ar": "Arabic",
          "ru": "Russian", "tr": "Turkish", "pl": "Polish", "sv": "Swedish", "uk": "Ukrainian", "vi": "Vietnamese"}


def language_name(code: str) -> str:
    c = code.strip()
    base = c.lower().replace("_", "-").split("-")[0]
    if len(base) == 2 and base in _LANGS:
        return _LANGS[base]
    return c[:1].upper() + c[1:]


# --- speech metrics -----------------------------------------------------------------

_WORD = re.compile(r"[^\W_]+(?:['’][^\W_]+)*", re.UNICODE)
_TAG = re.compile(r"\[[^\]]*\]")
FILLERS = {"um", "uh", "erm", "er", "uhm", "umm", "hmm", "mm", "mhm", "ah", "eh"}


def words(text: str) -> list[str]:
    return [w.lower() for w in _WORD.findall(_TAG.sub(" ", text))]


def speech_metrics(lst: Listening, duration: float) -> dict:
    lines = lst.lines
    if not lines:
        return {}
    all_words: list[str] = []
    per: dict[str, dict] = {}
    turns = 0
    prev = None
    longest_turn = 0.0
    run_start = run_end = 0.0
    for ln in lines:
        w = words(ln.text)
        all_words += w
        d = per.setdefault(ln.speaker, {"words": 0, "seconds": 0.0, "turns": 0, "questions": 0})
        d["words"] += len(w)
        d["seconds"] += max(0.0, ln.end - ln.start)
        d["questions"] += ln.text.count("?")
        if ln.speaker != prev:
            turns += 1
            d["turns"] += 1
            longest_turn = max(longest_turn, run_end - run_start)
            run_start, run_end = ln.start, ln.end
            prev = ln.speaker
        else:
            run_end = max(run_end, ln.end)
    longest_turn = max(longest_turn, run_end - run_start)
    total_words = len(all_words)
    speech_s = sum(d["seconds"] for d in per.values())
    fillers = sum(1 for w in all_words if w in FILLERS)
    text_lower = " ".join(all_words)
    fillers += len(re.findall(r"\byou know\b", text_lower))
    total_s = max(speech_s, 1e-9)
    for d in per.values():
        d["time_share"] = round(d["seconds"] / total_s, 4)
        d["word_share"] = round(d["words"] / max(total_words, 1), 4)
        d["wpm"] = round(d["words"] / (d["seconds"] / 60), 1) if d["seconds"] >= 20 else None
        d["seconds"] = round(d["seconds"], 1)
    return {
        "words": total_words,
        "speech_seconds": round(speech_s, 1),
        "speech_coverage": round(min(1.0, speech_s / duration), 4) if duration else None,
        "words_per_minute": round(total_words / (speech_s / 60), 1) if speech_s >= 20 else None,
        "speakers": len(per),
        "turns": turns,
        "turns_per_minute": round(turns / (duration / 60), 2) if duration else None,
        "longest_turn_s": round(longest_turn, 1),
        "questions": sum(d["questions"] for d in per.values()),
        "fillers_per_100_words": round(100 * fillers / total_words, 2) if total_words else None,
        "lexical_diversity": _mattr(all_words),
        "dominant_speaker_share": round(max(d["time_share"] for d in per.values()), 4),
        "per_speaker": per,
    }


def _mattr(ws: list[str], window: int = 50) -> float | None:
    """Moving-average type-token ratio: vocabulary variety independent of length."""
    if len(ws) < window:
        return round(len(set(ws)) / len(ws), 4) if len(ws) >= 10 else None
    counts: dict[str, int] = {}
    for w in ws[:window]:
        counts[w] = counts.get(w, 0) + 1
    ratios = [len(counts) / window]
    for i in range(window, len(ws)):
        old = ws[i - window]
        counts[old] -= 1
        if counts[old] == 0:
            del counts[old]
        counts[ws[i]] = counts.get(ws[i], 0) + 1
        ratios.append(len(counts) / window)
    return round(float(np.mean(ratios)), 4)


# --- quote verification ----------------------------------------------------------------

class QuoteIndex:
    """Find where a quoted passage occurs in a transcript."""

    def __init__(self, lines: list[TranscriptLine]) -> None:
        self.lines = lines
        self.tokens: list[str] = []
        self.owner: list[int] = []
        for i, ln in enumerate(lines):
            w = words(ln.text)
            self.tokens += w
            self.owner += [i] * len(w)
        self._joined = " " + " ".join(self.tokens) + " "
        # char offset of each token in _joined
        self._offsets = []
        pos = 1
        for t in self.tokens:
            self._offsets.append(pos)
            pos += len(t) + 1
        self._first: dict[str, list[int]] = {}
        for i, t in enumerate(self.tokens):
            self._first.setdefault(t, []).append(i)

    def _exact(self, q: list[str]) -> int | None:
        if not q:
            return None
        needle = " " + " ".join(q) + " "
        pos = self._joined.find(needle)
        if pos < 0:
            return None
        return int(np.searchsorted(self._offsets, pos + 1))

    def _fuzzy(self, q: list[str]) -> tuple[int | None, float]:
        n = len(q)
        if n < 4 or not self.tokens:
            return None, 0.0
        anchors: set[int] = set()
        for k in range(min(n, 6)):             # any of the first few words can anchor
            for pos in self._first.get(q[k], [])[:400]:
                anchors.add(max(0, pos - k))
        best, best_ratio = None, 0.0
        for s in anchors:
            window = self.tokens[s: s + n + max(2, n // 5)]
            blocks = [b for b in difflib.SequenceMatcher(None, q, window, autojunk=False).get_matching_blocks() if b.size]
            if not blocks:
                continue
            matched = sum(b.size for b in blocks)
            span = blocks[-1].b + blocks[-1].size - blocks[0].b      # aligned part of the transcript
            ratio = 2 * matched / (n + span)                          # Dice over the aligned span
            if ratio > best_ratio:
                best, best_ratio = s + blocks[0].b, ratio
        return best, best_ratio

    def find(self, quote: str) -> tuple[str, int | None]:
        """('exact' | 'close' | 'not found', line index)."""
        pieces = [words(p) for p in re.split(r"\.\.\.|…|\[\s*\.\.\.\s*\]", quote)]
        pieces = [p for p in pieces if p]
        if not pieces:
            return "not found", None
        first_line = None
        status = "exact"
        for p in pieces:
            if len(p) < 2 and len(pieces) > 1:
                continue
            at = self._exact(p)
            if at is None:
                at, ratio = self._fuzzy(p)
                if at is None or ratio < 0.8:
                    return "not found", None
                status = "close"
            if first_line is None and at is not None and at < len(self.owner):
                first_line = self.owner[at]
        return status, first_line


def _check(evidence: list[dict], index: QuoteIndex, text_key: str = "quote") -> None:
    for ev in evidence:
        status, line = index.find(ev.get(text_key, ""))
        ev["verified"] = status
        ev["model_time"] = ev.get("time", "")
        if line is not None:
            ln = index.lines[line]
            ev["time_s"] = ln.start
            ev["time"] = fmt_time(ln.start)
            if not ev.get("speaker"):
                ev["speaker"] = ln.speaker
        else:
            ev["time_s"] = parse_time(ev.get("time"))


def verify_file_analysis(analysis: dict, lst: Listening) -> dict:
    index = QuoteIndex(lst.lines)
    for theme in analysis.get("themes", []):
        _check(theme.get("evidence", []), index)
    _check(analysis.get("notable_quotes", []), index)
    for qa in analysis.get("research_answers", []):
        _check(qa.get("evidence", []), index)
    for tm in analysis.get("tone_moments", []):
        tm["time_s"] = parse_time(tm.get("time"))
    counts = {"exact": 0, "close": 0, "not found": 0}
    for ev in _all_evidence(analysis):
        counts[ev["verified"]] = counts.get(ev["verified"], 0) + 1
    analysis["quote_check"] = counts
    return analysis


def _all_evidence(analysis: dict) -> list[dict]:
    out = []
    for theme in analysis.get("themes", []):
        out += theme.get("evidence", [])
    out += analysis.get("notable_quotes", [])
    for qa in analysis.get("research_answers", []):
        out += qa.get("evidence", [])
    return out


# --- 2. per-file analysis -------------------------------------------------------------

ANALYSIS_SYSTEM = (
    "You are an expert analyst of recorded audio (interviews, meetings, lectures, calls, podcasts, "
    "music and soundscapes) and an experienced qualitative researcher. You receive a time-stamped "
    "transcript made from the recording, listening notes and measured acoustic metrics. Work only "
    "from this material. Quotes must be copied exactly from the transcript: you may shorten a quote "
    "with '...' but never reword it, and you give the timestamp of the transcript line it comes "
    "from. Write the analysis in English; keep quotes in their original language. Be specific and "
    "evidence-based and say when something is uncertain. Do not guess speakers' identity, health, "
    "ethnicity or other sensitive traits beyond what they say about themselves."
)


def _acoustic_brief(info, ac, flags) -> str:
    bits = [
        f"duration {fmt_time(ac.duration_s)}",
        f"loudness {ac.integrated_lufs:.1f} LUFS" if ac.integrated_lufs is not None else "loudness n/a",
        f"estimated SNR {ac.snr_db:.0f} dB" if not ac.steady_signal else "steady signal (no clear pauses)",
        f"sound activity {ac.activity_ratio * 100:.0f}%",
        f"{ac.pause_count} pauses of 0.5 s or more ({ac.pauses_per_min:.1f}/min, longest {ac.longest_pause_s:.1f} s)",
    ]
    if ac.pitch_median_hz:
        bits.append(f"median voice pitch {ac.pitch_median_hz:.0f} Hz (middle half {ac.pitch_p25_hz:.0f}-{ac.pitch_p75_hz:.0f} Hz)")
    bits.append(f"spectral centroid {ac.spectral_centroid_hz:.0f} Hz, effective bandwidth {ac.bandwidth_hz / 1000:.1f} kHz")
    if ac.tempo_bpm:
        bits.append(f"tempo estimate {ac.tempo_bpm:.0f} BPM (only meaningful for music)")
    bits.append("quality checks: " + " ".join(m for _, m in flags))
    return "; ".join(bits)


def _speech_brief(sm: dict) -> str:
    if not sm:
        return "no transcribed speech"
    shares = ", ".join(f"{k} {v['time_share'] * 100:.0f}%" for k, v in sorted(sm["per_speaker"].items(), key=lambda kv: -kv[1]["time_share"]))
    wpm = f"{sm['words_per_minute']:.0f} words/min" if sm.get("words_per_minute") else "speaking rate n/a"
    return (f"{sm['words']} words, {wpm}, {sm['speakers']} speaker label(s), {sm['turns']} turns, "
            f"{sm['questions']} question marks, fillers {sm.get('fillers_per_100_words')} per 100 words; "
            f"talk-time share: {shares}")


def _questions_block(questions: list[str]) -> str:
    if not questions:
        return "The user asked no specific questions: return an empty research_answers list."
    q = "\n".join(f"{i + 1}. {x}" for i, x in enumerate(questions))
    return ("Answer each of the user's questions for THIS recording in research_answers (same order, "
            "say 'Not discussed in this recording' when it does not apply):\n" + q)


def analyse_file(gem: Gemini, file_id: str, info, ac, flags, lst: Listening, sm: dict,
                 context: str, questions: list[str], models: list[str] | None = None) -> dict:
    notes = []
    for a, b, text in lst.descriptions:
        notes.append(f"- {fmt_time(a)}-{fmt_time(b)}: {text}")
    for a, b, text in lst.quality_notes:
        notes.append(f"- quality {fmt_time(a)}-{fmt_time(b)}: {text}")
    events = "\n".join(f"- [{fmt_time(t)}] {n}" for t, n in lst.sound_events[:200]) or "- none noted"
    tones = "\n".join(f"- [{fmt_time(t)}] {n}" for t, n in lst.vocal_tone[:200]) or "- none noted"
    speakers = "\n".join(f"- {k}: {v}" for k, v in lst.speakers.items()) or "- none"
    missing = ""
    if lst.chunk_errors:
        missing = ("\nParts of the recording could not be transcribed, so do not draw conclusions "
                   "about them:\n" + "\n".join(f"- {e}" for e in lst.chunk_errors))
    recorded = f"{info.recorded_at.replace('T', ' ')} (from {info.recorded_at_source})" if info.recorded_at else "unknown"
    prompt = f"""File {file_id}: "{info.name}"
Recorded: {recorded}
Language detected: {lst.language or 'n/a'}

User context: {context.strip()[:3000] or 'none given'}

{_questions_block(questions)}

Measured acoustics (signal processing, not AI): {_acoustic_brief(info, ac, flags)}
Speech metrics computed from the transcript: {_speech_brief(sm)}

Speakers heard:
{speakers}

Listening notes:
{chr(10).join(notes) or '- none'}
Sound events:
{events}
Vocal tone changes:
{tones}{missing}

Transcript (times are from the start of the recording):
{lst.transcript_text() or '(no speech was transcribed)'}
"""
    r = gem.generate([text_part(prompt)], FileAnalysis, ANALYSIS_SYSTEM, models=models, label=f"{info.name} analysis")
    return verify_file_analysis(r.model_dump(), lst)


# --- 3. cross-file synthesis ---------------------------------------------------------------

SYNTH_SYSTEM = (
    "You are a senior research analyst comparing several audio recordings to find the patterns "
    "across them. Base every statement on the material provided: per-file analyses, transcripts and "
    "measured metrics. Cite evidence as file id, transcript timestamp and an exact quote from that "
    "file's transcript (shorten with '...' only, never reword). Separate patterns that hold in all "
    "files from those found in some. With few recordings, say that the patterns are tentative and "
    "do not overgeneralise. Describe trends over time only when recording dates are given. Write in "
    "English and keep quotes in their original language."
)

MAX_SYNTH_TRANSCRIPT_CHARS = 600_000   # ~150k tokens of transcripts; above that, analyses only


def synthesise(gem: Gemini, files: list[dict], numeric_brief: str, context: str, questions: list[str],
               models: list[str] | None = None) -> dict:
    """files: [{id, name, recorded_at, duration, analysis, listening}] (successful files only)."""
    roster = []
    for f in files:
        a = f["analysis"] or {}
        roster.append(f"{f['id']} - \"{f['name']}\" | recorded {f.get('recorded_at') or 'unknown'} | "
                      f"{fmt_time(f['duration'])} | {a.get('content_type', 'n/a')} | {a.get('title', '')}")
    compact = []
    for f in files:
        a = copy.deepcopy(f["analysis"] or {})
        a.pop("quote_check", None)
        for ev in _all_evidence(a):
            for k in ("model_time", "time_s"):
                ev.pop(k, None)
            if ev.pop("verified", "") == "not found":
                ev["quote"] = "(unverified) " + ev.get("quote", "")
        compact.append({"file_id": f["id"], **{k: a.get(k) for k in (
            "title", "content_type", "summary", "speakers", "themes", "key_points", "questions_raised",
            "action_items", "sentiment", "keywords", "music", "research_answers")}})
    transcripts = ""
    total = sum(len(f["listening"].transcript_text()) for f in files if f.get("listening"))
    if 0 < total <= MAX_SYNTH_TRANSCRIPT_CHARS:
        blocks = [f"=== {f['id']} transcript ===\n{f['listening'].transcript_text()}" for f in files
                  if f.get("listening") and f["listening"].lines]
        transcripts = "\n\nFull transcripts (for finding and quoting cross-file patterns):\n" + "\n\n".join(blocks)
    else:
        transcripts = ("\n\n(Transcripts are too long to include; quote only from the evidence in the "
                       "per-file analyses.)" if total else "")
    ids = ", ".join(f["id"] for f in files)
    q_block = ("Answer each user question across ALL recordings in research_answers (same order):\n"
               + "\n".join(f"{i + 1}. {q}" for i, q in enumerate(questions))) if questions else \
        "The user asked no specific questions: return an empty research_answers list."
    prompt = f"""There are {len(files)} recordings (file ids {ids}):
{chr(10).join(roster)}

User context: {context.strip()[:3000] or 'none given'}

{q_block}

Measured comparison (signal processing and transcript statistics, not AI):
{numeric_brief}

Per-file analyses (JSON):
{json.dumps(compact, ensure_ascii=False, indent=1)}{transcripts}

Instructions:
- common_themes: themes present in two or more recordings. For each, 'presence' must contain one entry for every file id ({ids}).
- contrasts: meaningful differences in content, opinion, tone or setting.
- unique_points: one entry per file id with what only that recording contributes.
- trends_over_time: only if at least three recordings have known dates and there is a consistent change across them; otherwise return an empty list (describe two-file differences under contrasts).
- acoustic_patterns: interpret the measured comparison (level, noise, pauses, pitch, speaking rate...) and quote the numbers.
- outliers: recordings that differ strongly from the rest, with the reason.
- recommendations: concrete next steps or follow-up questions suggested by the patterns.
"""
    r = gem.generate([text_part(prompt)], CrossSynthesis, SYNTH_SYSTEM, models=models, label="cross-file synthesis")
    out = r.model_dump()
    valid = {f["id"] for f in files}
    if sum(1 for f in files if f.get("recorded_at")) < 3:
        out["trends_over_time"] = []        # two dated points are a contrast, not a trend
    by_id = {f["id"]: QuoteIndex(f["listening"].lines) for f in files if f.get("listening")}

    def check(evs: list[dict]) -> None:
        for ev in evs:
            fid = str(ev.get("file_id", "")).strip().upper()
            ev["file_id"] = fid
            if fid not in by_id:
                ev["verified"] = "not found"
                ev["time_s"] = parse_time(ev.get("time"))
                continue
            _check([ev], by_id[fid])

    for th in out["common_themes"]:
        seen = {}
        for p in th["presence"]:
            fid = str(p.get("file_id", "")).strip().upper()
            if fid in valid:
                seen[fid] = p["strength"]
        th["presence"] = [{"file_id": fid, "strength": seen.get(fid, "not assessed")} for fid in sorted(valid, key=_id_key)]
        check(th["evidence"])
    for qa in out["research_answers"]:
        check(qa["evidence"])
    for group in ("contrasts", "trends_over_time"):
        for item in out[group]:
            item["file_ids"] = [x.strip().upper() for x in item.get("file_ids", []) if x.strip().upper() in valid]
    out["unique_points"] = [u for u in out["unique_points"] if str(u.get("file_id", "")).strip().upper() in valid]
    for u in out["unique_points"]:
        u["file_id"] = u["file_id"].strip().upper()
    out["outliers"] = [o for o in out["outliers"] if str(o.get("file_id", "")).strip().upper() in valid]
    return out


def _id_key(fid: str) -> int:
    try:
        return int(fid.lstrip("F"))
    except ValueError:
        return 10 ** 6