luzadevandClaude Opus 4.7
850d3f4b6c
- main.py: titolo finestra pywebview -> 'MusicTools' - core/config.py: APP_NAME='MusicTools', cartella config bundled ora ~/Library/Application Support/MusicTools/ e %APPDATA%/MusicTools/. Migrazione one-shot: se non esiste la nuova ma esiste la legacy 'MusicDownload', il config.json viene copiato cosi' le impostazioni non vanno perse dopo l'aggiornamento - core/recorder.py: messaggio errore permesso microfono - api/bridge.py: User-Agent update check + guida Spotify menzionano MusicTools - build_macos.py / build_windows.py / .github/workflows/build.yml: prodotti rinominati (MusicTools.app, MusicTools.exe, MusicTools-macOS.zip, MusicTools-Windows.zip, Release 'Build MusicTools') - webui: <title>, brand-name sidebar, intestazioni commenti - README: tutti i riferimenti aggiornati + nota che il repo Git conserva il nome storico 'musicdownload' (URL stabili per release gia' pubblicate) Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
374 lines
12 KiB
Python
374 lines
12 KiB
Python
"""Registrazione audio da un dispositivo di input via ffmpeg.
|
|
|
|
macOS: usa AVFoundation (ingresso fisico + driver virtuali tipo BlackHole)
|
|
Windows: usa DirectShow
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import threading
|
|
import time
|
|
from pathlib import Path
|
|
from typing import Callable, Optional
|
|
|
|
from core.paths import find_ffmpeg
|
|
|
|
|
|
_IS_MAC = sys.platform == "darwin"
|
|
_IS_WIN = sys.platform == "win32"
|
|
|
|
|
|
# ============================================================
|
|
# Stato globale (un solo recording alla volta)
|
|
# ============================================================
|
|
_current_proc: Optional[subprocess.Popen] = None
|
|
_proc_lock = threading.Lock()
|
|
_stop_event = threading.Event()
|
|
_recording_thread: Optional[threading.Thread] = None
|
|
_stderr_buf: list = []
|
|
_stderr_lock = threading.Lock()
|
|
|
|
|
|
def is_recording() -> bool:
|
|
with _proc_lock:
|
|
return _current_proc is not None and _current_proc.poll() is None
|
|
|
|
|
|
# ============================================================
|
|
# Lista dispositivi
|
|
# ============================================================
|
|
def list_input_devices() -> list[dict]:
|
|
"""Ritorna la lista dei dispositivi audio di input.
|
|
Ogni voce: {"id": "0", "name": "...", "is_virtual": bool}.
|
|
|
|
is_virtual = True per driver di loopback noti (BlackHole, Soundflower,
|
|
Loopback, Audio Hijack, VB-Cable, ...)
|
|
"""
|
|
ffmpeg = find_ffmpeg()
|
|
if not ffmpeg:
|
|
return []
|
|
|
|
if _IS_MAC:
|
|
return _list_macos(ffmpeg)
|
|
if _IS_WIN:
|
|
return _list_windows(ffmpeg)
|
|
return []
|
|
|
|
|
|
_VIRTUAL_KEYWORDS = (
|
|
"blackhole", "soundflower", "loopback", "audio hijack",
|
|
"vb-cable", "vb-audio", "stereo mix", "what u hear",
|
|
"obs", "system audio",
|
|
)
|
|
|
|
|
|
def _is_virtual(name: str) -> bool:
|
|
n = (name or "").lower()
|
|
return any(k in n for k in _VIRTUAL_KEYWORDS)
|
|
|
|
|
|
def _list_macos(ffmpeg: str) -> list[dict]:
|
|
"""Parsa l'output di `ffmpeg -f avfoundation -list_devices true -i ""`.
|
|
L'output va su stderr."""
|
|
try:
|
|
proc = subprocess.run(
|
|
[ffmpeg, "-hide_banner", "-f", "avfoundation",
|
|
"-list_devices", "true", "-i", ""],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
except Exception:
|
|
return []
|
|
|
|
devices: list[dict] = []
|
|
in_audio = False
|
|
audio_section_re = re.compile(r"AVFoundation\s+audio\s+devices", re.IGNORECASE)
|
|
other_section_re = re.compile(r"AVFoundation\s+(?:video)\s+devices", re.IGNORECASE)
|
|
line_re = re.compile(r"\[(\d+)\]\s+(.+?)\s*$")
|
|
|
|
for raw in proc.stderr.splitlines():
|
|
line = raw.strip()
|
|
if audio_section_re.search(line):
|
|
in_audio = True
|
|
continue
|
|
if other_section_re.search(line):
|
|
in_audio = False
|
|
continue
|
|
if not in_audio:
|
|
continue
|
|
clean = re.sub(r"^\[AVFoundation[^\]]*\]\s*", "", line)
|
|
m = line_re.match(clean)
|
|
if not m:
|
|
continue
|
|
idx, name = m.group(1), m.group(2).strip()
|
|
devices.append({"id": idx, "name": name, "is_virtual": _is_virtual(name)})
|
|
|
|
return devices
|
|
|
|
|
|
def _list_windows(ffmpeg: str) -> list[dict]:
|
|
"""Parsa l'output di `ffmpeg -f dshow -list_devices true -i dummy`."""
|
|
try:
|
|
proc = subprocess.run(
|
|
[ffmpeg, "-hide_banner", "-f", "dshow",
|
|
"-list_devices", "true", "-i", "dummy"],
|
|
capture_output=True, text=True, timeout=10,
|
|
)
|
|
except Exception:
|
|
return []
|
|
|
|
devices: list[dict] = []
|
|
in_audio = False
|
|
line_re = re.compile(r'"([^"]+)"\s*\((audio|video)\)', re.IGNORECASE)
|
|
name_re = re.compile(r'"([^"]+)"')
|
|
|
|
for raw in proc.stderr.splitlines():
|
|
line = raw.strip()
|
|
low = line.lower()
|
|
if "directshow audio devices" in low:
|
|
in_audio = True
|
|
continue
|
|
if "directshow video devices" in low:
|
|
in_audio = False
|
|
continue
|
|
|
|
m = line_re.search(line)
|
|
if m:
|
|
name, kind = m.group(1), m.group(2).lower()
|
|
if kind == "audio":
|
|
devices.append({"id": name, "name": name, "is_virtual": _is_virtual(name)})
|
|
continue
|
|
|
|
if in_audio:
|
|
mn = name_re.search(line)
|
|
if mn:
|
|
name = mn.group(1)
|
|
devices.append({"id": name, "name": name, "is_virtual": _is_virtual(name)})
|
|
|
|
# Dedup preservando l'ordine
|
|
seen = set()
|
|
out = []
|
|
for d in devices:
|
|
if d["id"] in seen:
|
|
continue
|
|
seen.add(d["id"])
|
|
out.append(d)
|
|
return out
|
|
|
|
|
|
# ============================================================
|
|
# Registrazione
|
|
# ============================================================
|
|
def _build_record_cmd(ffmpeg: str, device_id: str, output_path: str,
|
|
bitrate: str) -> list[str]:
|
|
"""Comando ffmpeg per registrare su file MP3."""
|
|
if _IS_MAC:
|
|
return [
|
|
ffmpeg, "-hide_banner", "-y",
|
|
"-f", "avfoundation",
|
|
"-i", f":{device_id}",
|
|
"-acodec", "libmp3lame",
|
|
"-b:a", bitrate,
|
|
"-vn",
|
|
output_path,
|
|
]
|
|
if _IS_WIN:
|
|
return [
|
|
ffmpeg, "-hide_banner", "-y",
|
|
"-f", "dshow",
|
|
"-i", f"audio={device_id}",
|
|
"-acodec", "libmp3lame",
|
|
"-b:a", bitrate,
|
|
"-vn",
|
|
output_path,
|
|
]
|
|
raise RuntimeError("Sistema operativo non supportato per la registrazione")
|
|
|
|
|
|
def start_recording(
|
|
device_id: str,
|
|
output_path: str,
|
|
bitrate: str = "320k",
|
|
progress_callback: Optional[Callable] = None,
|
|
) -> dict:
|
|
"""Avvia una registrazione in un thread.
|
|
Ritorna {ok, error?, output_path?}.
|
|
|
|
progress_callback(status, payload) viene chiamato con:
|
|
("started", {"output_path": ...})
|
|
("tick", {"seconds": int})
|
|
("stopped", {"output_path": ..., "seconds": int})
|
|
("error", {"error": "..."})
|
|
"""
|
|
global _current_proc, _recording_thread
|
|
|
|
if is_recording():
|
|
return {"ok": False, "error": "Una registrazione e gia in corso"}
|
|
|
|
ffmpeg = find_ffmpeg()
|
|
if not ffmpeg:
|
|
return {"ok": False, "error": "ffmpeg non trovato"}
|
|
|
|
out = Path(output_path)
|
|
out.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
cmd = _build_record_cmd(ffmpeg, device_id, str(out), bitrate)
|
|
|
|
_stop_event.clear()
|
|
with _stderr_lock:
|
|
_stderr_buf.clear()
|
|
|
|
def _drain_stderr(proc):
|
|
"""Drena lo stderr di ffmpeg in modo continuo per evitare blocchi
|
|
del pipe e per salvare le ultime righe ai fini diagnostici."""
|
|
try:
|
|
for raw in iter(proc.stderr.readline, b""):
|
|
if not raw:
|
|
break
|
|
line = raw.decode("utf-8", errors="ignore").rstrip()
|
|
with _stderr_lock:
|
|
_stderr_buf.append(line)
|
|
# Mantieni solo le ultime 50 righe per non sforare la RAM
|
|
if len(_stderr_buf) > 50:
|
|
del _stderr_buf[0]
|
|
except Exception:
|
|
pass
|
|
|
|
def _worker():
|
|
global _current_proc
|
|
try:
|
|
with _proc_lock:
|
|
_current_proc = subprocess.Popen(
|
|
cmd,
|
|
stdin=subprocess.PIPE,
|
|
stdout=subprocess.DEVNULL,
|
|
stderr=subprocess.PIPE,
|
|
text=False,
|
|
)
|
|
|
|
drain_t = threading.Thread(target=_drain_stderr,
|
|
args=(_current_proc,), daemon=True)
|
|
drain_t.start()
|
|
|
|
# Aspetta brevemente che ffmpeg apra il device. Se fallisce nei
|
|
# primi ~1.5s, segnala errore prima di emettere 'started'.
|
|
start = time.monotonic()
|
|
opened = False
|
|
for _ in range(15):
|
|
if _current_proc.poll() is not None:
|
|
break
|
|
# ffmpeg stampa "Press [q] to stop" quando il device e' aperto
|
|
with _stderr_lock:
|
|
if any("Press [q]" in l for l in _stderr_buf):
|
|
opened = True
|
|
break
|
|
time.sleep(0.1)
|
|
|
|
rc_early = _current_proc.poll()
|
|
if not opened and rc_early is not None:
|
|
# ffmpeg e' uscito prima di iniziare a registrare
|
|
with _proc_lock:
|
|
_current_proc = None
|
|
if progress_callback:
|
|
progress_callback("error", {"error": _extract_friendly_error(rc_early)})
|
|
return
|
|
|
|
if progress_callback:
|
|
progress_callback("started", {"output_path": str(out)})
|
|
|
|
while True:
|
|
if _stop_event.is_set():
|
|
break
|
|
rc = _current_proc.poll()
|
|
if rc is not None:
|
|
break
|
|
elapsed = int(time.monotonic() - start)
|
|
if progress_callback:
|
|
progress_callback("tick", {"seconds": elapsed})
|
|
time.sleep(1.0)
|
|
|
|
# Stop gracefully
|
|
if _current_proc.poll() is None:
|
|
try:
|
|
_current_proc.stdin.write(b"q\n")
|
|
_current_proc.stdin.flush()
|
|
except Exception:
|
|
pass
|
|
try:
|
|
_current_proc.wait(timeout=5)
|
|
except subprocess.TimeoutExpired:
|
|
_current_proc.terminate()
|
|
try:
|
|
_current_proc.wait(timeout=3)
|
|
except subprocess.TimeoutExpired:
|
|
_current_proc.kill()
|
|
|
|
rc = _current_proc.returncode
|
|
seconds = int(time.monotonic() - start)
|
|
|
|
with _proc_lock:
|
|
_current_proc = None
|
|
|
|
if progress_callback:
|
|
if rc == 0 or out.exists():
|
|
progress_callback("stopped", {"output_path": str(out), "seconds": seconds})
|
|
else:
|
|
progress_callback("error", {"error": _extract_friendly_error(rc)})
|
|
except Exception as e:
|
|
with _proc_lock:
|
|
_current_proc = None
|
|
if progress_callback:
|
|
progress_callback("error", {"error": str(e)})
|
|
|
|
_recording_thread = threading.Thread(target=_worker, daemon=True)
|
|
_recording_thread.start()
|
|
|
|
return {"ok": True, "output_path": str(out)}
|
|
|
|
|
|
def _extract_friendly_error(rc: int) -> str:
|
|
"""Trasforma rc + stderr di ffmpeg in un messaggio leggibile.
|
|
Riconosce i casi tipici: permessi microfono, device occupato,
|
|
device non disponibile, ..."""
|
|
with _stderr_lock:
|
|
lines = list(_stderr_buf)
|
|
|
|
text = "\n".join(lines).lower()
|
|
|
|
if "abort" in text and rc == -6 and sys.platform == "darwin":
|
|
return (
|
|
"ffmpeg abortito (errore -6): probabilmente l'app non ha il "
|
|
"permesso 'Microfono' di macOS. Vai su Preferenze di Sistema "
|
|
"→ Privacy e sicurezza → Microfono e abilita Terminal/Python. "
|
|
"Poi riavvia l'app."
|
|
)
|
|
|
|
if any(s in text for s in ("input/output error", "errno 22", "device not configured")):
|
|
return "Dispositivo non disponibile (riavvia il servizio audio o ricontrolla la selezione)."
|
|
|
|
if "permission" in text or "not authorized" in text or "not permitted" in text:
|
|
return "Permesso microfono negato da macOS. Abilita Python/MusicTools in Privacy → Microfono."
|
|
|
|
if "no such device" in text or "no such audio device" in text:
|
|
return "Dispositivo non trovato. Premi '↻ Aggiorna' e riseleziona."
|
|
|
|
# Restituisci l'ultima riga utile dello stderr (esclude le righe di banner/progress)
|
|
for line in reversed(lines):
|
|
l = line.strip()
|
|
if not l:
|
|
continue
|
|
if l.startswith("[avfoundation") or l.startswith("[dshow") or "@ 0x" in l:
|
|
return l
|
|
if "error" in l.lower() or "fail" in l.lower() or "denied" in l.lower():
|
|
return l
|
|
return f"ffmpeg exit {rc}"
|
|
|
|
|
|
def stop_recording() -> dict:
|
|
"""Ferma la registrazione corrente (se attiva)."""
|
|
_stop_event.set()
|
|
return {"ok": True}
|