"""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