Spaces:
Sleeping
Sleeping
| """Builds the self-contained HTML report and the data exports. | |
| Every piece of text that came from a file name, a transcript or Gemini is | |
| HTML-escaped. AI-generated sections are labelled as such; measured sections are | |
| labelled "measured". | |
| """ | |
| from __future__ import annotations | |
| import csv | |
| import datetime as dt | |
| import html | |
| import io | |
| import json | |
| import os | |
| import re | |
| import zipfile | |
| import numpy as np | |
| import charts | |
| from ai_analysis import fmt_time | |
| from patterns import METRIC_INFO | |
| esc = charts.esc | |
| CSS = """ | |
| :root{color-scheme:light; | |
| --page:#ffffff;--surface:#ffffff;--card:#ffffff;--ink:#111111;--ink2:#333333;--muted:#666666;--grid:#e6e6e6;--axis:#bdbdbd; | |
| --border:#dcdcdc;--accent:#111111;--wash:rgba(0,0,0,.05);--mark:#222222; | |
| --q1:#f0f0f0;--q2:#d9d9d9;--q3:#bdbdbd;--q4:#969696;--q5:#737373;--q6:#4d4d4d;--q7:#252525; | |
| --hv-dark:#111111;--hv-light:#ffffff;--diag:#f5f5f5} | |
| *{box-sizing:border-box} | |
| body{margin:0;background:var(--page);color:var(--ink);font:15px/1.6 -apple-system,BlinkMacSystemFont,"Segoe UI","Helvetica Neue",Arial,sans-serif} | |
| .wrap{max-width:1040px;margin:0 auto;padding:28px 16px 64px} | |
| h1{font-size:28px;line-height:1.2;margin:0 0 6px} | |
| h2{font-size:21px;margin:40px 0 12px;padding-top:8px} | |
| h3{font-size:17px;margin:26px 0 8px} | |
| h4{font-size:15px;margin:18px 0 6px} | |
| p{margin:6px 0 10px} | |
| a{color:var(--accent);text-decoration:underline;text-underline-offset:2px} | |
| .sub{color:var(--ink2)} | |
| .muted{color:var(--muted)} | |
| .card{background:var(--card);border:1px solid var(--border);border-radius:6px;padding:18px 20px;margin:14px 0} | |
| .tiles{display:grid;grid-template-columns:repeat(auto-fit,minmax(170px,1fr));gap:12px;margin:16px 0} | |
| .tile{background:var(--card);border:1px solid var(--border);border-radius:6px;padding:14px 16px} | |
| .tile .v{font-size:26px;font-weight:600;line-height:1.2} | |
| .tile .l{color:var(--ink2);font-size:13px} | |
| nav.toc{display:flex;flex-wrap:wrap;gap:6px 14px;font-size:14px;margin:10px 0 0} | |
| .badge{display:inline-block;font-size:11.5px;line-height:1.4;padding:1px 7px;border-radius:3px;border:1px solid var(--border);color:var(--ink2);background:var(--surface);white-space:nowrap;font-weight:500} | |
| .badge.ai{border-color:var(--ink);color:var(--ink)} | |
| .status{display:inline-flex;align-items:center;gap:6px;font-size:13px;color:var(--ink)} | |
| .status i{font-style:normal;display:inline-grid;place-items:center;width:18px;height:18px;border-radius:50%;font-size:11px;font-weight:700; | |
| border:1.5px solid var(--ink);color:var(--ink);background:var(--card)} | |
| .status.serious i,.status.critical i{background:var(--ink);color:#ffffff} | |
| ul.flags{list-style:none;padding:0;margin:6px 0}ul.flags li{margin:4px 0;display:flex;gap:8px;align-items:flex-start} | |
| blockquote{margin:8px 0;padding:6px 12px;border-left:3px solid var(--axis);color:var(--ink)} | |
| blockquote .meta{display:block;color:var(--ink2);font-size:13px;margin-top:2px} | |
| table{border-collapse:collapse;width:100%;font-size:14px;margin:8px 0} | |
| th,td{text-align:left;padding:6px 8px;border-bottom:1px solid var(--grid);vertical-align:top} | |
| th{color:var(--ink2);font-weight:600} | |
| td.num,th.num{text-align:right;font-variant-numeric:tabular-nums;white-space:nowrap} | |
| .tablewrap{overflow-x:auto} | |
| details{margin:8px 0}summary{cursor:pointer;color:var(--ink2);font-size:14px} | |
| .chart{width:100%;height:auto;display:block;margin:6px 0} | |
| .chart .grid{stroke:var(--grid);stroke-width:1} | |
| .chart .axis{stroke:var(--axis);stroke-width:1} | |
| .chart text{fill:var(--ink2);font-size:11px} | |
| .chart .tick{fill:var(--muted);font-size:10.5px;font-variant-numeric:tabular-nums} | |
| .chart .lab{fill:var(--ink2);font-size:11.5px} | |
| .chart .val{fill:var(--ink);font-size:11.5px;font-variant-numeric:tabular-nums} | |
| .chart .line{fill:none;stroke-width:2;stroke-linejoin:round;stroke-linecap:round} | |
| .chart .area{opacity:.06} | |
| .chart .pause{fill:#f0f0f0} | |
| .chart .hit{fill:transparent} | |
| .chart .hit:hover{fill:var(--wash)} | |
| .chart .seg:hover,.chart .bar:hover,.chart .dot:hover,.chart [class^="q"]:hover{opacity:.8} | |
| .chart .evt{fill:var(--ink2);stroke:var(--surface);stroke-width:2} | |
| .chart .mk{stroke:var(--surface);stroke-width:2} | |
| .chart .ring{fill:none;stroke:var(--axis);stroke-width:1.5} | |
| .chart .diag{fill:var(--diag)} | |
| .chart .hv{font-size:11px;font-variant-numeric:tabular-nums} | |
| .chart .hv1,.chart .hv2,.chart .hv3{fill:var(--hv-dark)}.chart .hv4,.chart .hv5,.chart .hv6,.chart .hv7{fill:var(--hv-light)} | |
| .s1f,.s2f,.s3f,.s4f,.s5f,.s6f,.s7f,.s8f{fill:var(--mark)} | |
| .s1s{stroke:var(--mark)} | |
| .q1{fill:var(--q1)}.q2{fill:var(--q2)}.q3{fill:var(--q3)}.q4{fill:var(--q4)}.q5{fill:var(--q5)}.q6{fill:var(--q6)}.q7{fill:var(--q7)} | |
| .o2{fill:var(--q3)}.o3{fill:var(--q5)}.o4{fill:var(--q7)} | |
| .legend{display:flex;flex-wrap:wrap;gap:6px 16px;font-size:13px;color:var(--ink2);margin:4px 0} | |
| .legend span{display:inline-flex;align-items:center;gap:6px} | |
| .legend b{display:inline-block;width:12px;height:12px;border-radius:3px} | |
| .legend b.dot{border-radius:50%} | |
| .legend b.ringk{border-radius:50%;border:1.5px solid var(--axis);background:none;width:8px;height:8px} | |
| .grid2{display:grid;grid-template-columns:repeat(auto-fit,minmax(300px,1fr));gap:12px} | |
| .small{font-size:13px} | |
| .transcript{max-height:520px;overflow:auto;border:1px solid var(--grid);border-radius:8px;padding:8px 12px;background:var(--surface)} | |
| .transcript p{margin:4px 0}.transcript .ts{color:var(--muted);font-variant-numeric:tabular-nums;margin-right:6px} | |
| .transcript .sp{font-weight:600;margin-right:4px} | |
| .search{width:100%;max-width:360px;padding:6px 10px;border:1px solid var(--border);border-radius:8px;background:var(--card);color:var(--ink);font:inherit;margin:6px 0} | |
| .spec{width:100%;height:auto;border-radius:4px;border:1px solid var(--border)} | |
| #tip{position:fixed;z-index:10;pointer-events:none;max-width:340px;background:var(--card);color:var(--ink);border:1px solid var(--border); | |
| border-radius:4px;padding:8px 10px;font-size:12.5px;line-height:1.4;box-shadow:0 2px 10px rgba(0,0,0,.12);display:none} | |
| #tip strong{display:block;font-size:13.5px} | |
| .fid{display:inline-block;min-width:28px;font-weight:700} | |
| .sr{position:absolute;width:1px;height:1px;overflow:hidden;clip:rect(0 0 0 0);white-space:nowrap} | |
| .err{border-left:3px solid var(--ink);padding-left:10px} | |
| @media print{nav.toc,.search,#tip{display:none!important}.card{break-inside:avoid}body{background:#fff}} | |
| @media (max-width:600px){.wrap{padding:18px 16px 48px}h1{font-size:23px}.card{padding:14px}} | |
| """ | |
| JS = r""" | |
| (function(){ | |
| var tip=document.getElementById('tip'); | |
| function show(el,x,y){var t=el.getAttribute('data-tip');if(!t)return;var p=t.split('|'); | |
| tip.textContent='';var s=document.createElement('strong');s.textContent=p.length>1?p[1]:p[0];tip.appendChild(s); | |
| if(p.length>1){var d=document.createElement('span');d.textContent=p[0];tip.appendChild(d);} | |
| tip.style.display='block';var r=tip.getBoundingClientRect();var nx=x+14,ny=y+14; | |
| if(nx+r.width>innerWidth-8)nx=x-r.width-14;if(ny+r.height>innerHeight-8)ny=y-r.height-14; | |
| tip.style.left=Math.max(8,nx)+'px';tip.style.top=Math.max(8,ny)+'px';} | |
| document.addEventListener('pointermove',function(e){var el=e.target.closest&&e.target.closest('[data-tip]'); | |
| if(el)show(el,e.clientX,e.clientY);else tip.style.display='none';}); | |
| document.addEventListener('focusin',function(e){var el=e.target.closest&&e.target.closest('[data-tip]'); | |
| if(el){var r=el.getBoundingClientRect();show(el,r.right,r.top);}}); | |
| document.addEventListener('focusout',function(){tip.style.display='none';}); | |
| document.querySelectorAll('input.search').forEach(function(inp){inp.addEventListener('input',function(){ | |
| var q=inp.value.toLowerCase();var box=document.getElementById(inp.getAttribute('data-target')); | |
| box.querySelectorAll('p').forEach(function(p){p.style.display=(!q||p.textContent.toLowerCase().indexOf(q)>=0)?'':'none';});});}); | |
| window.addEventListener('beforeprint',function(){document.querySelectorAll('details').forEach(function(d){d.open=true;});}); | |
| })(); | |
| """ | |
| # --- small helpers --------------------------------------------------------------------- | |
| def num(v, dec: int = 1, unit: str = "") -> str: | |
| if v is None or (isinstance(v, float) and not np.isfinite(v)): | |
| return "n/a" | |
| s = f"{v:,.{dec}f}" | |
| return f"{s} {unit}".strip() if unit and unit not in ("%",) else s + unit | |
| def status(sev: str, text: str) -> str: | |
| icon = {"good": "✓", "warning": "!", "serious": "!", "critical": "✕"}.get(sev, "i") | |
| label = {"good": "OK", "warning": "Check", "serious": "Problem", "critical": "Problem"}.get(sev, "Info") | |
| return f'<span class="status {esc(sev)}"><i aria-hidden="true">{icon}</i><span class="sr">{label}:</span></span> {esc(text)}' | |
| # Measures where zero is not a meaningful baseline are drawn as dots on a shared axis. | |
| DOT_METRICS = {"lufs", "pitch_median_hz", "centroid_hz", "bandwidth_khz", "lexical_diversity"} | |
| VERIFY = { | |
| "exact": ("good", "exact match in transcript"), | |
| "close": ("warning", "close match in transcript"), | |
| "not found": ("critical", "not found in transcript"), | |
| } | |
| def verify_badge(ev: dict) -> str: | |
| sev, label = VERIFY.get(ev.get("verified", ""), ("warning", "unchecked")) | |
| icon = {"good": "✓", "warning": "~", "critical": "✕"}[sev] | |
| return f'<span class="status {sev} small"><i aria-hidden="true">{icon}</i>{esc(label)}</span>' | |
| def quote_html(ev: dict, file_label: str = "") -> str: | |
| who = ev.get("speaker") or "" | |
| when = ev.get("time") or "" | |
| meta = " · ".join(x for x in (file_label, when, who) if x) | |
| return (f'<blockquote>“{esc(ev.get("quote", ""))}”<span class="meta">{esc(meta)} ' | |
| f'{verify_badge(ev)}</span></blockquote>') | |
| def bullet_list(items: list[str]) -> str: | |
| items = [i for i in items if str(i).strip()] | |
| if not items: | |
| return '<p class="muted">None noted.</p>' | |
| return "<ul>" + "".join(f"<li>{esc(i)}</li>" for i in items) + "</ul>" | |
| def table(headers: list[str], rows: list[list], num_cols: set[int] = frozenset()) -> str: | |
| th = "".join(f'<th class="{"num" if i in num_cols else ""}">{esc(h)}</th>' for i, h in enumerate(headers)) | |
| body = "".join( | |
| "<tr>" + "".join(f'<td class="{"num" if i in num_cols else ""}">{c}</td>' for i, c in enumerate(r)) + "</tr>" | |
| for r in rows) | |
| return f'<div class="tablewrap"><table><thead><tr>{th}</tr></thead><tbody>{body}</tbody></table></div>' | |
| def data_table(summary: str, headers: list[str], rows: list[list], num_cols: set[int] = frozenset()) -> str: | |
| return f"<details><summary>{esc(summary)}</summary>{table(headers, rows, num_cols)}</details>" | |
| def ai_tag() -> str: | |
| return '<span class="badge ai" title="Generated by Gemini; check against the audio">AI-generated</span>' | |
| def measured_tag() -> str: | |
| return '<span class="badge" title="Computed by signal processing">measured</span>' | |
| def speaker_order(rec: dict) -> list[str]: | |
| lst = rec.get("listening") | |
| order: list[str] = [] | |
| if lst: | |
| for ln in lst.lines: | |
| if ln.speaker not in order: | |
| order.append(ln.speaker) | |
| return order | |
| # --- sections --------------------------------------------------------------------------- | |
| def _header(run: dict) -> str: | |
| ok = [r for r in run["files"] if not r.get("error")] | |
| total = sum(r["acoustics"].duration_s for r in ok) | |
| words = sum((r.get("speech") or {}).get("words", 0) for r in ok) | |
| gen = run["generated_at"] | |
| models = ", ".join(sorted(run.get("usage", {}).get("models", []))) or "none (measured analysis only)" | |
| tiles = [ | |
| (str(len(ok)), "recordings analysed" + (f" ({len(run['files']) - len(ok)} failed)" if len(ok) < len(run["files"]) else "")), | |
| (fmt_time(total), "total duration"), | |
| (f"{words:,}", "words transcribed"), | |
| ] | |
| syn = run.get("synthesis") | |
| if syn: | |
| tiles.append((str(len(syn.get("common_themes", []))), "themes shared across recordings")) | |
| tiles_html = "".join(f'<div class="tile"><div class="v">{esc(v)}</div><div class="l">{esc(l)}</div></div>' for v, l in tiles) | |
| toc = ['<a href="#summary">Summary</a>'] | |
| if len(ok) >= 2: | |
| toc.append('<a href="#patterns">Patterns across recordings</a>') | |
| for r in run["files"]: | |
| toc.append(f'<a href="#{esc(r["id"])}">{esc(r["id"])}</a>') | |
| toc.append('<a href="#methods">Methods and limits</a>') | |
| return f""" | |
| <header> | |
| <h1>{esc(run.get('title') or 'Audio analysis report')}</h1> | |
| <div class="sub">Generated {esc(gen)} · Gemini model: {esc(models)}</div> | |
| <nav class="toc">{''.join(toc)}</nav> | |
| </header> | |
| <div class="tiles">{tiles_html}</div>""" | |
| def _summary(run: dict) -> str: | |
| ok = [r for r in run["files"] if not r.get("error")] | |
| syn = run.get("synthesis") | |
| out = ['<section id="summary"><h2>Summary</h2>'] | |
| ctx = run["settings"].get("context", "").strip() | |
| if ctx: | |
| out.append(f'<p class="sub"><b>Context given:</b> {esc(ctx)}</p>') | |
| if syn: | |
| out.append(f'<div class="card">{ai_tag()}<p>{esc(syn.get("overview", ""))}</p></div>') | |
| elif len(ok) == 1 and ok[0].get("analysis"): | |
| out.append(f'<div class="card">{ai_tag()}<p>{esc(ok[0]["analysis"].get("summary", ""))}</p></div>') | |
| elif ok: | |
| out.append('<div class="card"><p>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).</p></div>') | |
| 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('<ul class="flags">' + "".join(f"<li>{status(s, t)}</li>" for s, t in problems) + "</ul>") | |
| if syn and syn.get("research_answers"): | |
| out.append(f"<h3>Your questions, across all recordings {ai_tag()}</h3>") | |
| for qa in syn["research_answers"]: | |
| out.append(f'<div class="card"><h4>{esc(qa.get("question", ""))}</h4><p>{esc(qa.get("answer", ""))}</p>' | |
| + "".join(quote_html(ev, ev.get("file_id", "")) for ev in qa.get("evidence", [])) + "</div>") | |
| # file list | |
| rows = [] | |
| for r in run["files"]: | |
| if r.get("error"): | |
| rows.append([f'<span class="fid">{esc(r["id"])}</span>', esc(r["name"]), "—", "—", "—", "failed"]) | |
| continue | |
| a = r.get("analysis") or {} | |
| info = r["info"] | |
| rows.append([f'<a class="fid" href="#{esc(r["id"])}">{esc(r["id"])}</a>', 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("</section>") | |
| 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 = ['<section id="patterns"><h2>Patterns across recordings</h2>'] | |
| if len(ok) <= 3: | |
| out.append(f'<p class="sub">With {len(ok)} recordings, patterns are indicative only; more recordings make them firmer.</p>') | |
| if syn: | |
| themes = syn.get("common_themes", []) | |
| out.append(f"<h3>Shared themes {ai_tag()}</h3>") | |
| if themes: | |
| out.append('<div class="legend"><span><b class="dot" style="background:var(--q7)"></b>strong</span>' | |
| '<span><b class="dot" style="background:var(--q5)"></b>moderate</span>' | |
| '<span><b class="dot" style="background:var(--q3)"></b>weak</span>' | |
| '<span><b class="ringk"></b>absent</span><span>? not assessed</span></div>') | |
| 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'<div class="card"><h4>{esc(t["theme"])} <span class="badge">{esc(", ".join(present))}</span></h4>' | |
| f'<p>{esc(t.get("description", ""))}</p>' | |
| + "".join(quote_html(ev, ev.get("file_id", "")) for ev in t.get("evidence", [])) + "</div>") | |
| else: | |
| out.append('<p class="muted">No theme recurs in two or more recordings.</p>') | |
| def id_list(xs): | |
| return " ".join(f'<span class="badge">{esc(x)}</span>' for x in xs) | |
| if syn.get("contrasts"): | |
| out.append(f"<h3>Differences {ai_tag()}</h3>") | |
| for c in syn["contrasts"]: | |
| out.append(f'<div class="card"><h4>{esc(c["aspect"])} {id_list(c.get("file_ids", []))}</h4><p>{esc(c["description"])}</p></div>') | |
| if syn.get("unique_points"): | |
| out.append(f"<h3>What each recording adds {ai_tag()}</h3>") | |
| rows = [[f'<span class="fid">{esc(u["file_id"])}</span>', 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"<h3>Changes over time {ai_tag()}</h3>") | |
| out.append("<ul>" + "".join(f"<li>{esc(t['description'])} {id_list(t.get('file_ids', []))}</li>" for t in syn["trends_over_time"]) + "</ul>") | |
| for key, title in (("speaker_patterns", "Speaker patterns"), ("acoustic_patterns", "Acoustic patterns"), | |
| ("recommendations", "Suggested next steps")): | |
| if syn.get(key): | |
| out.append(f"<h3>{title} {ai_tag()}</h3>{bullet_list(syn[key])}") | |
| if syn.get("outliers"): | |
| out.append(f"<h3>Outliers {ai_tag()}</h3>") | |
| out.append("<ul>" + "".join(f"<li><b>{esc(o['file_id'])}</b>: {esc(o['reason'])}</li>" for o in syn["outliers"]) + "</ul>") | |
| # --- measured comparison ------------------------------------------------------- | |
| out.append(f"<h3>Measured comparison {measured_tag()}</h3>") | |
| 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'<div class="card"><h4>{esc(row["label"])}{(" (" + esc(unit) + ")") if unit else ""}</h4>{svg}</div>') | |
| out.append('<div class="grid2">' + "".join(grid) + "</div>") | |
| 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("<h4>Similarity between recordings</h4>") | |
| out.append('<p class="sub small">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).</p>') | |
| blocks = [("Acoustic", cmp["acoustic"]["overall"])] | |
| if cmp.get("content"): | |
| blocks.append(("Vocabulary", cmp["content"]["matrix"])) | |
| hm = [] | |
| for label, m in blocks: | |
| hm.append(f'<div class="card"><h4>{label} similarity</h4>{charts.heatmap(m, ids, names, label + " similarity")}</div>') | |
| out.append('<div class="grid2">' + "".join(hm) + "</div>") | |
| out.append('<div class="legend">' + "".join( | |
| f'<span><b style="background:var(--q{k})"></b>{(k - 1) / 7:.2f}–{k / 7:.2f}</span>' for k in range(1, 8)) + "</div>") | |
| 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("<h4>Vocabulary</h4>") | |
| if content["shared_terms"]: | |
| out.append("<p><b>Shared terms:</b> " + ", ".join( | |
| f'{esc(t["term"])} <span class="muted small">({t["files"]} files, {t["count"]}×)</span>' for t in content["shared_terms"][:20]) + "</p>") | |
| rows = [[f'<span class="fid">{esc(fid)}</span>', 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("<h4>Consistent changes across recording dates</h4>") | |
| 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('<div class="grid2">' + "".join(f'<div class="card">{f}</div>' for f in figs) + "</div>") | |
| 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("</section>") | |
| return "".join(out) | |
| def _file_section(rec: dict, run: dict) -> str: | |
| fid, name = rec["id"], rec["name"] | |
| if rec.get("error"): | |
| return (f'<section id="{esc(fid)}"><h2>{esc(fid)} · {esc(name)}</h2>' | |
| f'<div class="card err">{status("critical", "Could not be analysed: " + rec["error"])}</div></section>') | |
| info, ac = rec["info"], rec["acoustics"] | |
| a = rec.get("analysis") or {} | |
| lst = rec.get("listening") | |
| sm = rec.get("speech") or {} | |
| out = [f'<section id="{esc(fid)}"><h2>{esc(fid)} · {esc(a.get("title") or name)}</h2>'] | |
| 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'<p class="sub">{" · ".join(esc(m) for m in meta)}</p>') | |
| if rec.get("ai_error"): | |
| out.append(f'<ul class="flags"><li>{status("warning", "AI analysis incomplete: " + rec["ai_error"])}</li></ul>') | |
| if a: | |
| out.append(f'<div class="card">{ai_tag()}<p>{esc(a.get("summary", ""))}</p>') | |
| s = a.get("sentiment") or {} | |
| if s: | |
| out.append(f'<p class="small"><b>Overall sentiment:</b> {esc(s.get("overall", ""))} ' | |
| f'({num(s.get("score"), 2)} on a -1 to +1 scale). {esc(s.get("explanation", ""))}</p>') | |
| out.append("</div>") | |
| # quality | |
| out.append(f"<h3>Recording quality {measured_tag()}</h3>") | |
| out.append('<ul class="flags">' + "".join(f"<li>{status(sev, msg)}</li>" for sev, msg in rec["flags"]) + "</ul>") | |
| if a.get("audio_quality_assessment"): | |
| out.append(f'<p class="small">{ai_tag()} {esc(a["audio_quality_assessment"])}</p>') | |
| # charts | |
| out.append(f"<h3>Loudness over time {measured_tag()}</h3>") | |
| out.append('<p class="sub small">One-second level in dBFS; grey bands mark pauses of one second or longer. Hover for values.</p>') | |
| 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"<h3>Who speaks when {ai_tag()}</h3>") | |
| out.append('<p class="sub small">One row per speaker; dots in the last row mark non-speech sounds. Hover for the words.</p>') | |
| out.append(charts.speaker_timeline(lst.lines, order, lst.sound_events, ac.duration_s)) | |
| png = charts.spectrogram_png(ac.mel_db, ac.duration_s) | |
| if png: | |
| out.append(f"<h3>Spectrogram {measured_tag()}</h3>") | |
| out.append('<p class="sub small">Frequency content over time (mel scale, 16 kHz analysis). Darker = louder.</p>') | |
| out.append(f'<img class="spec" alt="Spectrogram of {esc(name)}" src="data:image/png;base64,{png}">') | |
| # speakers | |
| if sm: | |
| out.append(f"<h3>Speakers</h3>") | |
| 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'<p class="sub small">Talk-time share, from transcript timings {measured_tag()}</p>') | |
| 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"<h3>Your questions {ai_tag()}</h3>") | |
| for qa in a["research_answers"]: | |
| out.append(f'<div class="card"><h4>{esc(qa.get("question", ""))}</h4><p>{esc(qa.get("answer", ""))}</p>' | |
| + "".join(quote_html(ev) for ev in qa.get("evidence", [])) + "</div>") | |
| if a.get("themes"): | |
| out.append(f"<h3>Themes {ai_tag()}</h3>") | |
| for t in a["themes"]: | |
| out.append(f'<div class="card"><h4>{esc(t.get("name", ""))}</h4><p>{esc(t.get("description", ""))}</p>' | |
| + "".join(quote_html(ev) for ev in t.get("evidence", [])) + "</div>") | |
| out.append(f'<div class="grid2"><div><h3>Key points {ai_tag()}</h3>{bullet_list(a.get("key_points", []))}</div>' | |
| f'<div><h3>Questions raised {ai_tag()}</h3>{bullet_list(a.get("questions_raised", []))}</div></div>') | |
| if a.get("action_items"): | |
| out.append(f"<h3>Decisions, requests and follow-ups {ai_tag()}</h3>{bullet_list(a['action_items'])}") | |
| if a.get("notable_quotes"): | |
| out.append(f"<h3>Notable quotes {ai_tag()}</h3>") | |
| for q in a["notable_quotes"]: | |
| out.append(quote_html(q).replace("</blockquote>", f'<span class="meta">{esc(q.get("why_it_matters", ""))}</span></blockquote>')) | |
| if a.get("tone_moments"): | |
| out.append(f"<h3>Tone and emotion {ai_tag()}</h3>") | |
| 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"<h3>Music {ai_tag()}</h3>") | |
| 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"<h3>Keywords and names {ai_tag()}</h3>") | |
| if kw: | |
| out.append("<p>" + " ".join(f'<span class="badge">{esc(k)}</span>' for k in kw) + "</p>") | |
| if ents: | |
| out.append("<p class='small'>" + "; ".join(f"{esc(e.get('name', ''))} <span class='muted'>({esc(e.get('type', ''))})</span>" for e in ents) + "</p>") | |
| qc = a.get("quote_check") | |
| if qc: | |
| out.append(f'<p class="small muted">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.</p>') | |
| # metrics table | |
| out.append(f"<h3>Measurements {measured_tag()}</h3>") | |
| 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"<h3>Transcript {ai_tag()}</h3>") | |
| out.append('<p class="sub small">Machine transcript by Gemini: check important passages against the audio. ' | |
| 'Times are approximate (usually within a few seconds).</p>') | |
| body = "".join(f'<p><span class="ts">{esc(fmt_time(ln.start))}</span><span class="sp">{esc(ln.speaker)}:</span>{esc(ln.text)}</p>' | |
| for ln in lst.lines) | |
| out.append(f'<details><summary>Show transcript ({sm.get("words", 0):,} words)</summary>' | |
| f'<input class="search" type="search" placeholder="Filter lines…" data-target="{esc(tid)}" aria-label="Filter transcript">' | |
| f'<div class="transcript" id="{esc(tid)}">{body}</div></details>') | |
| if lst and (lst.descriptions or lst.sound_events or lst.vocal_tone or lst.notes): | |
| parts = [] | |
| for s, e, text in lst.descriptions: | |
| parts.append(f"<p><b>{esc(fmt_time(s))}–{esc(fmt_time(e))}</b>: {esc(text)}</p>") | |
| 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'<p class="small muted">Timestamp repair: {esc(n)}</p>') | |
| out.append(f"<details><summary>Listening notes (Gemini)</summary>{''.join(parts)}</details>") | |
| out.append("</section>") | |
| return "".join(out) | |
| def _methods(run: dict) -> str: | |
| u = run.get("usage", {}) | |
| usage = "" | |
| if u.get("calls"): | |
| usage = (f"<p class='small muted'>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', []))))}).</p>") | |
| return f""" | |
| <section id="methods"><h2>Methods and limits</h2> | |
| <div class="card small"> | |
| <p><b>Measured</b> 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.</p> | |
| <p><b>AI-generated</b> 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.</p> | |
| <p><b>Checks on the AI.</b> 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 <i>exact</i>, <i>close</i> or | |
| <i>not found</i>; 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.</p> | |
| <p><b>Cross-file similarity.</b> 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.</p> | |
| <p><b>Privacy.</b> 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.</p> | |
| {usage} | |
| </div></section>""" | |
| 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'<!doctype html><html lang="en"><head><meta charset="utf-8">' | |
| f'<meta name="viewport" content="width=device-width,initial-scale=1"><title>{title}</title>' | |
| f'<style>{CSS}</style></head><body><div class="wrap">{"".join(body)}</div>' | |
| f'<div id="tip" role="status"></div><script>{JS}</script></body></html>') | |
| # --- 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 | |