Compare commits

...
2 Commits
Author SHA1 Message Date
luzadev dc6ae1fbe1 v1.7.15: patch_info_plist con plistlib (PlistBuddy falliva su apostrofi)
Crash report v1.7.14 ha rivelato che NSMicrophoneUsageDescription
NON era stato scritto nell'Info.plist distribuito. Causa: il valore
contiene l'apostrofo "l'audio" che rompe PlistBuddy con quote
shell. PlistBuddy falliva ma restituiva exit code != 0 con un
messaggio diverso da "already exists", quindi anche il fallback
Set non veniva chiamato -> chiave mai aggiunta.

Riscritto patch_info_plist con plistlib di Python: nessun shell
escaping, robusto a qualsiasi carattere. Aggiunto check di verifica
finale che logga errore critico se la chiave manca dopo write.
2026-06-09 23:17:08 +02:00
luzadev 0e14d161d0 v1.7.14: fix definitivo registrazione macOS — AVCaptureSession in-process
Dopo che codesign --info-plist e wrap in sub-bundle .app si sono
rivelati strade morte (macOS 14+ TCC e' troppo severo per ad-hoc
signed binaries), riscritta la registrazione macOS per usare
PyObjC + AVCaptureSession nel processo Python principale.

Il main MusicTools ha il TCC del bundle (NSMicrophoneUsageDescription
nell'Info.plist), e usando AVFoundation IN-PROCESS il permesso
viene applicato correttamente. ffmpeg subprocess viene usato SOLO
per la conversione CAF -> MP3 dopo lo stop (operazione su file,
niente accesso microfono = niente TCC).

Aggiunto pyobjc-framework-AVFoundation a requirements.
core/paths.py: ripristinato lookup semplice in Contents/Frameworks/
(non serve piu' cercare in sub-bundle).
build_macos.py: wrap_subprocess_in_bundle disabilitato.

Diagnosi crash report v1.7.13 + investigazione TCC behavior
confermano: PyObjC AVCaptureSession in-process e' l'unica
soluzione affidabile senza Developer ID Apple.

Vedi: https://www.qt.io/blog/the-curious-case-of-the-responsible-process
2026-06-09 23:04:16 +02:00
4 changed files with 385 additions and 256 deletions

No files matched your search

+34 -30
View File
@@ -239,7 +239,12 @@ def run_pyinstaller():
if app_path.exists(): if app_path.exists():
log(f"Build completata: {app_path}") log(f"Build completata: {app_path}")
patch_info_plist(app_path) patch_info_plist(app_path)
wrap_subprocess_in_bundle(app_path) # NOTA: dal v1.7.14 il microfono viene letto direttamente da
# PyObjC/AVCaptureSession nel processo Python (che ha il TCC del
# bundle), non da ffmpeg subprocess. ffmpeg fa solo la conversione
# CAF -> MP3 (no microfono = no TCC). Per questo non serve piu'
# incapsularlo in un sub-bundle .app con NSMicrophoneUsageDescription.
# wrap_subprocess_in_bundle(app_path)
adhoc_codesign(app_path) adhoc_codesign(app_path)
# Mostra dimensione # Mostra dimensione
size = sum(f.stat().st_size for f in app_path.rglob("*") if f.is_file()) size = sum(f.stat().st_size for f in app_path.rglob("*") if f.is_file())
@@ -253,45 +258,44 @@ def run_pyinstaller():
def patch_info_plist(app_path): def patch_info_plist(app_path):
"""Aggiunge le chiavi TCC (Transparency, Consent and Control) richieste """Aggiunge le chiavi TCC nell'Info.plist del bundle .app.
da macOS nel bundle .app, altrimenti macOS nega silenziosamente l'accesso
al microfono e l'app NON compare in Impostazioni > Privacy e sicurezza.
Senza NSMicrophoneUsageDescription, qualsiasi tentativo di ffmpeg di NSMicrophoneUsageDescription e' obbligatoria: senza, macOS uccide
aprire un dispositivo audio (incluso BlackHole) viene rifiutato dal l'app con SIGABRT quando AVFoundation tenta di aprire il microfono.
sistema e non viene mostrato nessun prompt all'utente.
Usiamo plistlib invece di PlistBuddy perche' i valori con apostrofi
(es. "l'audio") rompono il parser shell-based di PlistBuddy.
""" """
import plistlib
info_plist = app_path / "Contents" / "Info.plist" info_plist = app_path / "Contents" / "Info.plist"
if not info_plist.exists(): if not info_plist.exists():
log("ATTENZIONE: Info.plist non trovato, skip patch TCC") log("ATTENZIONE: Info.plist non trovato, skip patch TCC")
return return
pb = "/usr/libexec/PlistBuddy"
# Helper: aggiungi una chiave; se gia presente sovrascrivila.
def add_or_set(key, value, ptype="string"):
# Tenta Add; se esiste, fa Set.
r = subprocess.run(
[pb, "-c", f'Add :{key} {ptype} "{value}"', str(info_plist)],
capture_output=True, text=True,
)
if r.returncode != 0 and "already exists" in (r.stderr + r.stdout).lower():
subprocess.run(
[pb, "-c", f'Set :{key} "{value}"', str(info_plist)],
check=True,
)
log("Patching Info.plist con le chiavi TCC...") log("Patching Info.plist con le chiavi TCC...")
add_or_set( with open(info_plist, "rb") as f:
"NSMicrophoneUsageDescription", data = plistlib.load(f)
data["NSMicrophoneUsageDescription"] = (
"MusicTools usa il microfono per registrare l'audio del sistema " "MusicTools usa il microfono per registrare l'audio del sistema "
"(ad esempio tramite BlackHole o un altro dispositivo loopback).", "(ad esempio tramite BlackHole o un altro dispositivo loopback)."
) )
# Sicurezza extra: dichiarare bundle identifier stabile aiuta il # Bundle identifier stabile aiuta TCC a riconoscere l'app tra le sessioni
# database TCC a riconoscere coerentemente l'app tra le sessioni. data["CFBundleIdentifier"] = "com.djluza.musictools"
add_or_set("CFBundleIdentifier", "com.djluza.musictools") data["LSApplicationCategoryType"] = "public.app-category.music"
add_or_set("LSApplicationCategoryType", "public.app-category.music")
log("Info.plist aggiornato.") with open(info_plist, "wb") as f:
plistlib.dump(data, f)
log(f"Info.plist aggiornato. Chiavi: {sorted(data.keys())[-5:]}")
# Verifica
with open(info_plist, "rb") as f:
verify = plistlib.load(f)
if "NSMicrophoneUsageDescription" not in verify:
log("ERRORE CRITICO: NSMicrophoneUsageDescription NON e' stata scritta!")
else:
log("Verifica OK: NSMicrophoneUsageDescription presente.")
_BUNDLE_ID = "com.djluza.musictools" _BUNDLE_ID = "com.djluza.musictools"
+1 -1
View File
@@ -5,7 +5,7 @@ import os
import sys import sys
from pathlib import Path from pathlib import Path
VERSION = "v1.7.13" VERSION = "v1.7.15"
APP_NAME = "MusicTools" APP_NAME = "MusicTools"
+349 -225
View File
@@ -1,7 +1,10 @@
"""Registrazione audio da un dispositivo di input via ffmpeg. """Registrazione audio.
macOS: usa AVFoundation (ingresso fisico + driver virtuali tipo BlackHole) macOS: AVCaptureSession via PyObjC dentro al processo Python principale.
Windows: usa DirectShow Il bundle MusicTools.app ha NSMicrophoneUsageDescription e il TCC viene
concesso al main, quindi AVFoundation registra senza problemi. ffmpeg
viene usato SOLO per convertire il file CAF a MP3 (no microfono = no TCC).
Windows: ffmpeg DirectShow (qui il problema TCC non esiste).
""" """
from __future__ import annotations from __future__ import annotations
@@ -25,38 +28,33 @@ _IS_WIN = sys.platform == "win32"
# ============================================================ # ============================================================
# Stato globale (un solo recording alla volta) # Stato globale (un solo recording alla volta)
# ============================================================ # ============================================================
_current_proc: Optional[subprocess.Popen] = None _recording_lock = threading.Lock()
_proc_lock = threading.Lock() _recording_state: dict = {}
_stop_event = threading.Event()
_recording_thread: Optional[threading.Thread] = None
_stderr_buf: list = [] _stderr_buf: list = []
_stderr_lock = threading.Lock() _stderr_lock = threading.Lock()
def is_recording() -> bool: def is_recording() -> bool:
with _proc_lock: with _recording_lock:
return _current_proc is not None and _current_proc.poll() is None return bool(_recording_state.get("active"))
# ============================================================ def get_last_stderr(max_lines: int = 50) -> list[str]:
# Lista dispositivi """Espone le ultime righe di log diagnostico ai fini diagnostici.
# ============================================================
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, Su macOS contiene messaggi del backend AVCaptureSession + stderr di
Loopback, Audio Hijack, VB-Cable, ...) ffmpeg durante la conversione CAF->MP3. Su Windows contiene lo
stderr di ffmpeg per la registrazione DirectShow.
""" """
ffmpeg = find_ffmpeg() with _stderr_lock:
if not ffmpeg: return list(_stderr_buf[-max_lines:])
return []
if _IS_MAC:
return _list_macos(ffmpeg) def _log_diag(line: str) -> None:
if _IS_WIN: with _stderr_lock:
return _list_windows(ffmpeg) _stderr_buf.append(line)
return [] if len(_stderr_buf) > 200:
del _stderr_buf[0]
_VIRTUAL_KEYWORDS = ( _VIRTUAL_KEYWORDS = (
@@ -71,54 +69,44 @@ def _is_virtual(name: str) -> bool:
return any(k in n for k in _VIRTUAL_KEYWORDS) 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 ""`. # Lista dispositivi
L'output va su stderr. # ============================================================
def list_input_devices() -> list[dict]:
"""Ritorna la lista dei dispositivi audio di input.
IMPORTANTE: come 'id' usiamo il NOME del device (non l'index numerico). Ogni voce: {"id": "...", "name": "...", "is_virtual": bool}.
AVFoundation rinumera silenziosamente i device quando cambia il set di Su macOS l'`id` e' l'uniqueID di AVCaptureDevice (stabile).
hardware collegato (es. iPhone in continuita', Multi-Output Device, ecc). Su Windows e' il nome del device (richiesto da DirectShow).
Il nome e' invece stabile; ffmpeg AVFoundation accetta sia ':<index>'
sia ':<name>' come specifica del device input.
""" """
if _IS_MAC:
return _list_macos_avf()
if _IS_WIN:
ffmpeg = find_ffmpeg()
return _list_windows(ffmpeg) if ffmpeg else []
return []
def _list_macos_avf() -> list[dict]:
"""Enumera input audio via AVFoundation (no ffmpeg subprocess)."""
try: try:
proc = subprocess.run( from AVFoundation import AVCaptureDevice, AVMediaTypeAudio
[ffmpeg, "-hide_banner", "-f", "avfoundation", except ImportError as e:
"-list_devices", "true", "-i", ""], _log_diag(f"[avf] AVFoundation non disponibile: {e}")
capture_output=True, text=True, timeout=10,
**subprocess_flags(),
)
except Exception:
return [] return []
devices: list[dict] = [] devices: list[dict] = []
in_audio = False for d in AVCaptureDevice.devicesWithMediaType_(AVMediaTypeAudio):
audio_section_re = re.compile(r"AVFoundation\s+audio\s+devices", re.IGNORECASE) try:
other_section_re = re.compile(r"AVFoundation\s+(?:video)\s+devices", re.IGNORECASE) name = str(d.localizedName())
line_re = re.compile(r"\[(\d+)\]\s+(.+?)\s*$") uid = str(d.uniqueID())
except Exception:
for raw in proc.stderr.splitlines():
line = raw.strip()
if audio_section_re.search(line):
in_audio = True
continue 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({ devices.append({
"id": name, # nome stabile (vedi docstring) "id": uid,
"index": idx, # solo informativo per debug
"name": name, "name": name,
"is_virtual": _is_virtual(name), "is_virtual": _is_virtual(name),
}) })
return devices return devices
@@ -162,7 +150,6 @@ def _list_windows(ffmpeg: str) -> list[dict]:
name = mn.group(1) name = mn.group(1)
devices.append({"id": name, "name": name, "is_virtual": _is_virtual(name)}) devices.append({"id": name, "name": name, "is_virtual": _is_virtual(name)})
# Dedup preservando l'ordine
seen = set() seen = set()
out = [] out = []
for d in devices: for d in devices:
@@ -174,130 +161,295 @@ def _list_windows(ffmpeg: str) -> list[dict]:
# ============================================================ # ============================================================
# Registrazione # Permesso microfono (macOS)
# ============================================================ # ============================================================
def _build_record_cmd(ffmpeg: str, device_id: str, output_path: str, def _ensure_macos_mic_permission() -> tuple[bool, str]:
bitrate: str) -> list[str]: """Verifica/richiede il permesso microfono (TCC) per il processo corrente.
"""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")
Ritorna (granted, reason).
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 try:
from AVFoundation import AVCaptureDevice, AVMediaTypeAudio
except ImportError as e:
return False, f"AVFoundation non disponibile: {e}"
if is_recording(): # 0=notDetermined, 1=restricted, 2=denied, 3=authorized
return {"ok": False, "error": "Una registrazione e gia in corso"} status = AVCaptureDevice.authorizationStatusForMediaType_(AVMediaTypeAudio)
if status == 3:
return True, "authorized"
if status in (1, 2):
return False, (
"Permesso microfono negato.\n"
"Impostazioni di Sistema -> Privacy e sicurezza -> Microfono\n"
"Attiva MusicTools e riavvia l'app."
)
# notDetermined -> chiedi il prompt in modo sincrono.
ev = threading.Event()
granted = [False]
def cb(ok):
granted[0] = bool(ok)
ev.set()
AVCaptureDevice.requestAccessForMediaType_completionHandler_(AVMediaTypeAudio, cb)
# PyObjC pumpa il run loop automaticamente in attesa del callback.
ev.wait(60)
if not granted[0]:
return False, "Permesso microfono negato dall'utente."
return True, "granted"
# ============================================================
# Registrazione macOS via AVCaptureSession
# ============================================================
def _start_recording_macos(device_id: str, output_path: str, bitrate: str,
progress_callback: Optional[Callable]) -> dict:
from AVFoundation import (
AVCaptureSession, AVCaptureDevice, AVCaptureDeviceInput,
AVCaptureAudioFileOutput, AVMediaTypeAudio,
)
from Foundation import NSURL, NSObject
ok, reason = _ensure_macos_mic_permission()
if not ok:
return {"ok": False, "error": reason}
# Trova il device: prima per uniqueID, poi per nome (fallback)
dev = AVCaptureDevice.deviceWithUniqueID_(device_id)
if dev is None:
for d in AVCaptureDevice.devicesWithMediaType_(AVMediaTypeAudio):
if str(d.localizedName()) == device_id:
dev = d
break
if dev is None:
return {"ok": False, "error": "Dispositivo audio non trovato."}
session = AVCaptureSession.alloc().init()
inp, err = AVCaptureDeviceInput.deviceInputWithDevice_error_(dev, None)
if err is not None:
return {"ok": False, "error": f"Errore input: {err.localizedDescription()}"}
if not session.canAddInput_(inp):
return {"ok": False, "error": "Impossibile aggiungere il device alla session."}
session.addInput_(inp)
out = AVCaptureAudioFileOutput.alloc().init()
if not session.canAddOutput_(out):
return {"ok": False, "error": "Impossibile aggiungere l'output alla session."}
session.addOutput_(out)
out_path = Path(output_path)
out_path.parent.mkdir(parents=True, exist_ok=True)
caf_path = out_path.with_suffix(".caf")
# Delegate: aspetta che lo stop produca il file finale
class _Delegate(NSObject):
def captureOutput_didStartRecordingToOutputFileAtURL_fromConnections_(
self, output, url, conns):
self.started = True
def captureOutput_didFinishRecordingToOutputFileAtURL_fromConnections_error_(
self, output, url, conns, err):
self.done = True
self.err = err.localizedDescription() if err is not None else None
delegate = _Delegate.alloc().init()
delegate.started = False
delegate.done = False
delegate.err = None
session.startRunning()
out.startRecordingToOutputFileURL_outputFileType_recordingDelegate_(
NSURL.fileURLWithPath_(str(caf_path)),
"com.apple.coreaudio-format",
delegate,
)
with _recording_lock:
_recording_state.update({
"active": True,
"session": session,
"output": out,
"delegate": delegate,
"caf_path": str(caf_path),
"mp3_path": str(out_path),
"bitrate": bitrate,
"started_at": time.monotonic(),
"progress_cb": progress_callback,
})
if progress_callback:
progress_callback("started", {"output_path": str(out_path)})
# Tick thread per emettere "seconds" elapsed (la UI lo mostra)
threading.Thread(target=_tick_worker, daemon=True).start()
_log_diag(f"[avf] Recording started -> {caf_path}")
return {"ok": True, "output_path": str(out_path)}
def _tick_worker():
while is_recording():
with _recording_lock:
cb = _recording_state.get("progress_cb")
started_at = _recording_state.get("started_at", time.monotonic())
elapsed = int(time.monotonic() - started_at)
if cb:
cb("tick", {"seconds": elapsed})
time.sleep(1.0)
def _stop_recording_macos() -> dict:
with _recording_lock:
if not _recording_state.get("active"):
return {"ok": True}
state = dict(_recording_state)
_recording_state["active"] = False
session = state["session"]
out = state["output"]
delegate = state["delegate"]
caf_path = Path(state["caf_path"])
mp3_path = Path(state["mp3_path"])
bitrate = state["bitrate"]
cb = state["progress_cb"]
started_at = state["started_at"]
# Stop recording (chiama il delegate didFinish dopo aver finalizzato il file)
out.stopRecording()
# Aspetta il completamento del file
deadline = time.time() + 8
while not getattr(delegate, "done", False) and time.time() < deadline:
time.sleep(0.1)
session.stopRunning()
if getattr(delegate, "err", None):
_log_diag(f"[avf] delegate error: {delegate.err}")
seconds = int(time.monotonic() - started_at)
# Conversione CAF -> MP3 via ffmpeg (NO microfono: niente TCC)
if caf_path.exists():
ffmpeg = find_ffmpeg()
if ffmpeg:
try:
proc = subprocess.run(
[ffmpeg, "-hide_banner", "-y", "-i", str(caf_path),
"-acodec", "libmp3lame", "-b:a", bitrate,
"-vn", str(mp3_path)],
capture_output=True, text=True, timeout=120,
**subprocess_flags(),
)
if proc.returncode != 0:
_log_diag(f"[ffmpeg-convert] rc={proc.returncode}")
for ln in (proc.stderr or "").splitlines()[-10:]:
_log_diag(ln)
# Cleanup CAF temporaneo
try:
caf_path.unlink()
except OSError:
pass
except Exception as e:
_log_diag(f"[ffmpeg-convert] exception: {e}")
else:
_log_diag("[ffmpeg-convert] ffmpeg non trovato, lascio il file CAF.")
# Se ffmpeg non c'e', rinomina il CAF in MP3 (estensione errata
# ma l'utente ha qualcosa)
try:
caf_path.rename(mp3_path.with_suffix(".caf"))
except OSError:
pass
if cb:
if mp3_path.exists():
cb("stopped", {"output_path": str(mp3_path), "seconds": seconds})
else:
cb("error", {"error": "File registrato non trovato dopo lo stop."})
return {"ok": True}
# ============================================================
# Registrazione Windows via ffmpeg DirectShow (invariato)
# ============================================================
def _start_recording_windows(device_id: str, output_path: str, bitrate: str,
progress_callback: Optional[Callable]) -> dict:
ffmpeg = find_ffmpeg() ffmpeg = find_ffmpeg()
if not ffmpeg: if not ffmpeg:
return {"ok": False, "error": "ffmpeg non trovato"} return {"ok": False, "error": "ffmpeg non trovato"}
out = Path(output_path) out = Path(output_path)
out.parent.mkdir(parents=True, exist_ok=True) out.parent.mkdir(parents=True, exist_ok=True)
cmd = [
ffmpeg, "-hide_banner", "-y",
"-f", "dshow",
"-i", f"audio={device_id}",
"-acodec", "libmp3lame",
"-b:a", bitrate,
"-vn",
str(out),
]
cmd = _build_record_cmd(ffmpeg, device_id, str(out), bitrate)
_stop_event.clear()
with _stderr_lock: with _stderr_lock:
_stderr_buf.clear() _stderr_buf.clear()
def _drain_stderr(proc): def _drain(proc):
"""Drena lo stderr di ffmpeg in modo continuo per evitare blocchi
del pipe e per salvare le ultime righe ai fini diagnostici."""
try: try:
for raw in iter(proc.stderr.readline, b""): for raw in iter(proc.stderr.readline, b""):
if not raw: if not raw:
break break
line = raw.decode("utf-8", errors="ignore").rstrip() line = raw.decode("utf-8", errors="ignore").rstrip()
with _stderr_lock: _log_diag(line)
_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: except Exception:
pass pass
def _worker(): def _worker():
global _current_proc
try: try:
with _proc_lock: proc = subprocess.Popen(
_current_proc = subprocess.Popen( cmd,
cmd, stdin=subprocess.PIPE,
stdin=subprocess.PIPE, stdout=subprocess.DEVNULL,
stdout=subprocess.DEVNULL, stderr=subprocess.PIPE,
stderr=subprocess.PIPE, text=False,
text=False, **subprocess_flags(),
**subprocess_flags(), )
) with _recording_lock:
_recording_state.update({
"active": True,
"proc": proc,
"output_path": str(out),
"started_at": time.monotonic(),
"progress_cb": progress_callback,
})
threading.Thread(target=_drain, args=(proc,), daemon=True).start()
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() start = time.monotonic()
opened = False opened = False
for _ in range(15): for _ in range(15):
if _current_proc.poll() is not None: if proc.poll() is not None:
break break
# ffmpeg stampa "Press [q] to stop" quando il device e' aperto
with _stderr_lock: with _stderr_lock:
if any("Press [q]" in l for l in _stderr_buf): if any("Press [q]" in l for l in _stderr_buf):
opened = True opened = True
break break
time.sleep(0.1) time.sleep(0.1)
rc_early = _current_proc.poll() if not opened and proc.poll() is not None:
if not opened and rc_early is not None: with _recording_lock:
# ffmpeg e' uscito prima di iniziare a registrare _recording_state["active"] = False
with _proc_lock:
_current_proc = None
if progress_callback: if progress_callback:
progress_callback("error", {"error": _extract_friendly_error(rc_early)}) progress_callback("error", {
"error": _extract_friendly_error_windows(proc.returncode),
})
return return
if progress_callback: if progress_callback:
progress_callback("started", {"output_path": str(out)}) progress_callback("started", {"output_path": str(out)})
while True: while True:
if _stop_event.is_set(): with _recording_lock:
break if not _recording_state.get("active"):
rc = _current_proc.poll() break
rc = proc.poll()
if rc is not None: if rc is not None:
break break
elapsed = int(time.monotonic() - start) elapsed = int(time.monotonic() - start)
@@ -305,123 +457,95 @@ def start_recording(
progress_callback("tick", {"seconds": elapsed}) progress_callback("tick", {"seconds": elapsed})
time.sleep(1.0) time.sleep(1.0)
# Stop gracefully if proc.poll() is None:
if _current_proc.poll() is None:
try: try:
_current_proc.stdin.write(b"q\n") proc.stdin.write(b"q\n")
_current_proc.stdin.flush() proc.stdin.flush()
except Exception: except Exception:
pass pass
try: try:
_current_proc.wait(timeout=5) proc.wait(timeout=5)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
_current_proc.terminate() proc.terminate()
try: try:
_current_proc.wait(timeout=3) proc.wait(timeout=3)
except subprocess.TimeoutExpired: except subprocess.TimeoutExpired:
_current_proc.kill() proc.kill()
rc = _current_proc.returncode
seconds = int(time.monotonic() - start) seconds = int(time.monotonic() - start)
with _recording_lock:
with _proc_lock: _recording_state["active"] = False
_current_proc = None
if progress_callback: if progress_callback:
if rc == 0 or out.exists(): if proc.returncode == 0 or out.exists():
progress_callback("stopped", {"output_path": str(out), "seconds": seconds}) progress_callback("stopped", {"output_path": str(out), "seconds": seconds})
else: else:
progress_callback("error", {"error": _extract_friendly_error(rc)}) progress_callback("error", {
"error": _extract_friendly_error_windows(proc.returncode),
})
except Exception as e: except Exception as e:
with _proc_lock: with _recording_lock:
_current_proc = None _recording_state["active"] = False
if progress_callback: if progress_callback:
progress_callback("error", {"error": str(e)}) progress_callback("error", {"error": f"{type(e).__name__}: {e}"})
_recording_thread = threading.Thread(target=_worker, daemon=True)
_recording_thread.start()
threading.Thread(target=_worker, daemon=True).start()
return {"ok": True, "output_path": str(out)} return {"ok": True, "output_path": str(out)}
def _extract_friendly_error(rc: int) -> str: def _extract_friendly_error_windows(rc: int) -> str:
"""Trasforma rc + stderr di ffmpeg in un messaggio leggibile.
Riconosce i casi tipici: permessi microfono, device occupato,
device non disponibile, clock drift di BlackHole, ecc.
Fallback: se non riconosce il pattern, ritorna le ultime righe
significative dello stderr cosi' l'utente puo' diagnosticare
o aprire un ticket.
"""
with _stderr_lock: with _stderr_lock:
lines = list(_stderr_buf) lines = list(_stderr_buf)
text = "\n".join(lines).lower() text = "\n".join(lines).lower()
is_mac = sys.platform == "darwin" if "no such device" in text or "could not find" in text:
return "Dispositivo non trovato. Premi 'Aggiorna' e riseleziona."
# ---- Errori specifici (priorita' alta) ---- if "permission" in text or "denied" in text:
return "Permesso microfono mancante. Impostazioni Windows -> Privacy -> Microfono."
if "input/output error" in text or "errno 22" in text:
return (
"Dispositivo audio non pronto (BlackHole o scheda virtuale). "
"Spesso si risolve cosi':\n"
" 1) Smetti la registrazione e riprova fra 5 secondi\n"
" 2) Se persiste, da Terminale: 'sudo killall coreaudiod'\n"
" 3) Verifica che nessun'altra app stia gia' registrando il device"
)
if "device not configured" in text or "device not available" in text:
return "Dispositivo non configurato. Premi 'Aggiorna' e riseleziona."
if "permission" in text or "not authorized" in text or "not permitted" in text \
or "tcc" in text or "denied" in text:
return _permission_message()
if "no such device" in text or "no such audio device" in text or "invalid device" in text:
return "Dispositivo non trovato. Premi 'Aggiorna' e riseleziona dalla lista."
# ---- Macro per rc=-6 (SIGABRT) su macOS ----
# Su macOS rc=-6 e' quasi sempre il sintomo di un crash di ffmpeg
# dovuto al permesso Microfono mancante. La parola 'abort' non
# sempre compare nello stderr di ffmpeg recenti.
if rc == -6 and is_mac:
return _permission_message()
# ---- Fallback: cerca le ultime righe significative ----
for line in reversed(lines): for line in reversed(lines):
l = line.strip() l = line.strip()
if not l: if not l:
continue continue
low = l.lower() low = l.lower()
# Riga AVFoundation/dshow specifica (es. "[avfoundation @ 0x...] Could not...") if "error" in low or "fail" in low or "denied" in low:
if l.startswith("[avfoundation") or l.startswith("[dshow") or "@ 0x" in l:
return l return l
if "error" in low or "fail" in low or "denied" in low or "cannot" in low:
return l
# Ultima spiaggia: rc + ultime 3 righe di stderr per debug
tail = " | ".join(line for line in lines[-3:] if line.strip()) or "(nessuno)" tail = " | ".join(line for line in lines[-3:] if line.strip()) or "(nessuno)"
return f"ffmpeg exit {rc}. Ultimo stderr: {tail}" return f"ffmpeg exit {rc}. Ultimo stderr: {tail}"
def _permission_message() -> str: def _stop_recording_windows() -> dict:
"""Messaggio standardizzato per il problema permesso Microfono macOS.""" with _recording_lock:
return ( if not _recording_state.get("active"):
"Permesso Microfono mancante. Su macOS:\n" return {"ok": True}
" Impostazioni di Sistema -> Privacy e sicurezza -> Microfono\n" # Il worker thread vede 'active'=False e termina ffmpeg con 'q'.
" Abilita 'Terminal' (se lanci l'app da terminale) o 'MusicTools'\n" _recording_state["active"] = False
" (se usi l'app installata). Poi riavvia l'app." return {"ok": True}
)
def get_last_stderr(max_lines: int = 50) -> list[str]: # ============================================================
"""Espone le ultime righe di stderr ai fini diagnostici. # API pubblica
Chiamato dal bridge per la UI 'Mostra log tecnici'.""" # ============================================================
def start_recording(
device_id: str,
output_path: str,
bitrate: str = "320k",
progress_callback: Optional[Callable] = None,
) -> dict:
if is_recording():
return {"ok": False, "error": "Una registrazione e gia in corso"}
with _stderr_lock: with _stderr_lock:
return list(_stderr_buf[-max_lines:]) _stderr_buf.clear()
if _IS_MAC:
return _start_recording_macos(device_id, output_path, bitrate, progress_callback)
if _IS_WIN:
return _start_recording_windows(device_id, output_path, bitrate, progress_callback)
return {"ok": False, "error": "Sistema operativo non supportato per la registrazione"}
def stop_recording() -> dict: def stop_recording() -> dict:
"""Ferma la registrazione corrente (se attiva).""" if _IS_MAC:
_stop_event.set() return _stop_recording_macos()
if _IS_WIN:
return _stop_recording_windows()
return {"ok": True} return {"ok": True}
+1
View File
@@ -5,5 +5,6 @@ mutagen>=1.47.0
pyobjc-core>=10.0 ; sys_platform == "darwin" pyobjc-core>=10.0 ; sys_platform == "darwin"
pyobjc-framework-Cocoa>=10.0 ; sys_platform == "darwin" pyobjc-framework-Cocoa>=10.0 ; sys_platform == "darwin"
pyobjc-framework-WebKit>=10.0 ; sys_platform == "darwin" pyobjc-framework-WebKit>=10.0 ; sys_platform == "darwin"
pyobjc-framework-AVFoundation>=10.0 ; sys_platform == "darwin"
pythonnet==3.0.5 ; sys_platform == "win32" pythonnet==3.0.5 ; sys_platform == "win32"
clr-loader==0.2.7.post0 ; sys_platform == "win32" clr-loader==0.2.7.post0 ; sys_platform == "win32"