297 lines
12 KiB
Python
Executable File
297 lines
12 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""STT + Speaker Diarization pipeline (frigate.vpn).
|
|
ASR: whisper.cpp (Vulkan) | Diarization: pyannote Community-1 (CPU)
|
|
Uso: transcribe.py <audio> [--lang it] [--model large-v3-turbo] [--out base]
|
|
[--prompt \"frase personale\"] [--glossary glossario.json]
|
|
"""
|
|
import argparse, json, os, re, subprocess, sys, tempfile
|
|
|
|
WHISPER_CLI = os.environ.get('WHISPER_CLI', os.path.expanduser('~/dev/whisper.cpp/build/bin/whisper-cli'))
|
|
MODELS_DIR = os.environ.get('MODELS_DIR', os.path.expanduser('~/dev/whisper.cpp/models'))
|
|
PYTHON = os.environ.get('STT_PYTHON', os.path.expanduser('~/dev/stt-venv/bin/python'))
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Trim del silenzio iniziale/finale (ottimizzazione)
|
|
# ---------------------------------------------------------------------------
|
|
# Cancella le parti vuote prima della prima voce e dopo l'ultima: meno audio
|
|
# da processare (spettrogramma + pyannote), meno rischio di allucinazioni.
|
|
# VAD a energia (RMS) in puro numpy — nessuna dipendenza aggiuntiva.
|
|
# Configurabile: TRIM_SILENCE=0 per disattivare, TRIM_THRESHOLD_DB (default
|
|
# -40), TRIM_PADDING_MS (default 150). I timestamp restano allineati alla
|
|
# timeline originale (offset somato ai segmenti).
|
|
|
|
try:
|
|
import numpy as _np
|
|
import soundfile as _sf
|
|
_HAS_NP = True
|
|
except ImportError:
|
|
_HAS_NP = False
|
|
|
|
|
|
def _frame_rms(samples, frame):
|
|
n = (len(samples) // frame) * frame
|
|
return _np.sqrt(_np.mean(samples[:n].reshape(-1, frame) ** 2, axis=1))
|
|
|
|
|
|
def detect_trim_bounds(samples, sr, frame_ms=30, threshold_db=-40.0,
|
|
min_speech_ms=300):
|
|
"""Trova (start_sample, end_sample) del parlato (primo/ultimo frame con
|
|
RMS >= soglia). None se non c'è parlato sufficiente."""
|
|
frame = max(1, int(sr * frame_ms / 1000))
|
|
thr = 10 ** (threshold_db / 20)
|
|
rms = _frame_rms(samples, frame)
|
|
speech = rms >= thr
|
|
if not speech.any():
|
|
return None
|
|
min_frames = max(1, int(min_speech_ms / frame_ms))
|
|
# Indici dei frame sopra soglia
|
|
idx = _np.where(speech)[0]
|
|
# Gruppi contigui (per ignorare picchi spuri isolati)
|
|
groups = _np.split(idx, _np.where(_np.diff(idx) > 1)[0] + 1)
|
|
long = [g for g in groups if len(g) >= min_frames]
|
|
if not long:
|
|
return None
|
|
start = long[0][0] * frame
|
|
end = (long[-1][-1] + 1) * frame
|
|
return int(start), int(end)
|
|
|
|
|
|
def trim_silence(audio_path):
|
|
"""Ritorna (path_wav_ripulito, offset_secondi) usando un file temporaneo
|
|
accanto all'originale; None se il silenzio iniziale/finale è già minimo
|
|
(o mancano numpy/soundfile, o non c'è parlato)."""
|
|
if not _HAS_NP:
|
|
return None
|
|
try:
|
|
threshold_db = float(os.environ.get('TRIM_THRESHOLD_DB', '-40'))
|
|
padding_ms = int(os.environ.get('TRIM_PADDING_MS', '150'))
|
|
data, sr = _sf.read(audio_path, dtype='float32')
|
|
if data.ndim > 1:
|
|
data = data.mean(axis=1)
|
|
if len(data) == 0:
|
|
return None
|
|
bounds = detect_trim_bounds(data, sr, threshold_db=threshold_db)
|
|
if bounds is None:
|
|
# Nessun parlato: non trascrivere il vuoto
|
|
return ('', 0.0)
|
|
start, end = bounds
|
|
pad = int(padding_ms / 1000 * sr)
|
|
start = max(0, start - pad)
|
|
end = min(len(data), end + pad)
|
|
# Se il taglio è trascurabile, lascia stare
|
|
if start < 0.05 * sr and end > len(data) - 0.05 * sr:
|
|
return None
|
|
out_path = audio_path + '.trim.wav'
|
|
_sf.write(out_path, data[start:end], sr)
|
|
return (out_path, start / sr)
|
|
except Exception:
|
|
return None
|
|
|
|
|
|
# ---------------------------------------------------------------------------
|
|
# Glossario personale: estensione vocabolario + normalizzazione deterministica
|
|
# ---------------------------------------------------------------------------
|
|
# Formato (JSON):
|
|
# {
|
|
# "people": [{"canonical": "Giulia Bianchi", "aliases": ["Giulia Bianci", ...]}],
|
|
# "organizations": [{"canonical": "...", "aliases": [...]}],
|
|
# "terms": [{"canonical": "fine-tuning", "aliases": ["fine tuning", "fain tuning"]}]
|
|
# }
|
|
# La normalizzazione sostituisce gli alias con la forma canonica SOLO su match
|
|
# esatto (case-insensitive, confini di parola). Le sostituzioni sono annotate
|
|
# nel campo "raw" del segmento per audit/rollback.
|
|
|
|
def load_glossary(path_or_json):
|
|
"""Accetta un path JSON o una stringa JSON; restituisce la mappa
|
|
alias->canonical (alias in lowercase). Invalido/vuoto -> {}."""
|
|
if not path_or_json:
|
|
return {}
|
|
try:
|
|
data = path_or_json
|
|
if isinstance(data, str):
|
|
stripped = data.strip()
|
|
if stripped.startswith('{'):
|
|
data = json.loads(stripped) # stringa JSON
|
|
else:
|
|
with open(stripped, encoding='utf-8') as f: # path
|
|
data = json.load(f)
|
|
flat = {}
|
|
for group in ('people', 'organizations', 'terms', 'custom'):
|
|
for entry in (data.get(group) or []):
|
|
canon = str(entry.get('canonical') or '').strip()
|
|
if not canon:
|
|
continue
|
|
for alias in (entry.get('aliases') or []):
|
|
alias = str(alias).strip()
|
|
if alias and alias.lower() != canon.lower():
|
|
flat[alias.lower()] = canon
|
|
# Formato alternativo semplice: {"alias": "canonical"}
|
|
if not flat:
|
|
for alias, canon in data.items():
|
|
if isinstance(canon, str) and isinstance(alias, str):
|
|
flat[alias.lower()] = canon
|
|
return flat
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
def normalize_text(text, flat_map):
|
|
"""Sostituisce alias (case-insensitive, confini parola) con la forma
|
|
canonica. Le sostituzioni avvengono su segnaposto univoci: un alias non
|
|
può ri-matchare testo generato da un'altra sostituzione (no cascate).
|
|
Restituisce (nuovo_testo, sostituzioni_fatte)."""
|
|
if not flat_map or not text:
|
|
return text, 0
|
|
out, count = text, 0
|
|
placeholders = []
|
|
# Passo 1: alias -> segnaposto (alias più lunghi prima, evita match parziali)
|
|
for alias in sorted(flat_map, key=len, reverse=True):
|
|
canon = flat_map[alias]
|
|
pattern = r'(?<!\w)' + re.escape(alias) + r'(?!\w)'
|
|
ph = f'\x00PH{len(placeholders)}\x00'
|
|
new_out, n = re.subn(pattern, ph, out, flags=re.IGNORECASE)
|
|
if n:
|
|
placeholders.append((ph, canon))
|
|
count += n
|
|
out = new_out
|
|
# Passo 2: segnaposto -> forma canonica
|
|
for ph, canon in placeholders:
|
|
out = out.replace(ph, canon)
|
|
return out, count
|
|
|
|
|
|
def asr(audio, model, lang, tmp, prompt=None):
|
|
model_path = os.path.join(MODELS_DIR, f'ggml-{model}.bin')
|
|
out_json = os.path.join(tmp, 'asr.json')
|
|
vad_model = os.path.join(MODELS_DIR, 'ggml-silero-v6.2.0.bin')
|
|
# Trim del silenzio iniziale/finale (ottimizzazione, default attivo)
|
|
src = audio
|
|
offset = 0.0
|
|
if os.environ.get('TRIM_SILENCE', '1') != '0':
|
|
res = trim_silence(audio)
|
|
if res is not None:
|
|
trimmed, offset = res
|
|
if not trimmed:
|
|
return [] # nessun parlato: niente da trascrivere
|
|
src = trimmed
|
|
cmd = [WHISPER_CLI, '-m', model_path, '-f', src, '-l', lang,
|
|
'-oj', '-of', os.path.join(tmp, 'asr'), '--no-prints',
|
|
'-vm', vad_model, '--vad',
|
|
'--suppress-regex',
|
|
r'(Grazie a tutti|Grazie per l.attenzione|Thank you|Thanks for watching|Sottotitoli creati|Sottotitoli)']
|
|
# Initial prompt (bias vocabolario personale, nomi, gergo): boost
|
|
# zero-training. --carry-initial-prompt lo riapplica su ogni finestra
|
|
# (necessario per audio lunghi; verificare con whisper-cli -h).
|
|
if prompt:
|
|
cmd += ['--prompt', prompt, '--carry-initial-prompt']
|
|
subprocess.run(cmd, check=True, capture_output=True)
|
|
data = json.load(open(out_json))
|
|
segs = []
|
|
for s in data['transcription']:
|
|
text = s['text'].strip()
|
|
# Filtro finale: scarta segmenti che sono solo frasi allucinate
|
|
# (rete di sicurezza oltre a VAD e suppress-regex)
|
|
if text.lower() in {'grazie', 'grazie a tutti', 'thank you', 'thanks',
|
|
"grazie per l'attenzione", 'sottotitoli',
|
|
'sottotitoli creati', 'sottotitoli creati da'}:
|
|
continue
|
|
# Riallinea i timestamp alla timeline originale (dopo il trim)
|
|
segs.append({'start': s['offsets']['from']/1000.0 + offset,
|
|
'end': s['offsets']['to']/1000.0 + offset,
|
|
'text': text})
|
|
return segs
|
|
|
|
|
|
def apply_glossary(segs, flat_map):
|
|
"""Normalizza i testi dei segmenti con il glossario. I segmenti modificati
|
|
ricevono 'raw' = testo ASR originale (audit/rollback)."""
|
|
for seg in segs:
|
|
new_text, count = normalize_text(seg['text'], flat_map)
|
|
if count:
|
|
seg['raw'] = seg['text']
|
|
seg['text'] = new_text
|
|
return segs
|
|
|
|
|
|
def diarize(audio, tmp):
|
|
script = f'''
|
|
import json, sys
|
|
import numpy as np, soundfile as sf, torch
|
|
from pyannote.audio import Pipeline
|
|
data, sr = sf.read({audio!r}, dtype='float32')
|
|
wav = torch.from_numpy(data.T if data.ndim > 1 else data[None, :]).float()
|
|
p = Pipeline.from_pretrained('pyannote/speaker-diarization-community-1')
|
|
out = p({{'waveform': wav, 'sample_rate': sr}})
|
|
diar = out.speaker_diarization if hasattr(out, 'speaker_diarization') else out
|
|
turns = [{{'start': t.start, 'end': t.end, 'speaker': sp}}
|
|
for t, _, sp in diar.itertracks(yield_label=True)]
|
|
json.dump(turns, open({os.path.join(tmp, 'diar.json')!r}, 'w'))
|
|
'''
|
|
sp = os.path.join(tmp, 'diar.py')
|
|
open(sp, 'w').write(script)
|
|
subprocess.run([PYTHON, sp], check=True)
|
|
return json.load(open(os.path.join(tmp, 'diar.json')))
|
|
|
|
def overlap(a, b):
|
|
return max(0.0, min(a['end'], b['end']) - max(a['start'], b['start']))
|
|
|
|
def merge(segs, turns):
|
|
for s in segs:
|
|
best, best_ov = None, 0.0
|
|
for t in turns:
|
|
ov = overlap(s, t)
|
|
if ov > best_ov:
|
|
best, best_ov = t['speaker'], ov
|
|
s['speaker'] = best if best_ov > 0.05 else 'UNKNOWN'
|
|
return segs
|
|
|
|
def to_srt(segs, path):
|
|
def ts(x):
|
|
h, m = int(x//3600), int(x%3600//60)
|
|
s, ms = int(x%60), int((x-int(x))*1000)
|
|
return f'{h:02d}:{m:02d}:{s:02d},{ms:03d}'
|
|
with open(path, 'w') as f:
|
|
for i, s in enumerate(segs, 1):
|
|
f.write(f'{i}\n{ts(s["start"])} --> {ts(s["end"])}\n')
|
|
f.write(f'[{s["speaker"]}] {s["text"]}\n\n')
|
|
|
|
def main():
|
|
ap = argparse.ArgumentParser()
|
|
ap.add_argument('audio')
|
|
ap.add_argument('--lang', default='it')
|
|
ap.add_argument('--model', default='large-v3-turbo')
|
|
ap.add_argument('--out', default=None)
|
|
ap.add_argument('--prompt', default=None,
|
|
help='initial prompt personale (nomi, termini, gergo)')
|
|
ap.add_argument('--glossary', default=None,
|
|
help='path a glossario JSON (alias -> forma canonica)')
|
|
ap.add_argument('--no-trim', action='store_true',
|
|
help='disattiva la cancellazione del silenzio iniziale/finale')
|
|
args = ap.parse_args()
|
|
if args.no_trim:
|
|
os.environ['TRIM_SILENCE'] = '0'
|
|
base = args.out or os.path.splitext(args.audio)[0]
|
|
flat_map = load_glossary(args.glossary)
|
|
with tempfile.TemporaryDirectory() as tmp:
|
|
print(f'[1/3] ASR whisper.cpp ({args.model}, {args.lang})...', file=sys.stderr)
|
|
segs = asr(args.audio, args.model, args.lang, tmp, prompt=args.prompt)
|
|
print(f' {len(segs)} segmenti', file=sys.stderr)
|
|
if flat_map:
|
|
segs = apply_glossary(segs, flat_map)
|
|
print(f' glossario: {len(flat_map)} alias applicati', file=sys.stderr)
|
|
print('[2/3] Diarization pyannote Community-1...', file=sys.stderr)
|
|
turns = diarize(args.audio, tmp)
|
|
print(f' {len(turns)} turni parlante', file=sys.stderr)
|
|
print('[3/3] Merge ASR + diarization...', file=sys.stderr)
|
|
segs = merge(segs, turns)
|
|
out_json = base + '.json'
|
|
out_srt = base + '.srt'
|
|
json.dump({'segments': segs}, open(out_json, 'w'), ensure_ascii=False, indent=2)
|
|
to_srt(segs, out_srt)
|
|
print(f'OK: {out_json} + {out_srt}', file=sys.stderr)
|
|
for s in segs:
|
|
print(f'{s["start"]:6.2f}-{s["end"]:6.2f} [{s["speaker"]}] {s["text"]}')
|
|
|
|
if __name__ == '__main__':
|
|
main()
|