Sascha
810cee1dba
#79 verlor einen von vier Sprechern und lieferte eine 1082-Zeichen-Zusammenfassung. Diarisierung: ein zweiter, falsch verankerter Marker verschmolz Saschas Cluster in Merles (Kosinus -0,727). must-link prueft jetzt die akustische Aehnlichkeit, Anker werden nach Trefferqualitaet gewichtet, und enrolled Stimmen wirken als cannot-link -- damit ist der Box-Besitzer immer ein eigener Sprecher. LLM: der Whole-Pass komprimierte das erste Thema weg. Sektions-Pfad jetzt ab 9000 Zeichen, kleinere Chunks, woertlicher Kontext aus dem Vorgaenger-Chunk und ein gezielter Pruef-Aufruf gegen ueberdehnte Einzelaussagen. #79: 4 Sprecher, Zusammenfassung ~5300 Zeichen, 7 Themenbloecke. Regression: #70 -> 4, #68 -> 2, #74 -> 20 Sprecher unveraendert.
636 lines
34 KiB
JavaScript
636 lines
34 KiB
JavaScript
import express from 'express';
|
||
import path from 'node:path';
|
||
import fs from 'node:fs';
|
||
import { execFile } from 'node:child_process';
|
||
import { promisify } from 'node:util';
|
||
import { fileURLToPath } from 'node:url';
|
||
import * as bt from './bt.js';
|
||
import * as rec from './recorder.js';
|
||
import * as db from './db.js';
|
||
import * as llm from './llm.js';
|
||
import * as diar from './diar.js';
|
||
import * as spk from './spk.js';
|
||
const pexec = promisify(execFile);
|
||
|
||
// Glossary → { soundalike(lowercased): correctTerm }. Only entries WITH
|
||
// soundalikes get a deterministic swap; term-only entries are left to the LLM hint.
|
||
function glossaryMap(glossary) {
|
||
const m = {};
|
||
for (const g of glossary || []) {
|
||
if (!g.term || !g.soundalikes) continue;
|
||
for (const s of String(g.soundalikes).split(',')) {
|
||
const k = s.trim().toLowerCase();
|
||
if (k) m[k] = String(g.term).trim();
|
||
}
|
||
}
|
||
return m;
|
||
}
|
||
// Whole-word replace (Unicode-aware, case-insensitive match): swaps any token
|
||
// equal to a known soundalike for its correct term, leaving punctuation intact.
|
||
function applyGlossaryText(text, m) {
|
||
if (!text || !m || !Object.keys(m).length) return text;
|
||
return String(text).replace(/[\p{L}\p{N}]+/gu, (tok) => m[tok.toLowerCase()] || tok);
|
||
}
|
||
|
||
// Build the LLM input from stored words. If acoustic diarization tagged speakers,
|
||
// produce a labelled dialogue ("Sprecher 1: …") so the LLM keeps real speakers
|
||
// instead of guessing; otherwise the plain transcript.
|
||
// Also returns `labels` = the exact label every speaker carries in the text the
|
||
// LLM sees ({speaker_07: "Sprecher 5"}); it is stored with the analysis so a
|
||
// rename afterwards can be patched into the finished summary (relabelAnalysis).
|
||
function buildBasis(rec, nameMap = {}) {
|
||
const words = rec.words || [];
|
||
const hasSpk = words.some(w => w.speaker);
|
||
if (!hasSpk) return { text: rec.transcript || '', labeled: false, labels: {} };
|
||
const turns = diar.toTurns(words);
|
||
const labels = {}; let n = 0;
|
||
// Labels MUST be unique. On #79 a mis-anchored marker named Sascha's cluster
|
||
// "Markus" while the voiceprint named Markus' own cluster "Markus" too — the
|
||
// LLM then read two different people under one name and blended them into one
|
||
// person, and relabelAnalysis refuses to patch an ambiguous label afterwards.
|
||
const used = new Set();
|
||
const uniq = (name) => {
|
||
if (!used.has(name)) { used.add(name); return name; }
|
||
for (let k = 2; ; k++) if (!used.has(`${name} ${k}`)) { used.add(`${name} ${k}`); return `${name} ${k}`; }
|
||
};
|
||
const label = sp => labels[sp] || (labels[sp] = uniq(nameMap[sp] || `Sprecher ${++n}`));
|
||
const text = turns.map(t => `${label(t.speaker)}: ${t.text}`).join('\n');
|
||
return { text, labeled: true, labels };
|
||
}
|
||
|
||
// Swap speaker labels for real names in the ALREADY FINISHED analysis text.
|
||
// Renaming a speaker after the analysis used to leave "Sprecher 5" all over the
|
||
// summary/todos; re-running the LLM costs ~20 min on this board, while the label
|
||
// each speaker had is known exactly (recordings.speaker_labels). Whole-word
|
||
// replacement in summary, todos, transcript_clean and participants; the stored
|
||
// label is then updated to the new name so a second rename works too.
|
||
async function relabelAnalysis(id) {
|
||
const rec = await db.getRecording(id);
|
||
if (!rec) return { changed: 0 };
|
||
const names = rec.speaker_names || {};
|
||
const labels = await db.getSpeakerLabels(id);
|
||
if (!Object.keys(names).length || !Object.keys(labels).length) return { changed: 0 };
|
||
// A label that TWO speakers carried (the voice-ID once matched "Fabio" onto two
|
||
// clusters) is ambiguous in the text — leave those alone instead of renaming
|
||
// both occurrences to one person.
|
||
const labelCount = {};
|
||
for (const l of Object.values(labels)) labelCount[l] = (labelCount[l] || 0) + 1;
|
||
const pairs = Object.entries(names)
|
||
.filter(([sp, name]) => name && labels[sp] && labels[sp] !== name && labelCount[labels[sp]] === 1)
|
||
.map(([sp, name]) => ({ sp, from: labels[sp], to: name }))
|
||
// longest label first: "Sprecher 1" must not eat "Sprecher 12"
|
||
.sort((a, b) => b.from.length - a.from.length);
|
||
const esc = s => s.replace(/[.*+?^${}()|[\]\\]/g, '\\$&');
|
||
const swap = (text) => {
|
||
if (!text) return text;
|
||
let out = String(text);
|
||
for (const p of pairs) out = out.replace(new RegExp(`(?<![\\p{L}\\p{N}])${esc(p.from)}(?![\\p{L}\\p{N}])`, 'gu'), p.to);
|
||
return out;
|
||
};
|
||
const todos = Array.isArray(rec.todos) ? rec.todos.map(t => swap(String(t))) : rec.todos;
|
||
// Participants are REBUILT (not string-swapped) from the diarized speakers in
|
||
// order of first appearance — a swap would leave the entries of speakers that
|
||
// were never mentioned in the text untouched and can duplicate names.
|
||
const spkOrder = [];
|
||
for (const w of (rec.words || [])) if (w.speaker && !spkOrder.includes(w.speaker)) spkOrder.push(w.speaker);
|
||
const participants = spkOrder.length
|
||
? spkOrder.map((sp, i) => names[sp] || labels[sp] || `Sprecher ${i + 1}`)
|
||
: (Array.isArray(rec.participants) ? rec.participants.map(p => swap(String(p))) : rec.participants);
|
||
const partChanged = JSON.stringify(participants || null) !== JSON.stringify(rec.participants || null);
|
||
if (!pairs.length && !partChanged) return { changed: 0 };
|
||
await db.pool.execute(
|
||
'UPDATE recordings SET summary=?, todos=?, transcript_clean=?, participants=? WHERE id=?',
|
||
[swap(rec.summary), todos ? JSON.stringify(todos) : null, swap(rec.transcript_clean),
|
||
participants ? JSON.stringify(participants) : null, id]);
|
||
const next = { ...labels };
|
||
for (const p of pairs) next[p.sp] = p.to;
|
||
await db.setSpeakerLabels(id, next);
|
||
console.log(`relabel #${id}: ${pairs.length ? pairs.map(p => `${p.from}→${p.to}`).join(', ') : 'nur Teilnehmerliste'}`);
|
||
return { changed: pairs.length || (partChanged ? 1 : 0) };
|
||
}
|
||
|
||
const REC_DIR2 = path.join(path.dirname(fileURLToPath(import.meta.url)), 'recordings');
|
||
// Match each diarization cluster against enrolled voiceprints → { speaker: name }.
|
||
// Cuts the speaker's longest contiguous run from the kept wav, embeds it, cosine-
|
||
// matches. Returns only confident matches (others stay "Sprecher N").
|
||
async function identifySpeakers(recData) {
|
||
const words = recData.words || [];
|
||
const speakers = [...new Set(words.filter(w => w.speaker).map(w => w.speaker))];
|
||
if (!speakers.length || !recData.wav_file) return {};
|
||
const wavPath = path.join(REC_DIR2, recData.wav_file);
|
||
if (!fs.existsSync(wavPath)) return {};
|
||
const vps = await db.listVoiceprints(spk.modelId());
|
||
if (!vps.length) return {};
|
||
const best = {}; let cur = null;
|
||
for (const w of words) {
|
||
if (!cur || cur.speaker !== w.speaker) cur = { speaker: w.speaker, start: w.t_start_ms, end: w.t_end_ms };
|
||
else cur.end = w.t_end_ms;
|
||
const len = cur.end - cur.start;
|
||
if (!best[cur.speaker] || len > best[cur.speaker].len) best[cur.speaker] = { start: cur.start, end: cur.end, len };
|
||
}
|
||
// Cut at most ID_CUT_SEC: onnxruntime ABORTS the embedding helper on very long
|
||
// inputs (reproduced: 180 s -> "Non-zero status code returned while running
|
||
// Where"; 20/60 s fine). On #74 the longest run of a moderator was minutes
|
||
// long, which killed the helper on the first speaker and — through the then
|
||
// poisoned promise chain in spk.js — every speaker after it ("spk beendet"),
|
||
// i.e. voice auto-naming was dead for the whole recording.
|
||
const ID_CUT_SEC = 25;
|
||
const hits = [];
|
||
for (const sp of speakers) {
|
||
const r = best[sp]; if (!r) continue;
|
||
const seg = `/tmp/idseg_${recData.id}_${sp}.wav`;
|
||
const from = r.start / 1000;
|
||
const to = Math.min(r.end / 1000, from + ID_CUT_SEC);
|
||
try {
|
||
await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(from), '-to', String(to), '-ar', '16000', '-ac', '1', seg]);
|
||
const emb = await spk.embed(seg);
|
||
const m = spk.bestMatch(emb, vps);
|
||
if (m) hits.push({ speaker: sp, name: m.name, score: m.score });
|
||
} catch (e) { console.error('identify', sp, e.message); }
|
||
finally { try { fs.unlinkSync(seg); } catch {} }
|
||
}
|
||
// One enrolled person can only BE one cluster: keep the best-scoring match per
|
||
// name (#74 handed "Fabio" to two clusters at once).
|
||
const bestByName = {};
|
||
for (const h of hits) if (!bestByName[h.name] || h.score > bestByName[h.name].score) bestByName[h.name] = h;
|
||
const map = {};
|
||
for (const h of Object.values(bestByName)) map[h.speaker] = h.name;
|
||
return map;
|
||
}
|
||
|
||
// Fold a speaker's longest clean run into the named voiceprint centroid, so the
|
||
// next recording can auto-identify them. Shared by the manual rename popup and
|
||
// the live-marker resolution. Returns true if a sample was learned.
|
||
async function enrollSpeakerVoice(recData, speaker, name) {
|
||
if (!name || !recData.wav_file) return false;
|
||
const seg = longestRun(recData.words, speaker);
|
||
if (!seg || (seg.end - seg.start) < 1500) return false;
|
||
const wavPath = path.join(REC_DIR2, recData.wav_file);
|
||
if (!fs.existsSync(wavPath)) return false;
|
||
const cut = `/tmp/enr_${recData.id}_${speaker}.wav`;
|
||
try {
|
||
await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(seg.start / 1000),
|
||
'-to', String(Math.min(seg.end, seg.start + 20000) / 1000), '-ar', '16000', '-ac', '1', cut]);
|
||
const emb = await spk.embed(cut);
|
||
await db.upsertVoiceprint(name, emb, spk.modelId());
|
||
return true;
|
||
} catch (e) { console.error('enroll failed:', e.message); return false; }
|
||
finally { try { fs.unlinkSync(cut); } catch {} }
|
||
}
|
||
|
||
// Apply the names the user typed DURING the recording to the diarized clusters.
|
||
// Runs after every (re-)diarization, before the LLM analysis, so the summary
|
||
// already talks about real people. Two effects:
|
||
// 1. each resolved cluster gets its typed name (authoritative, manual-level),
|
||
// 2. if ONE name confidently lands on SEVERAL clusters, those clusters are the
|
||
// same person split by the diarizer → they are merged (user-supervised fix
|
||
// for the residual over-segmentation).
|
||
async function applyMarkers(id) {
|
||
const rec = await db.getRecording(id);
|
||
if (!rec || !(rec.markers || []).length || !(rec.words || []).some(w => w.speaker)) return null;
|
||
const resolved = diar.resolveMarkers(rec.words, rec.markers);
|
||
for (const m of resolved) await db.setMarkerSpeaker(m.id, m.speaker, m.conf).catch(() => {});
|
||
|
||
// name -> clusters. NAMING may follow a 'mid' hit, but the cluster MERGE below
|
||
// is destructive and unguarded (no acoustic check at this level), so only a
|
||
// 'high' hit — marker inside the turn, that turn clearly dominant — is allowed
|
||
// to fold two clusters together. #79 lost a speaker exactly this way one layer
|
||
// deeper, in the diarizer's must-link.
|
||
const byName = {};
|
||
for (const m of resolved) {
|
||
if (!m.speaker || m.conf !== 'high') continue;
|
||
(byName[m.name] = byName[m.name] || new Set()).add(m.speaker);
|
||
}
|
||
const nameOnly = {};
|
||
for (const m of resolved) {
|
||
if (!m.speaker || m.conf === 'low' || m.conf === null) continue;
|
||
(nameOnly[m.name] = nameOnly[m.name] || new Set()).add(m.speaker);
|
||
}
|
||
const talk = {};
|
||
for (const w of rec.words) if (w.speaker) talk[w.speaker] = (talk[w.speaker] || 0) + (w.t_end_ms - w.t_start_ms);
|
||
|
||
const fold = {}; // extra cluster -> primary cluster
|
||
const nameOf = {}; // cluster -> typed name
|
||
for (const [name, set] of Object.entries(byName)) {
|
||
const cls = [...set].sort((a, b) => (talk[b] || 0) - (talk[a] || 0));
|
||
const primary = cls[0];
|
||
nameOf[primary] = name;
|
||
for (const c of cls.slice(1)) fold[c] = primary;
|
||
}
|
||
// Naming without merging: a 'mid' hit labels the cluster it landed on, as long
|
||
// as that cluster is not already claimed by somebody else.
|
||
for (const [name, set] of Object.entries(nameOnly)) {
|
||
if (byName[name]) continue;
|
||
const cls = [...set].sort((a, b) => (talk[b] || 0) - (talk[a] || 0));
|
||
if (cls[0] && !nameOf[cls[0]]) nameOf[cls[0]] = name;
|
||
}
|
||
let merged = 0;
|
||
if (Object.keys(fold).length) {
|
||
for (const w of rec.words) if (fold[w.speaker]) w.speaker = fold[w.speaker];
|
||
await db.replaceWords(id, rec.words);
|
||
merged = Object.keys(fold).length;
|
||
console.log(`markers: merged ${merged} cluster(s) into marker-named speakers on #${id}`);
|
||
}
|
||
for (const [sp, name] of Object.entries(nameOf)) await db.setSpeakerName(id, sp, name).catch(() => {});
|
||
// Learn the voices so the NEXT recording identifies them automatically.
|
||
const rec2 = await db.getRecording(id);
|
||
let learned = 0;
|
||
for (const [sp, name] of Object.entries(nameOf)) if (await enrollSpeakerVoice(rec2, sp, name)) learned++;
|
||
const named = Object.keys(nameOf).length;
|
||
await relabelAnalysis(id).catch(() => {});
|
||
console.log(`markers on #${id}: ${resolved.length} marker(s) → ${named} named speaker(s), ${merged} merged, ${learned} voice(s) learned`);
|
||
return { resolved, named, merged, learned };
|
||
}
|
||
|
||
// Run LLM analysis for a recording and persist title/summary/todos/participants/
|
||
// clean dialogue. Background (NPU ~30-90s) so record/stop stays fast.
|
||
// A76 clock cap for the whole run: rkllm burns ~1.7 CPU cores alongside the NPU
|
||
// and a long Gemma generation hit 95-97°C uncapped on the fanless board (#70,
|
||
// 16-min run) — same shutdown risk transcribeWav/rediarize already guard against.
|
||
async function analyzeRecording(id) {
|
||
// Free the resident rkllm (~5.9GB) BEFORE identifySpeakers: the TitaNet
|
||
// spk-embed helper OOM-dies next to it ("spk beendet" per cluster, #71 got
|
||
// 48 raw speakers). llm.analyze() below respawns it (~96s load, noise
|
||
// against a 15-min analysis).
|
||
llm.stopServer();
|
||
await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {});
|
||
try {
|
||
const rec = await db.getRecording(id);
|
||
if (!rec) return;
|
||
const nameMap = await identifySpeakers(rec).catch(() => ({}));
|
||
// Seed the per-recording speaker names from acoustic identification (fills
|
||
// blanks only; manual edits stay). Manual names win for the LLM labelling.
|
||
await db.mergeSpeakerNames(id, nameMap).catch(() => {});
|
||
// Stored names normally win over acoustic identification (they are typed by
|
||
// hand). Exception: a name that came from a NON-'high' marker is only a
|
||
// guess about WHICH cluster was speaking — resolveMarkers looks back up to
|
||
// 25 s, so in a fast four-way discussion it lands on the wrong voice (#79
|
||
// put "Markus" on Sascha's cluster). If an enrolled voiceprint disagrees
|
||
// with such a weak marker name, the voiceprint wins.
|
||
const markers = await db.listMarkers(id).catch(() => []);
|
||
const weakName = new Map();
|
||
for (const m of markers) if (m.speaker && m.conf !== 'high') weakName.set(m.speaker, m.name);
|
||
const names = { ...nameMap, ...(rec.speaker_names || {}) };
|
||
for (const [sp, vname] of Object.entries(nameMap)) {
|
||
if (names[sp] !== vname && weakName.get(sp) === names[sp]) {
|
||
console.log(`identify #${id}: Stimmabdruck "${vname}" schlägt unsicheren Marker "${names[sp]}" auf ${sp}`);
|
||
names[sp] = vname;
|
||
await db.setSpeakerName(id, sp, vname).catch(() => {});
|
||
}
|
||
}
|
||
const glossary = (await db.listGlossary().catch(() => [])).filter(g => g.enabled);
|
||
// Deterministic soundalike replacement: the 3B model is unreliable at obeying
|
||
// a "map Amy->Aime" instruction (it treats "Amy" as a valid name), so entries
|
||
// WITH soundalikes are applied as a hard whole-word swap on the words FIRST —
|
||
// then the LLM sees the correct term and the turn view shows it too. Entries
|
||
// without soundalikes still rely on the (contextual) LLM glossary hint.
|
||
const gmap = glossaryMap(glossary);
|
||
if (Object.keys(gmap).length && Array.isArray(rec.words) && rec.words.length) {
|
||
let changed = false;
|
||
for (const w of rec.words) { const nt = applyGlossaryText(w.word, gmap); if (nt !== w.word) { w.word = nt; changed = true; } }
|
||
if (changed) { await db.replaceWords(id, rec.words); console.log(`glossary: applied soundalike swaps to #${id} words`); }
|
||
}
|
||
const { text, labeled, labels } = buildBasis(rec, names);
|
||
// Remember which label each speaker had in the LLM input, so renaming a
|
||
// speaker later can patch the finished text (relabelAnalysis).
|
||
await db.setSpeakerLabels(id, labels).catch(() => {});
|
||
const a = await llm.analyze(text, { labeled, glossary });
|
||
// Participants = the ACOUSTICALLY diarized speakers (named where known),
|
||
// not the LLM's list — the model returned every name MENTIONED in the
|
||
// conversation as a "Teilnehmer" (11 names on a 3-person call, #69).
|
||
// LLM participants stay only as fallback when there is no diarization.
|
||
const spkOrder = [];
|
||
for (const w of (rec.words || [])) if (w.speaker && !spkOrder.includes(w.speaker)) spkOrder.push(w.speaker);
|
||
// Same labels as in the dialogue the LLM read — otherwise the participant
|
||
// list numbers the speakers differently than the summary talks about them.
|
||
if (spkOrder.length) a.participants = spkOrder.map((sp, i) => names[sp] || labels[sp] || `Sprecher ${i + 1}`);
|
||
// Belt-and-suspenders: also swap in the LLM outputs (cheap, catches any the
|
||
// model reintroduced).
|
||
for (const k of ['title', 'summary', 'transcriptClean']) if (a[k]) a[k] = applyGlossaryText(a[k], gmap);
|
||
a.participants = (a.participants || []).map(p => applyGlossaryText(p, gmap));
|
||
a.todos = (a.todos || []).map(t => applyGlossaryText(t, gmap));
|
||
await db.updateAnalysis(id, { ...a, status: 'done' });
|
||
// A speaker can be named WHILE the analysis runs (it takes ~15 min, and the
|
||
// marker/voiceprint passes themselves rename clusters). The finished text
|
||
// still carries the label from analysis start ("Sprecher 1"), so patch it
|
||
// once at the end — same mechanism as renaming a speaker by hand.
|
||
await relabelAnalysis(id).catch(() => {});
|
||
console.log(`analysis done for #${id}: ${a.title}`);
|
||
} catch (e) {
|
||
console.error(`analysis failed for #${id}:`, e.message);
|
||
await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {});
|
||
} finally {
|
||
await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {});
|
||
}
|
||
}
|
||
|
||
const __dirname = path.dirname(fileURLToPath(import.meta.url));
|
||
const app = express();
|
||
app.use(express.json());
|
||
|
||
const REC_DIR = path.join(__dirname, 'recordings');
|
||
app.use('/media', express.static(REC_DIR));
|
||
app.use(express.static(path.join(__dirname, 'web/dist')));
|
||
|
||
const wrap = fn => (req, res) => fn(req, res).catch(e => res.status(500).json({ error: String(e && e.message || e) }));
|
||
|
||
app.get('/api/health', (req, res) => res.json({ ok: true, ...rec.status() }));
|
||
|
||
// --- Bluetooth ---
|
||
app.post('/api/bt/scan', wrap(async (req, res) => {
|
||
const secs = Math.min(+(req.body && req.body.seconds) || 10, 30);
|
||
res.json(await bt.scan(secs));
|
||
}));
|
||
app.get('/api/bt/devices', wrap(async (req, res) => res.json(await bt.knownDevices())));
|
||
app.post('/api/bt/pair', wrap(async (req, res) => res.json(await bt.pair(req.body.mac))));
|
||
app.post('/api/bt/connect', wrap(async (req, res) => res.json(await bt.connect(req.body.mac))));
|
||
app.post('/api/bt/repair', wrap(async (req, res) => res.json(await bt.repair(req.body.mac))));
|
||
app.post('/api/bt/disconnect', wrap(async (req, res) => res.json(await bt.disconnect(req.body.mac))));
|
||
|
||
// --- Audio ---
|
||
app.get('/api/audio/sources', wrap(async (req, res) => res.json(await rec.listSources())));
|
||
app.post('/api/audio/headset-mic', wrap(async (req, res) => res.json(await rec.enableHeadsetMic((req.body && req.body.codec) || 'auto'))));
|
||
|
||
// --- Global NPU pipeline serializer ---------------------------------------
|
||
// Transcription (whisper-turbo ENCODER on the NPU) and LLM analysis (rkllm on
|
||
// the NPU) cannot run simultaneously: while the whisper encoder holds the NPU,
|
||
// rkllm_init fails with "failed to allocate handle, ret:-1, errno:14, Bad
|
||
// address", which surfaces as "rkllm READY timeout" and a failed/empty
|
||
// analysis. This bit when two recordings finished within seconds of each other
|
||
// (both fired transcribe+analyze fire-and-forget with no coordination). Every
|
||
// heavy per-recording job (transcribe / analyze / re-diarize) now runs through
|
||
// this single FIFO queue, so exactly one touches the NPU at a time; a second
|
||
// recording simply waits its turn instead of colliding.
|
||
let npuChain = Promise.resolve();
|
||
let npuDepth = 0;
|
||
function withNpuLock(label, fn) {
|
||
npuDepth++;
|
||
const waiting = npuDepth - 1;
|
||
const run = npuChain.then(async () => {
|
||
if (waiting > 0) console.log(`npu-queue: ${label} start (${waiting} ahead done)`);
|
||
try { return await fn(); }
|
||
finally { npuDepth--; }
|
||
});
|
||
npuChain = run.catch(() => {}); // keep the chain alive across a job's failure
|
||
return run;
|
||
}
|
||
|
||
// --- Recording ---
|
||
app.post('/api/record/start', wrap(async (req, res) => res.json(await rec.startRecording(req.body || {}))));
|
||
// Live speaker markers: type the name of whoever is talking WHILE recording.
|
||
app.post('/api/record/marker', wrap(async (req, res) => res.json(rec.addMarker((req.body && req.body.name) || ''))));
|
||
app.get('/api/record/markers', (req, res) => res.json(rec.listMarkers()));
|
||
app.delete('/api/record/marker/:index', wrap(async (req, res) => res.json({ markers: rec.removeMarker(req.params.index) })));
|
||
// Transcribe (thermal-guarded whisper + diarize) then analyze, in the background.
|
||
// The wav + DB row already exist, so a crash here loses nothing — the recording
|
||
// stays with status 'transcribing' and can be recovered via /retranscribe.
|
||
async function transcribeAndAnalyze(id, wavFile, numSpeakers) {
|
||
await db.markAnalysisStart(id).catch(() => {});
|
||
try {
|
||
// Live markers constrain the diarization itself (same name = one speaker,
|
||
// different names = never merged), so they are loaded before transcribing.
|
||
const markers = await db.listMarkers(id).catch(() => []);
|
||
// Enrolled voices do the same for people who were never typed — the box
|
||
// owner above all (see diar.clusterWindows).
|
||
const voiceprints = await db.listVoiceprints(spk.modelId()).catch(() => []);
|
||
const { transcript, words } = await rec.transcribeWav(wavFile, numSpeakers, { markers, voiceprints });
|
||
await db.setTranscript(id, transcript, words);
|
||
console.log(`transcription done for #${id}: ${words.length} segments`);
|
||
// Names typed live → cluster names (+ user-supervised cluster merge) BEFORE
|
||
// the LLM sees the dialogue.
|
||
await applyMarkers(id).catch(e => console.error('applyMarkers:', e.message));
|
||
} catch (e) {
|
||
console.error(`transcription failed for #${id}:`, e.message);
|
||
await db.setStatus(id, 'error').catch(() => {});
|
||
// also mark analysis 'error' so the UI stops polling and shows a retry
|
||
await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'error' }).catch(() => {});
|
||
return; // no transcript -> skip LLM analysis
|
||
}
|
||
analyzeRecording(id);
|
||
}
|
||
|
||
app.post('/api/record/stop', wrap(async (req, res) => {
|
||
// Persist wav + mp3 + DB row FIRST (crash-safe), respond immediately, then
|
||
// transcribe/analyze in the background. The frontend already polls while
|
||
// analysis_status === 'pending', so the transcript + analysis fill in live.
|
||
const r = await rec.stopAndSave();
|
||
const id = await db.addRecording({ ...r, transcript: '', status: 'transcribing' });
|
||
await db.addMarkers(id, r.markers).catch(e => console.error('addMarkers:', e.message));
|
||
res.json({ id, title: r.title, mp3_file: r.mp3_file, duration_ms: r.duration_ms,
|
||
transcript: '', status: 'transcribing', analysis_status: 'pending',
|
||
markers: (r.markers || []).length });
|
||
withNpuLock(`transcribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, r.numSpeakers));
|
||
}));
|
||
|
||
// --- Recordings ---
|
||
app.get('/api/recordings', wrap(async (req, res) => res.json(await db.listRecordings())));
|
||
app.get('/api/recordings/:id', wrap(async (req, res) => {
|
||
const r = await db.getRecording(+req.params.id);
|
||
r ? res.json(r) : res.status(404).json({ error: 'not found' });
|
||
}));
|
||
app.delete('/api/recordings/:id', wrap(async (req, res) => { await db.deleteRecording(+req.params.id); res.json({ ok: true }); }));
|
||
|
||
// --- Voiceprints / enrollment ---
|
||
app.get('/api/voiceprints', wrap(async (req, res) => {
|
||
const vps = await db.listVoiceprints();
|
||
const cur = spk.modelId();
|
||
res.json(vps.map(v => ({ id: v.id, name: v.name, dim: v.dim, model: v.model,
|
||
stale: v.model !== cur, created_at: v.created_at })));
|
||
}));
|
||
app.delete('/api/voiceprints/:id', wrap(async (req, res) => { await db.deleteVoiceprint(+req.params.id); res.json({ ok: true }); }));
|
||
app.post('/api/enroll/start', wrap(async (req, res) => res.json(await rec.startRecording({ source: req.body && req.body.source }))));
|
||
app.post('/api/enroll/stop', wrap(async (req, res) => {
|
||
const name = (req.body && req.body.name || '').trim();
|
||
if (!name) return res.status(400).json({ error: 'Name fehlt' });
|
||
const wav = await rec.stopRaw();
|
||
try {
|
||
const emb = await spk.embed(wav);
|
||
const id = await db.addVoiceprint(name, emb, spk.modelId());
|
||
res.json({ id, name, dim: emb.length });
|
||
} finally { try { fs.unlinkSync(wav); } catch {} }
|
||
}));
|
||
|
||
// --- Glossary (known proper nouns / terms; fed into LLM correction) ---
|
||
app.get('/api/glossary', wrap(async (req, res) => res.json(await db.listGlossary())));
|
||
app.post('/api/glossary', wrap(async (req, res) => {
|
||
const term = (req.body && req.body.term || '').trim();
|
||
const soundalikes = (req.body && req.body.soundalikes || '').trim();
|
||
if (!term) return res.status(400).json({ error: 'Begriff fehlt' });
|
||
res.json(await db.addGlossary(term, soundalikes));
|
||
}));
|
||
app.delete('/api/glossary/:id', wrap(async (req, res) => { await db.deleteGlossary(+req.params.id); res.json({ ok: true }); }));
|
||
app.put('/api/glossary/:id', wrap(async (req, res) => {
|
||
const term = (req.body && req.body.term || '').trim();
|
||
const soundalikes = (req.body && req.body.soundalikes || '').trim();
|
||
if (!term) return res.status(400).json({ error: 'Begriff fehlt' });
|
||
const g = await db.updateGlossary(+req.params.id, term, soundalikes);
|
||
if (!g) return res.status(404).json({ error: 'Eintrag nicht gefunden' });
|
||
res.json(g);
|
||
}));
|
||
|
||
// --- LLM ---
|
||
app.get('/api/llm/status', (req, res) => res.json(llm.status()));
|
||
app.get('/api/spk/status', (req, res) => res.json(spk.status()));
|
||
// Re-diarize a recording's words with an (optional) fixed speaker count, then
|
||
// persist the new speaker labels. Text/timestamps are untouched. Needs the wav.
|
||
async function rediarize(id, numSpeakers) {
|
||
const r = await db.getRecording(id);
|
||
if (!r || !r.wav_file) throw new Error('wav für Re-Diarisierung fehlt');
|
||
const wav = path.join(REC_DIR, r.wav_file);
|
||
if (!fs.existsSync(wav)) throw new Error('wav-Datei nicht gefunden: ' + r.wav_file);
|
||
const words = r.words || [];
|
||
// Free the resident rkllm first — diar's consolidation embeds every cluster
|
||
// via TitaNet, which OOM-dies next to a resident 5.9GB LLM (see
|
||
// analyzeRecording; on #71 that silently yielded 48 raw speakers).
|
||
llm.stopServer();
|
||
// Diarization runs on the A76 cores; the board is fanless, so cap the clock
|
||
// for the whole run (transcribeWav does this for the record path — rediarize
|
||
// is the other entry point and must do it too, or a long re-diar overheats).
|
||
await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '1608000']).catch(() => {});
|
||
try {
|
||
// Enrolled voices go IN, not just out: they keep two known people (above all
|
||
// the box owner, who is enrolled and talks in every recording) from ending up
|
||
// in one cluster. See the cannot-link block in diar.clusterWindows.
|
||
const voiceprints = await db.listVoiceprints(spk.modelId()).catch(() => []);
|
||
const segs = await diar.diarize(wav, numSpeakers, { markers: r.markers || [], voiceprints });
|
||
diar.assignSpeakers(words, segs);
|
||
await db.replaceWords(id, words);
|
||
// Re-diarization RENUMBERS the clusters, so every previously stored name
|
||
// ("speaker_02" = Fabio) would now point at a different voice. Drop them; the
|
||
// live markers and the voiceprint identification re-fill them right after.
|
||
await db.pool.query('UPDATE recordings SET speaker_names=NULL WHERE id=?', [id]).catch(() => {});
|
||
console.log(`re-diarized #${id} (numSpeakers=${numSpeakers || 'auto'}) → ${new Set(segs.map(s => s.speaker)).size} speakers`);
|
||
} finally {
|
||
await pexec('sudo', ['-n', '/opt/callrec/aime-cpucap.sh', '2256000']).catch(() => {});
|
||
}
|
||
}
|
||
|
||
// Re-run analysis (e.g. the model wasn't downloaded yet at record time). Optional
|
||
// body { numSpeakers } re-diarizes first (fixes over-segmentation on a known count).
|
||
app.post('/api/recordings/:id/reanalyze', wrap(async (req, res) => {
|
||
const id = +req.params.id;
|
||
const r = await db.getRecording(id);
|
||
if (!r) return res.status(404).json({ error: 'not found' });
|
||
const numSpeakers = +(req.body && req.body.numSpeakers) || 0;
|
||
await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
|
||
await db.markAnalysisStart(id).catch(() => {});
|
||
res.json({ ok: true, status: 'pending', reDiarize: true });
|
||
// Always re-diarize: numSpeakers>0 pins the count, 0 = auto (now consolidated,
|
||
// so auto no longer over-segments). This is the "fix my speakers" path.
|
||
withNpuLock(`reanalyze #${id}`, async () => {
|
||
try { await rediarize(id, numSpeakers); }
|
||
catch (e) { console.error(`re-diarize failed for #${id}:`, e.message); }
|
||
await applyMarkers(id).catch(e => console.error('applyMarkers:', e.message));
|
||
await analyzeRecording(id);
|
||
});
|
||
}));
|
||
|
||
// --- markers of a stored recording -----------------------------------------
|
||
app.get('/api/recordings/:id/markers', wrap(async (req, res) => res.json(await db.listMarkers(+req.params.id))));
|
||
// Add a marker after the fact (e.g. while listening back) — same resolution path.
|
||
app.post('/api/recordings/:id/markers', wrap(async (req, res) => {
|
||
const name = String((req.body && req.body.name) || '').trim();
|
||
const tMs = +(req.body && req.body.t_ms);
|
||
if (!name || !Number.isFinite(tMs)) return res.status(400).json({ error: 'name/t_ms fehlt' });
|
||
await db.addMarker(+req.params.id, tMs, name);
|
||
res.json({ ok: true, markers: await db.listMarkers(+req.params.id) });
|
||
}));
|
||
app.delete('/api/recordings/:id/markers/:mid', wrap(async (req, res) => {
|
||
await db.deleteMarker(+req.params.mid);
|
||
res.json({ ok: true, markers: await db.listMarkers(+req.params.id) });
|
||
}));
|
||
// Re-resolve the markers against the current diarization (no LLM, no sherpa).
|
||
app.post('/api/recordings/:id/apply-markers', wrap(async (req, res) => {
|
||
const out = await applyMarkers(+req.params.id);
|
||
const r = await db.getRecording(+req.params.id);
|
||
res.json({ ok: true, ...(out || { named: 0, merged: 0, learned: 0 }),
|
||
speaker_names: (r && r.speaker_names) || {}, markers: (r && r.markers) || [] });
|
||
}));
|
||
|
||
// LLM-only re-analysis: re-run title/summary/todos on the EXISTING words +
|
||
// speaker labels, WITHOUT re-diarizing (fast — skips the ~40min sherpa pass when
|
||
// the speakers are already correct and only the LLM/prompts changed).
|
||
app.post('/api/recordings/:id/reanalyze-llm', wrap(async (req, res) => {
|
||
const id = +req.params.id;
|
||
const r = await db.getRecording(id);
|
||
if (!r) return res.status(404).json({ error: 'not found' });
|
||
await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
|
||
await db.markAnalysisStart(id).catch(() => {});
|
||
res.json({ ok: true, status: 'pending', reDiarize: false });
|
||
withNpuLock(`reanalyze-llm #${id}`, async () => { await analyzeRecording(id); });
|
||
}));
|
||
|
||
// Re-run transcription from the saved wav (recovery: a crash/thermal-shutdown
|
||
// during whisper leaves the row at status 'transcribing' with an empty
|
||
// transcript, or 'error'; the wav is on disk). Optional { numSpeakers }.
|
||
app.post('/api/recordings/:id/retranscribe', wrap(async (req, res) => {
|
||
const id = +req.params.id;
|
||
const r = await db.getRecording(id);
|
||
if (!r) return res.status(404).json({ error: 'not found' });
|
||
if (!r.wav_file) return res.status(400).json({ error: 'kein wav für Re-Transkription' });
|
||
const numSpeakers = +(req.body && req.body.numSpeakers) || 0;
|
||
await db.setStatus(id, 'transcribing');
|
||
await db.updateAnalysis(id, { title: null, summary: null, todos: null, status: 'pending' });
|
||
res.json({ ok: true, status: 'transcribing' });
|
||
withNpuLock(`retranscribe #${id}`, () => transcribeAndAnalyze(id, r.wav_file, numSpeakers));
|
||
}));
|
||
|
||
// Longest contiguous run of one diarization cluster (best clean sample of a voice).
|
||
function longestRun(words, speaker) {
|
||
let best = null, cur = null;
|
||
for (const w of words || []) {
|
||
if (w.speaker !== speaker) { cur = null; continue; }
|
||
if (!cur) cur = { start: w.t_start_ms, end: w.t_end_ms };
|
||
else cur.end = w.t_end_ms;
|
||
if (!best || (cur.end - cur.start) > (best.end - best.start)) best = { start: cur.start, end: cur.end };
|
||
}
|
||
return best;
|
||
}
|
||
|
||
// Name a diarized speaker in a recording and (default) LEARN their voice: fold
|
||
// the speaker's longest clean segment into the named voiceprint centroid, so
|
||
// future recordings auto-identify them. Empty name clears the label.
|
||
app.post('/api/recordings/:id/speaker', wrap(async (req, res) => {
|
||
const id = +req.params.id;
|
||
const speaker = String((req.body && req.body.speaker) || '').trim();
|
||
const name = String((req.body && req.body.name) || '').trim();
|
||
const enroll = !(req.body && req.body.enroll === false);
|
||
if (!speaker) return res.status(400).json({ error: 'speaker fehlt' });
|
||
const r = await db.getRecording(id);
|
||
if (!r) return res.status(404).json({ error: 'not found' });
|
||
await db.setSpeakerName(id, speaker, name);
|
||
// Patch the name into the already generated summary/todos/participants.
|
||
const relabel = await relabelAnalysis(id).catch(e => { console.error('relabel:', e.message); return { changed: 0 }; });
|
||
|
||
let enrolled = false, learnMsg = null;
|
||
if (enroll && name && r.wav_file) {
|
||
const seg = longestRun(r.words, speaker);
|
||
const wavPath = path.join(REC_DIR, r.wav_file);
|
||
if (!seg || (seg.end - seg.start) < 1500) {
|
||
learnMsg = 'Segment zu kurz zum Lernen (<1,5s)';
|
||
} else if (!fs.existsSync(wavPath)) {
|
||
learnMsg = 'wav fehlt – nicht gelernt';
|
||
} else {
|
||
const cut = `/tmp/enr_${id}_${speaker}.wav`;
|
||
try {
|
||
await pexec('ffmpeg', ['-y', '-i', wavPath, '-ss', String(seg.start / 1000),
|
||
'-to', String(seg.end / 1000), '-ar', '16000', '-ac', '1', cut]);
|
||
const emb = await spk.embed(cut);
|
||
await db.upsertVoiceprint(name, emb, spk.modelId());
|
||
enrolled = true;
|
||
} catch (e) { console.error('enroll-from-recording failed:', e.message); learnMsg = 'Lernen fehlgeschlagen'; }
|
||
finally { try { fs.unlinkSync(cut); } catch {} }
|
||
}
|
||
}
|
||
const r2 = await db.getRecording(id);
|
||
res.json({ ok: true, speaker_names: r2.speaker_names || {}, enrolled, learnMsg,
|
||
relabeled: relabel.changed });
|
||
}));
|
||
|
||
app.get(/^(?!\/api|\/media).*/, (req, res) => res.sendFile(path.join(__dirname, 'web/dist/index.html')));
|
||
|
||
const PORT = process.env.PORT || 3000;
|
||
db.ensureSchema()
|
||
.catch(e => console.error('ensureSchema:', e.message))
|
||
.finally(() => app.listen(PORT, '0.0.0.0', () => console.log(`callrec listening on :${PORT}`)));
|