"""
def _summary(run: dict) -> str:
ok = [r for r in run["files"] if not r.get("error")]
syn = run.get("synthesis")
out = ['
Summary
']
ctx = run["settings"].get("context", "").strip()
if ctx:
out.append(f'
Context given: {esc(ctx)}
')
if syn:
out.append(f'
{ai_tag()}
{esc(syn.get("overview", ""))}
')
elif len(ok) == 1 and ok[0].get("analysis"):
out.append(f'
{ai_tag()}
{esc(ok[0]["analysis"].get("summary", ""))}
')
elif ok:
out.append('
AI analysis was not available, so this report contains the measured '
'audio analysis only (loudness, noise, pauses, pitch, spectrum, quality checks and '
'cross-file acoustic comparison).
')
problems = []
for r in run["files"]:
if r.get("error"):
problems.append(("critical", f"{r['id']} {r['name']}: could not be analysed - {r['error']}"))
elif r.get("ai_error"):
problems.append(("warning", f"{r['id']} {r['name']}: AI analysis incomplete - {r['ai_error']}"))
lst = r.get("listening")
if lst and lst.chunk_errors:
for e in lst.chunk_errors:
problems.append(("warning", f"{r['id']}: part {e} was not transcribed"))
if run.get("synthesis_error"):
problems.append(("warning", f"Cross-file synthesis failed: {run['synthesis_error']}"))
for w in run.get("warnings", []):
problems.append(("warning", w))
if problems:
out.append('
' + "".join(f"
{status(s, t)}
" for s, t in problems) + "
")
if syn and syn.get("research_answers"):
out.append(f"
Your questions, across all recordings {ai_tag()}
")
for qa in syn["research_answers"]:
out.append(f'
{esc(qa.get("question", ""))}
{esc(qa.get("answer", ""))}
'
+ "".join(quote_html(ev, ev.get("file_id", "")) for ev in qa.get("evidence", [])) + "
")
# file list
rows = []
for r in run["files"]:
if r.get("error"):
rows.append([f'{esc(r["id"])}', esc(r["name"]), "—", "—", "—", "failed"])
continue
a = r.get("analysis") or {}
info = r["info"]
rows.append([f'{esc(r["id"])}', esc(r["name"]),
esc((info.recorded_at or "").replace("T", " ")[:16] or "—"),
esc(fmt_time(r["acoustics"].duration_s)), esc(a.get("content_type", "—")), esc(a.get("title", "")) or "—"])
out.append(table(["ID", "File", "Recorded", "Length", "Type (AI)", "Title (AI)"], rows, {3}))
out.append("")
return "".join(out)
def _patterns(run: dict) -> str:
cmp = run.get("compare") or {}
ok = [r for r in run["files"] if not r.get("error")]
if len(ok) < 2:
return ""
ids = [r["id"] for r in ok]
names = {r["id"]: r["name"] for r in ok}
syn = run.get("synthesis") or {}
out = ['
Patterns across recordings
']
if len(ok) <= 3:
out.append(f'
With {len(ok)} recordings, patterns are indicative only; more recordings make them firmer.
')
if syn:
themes = syn.get("common_themes", [])
out.append(f"
Shared themes {ai_tag()}
")
if themes:
out.append('
strong'
'moderate'
'weak'
'absent? not assessed
')
out.append(charts.theme_matrix(themes, ids))
out.append(data_table("Theme table", ["Theme"] + ids,
[[esc(t["theme"])] + [esc({p["file_id"]: p["strength"] for p in t["presence"]}.get(i, "")) for i in ids] for t in themes]))
for t in themes:
present = [p["file_id"] for p in t["presence"] if p["strength"] in ("strong", "moderate", "weak")]
out.append(f'
{esc(t["theme"])} {esc(", ".join(present))}
'
f'
{esc(t.get("description", ""))}
'
+ "".join(quote_html(ev, ev.get("file_id", "")) for ev in t.get("evidence", [])) + "
")
else:
out.append('
No theme recurs in two or more recordings.
')
def id_list(xs):
return " ".join(f'{esc(x)}' for x in xs)
if syn.get("contrasts"):
out.append(f"
")
rows = [[f'{esc(u["file_id"])}', bullet_list(u.get("distinctive_points", []))] for u in syn["unique_points"]]
out.append(table(["File", "Distinctive points"], rows))
if syn.get("trends_over_time"):
out.append(f"
")
for key, title in (("speaker_patterns", "Speaker patterns"), ("acoustic_patterns", "Acoustic patterns"),
("recommendations", "Suggested next steps")):
if syn.get(key):
out.append(f"
{title} {ai_tag()}
{bullet_list(syn[key])}")
if syn.get("outliers"):
out.append(f"
")
table_rows = cmp.get("table", [])
grid = []
for row in table_rows:
key = row["key"]
vals = [(fid, row["values"].get(fid)) for fid in ids]
if sum(v is not None for _, v in vals) < 2:
continue
unit = row["unit"]
tips = {fid: f"{fid} · {names[fid]}|{num(v, row['decimals'])} {unit}".strip() for fid, v in vals}
dec = row["decimals"]
def fmt(v, dec=dec):
return num(v, dec)
if key in DOT_METRICS:
svg = charts.dotplot([(fid, v, tips[fid]) for fid, v in vals], fmt=fmt)
else:
domain = (-1.0, 1.0) if key == "sentiment_score" else None
svg = charts.hbars([(fid, v, tips[fid], 1) for fid, v in vals], "", domain=domain, label_w=36,
diverging=(key == "sentiment_score"), width=320, fmt=fmt)
grid.append(f'
{esc(row["label"])}{(" (" + esc(unit) + ")") if unit else ""}
{svg}
')
out.append('
' + "".join(grid) + "
")
out.append(data_table("Comparison table", ["Metric"] + ids,
[[esc(r["label"] + (f" ({r['unit']})" if r["unit"] else ""))] +
[esc(num(r["values"].get(i), r["decimals"])) for i in ids] for r in table_rows],
set(range(1, len(ids) + 1))))
if cmp.get("acoustic"):
out.append("
Similarity between recordings
")
out.append('
1.00 = identical. Acoustic similarity averages timbre (MFCC), '
'spectral balance and voice-pitch distributions: high values suggest the same room, '
'microphone or voices. Vocabulary similarity compares the transcripts\' word use (TF-IDF).
')
blocks = [("Acoustic", cmp["acoustic"]["overall"])]
if cmp.get("content"):
blocks.append(("Vocabulary", cmp["content"]["matrix"]))
hm = []
for label, m in blocks:
hm.append(f'
' + "".join(
f'{(k - 1) / 7:.2f}–{k / 7:.2f}' for k in range(1, 8)) + "
")
comp_rows = []
n = len(ids)
for i in range(n):
for j in range(i + 1, n):
row = [esc(f"{ids[i]} ↔ {ids[j]}")]
for k in ("timbre", "spectral balance", "voice pitch"):
v = cmp["acoustic"]["components"][k][i, j]
row.append(esc(num(v, 2)))
row.append(esc(num(cmp["acoustic"]["overall"][i, j], 2)))
row.append(esc(num(cmp["content"]["matrix"][i, j], 2)) if cmp.get("content") else "n/a")
comp_rows.append(row)
out.append(data_table("Similarity components", ["Pair", "Timbre", "Spectral balance", "Voice pitch", "Acoustic", "Vocabulary"],
comp_rows, {1, 2, 3, 4, 5}))
notes = []
if cmp.get("closest_pair"):
cp = cmp["closest_pair"]
notes.append(f"Most similar pair: {cp['ids'][0]} and {cp['ids'][1]} ({cp['similarity']:.2f}).")
if cmp.get("most_distinct"):
md = cmp["most_distinct"]
notes.append(f"Most distinct recording: {md['id']} (mean similarity {md['mean_similarity']:.2f}).")
if cmp.get("groups"):
notes.append("Similarity groups: " + "; ".join("{" + ", ".join(g) + "}" for g in cmp["groups"]) + ".")
for o in cmp.get("outliers", []):
notes.append(f"{o['file_id']} is an outlier for {o['label'].lower()}: {o['value']} vs median {o['median']}.")
if notes:
out.append(bullet_list(notes))
content = cmp.get("content")
if content:
out.append("
Vocabulary
")
if content["shared_terms"]:
out.append("
Shared terms: " + ", ".join(
f'{esc(t["term"])} ({t["files"]} files, {t["count"]}×)' for t in content["shared_terms"][:20]) + "
")
rows = [[f'{esc(fid)}', esc(", ".join(terms))] for fid, terms in zip(ids, content["distinctive_terms"])]
out.append(table(["File", "Distinctive terms"], rows))
if cmp.get("trends"):
out.append("
Consistent changes across recording dates
")
by_id = {r["id"]: r for r in ok}
figs = []
for t in cmp["trends"]:
pts = [(by_id[fid]["info"].recorded_at, cmp["values"][ids.index(fid)][t["metric"]], fid) for fid in t["file_ids"]]
figs.append(charts.trend_chart(t["label"], t["unit"], pts))
out.append('
' + "".join(f'
{f}
' for f in figs) + "
")
out.append(bullet_list([f"{t['label']} {t['direction']} in every later recording ({t['first']} → {t['last']}, n = {t['n']})."
for t in cmp["trends"]]))
out.append("")
return "".join(out)
def _file_section(rec: dict, run: dict) -> str:
fid, name = rec["id"], rec["name"]
if rec.get("error"):
return (f'
{esc(fid)} · {esc(name)}
'
f'
{status("critical", "Could not be analysed: " + rec["error"])}
')
info, ac = rec["info"], rec["acoustics"]
a = rec.get("analysis") or {}
lst = rec.get("listening")
sm = rec.get("speech") or {}
out = [f'
{esc(fid)} · {esc(a.get("title") or name)}
']
meta = [name, fmt_time(ac.duration_s)]
if info.recorded_at:
meta.append("recorded " + info.recorded_at.replace("T", " ")[:16] + f" ({info.recorded_at_source})")
meta.append(f"{info.codec.upper()} {info.native_sr / 1000:g} kHz, {info.channels} ch" + (f", {info.bit_rate // 1000} kb/s" if info.bit_rate else ""))
if a.get("content_type"):
meta.append(a["content_type"])
if lst and lst.language:
meta.append(lst.language)
out.append(f'
')
s = a.get("sentiment") or {}
if s:
out.append(f'
Overall sentiment: {esc(s.get("overall", ""))} '
f'({num(s.get("score"), 2)} on a -1 to +1 scale). {esc(s.get("explanation", ""))}
')
out.append("
")
# quality
out.append(f"
Recording quality {measured_tag()}
")
out.append('
' + "".join(f"
{status(sev, msg)}
" for sev, msg in rec["flags"]) + "
")
if a.get("audio_quality_assessment"):
out.append(f'
{ai_tag()} {esc(a["audio_quality_assessment"])}
')
# charts
out.append(f"
Loudness over time {measured_tag()}
")
out.append('
One-second level in dBFS; grey bands mark pauses of one second or longer. Hover for values.
')
out.append(charts.level_timeline(ac.level_timeline, ac.pauses, ac.duration_s))
order = speaker_order(rec)
if lst and (lst.lines or lst.sound_events):
out.append(f"
Who speaks when {ai_tag()}
")
out.append('
One row per speaker; dots in the last row mark non-speech sounds. Hover for the words.
Frequency content over time (mel scale, 16 kHz analysis). Darker = louder.
')
out.append(f'')
# speakers
if sm:
out.append(f"
Speakers
")
per = sm.get("per_speaker", {})
profiles = {p.get("label"): p for p in a.get("speakers", [])} if a else {}
slots = charts.speaker_slots(order)
items = [(sp, per[sp]["time_share"] * 100, f"{sp}|{per[sp]['time_share'] * 100:.0f}% of talk time · {per[sp]['words']:,} words", slots.get(sp, 8))
for sp in order if sp in per]
out.append(f'
Talk-time share, from transcript timings {measured_tag()}
')
out.append(charts.hbars(items, "%", domain=(0, 100), label_w=150, fmt=lambda v: f"{v:.0f}"))
rows = []
for sp in order:
d = per.get(sp, {})
p = profiles.get(sp, {})
rows.append([esc(sp), esc(p.get("role", "")), esc(p.get("description", "") or (lst.speakers.get(sp, "") if lst else "")),
esc(num(d.get("time_share", 0) * 100, 0) + "%"), esc(f"{d.get('words', 0):,}"),
esc(num(d.get("wpm"), 0)), esc(str(d.get("turns", "")))])
out.append(table(["Speaker", "Role (AI)", "Description (AI)", "Talk time", "Words", "Words/min", "Turns"], rows, {3, 4, 5, 6}))
if a:
if a.get("research_answers"):
out.append(f"
Your questions {ai_tag()}
")
for qa in a["research_answers"]:
out.append(f'
{esc(qa.get("question", ""))}
{esc(qa.get("answer", ""))}
'
+ "".join(quote_html(ev) for ev in qa.get("evidence", [])) + "
")
if a.get("themes"):
out.append(f"
Themes {ai_tag()}
")
for t in a["themes"]:
out.append(f'
{esc(t.get("name", ""))}
{esc(t.get("description", ""))}
'
+ "".join(quote_html(ev) for ev in t.get("evidence", [])) + "
")
out.append(f'
Key points {ai_tag()}
{bullet_list(a.get("key_points", []))}
'
f'
Questions raised {ai_tag()}
{bullet_list(a.get("questions_raised", []))}
')
if a.get("action_items"):
out.append(f"
Decisions, requests and follow-ups {ai_tag()}
{bullet_list(a['action_items'])}")
if a.get("notable_quotes"):
out.append(f"
Notable quotes {ai_tag()}
")
for q in a["notable_quotes"]:
out.append(quote_html(q).replace("", f'{esc(q.get("why_it_matters", ""))}'))
if a.get("tone_moments"):
out.append(f"
Tone and emotion {ai_tag()}
")
out.append(table(["Time", "Tone", "Note"], [[esc(t.get("time", "")), esc(t.get("tone", "")), esc(t.get("note", ""))] for t in a["tone_moments"]]))
music = a.get("music") or {}
if music.get("present"):
out.append(f"
Music {ai_tag()}
")
out.append(table(["Genre", "Mood", "Instruments", "Tempo feel"],
[[esc(music.get("genre", "")), esc(music.get("mood", "")), esc(", ".join(music.get("instruments", []))), esc(music.get("tempo_feel", ""))]]))
if music.get("structure"):
out.append(table(["Time", "Section"], [[esc(s.get("time", "")), esc(s.get("note", ""))] for s in music["structure"]]))
kw = a.get("keywords") or []
ents = a.get("named_entities") or []
if kw or ents:
out.append(f"
Keywords and names {ai_tag()}
")
if kw:
out.append("
" + " ".join(f'{esc(k)}' for k in kw) + "
")
if ents:
out.append("
" + "; ".join(f"{esc(e.get('name', ''))} ({esc(e.get('type', ''))})" for e in ents) + "
")
qc = a.get("quote_check")
if qc:
out.append(f'
Quote check: {qc.get("exact", 0)} exact, {qc.get("close", 0)} close, '
f'{qc.get("not found", 0)} not found in the transcript. Quote times come from the matching transcript line.
')
# metrics table
out.append(f"
Measurements {measured_tag()}
")
rows = [
["Integrated loudness", num(ac.integrated_lufs, 1, "LUFS")],
["Peak level (original file)", num(info.peak_dbfs, 1, "dBFS")],
["Average level (RMS)", num(ac.rms_dbfs, 1, "dBFS")],
["Noise floor (quietest 10%)", num(ac.noise_floor_dbfs, 1, "dBFS")],
["Estimated signal-to-noise", "n/a (steady signal)" if ac.steady_signal else num(ac.snr_db, 0, "dB")],
["Sound activity", num(ac.activity_ratio * 100, 0, "%")],
["Pauses of 0.5 s or more", f"{ac.pause_count} ({num(ac.pauses_per_min, 1)} per min; {ac.long_pause_count} over 2 s)"],
["Median / longest pause", f"{num(ac.median_pause_s, 2, 's')} / {num(ac.longest_pause_s, 1, 's')}"],
["Leading / trailing silence", f"{num(ac.leading_silence_s, 1, 's')} / {num(ac.trailing_silence_s, 1, 's')}"],
["Median voice pitch (middle 50%)", "n/a" if not ac.pitch_median_hz else f"{ac.pitch_median_hz:.0f} Hz ({ac.pitch_p25_hz:.0f}–{ac.pitch_p75_hz:.0f} Hz)"],
["Voiced share of active sound", num(ac.voiced_ratio * 100, 0, "%")],
["Spectral centroid / bandwidth", f"{ac.spectral_centroid_hz:.0f} Hz / {ac.spectral_bandwidth_hz:.0f} Hz"],
["Effective bandwidth", f"{ac.bandwidth_hz / 1000:.1f} kHz" + (" (analysis limit)" if ac.bandwidth_hz >= 7800 else "")],
["Onsets per second", num(ac.onset_rate_per_s, 2)],
["Tempo estimate (music only)", num(ac.tempo_bpm, 0, "BPM") if (not a or (a.get("music") or {}).get("present") or a.get("content_type") == "music") else "not applicable (no music detected)"],
["Clipped samples", f"{info.clipped_fraction * 100:.3f}%"],
["Channels: balance / correlation", "mono" if info.channel_balance_db is None else f"{info.channel_balance_db:+.1f} dB / {num(info.channel_correlation, 3)}"],
]
if sm:
rows += [
["Words transcribed", f"{sm['words']:,}"],
["Speaking rate", num(sm.get("words_per_minute"), 0, "words/min")],
["Speaker turns", f"{sm['turns']} ({num(sm.get('turns_per_minute'), 1)} per min)"],
["Longest single turn", num(sm.get("longest_turn_s"), 0, "s")],
["Questions (question marks)", str(sm.get("questions", 0))],
["Fillers per 100 words (um, uh...)", num(sm.get("fillers_per_100_words"), 1)],
["Vocabulary variety (MATTR, 0-1)", num(sm.get("lexical_diversity"), 2)],
]
band = "; ".join(f"{k} {v * 100:.0f}%" for k, v in ac.band_fractions.items())
rows.append(["Energy by frequency band", band])
out.append(table(["Measure", "Value"], [[esc(k), esc(v)] for k, v in rows]))
# transcript and listening notes
if lst and lst.lines:
tid = f"tr-{fid}"
out.append(f"
Transcript {ai_tag()}
")
out.append('
Machine transcript by Gemini: check important passages against the audio. '
'Times are approximate (usually within a few seconds).
")
if lst.sound_events:
parts.append(table(["Time", "Sound"], [[esc(fmt_time(t)), esc(n)] for t, n in lst.sound_events]))
if lst.vocal_tone:
parts.append(table(["Time", "How it sounds"], [[esc(fmt_time(t)), esc(n)] for t, n in lst.vocal_tone]))
for n in lst.notes:
parts.append(f'
Gemini usage for this report: {u['calls']} requests, "
f"{u.get('prompt_tokens', 0):,} input and {u.get('output_tokens', 0):,} output tokens "
f"(models: {esc(', '.join(sorted(u.get('models', []))))}).
")
return f"""
Methods and limits
Measured sections come from signal processing of a 16 kHz mono mix of each file (the average of its channels):
integrated loudness (ITU-R BS.1770), noise floor (10th percentile of 20 ms frame levels), sound activity and pauses
(adaptive threshold between noise floor and speech level), voice pitch (normalised autocorrelation, 60–450 Hz),
spectral shape, effective bandwidth (long-term spectrum within 40 dB of the 0.3–3 kHz level), mains-hum detection, and
peak / clipping / channel checks on the original samples. With 16 kHz analysis, content above 8 kHz is not measured.
AI-generated sections come from Google Gemini. Gemini listens to each recording in clips of about ten minutes cut
at pauses (long inputs make its clock drift), producing a verbatim transcript with speaker labels, non-speech sounds
and tone notes. A second, text-only step writes the per-recording analysis, and a third compares all recordings.
Speaker labels are inferred from voices and context and can be wrong, especially for similar voices or cross-talk.
Checks on the AI. Transcript times are re-based per clip, clamped to the clip and rescaled if they run past its
end. Every quote Gemini cites is searched for in the transcript and marked exact, close or
not found; the quote's time is taken from the matching transcript line. Speaking rate, talk share, turns,
questions and fillers are computed from the transcript, not estimated by the model.
Cross-file similarity. Acoustic similarity averages three 0–1 scores: timbre (distance between mean MFCCs,
scaled by each recording's own variability), spectral balance (Hellinger affinity of six band energies) and voice
pitch (overlap of pitch histograms). Vocabulary similarity is the TF-IDF cosine similarity of the transcripts.
Trends are reported only when a measure rises or falls in every later recording (3+ dated files); outliers use a
robust z-score above 3.5 (4+ files). With few recordings these are descriptive, not statistical, findings.
Privacy. Audio is sent to the Gemini API for analysis. Under Google's terms, content sent with an unpaid
(free-tier) key may be used to improve Google's products and seen by human reviewers; use a paid-tier key for
confidential or research-participant recordings, and follow your ethics approval.
{usage}
"""
def build_html(run: dict) -> str:
ok_files = [r for r in run["files"]]
body = [_header(run), _summary(run), _patterns(run)]
body += [_file_section(r, run) for r in ok_files]
body.append(_methods(run))
title = esc(run.get("title") or "Audio analysis report")
return (f''
f'{title}'
f'
{"".join(body)}
'
f'')
# --- exports --------------------------------------------------------------------------------
def _jsonable(obj):
if isinstance(obj, np.ndarray):
return [_jsonable(x) for x in obj.tolist()]
if isinstance(obj, (np.floating, float)):
f = float(obj)
return f if np.isfinite(f) else None
if isinstance(obj, (np.integer,)):
return int(obj)
if isinstance(obj, dict):
return {str(k): _jsonable(v) for k, v in obj.items()}
if isinstance(obj, (list, tuple, set)):
return [_jsonable(x) for x in obj]
return obj
def to_json(run: dict) -> dict:
files = []
for r in run["files"]:
d = {"id": r["id"], "name": r["name"], "error": r.get("error"), "ai_error": r.get("ai_error")}
if not r.get("error"):
d["info"] = r["info"].to_dict()
d["acoustics"] = r["acoustics"].summary()
d["quality_flags"] = [{"severity": s, "message": m} for s, m in r["flags"]]
d["chunks"] = r.get("chunks")
d["speech_metrics"] = r.get("speech")
d["analysis"] = r.get("analysis")
if r.get("listening"):
lst = r["listening"].to_dict()
d["listening"] = lst
files.append(d)
cmp = run.get("compare") or {}
cmp_out = {k: v for k, v in cmp.items() if k not in ("acoustic", "combined")}
if "acoustic" in cmp:
cmp_out["acoustic_similarity"] = cmp["acoustic"]["overall"]
cmp_out["acoustic_components"] = cmp["acoustic"]["components"]
if cmp.get("content"):
cmp_out["content"] = dict(cmp["content"])
return _jsonable({
"title": run.get("title"), "generated_at": run["generated_at"], "settings": run["settings"],
"usage": {k: v for k, v in run.get("usage", {}).items()}, "warnings": run.get("warnings", []),
"files": files, "comparison": cmp_out, "synthesis": run.get("synthesis"),
"synthesis_error": run.get("synthesis_error"),
})
def metrics_csv(run: dict) -> str:
buf = io.StringIO()
w = csv.writer(buf)
keys = [k for k in METRIC_INFO]
w.writerow(["file_id", "file_name", "recorded_at", "content_type_ai", "title_ai"] +
[f"{METRIC_INFO[k][0]} ({METRIC_INFO[k][1]})" if METRIC_INFO[k][1] else METRIC_INFO[k][0] for k in keys] +
["peak_dbfs", "clipped_fraction", "noise_floor_dbfs", "quality_flags", "error"])
values = {r["id"]: v for r, v in zip([r for r in run["files"] if not r.get("error")], (run.get("compare") or {}).get("values", []))}
for r in run["files"]:
if r.get("error"):
w.writerow([r["id"], r["name"], "", "", ""] + [""] * len(keys) + ["", "", "", "", r["error"]])
continue
v = values.get(r["id"], {})
a = r.get("analysis") or {}
w.writerow([r["id"], r["name"], r["info"].recorded_at or "", a.get("content_type", ""), a.get("title", "")] +
["" if v.get(k) is None else round(v[k], 4) for k in keys] +
[round(r["info"].peak_dbfs, 2) if np.isfinite(r["info"].peak_dbfs) else "",
r["info"].clipped_fraction, r["acoustics"].noise_floor_dbfs,
" | ".join(m for s, m in r["flags"] if s != "good"), ""])
return buf.getvalue()
def _srt_time(t: float) -> str:
ms = int(round(max(0.0, t) * 1000))
h, rem = divmod(ms, 3_600_000)
m, rem = divmod(rem, 60_000)
s, ms = divmod(rem, 1000)
return f"{h:02d}:{m:02d}:{s:02d},{ms:03d}"
def transcript_txt(rec: dict) -> str:
lst = rec["listening"]
head = f"Transcript of {rec['name']} ({rec['id']}), generated by Gemini - check against the audio.\n\n"
return head + "\n".join(f"[{fmt_time(ln.start)}] {ln.speaker}: {ln.text}" for ln in lst.lines) + "\n"
def transcript_srt(rec: dict) -> str:
out = []
lines = rec["listening"].lines
for i, ln in enumerate(lines, 1):
end = ln.end if ln.end > ln.start else (lines[i].start if i < len(lines) else ln.start + 3)
out.append(f"{i}\n{_srt_time(ln.start)} --> {_srt_time(max(end, ln.start + 0.5))}\n{ln.speaker}: {ln.text}\n")
return "\n".join(out)
def safe_name(name: str) -> str:
base = os.path.splitext(os.path.basename(name))[0]
return re.sub(r"[^A-Za-z0-9._-]+", "_", base).strip("_")[:60] or "file"
def write_outputs(run: dict, out_dir: str) -> dict:
"""Write report.html, report.json, metrics.csv, transcripts and a zip of all."""
os.makedirs(out_dir, exist_ok=True)
stamp = dt.datetime.now().strftime("%Y%m%d-%H%M")
paths = {}
html_text = build_html(run)
paths["html"] = os.path.join(out_dir, f"audio-report-{stamp}.html")
with open(paths["html"], "w", encoding="utf-8") as fh:
fh.write(html_text)
paths["json"] = os.path.join(out_dir, f"audio-report-{stamp}.json")
with open(paths["json"], "w", encoding="utf-8") as fh:
json.dump(to_json(run), fh, ensure_ascii=False, indent=1)
paths["csv"] = os.path.join(out_dir, f"audio-metrics-{stamp}.csv")
with open(paths["csv"], "w", encoding="utf-8", newline="") as fh:
fh.write(metrics_csv(run))
transcripts = []
for r in run["files"]:
if r.get("listening") and r["listening"].lines:
base = f"{r['id']}_{safe_name(r['name'])}"
for ext, fn in (("txt", transcript_txt), ("srt", transcript_srt)):
p = os.path.join(out_dir, f"{base}.{ext}")
with open(p, "w", encoding="utf-8") as fh:
fh.write(fn(r))
transcripts.append(p)
paths["transcripts"] = transcripts
paths["zip"] = os.path.join(out_dir, f"audio-analysis-{stamp}.zip")
with zipfile.ZipFile(paths["zip"], "w", zipfile.ZIP_DEFLATED) as z:
for p in [paths["html"], paths["json"], paths["csv"], *transcripts]:
z.write(p, os.path.basename(p))
paths["html_text"] = html_text
return paths
')
# speakers
if sm:
out.append(f"