28 KiB
28 KiB
In [5]:
# 1. Opzioni
MODELLO = "large-v3" # "large-v3" per qualità; "turbo" per maggiore velocità
LINGUA = "it" # "it" per italiano; None per rilevamento automatico
PRECISIONE = "float16" # adatta alla T4; "int8_float16" usa meno memoria
CONTESTO = "" # nomi/termini da riconoscere bene, es. "Kiev, Chisinau, Bucarest"
ORIGINE = "upload" # "upload" oppure "drive"
CARTELLA_DRIVE = "/content/drive/MyDrive/Audio da trascrivere"
SALVA_SU_DRIVE = False # True per conservare i risultati anche su Drive
CARTELLA_RISULTATI_DRIVE = "/content/drive/MyDrive/Trascrizioni"
FILTRO_SILENZI = False # False: nessun filtro VAD; True: filtra il non-parlato stimato
MINUTI_PER_SEZIONE = 5
SOGLIA_PAROLA_INCERTA = 0.45
SCARICA_ZIP = True
In [6]:
# 2. Dipendenze
import subprocess, sys
gpu = subprocess.run(["nvidia-smi", "--query-gpu=name,memory.total", "--format=csv,noheader"],
capture_output=True, text=True) if __import__("shutil").which("nvidia-smi") else None
if gpu is None or gpu.returncode != 0:
raise RuntimeError("GPU assente. Seleziona Runtime > Cambia tipo di runtime > GPU T4, poi riesegui tutto.")
print("GPU assegnata:", gpu.stdout.strip())
if "T4" not in gpu.stdout:
print("Colab ha assegnato una GPU diversa dalla T4; il notebook userà quella disponibile.")
subprocess.check_call([sys.executable, "-m", "pip", "install", "-q",
"faster-whisper==1.2.1", "ctranslate2>=4.5,<5", "av>=11,<16",
"nvidia-cublas-cu12", "nvidia-cudnn-cu12>=9,<10"])
print("Installazione completata.")
GPU assegnata: Tesla T4, 15360 MiB Installazione completata.
In [ ]:
# 3. Sorgenti e cartelle
from pathlib import Path
from datetime import datetime, timezone
import json, os, shutil
from google.colab import files, drive
if ORIGINE not in {"upload", "drive"}:
raise ValueError("ORIGINE deve essere 'upload' oppure 'drive'.")
if ORIGINE == "drive" or SALVA_SU_DRIVE:
drive.mount("/content/drive")
RUN_ID = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M%S_%f")
RUN_DIR = Path("/content/trascrizioni") / RUN_ID
RUN_DIR.mkdir(parents=True, exist_ok=True)
if ORIGINE == "upload":
INPUT_DIR = Path("/content/audio_caricati") / RUN_ID
INPUT_DIR.mkdir(parents=True, exist_ok=True)
old_cwd = Path.cwd()
try:
os.chdir(INPUT_DIR)
uploaded = files.upload()
finally:
os.chdir(old_cwd)
del uploaded # libera la copia in memoria
else:
INPUT_DIR = Path(CARTELLA_DRIVE)
if not INPUT_DIR.is_dir():
raise FileNotFoundError(f"Cartella non trovata: {INPUT_DIR}")
extensions = {".mp3", ".wav", ".m4a", ".flac", ".ogg", ".opus", ".aac"}
AUDIO_FILES = sorted(p for p in INPUT_DIR.iterdir() if p.is_file() and p.suffix.lower() in extensions)
if not AUDIO_FILES:
raise ValueError("Nessun audio supportato nella cartella selezionata.")
if MINUTI_PER_SEZIONE <= 0 or not 0 <= SOGLIA_PAROLA_INCERTA <= 1:
raise ValueError("Controlla MINUTI_PER_SEZIONE e SOGLIA_PAROLA_INCERTA.")
for p in AUDIO_FILES:
print(f"• {p.name} ({p.stat().st_size / 1024**2:.1f} MB)")
print("Risultati:", RUN_DIR)
Upload widget is only available when the cell has been executed in the
current browser session. Please rerun this cell to enable.
In [ ]:
# 4. Prepara il programma di trascrizione
WORKER_SOURCE = 'import os, sys, json, traceback, re\nfrom pathlib import Path\nfrom dataclasses import asdict\nfrom importlib.metadata import version\nfrom faster_whisper import WhisperModel\nimport ctranslate2\n\ndef timestamp(seconds, srt=False):\n ms = max(0, round(seconds * 1000))\n h, ms = divmod(ms, 3600000)\n m, ms = divmod(ms, 60000)\n s, ms = divmod(ms, 1000)\n return f"{h:02}:{m:02}:{s:02}" + (f",{ms:03}" if srt else "")\n\ndef export_result(source, base, segments, info, cfg):\n md = [f"# Trascrizione — {source.name}", "",\n f"**Lingua rilevata/impostata:** {info.language} ",\n f"**Durata:** {timestamp(info.duration)} ",\n f"**Modello:** {cfg[\'model\']} ",\n "**Interlocutori:** non identificati automaticamente.", "",\n "> Trascrizione automatica, non verificata mediante ascolto. Nessun riassunto. "\n "Le parole dubbie sono conservate e segnalate, non sostituite con ipotesi. "\n "I timestamp e i segnali di affidabilità sono approssimativi.", ""]\n txt, srt = [], []\n previous_section = None\n for i, seg in enumerate(segments, 1):\n text = seg[\'text\'].strip()\n section = int(seg[\'start\'] // (cfg[\'section_minutes\'] * 60))\n if section != previous_section:\n start = section * cfg[\'section_minutes\'] * 60\n end = min(start + cfg[\'section_minutes\'] * 60, info.duration)\n md += [f"## {timestamp(start)}–{timestamp(end)}", ""]\n previous_section = section\n md += [f"**[{timestamp(seg[\'start\'])}–{timestamp(seg[\'end\'])}]**", "", text, ""]\n uncertain = [w for w in (seg.get(\'words\') or []) if w[\'probability\'] < cfg[\'threshold\']]\n if uncertain:\n detail = "; ".join(f"{timestamp(w[\'start\'])}: «{w[\'word\'].strip()}»" for w in uncertain)\n md += [f"*[Da riascoltare: letture automatiche incerte — {detail}]*", ""]\n if seg.get(\'avg_logprob\', 0) < -1 or seg.get(\'compression_ratio\', 0) > 2.4:\n md += ["*[Segmento da verificare: possibile riconoscimento errato o ripetitivo.]*", ""]\n txt.append(text)\n srt.append(f"{i}\\n{timestamp(seg[\'start\'], True)} --> {timestamp(seg[\'end\'], True)}\\n{text}\\n")\n if not segments:\n md += ["[Nessun parlato riconosciuto automaticamente: verificare l’audio.]", ""]\n for ext, content in [(\'md\', \'\\n\'.join(md)), (\'txt\', \'\\n\'.join(txt)), (\'srt\', \'\\n\'.join(srt))]:\n base.with_suffix(\'.\' + ext).write_text(content, encoding=\'utf-8\')\n metadata = {\'source\': str(source), \'duration_seconds\': info.duration,\n \'language\': info.language, \'language_probability\': info.language_probability,\n \'settings\': cfg, \'segments\': segments,\n \'versions\': {p: version(p) for p in [\'faster-whisper\',\'ctranslate2\',\'av\']}}\n base.with_suffix(\'.json\').write_text(json.dumps(metadata, ensure_ascii=False, indent=2), encoding=\'utf-8\')\n\ndef main():\n cfg = json.loads(Path(sys.argv[1]).read_text(encoding=\'utf-8\'))\n if ctranslate2.get_cuda_device_count() < 1:\n raise RuntimeError(\'GPU CUDA non utilizzabile. Verifica il runtime Colab.\')\n print(\'Caricamento modello \' + cfg[\'model\'] + \' sulla GPU...\', flush=True)\n model = WhisperModel(cfg[\'model\'], device=\'cuda\', compute_type=cfg[\'precision\'])\n results=[]\n for index, name in enumerate(cfg[\'files\'], 1):\n source = Path(name)\n # Prefix and suffix protect different inputs with the same stem.\n safe = re.sub(r\'[^\\w\\- ]\', \'_\', source.stem)[:100] or \'audio\'\n base = Path(cfg[\'output\']) / f\'{index:03}_{safe}_trascrizione\'\n print(f\'\\n[{index}/{len(cfg["files"])}] {source.name}\', flush=True)\n partial = base.with_suffix(\'.partial.jsonl\')\n try:\n iterator, info = model.transcribe(str(source), language=cfg[\'language\'],\n task=\'transcribe\', beam_size=5, word_timestamps=True,\n vad_filter=cfg[\'vad\'], hotwords=cfg[\'context\'] or None, condition_on_previous_text=False)\n segments=[]\n with partial.open(\'w\', encoding=\'utf-8\') as f:\n for segment in iterator:\n record=asdict(segment)\n segments.append(record)\n f.write(json.dumps(record, ensure_ascii=False)+\'\\n\'); f.flush()\n print(f\' {timestamp(segment.end)} / {timestamp(info.duration)}\', flush=True)\n export_result(source, base, segments, info, cfg)\n partial.unlink()\n results.append({\'file\': source.name, \'status\': \'ok\', \'segments\': len(segments)})\n print(\' Esportazione completata.\', flush=True)\n except Exception as exc:\n base.with_suffix(\'.errore.txt\').write_text(traceback.format_exc(), encoding=\'utf-8\')\n results.append({\'file\': source.name, \'status\': \'errore\', \'error\': str(exc)})\n print(\' ERRORE: \' + str(exc) + \' — proseguo con il file successivo.\', flush=True)\n Path(cfg[\'output\'], \'riepilogo.json\').write_text(json.dumps(results, ensure_ascii=False, indent=2), encoding=\'utf-8\')\n print(\'\\nCompletati:\', sum(r[\'status\']==\'ok\' for r in results), \'/\', len(results), flush=True)\n if not any(r[\'status\']==\'ok\' for r in results):\n sys.exit(2)\n\nif __name__ == \'__main__\':\n main()\n'
WORKER = Path("/content/trascrivi_gpu.py")
WORKER.write_text(WORKER_SOURCE, encoding="utf-8")
CONFIG = {
"model": MODELLO, "language": LINGUA, "precision": PRECISIONE,
"vad": FILTRO_SILENZI, "context": CONTESTO.strip(), "section_minutes": MINUTI_PER_SEZIONE,
"threshold": SOGLIA_PAROLA_INCERTA,
"files": [str(p) for p in AUDIO_FILES], "output": str(RUN_DIR)
}
CONFIG_PATH = RUN_DIR / "configurazione.json"
CONFIG_PATH.write_text(json.dumps(CONFIG, ensure_ascii=False, indent=2), encoding="utf-8")
In [ ]:
# 5. Esegui il riconoscimento e conserva il log
import site
library_dirs = []
for folder in site.getsitepackages():
for package in ("cublas", "cudnn"):
candidate = Path(folder) / "nvidia" / package / "lib"
if candidate.is_dir():
library_dirs.append(str(candidate))
if len(library_dirs) < 2:
raise RuntimeError("Librerie CUDA/cuDNN non trovate: riesegui la cella 2.")
env = os.environ.copy()
env["LD_LIBRARY_PATH"] = ":".join(library_dirs + [env.get("LD_LIBRARY_PATH", "")])
env["PYTHONUNBUFFERED"] = "1"
with (RUN_DIR / "esecuzione.log").open("w", encoding="utf-8") as log:
process = subprocess.Popen([sys.executable, "-u", str(WORKER), str(CONFIG_PATH)],
env=env, stdout=subprocess.PIPE, stderr=subprocess.STDOUT,
text=True, bufsize=1)
try:
for line in process.stdout:
print(line, end="")
log.write(line); log.flush()
returncode = process.wait()
except KeyboardInterrupt:
process.terminate()
process.wait()
print("Interrotto: i segmenti già elaborati restano nei file partial.jsonl.")
raise
if returncode:
print(f"Processo terminato con codice {returncode}. Controlla il log e gli eventuali errori nello ZIP.")
In [ ]:
# 6. Archivio ZIP e copia facoltativa su Drive
archive = Path(shutil.make_archive(str(RUN_DIR) + "_risultati", "zip", root_dir=RUN_DIR))
if SALVA_SU_DRIVE:
destination = Path(CARTELLA_RISULTATI_DRIVE) / RUN_ID
destination.mkdir(parents=True, exist_ok=True)
shutil.copytree(RUN_DIR, destination, dirs_exist_ok=True)
shutil.copy2(archive, destination / archive.name)
print("Copia salvata su Drive:", destination)
print("Archivio:", archive)
print("Trascrizioni Markdown:")
for p in sorted(RUN_DIR.glob("*.md")):
print("•", p.name)
if SCARICA_ZIP:
files.download(str(archive))