lintunes/lintunes/importers/itunes_importer.py

import os
import plistlib
import unicodedata
from dataclasses import dataclass, field
from datetime import datetime
from pathlib import Path

from lintunes.models import Track, Playlist, PlaylistType, Library
from lintunes.models.track import decode_location_url


@dataclass
class ImportReport:
    track_count: int = 0
    missing_files: list[str] = field(default_factory=list)
    unmapped_locations: list[str] = field(default_factory=list)
    skipped_tracks: int = 0  # non-file tracks (streams, remote)
    skipped_video: int = 0   # movies / TV shows / home videos
    case_fixed: int = 0      # locations fixed by case/normalization matching
    playlist_count: int = 0
    folder_count: int = 0
    smart_count: int = 0           # smart playlists imported
    smart_unsupported: int = 0     # of those, criteria kept as a static snapshot
    skipped_system: int = 0

    def summary(self) -> str:
        lines = [
            f"Tracks imported:    {self.track_count}",
            f"Tracks skipped:     {self.skipped_tracks} (streams/remote), "
            f"{self.skipped_video} (video)",
            f"Locations adjusted: {self.case_fixed} (case/unicode normalization)",
            f"Missing files:      {len(self.missing_files)}",
            f"Unmapped locations: {len(self.unmapped_locations)}",
            f"Playlists imported: {self.playlist_count} (+ {self.folder_count} folders)",
            f"Smart playlists:    {self.smart_count} imported "
            f"({self.smart_unsupported} kept as snapshot)",
            f"Playlists skipped:  {self.skipped_system} system",
        ]
        return "\n".join(lines)


def make_location_remapper(music_folder_url: str, music_root: Path):
    """Map decoded iTunes locations under the XML's Music Folder onto music_root.

    music_folder_url is the raw 'Music Folder' value from the XML, e.g.
    file:///Volumes/Lucia/iTunes/iTunes%20Media/ — paths under it are
    rewritten to live under music_root. Paths outside it are returned
    unchanged (and end up in the unmapped report).
    """
    prefix = decode_location_url(music_folder_url).rstrip("/")
    root = str(music_root).rstrip("/")

    def remap(location: str) -> str:
        if prefix and location.startswith(prefix + "/"):
            return root + location[len(prefix):]
        return location

    return remap


class FuzzyPathResolver:
    """Resolve paths whose case or Unicode normalization differs on disk.

    macOS filesystems are case-insensitive and store names NFD-decomposed;
    on Linux the iTunes XML's exact path often misses the real file. Directory
    listings are cached, keyed by the folded (NFC + casefold) child name.
    """

    def __init__(self):
        self._cache: dict[str, dict[str, str] | None] = {}

    @staticmethod
    def _fold(name: str) -> str:
        return unicodedata.normalize("NFC", name).casefold()

    def _listing(self, directory: str) -> dict[str, str] | None:
        if directory not in self._cache:
            try:
                self._cache[directory] = {
                    self._fold(entry): entry for entry in os.listdir(directory)
                }
            except OSError:
                self._cache[directory] = None
        return self._cache[directory]

    def resolve(self, path_str: str) -> str | None:
        """Return the on-disk path matching path_str, or None."""
        if os.path.exists(path_str):
            return path_str
        parts = Path(path_str).parts
        if not parts:
            return None
        current = parts[0]
        for part in parts[1:]:
            candidate = os.path.join(current, part)
            if os.path.exists(candidate):
                current = candidate
                continue
            listing = self._listing(current)
            if listing is None:
                return None
            match = listing.get(self._fold(part))
            if match is None:
                return None
            current = os.path.join(current, match)
        return current


def import_itunes_xml(xml_path: Path, music_root: Path,
                      check_files: bool = True) -> tuple[Library, ImportReport]:
    with open(xml_path, "rb") as f:
        plist = plistlib.load(f)

    report = ImportReport()
    music_folder_url = plist.get("Music Folder", "")
    remap = make_location_remapper(music_folder_url, music_root)
    music_root_str = str(music_root).rstrip("/")

    # Import tracks (audio file tracks only — no streams, no video)
    tracks = {}
    resolver = FuzzyPathResolver()
    for track_id_str, itunes_track in plist.get("Tracks", {}).items():
        if itunes_track.get("Track Type", "File") != "File":
            report.skipped_tracks += 1
            continue
        if (itunes_track.get("Movie") or itunes_track.get("TV Show")
                or itunes_track.get("Has Video")):
            report.skipped_video += 1
            continue
        track = Track.from_itunes_dict(itunes_track, remap_location=remap)
        if not track.location:
            report.skipped_tracks += 1
            continue
        if not track.location.startswith(music_root_str + "/"):
            report.unmapped_locations.append(track.location)
        elif check_files and not Path(track.location).exists():
            resolved = resolver.resolve(track.location)
            if resolved is not None:
                track.location = resolved
                report.case_fixed += 1
            else:
                report.missing_files.append(track.location)
        tracks[track.track_id] = track
    report.track_count = len(tracks)

    # Import playlists: user playlists and folders only
    playlists = {}
    kept_folder_ids = set()
    for itunes_playlist in plist.get("Playlists", []):
        playlist = Playlist.from_itunes_dict(itunes_playlist)
        if playlist.is_system or playlist.playlist_type == PlaylistType.SYSTEM:
            report.skipped_system += 1
            continue
        if playlist.playlist_type == PlaylistType.SMART:
            from lintunes import smart
            playlist.smart_criteria = smart.parse_itunes_smart(
                itunes_playlist.get("Smart Info"),
                itunes_playlist.get("Smart Criteria"))
            report.smart_count += 1
            if playlist.smart_criteria.unsupported:
                report.smart_unsupported += 1
        # Drop references to tracks we didn't import. For smart playlists this
        # snapshot is the fallback membership until the first recompute (and the
        # permanent membership when the criteria are unsupported).
        playlist.track_ids = [tid for tid in playlist.track_ids if tid in tracks]
        playlists[playlist.persistent_id] = playlist
        if playlist.playlist_type == PlaylistType.FOLDER:
            kept_folder_ids.add(playlist.persistent_id)

    # Detach children whose parent folder wasn't imported
    for playlist in playlists.values():
        if playlist.parent_persistent_id and playlist.parent_persistent_id not in kept_folder_ids:
            playlist.parent_persistent_id = ""

    report.folder_count = len(kept_folder_ids)
    # Regular playlists only; folders and smart playlists are counted separately.
    report.playlist_count = len(playlists) - report.folder_count - report.smart_count

    library = Library(
        tracks=tracks,
        playlists=playlists,
        music_folder=str(music_root),
        import_date=datetime.now().isoformat(),
    )
    return library, report