Ho trasformato le mie playlist Apple Music in MP3

numpy>=2.0
scipy>=1.14
tqdm>=4.66

#!/usr/bin/env python3
"""Analizza una registrazione continua e la divide secondo una playlist iTunes."""

from __future__ import annotations

import argparse
import csv
import json
import logging
import math
import shutil
import subprocess
import sys
from dataclasses import dataclass
from pathlib import Path

import numpy as np
from scipy.ndimage import gaussian_filter1d
from scipy.signal import find_peaks
from tqdm import tqdm

LOG = logging.getLogger("playlist_splitter")
CSV_FIELDS = ["indice", "titolo", "start_previsto", "start_rilevato", "correzione", "durata_rilevata", "affidabilità"]


@dataclass(frozen=True)
class Song:
    index: int
    title: str
    artist: str
    album: str
    duration: float
    track: str
    year: str
    genre: str


def detect_text_encoding(path: Path) -> str:
    head = path.read_bytes()[:4]
    if head.startswith((b"\xff\xfe", b"\xfe\xff")):
        return "utf-16"
    return "utf-8-sig"  # gestisce UTF-8 normale e UTF-8 con BOM


def load_playlist(path: Path) -> list[Song]:
    if not path.is_file():
        raise FileNotFoundError(f"Playlist non trovata: {path}")
    encoding = detect_text_encoding(path)
    try:
        with path.open("r", encoding=encoding, newline="") as handle:
            rows = list(csv.DictReader(handle, delimiter="\t"))
    except UnicodeError as exc:
        raise ValueError(f"Impossibile leggere {path.name}: encoding non riconosciuto ({exc})") from exc
    required = {"Nome", "Artista", "Album", "Durata", "Numero traccia", "Anno", "Genere"}
    missing = required.difference(rows[0].keys() if rows else ())
    if missing:
        raise ValueError("Colonne iTunes mancanti: " + ", ".join(sorted(missing)))
    songs: list[Song] = []
    for index, row in enumerate(rows, 1):
        try:
            duration = float((row.get("Durata") or "").replace(",", "."))
        except ValueError as exc:
            raise ValueError(f"Durata non valida alla riga {index + 1}: {row.get('Durata')!r}") from exc
        if duration <= 0 or not (row.get("Nome") or "").strip():
            raise ValueError(f"Titolo o durata non validi alla riga {index + 1}")
        songs.append(Song(index, row["Nome"].strip(), (row["Artista"] or "").strip(),
                          (row["Album"] or "").strip(), duration,
                          (row["Numero traccia"] or "").strip(), (row["Anno"] or "").strip(),
                          (row["Genere"] or "").strip()))
    LOG.info("Playlist letta come %s: %d canzoni (ordine originale conservato)", encoding, len(songs))
    return songs


def require_program(name: str) -> str:
    executable = shutil.which(name)
    if not executable:
        raise RuntimeError(f"{name} non è disponibile nel PATH. Installarlo e riaprire il terminale.")
    return executable


def resolve_audio(requested: Path) -> Path:
    if requested.is_file():
        return requested
    mp3s = list(Path.cwd().glob("*.mp3"))
    if requested.name.lower() == "album.mp3" and len(mp3s) == 1:
        LOG.warning("album.mp3 non esiste: uso l'unico MP3 presente, %s", mp3s[0].name)
        return mp3s[0]
    raise FileNotFoundError(f"Audio non trovato: {requested}")


def probe_audio(path: Path) -> dict:
    command = [require_program("ffprobe"), "-v", "error", "-show_entries",
               "format=duration,size,bit_rate:stream=codec_name,sample_rate,channels", "-of", "json", str(path)]
    try:
        data = json.loads(subprocess.check_output(command, text=True, encoding="utf-8", errors="replace"))
        duration = float(data["format"]["duration"])
    except (subprocess.CalledProcessError, KeyError, ValueError, json.JSONDecodeError) as exc:
        raise RuntimeError(f"ffprobe non riesce ad analizzare {path}: {exc}") from exc
    if duration <= 0:
        raise RuntimeError("La durata audio rilevata non è valida.")
    data["duration"] = duration
    LOG.info("Audio: %s, %.3f s, codec=%s, campionamento=%s Hz, canali=%s", path.name, duration,
             data["streams"][0].get("codec_name", "?"), data["streams"][0].get("sample_rate", "?"),
             data["streams"][0].get("channels", "?"))
    return data


def extract_features(audio: Path, duration: float, rate: int = 8000, frame: int = 1024, hop: int = 512):
    """Decodifica in streaming e calcola RMS, energia e novità spettrale."""
    command = [require_program("ffmpeg"), "-v", "error", "-i", str(audio), "-vn", "-ac", "1",
               "-ar", str(rate), "-f", "f32le", "pipe:1"]
    process = subprocess.Popen(command, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
    window = np.hanning(frame).astype(np.float32)
    carry = np.empty(0, dtype=np.float32)
    rms_values, flux_values, energy_changes = [], [], []
    previous_spectrum = None
    previous_db = None
    block_bytes = 4 * rate * 60
    with tqdm(total=duration, unit="s", desc="Analisi audio", dynamic_ncols=True) as bar:
        processed_samples = 0
        while True:
            raw = process.stdout.read(block_bytes) if process.stdout else b""
            if not raw:
                break
            samples = np.frombuffer(raw, dtype="<f4")
            data = np.concatenate((carry, samples))
            if data.size < frame:
                carry = data
                continue
            count = 1 + (data.size - frame) // hop
            frames = np.lib.stride_tricks.sliding_window_view(data, frame)[::hop][:count]
            rms = np.sqrt(np.mean(frames * frames, axis=1) + 1e-12)
            db = 20.0 * np.log10(rms + 1e-12)
            spectra = np.abs(np.fft.rfft(frames * window, axis=1)).astype(np.float32)
            spectra /= np.maximum(spectra.sum(axis=1, keepdims=True), 1e-12)
            if previous_spectrum is None:
                flux = np.r_[0.0, np.maximum(np.diff(spectra, axis=0), 0).sum(axis=1)]
                change = np.r_[0.0, np.abs(np.diff(db))]
            else:
                joined = np.vstack((previous_spectrum, spectra))
                flux = np.maximum(np.diff(joined, axis=0), 0).sum(axis=1)
                change = np.abs(np.diff(np.r_[previous_db, db]))
            rms_values.extend(db.tolist())
            flux_values.extend(flux.tolist())
            energy_changes.extend(change.tolist())
            previous_spectrum, previous_db = spectra[-1:], float(db[-1])
            consumed = count * hop
            carry = data[consumed:]
            processed_samples += consumed
            bar.update(min(duration - bar.n, consumed / rate))
    stderr = process.stderr.read().decode("utf-8", "replace") if process.stderr else ""
    if process.wait() != 0:
        raise RuntimeError("ffmpeg ha fallito durante la decodifica: " + stderr.strip())
    times = np.arange(len(rms_values), dtype=np.float64) * hop / rate + frame / (2 * rate)
    LOG.info("Estratti %d fotogrammi di analisi (%.2f fotogrammi/s)", len(times), rate / hop)
    return times, np.asarray(rms_values), np.asarray(flux_values), np.asarray(energy_changes)


def robust01(values: np.ndarray) -> np.ndarray:
    lo, hi = np.percentile(values, [10, 95])
    return np.clip((values - lo) / max(hi - lo, 1e-9), 0, 1)


def match_recorded_prefix(songs: list[Song], audio_duration: float) -> list[Song]:
    """Sceglie il prefisso della playlist la cui durata è più vicina alla registrazione."""
    cumulative = np.cumsum([song.duration for song in songs])
    count = int(np.argmin(np.abs(cumulative - audio_duration))) + 1
    matched = songs[:count]
    difference = float(cumulative[count - 1] - audio_duration)
    if count < len(songs):
        LOG.warning("La registrazione corrisponde al prefisso di %d/%d canzoni: le ultime %d righe "
                    "non sono presenti nell'audio e non saranno generate", count, len(songs), len(songs) - count)
    LOG.info("Durata prefisso iTunes %.3f s; audio %.3f s; differenza %+.3f s",
             cumulative[count - 1], audio_duration, difference)
    return matched

def choose_boundaries(songs: list[Song], audio_duration: float, times: np.ndarray,
                      rms_db: np.ndarray, flux: np.ndarray, energy_change: np.ndarray, window_seconds: float, anti_bleed: float, end_guard: float):
    durations = np.asarray([s.duration for s in songs], dtype=float)

    predicted = np.r_[0.0, np.cumsum(durations)[:-1]]
    LOG.info("Previsioni iTunes non riscalate; ogni confine riparte dal precedente più la durata iTunes")

    # Transizioni: novità spettrale, salto energetico, onset dopo quiete e minimo RMS locale.
    sigma = max(1.0, 0.15 / np.median(np.diff(times)))
    smooth_rms = gaussian_filter1d(rms_db, sigma)
    novelty = gaussian_filter1d(robust01(flux), sigma)
    energy = gaussian_filter1d(robust01(energy_change), sigma)
    quiet = 1.0 - robust01(smooth_rms)
    onset = robust01(np.maximum(np.gradient(smooth_rms), 0))
    score = 0.40 * novelty + 0.22 * energy + 0.23 * onset + 0.15 * quiet

    detected = np.zeros(len(songs), dtype=float)
    confidence = np.ones(len(songs), dtype=float)
    corrections = np.zeros(len(songs), dtype=float)
    detected[0] = 0.0
    opening = np.flatnonzero((times <= 4.0) & (smooth_rms > -42.0))
    if len(opening) and times[opening[0]] > 0.08:
        detected[0] = float(times[opening[0]])
        LOG.info("Pausa tecnica iniziale rimossa: primo attacco a %.3f s", detected[0])
    for i in tqdm(range(1, len(songs)), desc="Ricerca confini", dynamic_ncols=True):
        center = detected[i - 1] + durations[i - 1]
        mask = (times >= center - window_seconds) & (times <= center + window_seconds)
        indices = np.flatnonzero(mask)
        if not len(indices):
            detected[i], confidence[i] = center, 0.0
            continue
        local = score[indices]
        frame_seconds = float(np.median(np.diff(times)))
        peaks, _ = find_peaks(local, distance=max(1, int(0.35 / frame_seconds)))
        candidates = indices[peaks] if len(peaks) else indices
        local_silent = smooth_rms[indices] < -38.0
        min_silent_frames = max(1, int(0.08 / frame_seconds))
        silence_ends, run_start = [], None
        for pos, silent in enumerate(np.r_[local_silent, False]):
            if silent and run_start is None:
                run_start = pos
            elif not silent and run_start is not None:
                if pos - run_start >= min_silent_frames and pos < len(indices):
                    silence_ends.append(indices[pos])
                run_start = None
        if silence_ends:
            candidates = np.unique(np.r_[candidates, silence_ends]).astype(int)
        offset = times[candidates] - center
        objective = score[candidates] - 0.060 * np.abs(offset)
        if silence_ends:
            objective = objective + 0.55 * np.isin(candidates, silence_ends)
        valid = times[candidates] > detected[i - 1] + 1.0
        if valid.any():
            candidates, objective, offset = candidates[valid], objective[valid], offset[valid]
        best_pos = int(np.argmax(objective))
        best = candidates[best_pos]
        detected[i] = float(times[best])
        corrections[i] = detected[i] - predicted[i]
        baseline = float(np.median(local))
        spread = float(np.percentile(local, 90) - baseline) + 1e-6
        confidence[i] = float(np.clip((score[best] - baseline) / (2 * spread), 0, 1))

    # Margine anti-bleed: privilegia l'assenza di audio della traccia precedente.
    # È uniforme e configurabile, quindi non introduce deriva cumulativa.
    if anti_bleed > 0 and len(detected) > 1:
        detected[1:] = np.minimum(detected[1:] + anti_bleed, audio_duration)
        corrections[1:] = detected[1:] - predicted[1:]
    ends = np.r_[detected[1:], audio_duration]
    useful_ends = ends.copy()
    track_durations = useful_ends - detected
    if end_guard > 0 and len(track_durations) > 1:
        track_durations[:-1] = np.maximum(0.05, track_durations[:-1] - end_guard)
    return predicted, detected, corrections, track_durations, confidence


def write_boundaries(path: Path, songs: list[Song], predicted, detected, corrections, durations, confidence, force: bool):
    if path.exists() and not force:
        if not sys.stdin.isatty():
            raise FileExistsError(f"{path} esiste già: non lo sovrascrivo. Usare --force dopo averne salvato le correzioni.")
        answer = input(f"{path} esiste e potrebbe essere stato corretto manualmente. Sovrascrivere? [s/N] ").strip().lower()
        if answer not in {"s", "si", "sì"}:
            raise RuntimeError("Operazione annullata: boundaries.csv è stato conservato.")
    with path.open("w", encoding="utf-8-sig", newline="") as handle:
        writer = csv.DictWriter(handle, fieldnames=CSV_FIELDS)
        writer.writeheader()
        for i, song in enumerate(songs):
            writer.writerow({"indice": song.index, "titolo": song.title,
                             "start_previsto": f"{predicted[i]:.3f}", "start_rilevato": f"{detected[i]:.3f}",
                             "correzione": f"{corrections[i]:+.3f}", "durata_rilevata": f"{durations[i]:.3f}",
                             "affidabilità": f"{confidence[i]:.3f}"})
    LOG.info("Creato %s", path)


def read_boundaries(path: Path, songs: list[Song]):
    with path.open("r", encoding="utf-8-sig", newline="") as handle:
        rows = list(csv.DictReader(handle))
    if len(rows) != len(songs):
        raise ValueError(f"{path} contiene {len(rows)} righe, ma la playlist contiene {len(songs)} canzoni.")
    result = []
    for song, row in zip(songs, rows):
        if int(row["indice"]) != song.index:
            raise ValueError(f"Indice non coerente in boundaries.csv alla canzone {song.index}.")
        try:
            start, duration = float(row["start_rilevato"].replace(",", ".")), float(row["durata_rilevata"].replace(",", "."))
        except (KeyError, ValueError) as exc:
            raise ValueError(f"Valori temporali non validi alla riga {song.index + 1} di {path}") from exc
        if start < 0 or duration <= 0:
            raise ValueError(f"Inizio/durata non validi per la canzone {song.index}.")
        result.append((start, duration))
    return result


def safe_filename(value: str) -> str:
    invalid = '<>:"/\\|?*'
    cleaned = "".join("_" if char in invalid or ord(char) < 32 else char for char in value).rstrip(" .")
    reserved = {"CON", "PRN", "AUX", "NUL", *(f"COM{i}" for i in range(1, 10)), *(f"LPT{i}" for i in range(1, 10))}
    if cleaned.upper() in reserved:
        cleaned = "_" + cleaned
    return cleaned or "Senza titolo"


def find_long_internal_silence(audio: Path, start: float, duration: float):
    """Restituisce la pausa quasi muta più lunga, se dura almeno 8 s ed è seguita da musica."""
    rate = 200
    command = [require_program("ffmpeg"), "-v", "error", "-ss", f"{start:.3f}", "-t", f"{duration:.3f}",
               "-i", str(audio), "-vn", "-ac", "1", "-ar", str(rate), "-f", "f32le", "pipe:1"]
    raw = subprocess.check_output(command)
    samples = np.frombuffer(raw, dtype="<f4")
    seconds = samples.size // rate
    if seconds < 60:
        return None
    frames = samples[:seconds * rate].reshape(seconds, rate)
    db = 20 * np.log10(np.sqrt(np.mean(frames * frames, axis=1)) + 1e-12)
    silent = db < -45.0
    runs, run_start = [], None
    for pos, value in enumerate(np.r_[silent, False]):
        if value and run_start is None:
            run_start = pos
        elif not value and run_start is not None:
            if pos - run_start >= 8 and run_start >= 30 and seconds - pos >= 30:
                runs.append((run_start, pos))
            run_start = None
    return max(runs, key=lambda item: item[1] - item[0]) if runs else None

def split_tracks(audio: Path, songs: list[Song], boundaries, output_dir: Path, overwrite: bool,
                 truncate_tracks: set[int] | None = None):
    truncate_tracks = truncate_tracks or set()
    output_dir.mkdir(parents=True, exist_ok=True)
    ffmpeg = require_program("ffmpeg")
    for song, (start, duration) in tqdm(list(zip(songs, boundaries)), desc="Estrazione brani", dynamic_ncols=True):
        output = output_dir / f"{song.index:03d} - {safe_filename(song.title)}.mp3"
        if output.exists() and not overwrite:
            raise FileExistsError(f"Il file {output} esiste già. Usare --overwrite per sostituirlo.")
        track = song.track or str(song.index)
        metadata = {"title": song.title, "artist": song.artist, "author": song.artist,
                    "album_artist": song.artist, "album": song.album, "date": song.year,
                    "track": track, "genre": song.genre}
        command = [ffmpeg, "-hide_banner", "-loglevel", "error", "-ss", f"{start:.3f}", "-t", f"{duration:.3f}",
                   "-i", str(audio), "-vn", "-map_metadata", "-1"]
        special = song.index in truncate_tracks
        silence = find_long_internal_silence(audio, start, duration) if special else None
        if silence:
            silence_start, silence_end = silence
            cut_at = max(0.05, silence_start - 0.25)
            graph = f"[0:a]atrim=start=0:end={cut_at:.3f},asetpts=PTS-STARTPTS[outa]"
            command.extend(["-filter_complex", graph, "-map", "[outa]"])
            LOG.info("Traccia %d: termino al primo silenzio lungo, taglio a +%.3f s", song.index, cut_at)
        else:
            command.extend(["-map", "0:a:0"])
            if special:
                LOG.warning("Traccia %d: pausa interna lunga non trovata durante il taglio", song.index)
        command.extend(["-c:a", "libmp3lame", "-q:a", "2"])
        for key, value in metadata.items():
            command.extend(["-metadata", f"{key}={value}"])
        command.extend(["-y" if overwrite else "-n", str(output)])
        completed = subprocess.run(command, capture_output=True, text=True, encoding="utf-8", errors="replace")
        if completed.returncode:
            raise RuntimeError(f"ffmpeg ha fallito su {song.index:03d} - {song.title}: {completed.stderr.strip()}")
    LOG.info("Creati %d file in %s", len(songs), output_dir)


def build_parser() -> argparse.ArgumentParser:
    parser = argparse.ArgumentParser(description="Divide una registrazione continua usando una playlist iTunes TSV.")
    parser.add_argument("--audio", type=Path, default=Path("album.mp3"), help="registrazione MP3 (default: album.mp3)")
    parser.add_argument("--playlist", type=Path, default=Path("playlist.txt"), help="esportazione iTunes TSV")
    parser.add_argument("--boundaries", type=Path, default=Path("boundaries.csv"), help="CSV dei confini")
    parser.add_argument("--window", type=float, default=8.0, help="semifinestra locale in secondi (default: 8)")
    parser.add_argument("--anti-bleed", type=float, default=0.20, help="margine dopo l'attacco in secondi (default: 0.20)")
    parser.add_argument("--end-guard", type=float, default=0.45, help="anticipo della fine rispetto al prossimo inizio (default: 0.45)")
    parser.add_argument("--analyze-only", action="store_true", help="analizza e crea il CSV senza tagliare")
    parser.add_argument("--split-from", type=Path, help="taglia usando un boundaries.csv esistente/corretto")
    parser.add_argument("--output-dir", type=Path, default=Path("Brani"), help="cartella di destinazione")
    parser.add_argument("--force", action="store_true", help="consente di sovrascrivere boundaries.csv")
    parser.add_argument("--overwrite", action="store_true", help="consente di sovrascrivere MP3 già estratti")
    parser.add_argument("--truncate-at-long-silence", type=int, action="append", default=[], metavar="INDICE",
                        help="termina questa traccia al primo silenzio lungo; ripetibile")
    parser.add_argument("--log-file", type=Path, default=Path("playlist_splitter.log"), help="file di log dettagliato")
    return parser


def main() -> int:
    args = build_parser().parse_args()
    logging.basicConfig(level=logging.INFO, format="%(asctime)s | %(levelname)s | %(message)s",
                        handlers=[logging.StreamHandler(), logging.FileHandler(args.log_file, encoding="utf-8")])
    try:
        if sys.version_info < (3, 10):
            raise RuntimeError("È richiesto Python 3.10 o successivo.")
        require_program("ffmpeg")
        require_program("ffprobe")
        all_songs = load_playlist(args.playlist)
        print(f"Trovate {len(all_songs)} canzoni in {args.playlist}.")
        audio = resolve_audio(args.audio)
        info = probe_audio(audio)
        if args.split_from:
            with args.split_from.open("r", encoding="utf-8-sig", newline="") as handle:
                boundary_count = sum(1 for _ in csv.DictReader(handle))
            if not 0 < boundary_count <= len(all_songs):
                raise ValueError(
                    f"{args.split_from} contiene {boundary_count} righe, ma la playlist ne contiene "
                    f"{len(all_songs)}."
                )
            # Un CSV corretto manualmente è la fonte autorevole: non tentare di
            # dedurne il numero di tracce dalla somma delle durate iTunes.
            songs = all_songs[:boundary_count]
            print(f"Uso le {len(songs)} tracce indicate da {args.split_from}.")
            boundaries = read_boundaries(args.split_from, songs)
            split_tracks(audio, songs, boundaries, args.output_dir, args.overwrite,
                         set(args.truncate_at_long_silence))
            return 0
        songs = match_recorded_prefix(all_songs, info["duration"])
        print(f"La registrazione contiene circa le prime {len(songs)} canzoni.")
        times, rms, flux, energy = extract_features(audio, info["duration"])
        result = choose_boundaries(songs, info["duration"], times, rms, flux, energy, args.window, args.anti_bleed, args.end_guard)
        write_boundaries(args.boundaries, songs, *result, force=args.force)
        if not args.analyze_only:
            split_tracks(audio, songs, list(zip(result[1], result[3])), args.output_dir, args.overwrite,
                         set(args.truncate_at_long_silence))
        return 0
    except (FileNotFoundError, FileExistsError, ValueError, RuntimeError, OSError) as exc:
        LOG.error("%s", exc)
        return 2


if __name__ == "__main__":
    raise SystemExit(main())

py -m pip install -r requirements.txt

py playlist_splitter.py --analyze-only

py playlist_splitter.py --split-from boundaries.csv