Import existing LinTunes project

Snapshot of the existing codebase before working through the TASKS.md
backlog. Real library data (data/) and the iTunes import fixture
(itunes-test-library/) are gitignored.

Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>

lintunes/storage/conflict_resolver.py

@@ -0,0 +1,165 @@
+import json
+import re
+import shutil
+from datetime import datetime
+from pathlib import Path
+
+from lintunes.models import Track, Playlist
+
+
+# Syncthing conflict file pattern: filename.sync-conflict-20240101-123456-ABCDEFG.json
+CONFLICT_PATTERN = re.compile(r"^(.+)\.sync-conflict-(\d{8}-\d{6})-\w+(\.\w+)$")
+
+
+def resolve_conflicts(data_dir: Path):
+    if not data_dir.exists():
+        return
+
+    resolved_dir = data_dir / ".resolved"
+    conflict_files = _find_conflict_files(data_dir)
+
+    if not conflict_files:
+        return
+
+    print(f"Found {len(conflict_files)} Syncthing conflict(s), resolving...")
+    resolved_dir.mkdir(exist_ok=True)
+
+    for conflict_path, original_path in conflict_files:
+        if not original_path.exists():
+            # Original was deleted; just archive the conflict
+            _archive(conflict_path, resolved_dir)
+            continue
+
+        try:
+            if original_path.name == "library.json":
+                _merge_library(original_path, conflict_path)
+            elif original_path.name == "library_metadata.json":
+                _merge_metadata(original_path, conflict_path)
+            elif original_path.parent.name == "playlists":
+                _merge_playlist(original_path, conflict_path)
+            else:
+                # Unknown file type — keep the newer one
+                _keep_newer(original_path, conflict_path)
+        except Exception as e:
+            print(f"  Warning: Failed to merge {conflict_path.name}: {e}")
+
+        _archive(conflict_path, resolved_dir)
+
+    print(f"  Resolved {len(conflict_files)} conflict(s)")
+
+
+def _find_conflict_files(data_dir: Path) -> list[tuple[Path, Path]]:
+    results = []
+    for path in data_dir.rglob("*.sync-conflict-*"):
+        match = CONFLICT_PATTERN.match(path.name)
+        if match:
+            original_name = match.group(1) + match.group(3)
+            original_path = path.parent / original_name
+            results.append((path, original_path))
+    return results
+
+
+def _merge_library(original_path: Path, conflict_path: Path):
+    original = _load_json(original_path)
+    conflict = _load_json(conflict_path)
+
+    for tid, conflict_track_data in conflict.items():
+        if tid not in original:
+            # New track from conflict side
+            original[tid] = conflict_track_data
+            continue
+
+        orig_track = original[tid]
+        _merge_track_fields(orig_track, conflict_track_data)
+
+    _save_json(original_path, original)
+
+
+def _merge_track_fields(orig: dict, conflict: dict):
+    # play_count, skip_count: take max
+    for field in ("play_count", "skip_count"):
+        if field in conflict:
+            orig[field] = max(orig.get(field, 0), conflict[field])
+
+    # play_date_utc: take most recent
+    for field in ("play_date_utc", "skip_date"):
+        if field in conflict:
+            orig_val = orig.get(field)
+            conf_val = conflict[field]
+            if orig_val is None or (conf_val and conf_val > orig_val):
+                orig[field] = conf_val
+
+    # date_added: take earliest
+    if "date_added" in conflict:
+        orig_val = orig.get("date_added")
+        conf_val = conflict["date_added"]
+        if orig_val is None or (conf_val and conf_val < orig_val):
+            orig["date_added"] = conf_val
+
+    # loved: OR (true wins)
+    if conflict.get("loved", False):
+        orig["loved"] = True
+
+    # rating, metadata: newest date_modified wins
+    orig_mod = orig.get("date_modified")
+    conf_mod = conflict.get("date_modified")
+    if conf_mod and (not orig_mod or conf_mod > orig_mod):
+        # Conflict is newer — take its metadata fields
+        for field in ("rating", "name", "artist", "album_artist", "album",
+                       "genre", "composer", "grouping", "comments",
+                       "sort_name", "sort_artist", "sort_album_artist",
+                       "sort_album", "sort_composer", "date_modified"):
+            if field in conflict:
+                orig[field] = conflict[field]
+
+
+def _merge_playlist(original_path: Path, conflict_path: Path):
+    original = _load_json(original_path)
+    conflict = _load_json(conflict_path)
+
+    # Union track_ids, order from the newer file
+    orig_ids = set(original.get("track_ids", []))
+    conf_ids = set(conflict.get("track_ids", []))
+    all_ids = orig_ids | conf_ids
+
+    # Determine which file is newer by checking file mtime
+    if conflict_path.stat().st_mtime > original_path.stat().st_mtime:
+        # Use conflict's order as base, append any missing from original
+        base_order = conflict.get("track_ids", [])
+        extra = [tid for tid in original.get("track_ids", []) if tid not in conf_ids]
+        original["track_ids"] = base_order + extra
+        # Settings from newer file
+        if "settings" in conflict:
+            original["settings"] = conflict["settings"]
+    else:
+        # Original is newer — append missing from conflict
+        extra = [tid for tid in conflict.get("track_ids", []) if tid not in orig_ids]
+        original["track_ids"] = original.get("track_ids", []) + extra
+
+    _save_json(original_path, original)
+
+
+def _merge_metadata(original_path: Path, conflict_path: Path):
+    # Simple: keep the newer file's content
+    _keep_newer(original_path, conflict_path)
+
+
+def _keep_newer(original_path: Path, conflict_path: Path):
+    if conflict_path.stat().st_mtime > original_path.stat().st_mtime:
+        shutil.copy2(conflict_path, original_path)
+
+
+def _archive(conflict_path: Path, resolved_dir: Path):
+    timestamp = datetime.now().strftime("%Y%m%d-%H%M%S")
+    archive_name = f"{timestamp}_{conflict_path.name}"
+    shutil.move(str(conflict_path), str(resolved_dir / archive_name))
+
+
+def _load_json(path: Path) -> dict:
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)
+
+
+def _save_json(path: Path, data: dict):
+    with open(path, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)

lintunes/storage/json_storage.py

@@ -0,0 +1,106 @@
+import json
+from pathlib import Path
+
+from lintunes.models import Track, Playlist, PlaylistSettings, Library
+
+
+def save_library(library: Library, data_dir: Path):
+    data_dir.mkdir(parents=True, exist_ok=True)
+    save_tracks(library, data_dir)
+    save_metadata(library, data_dir)
+
+    playlists_dir = data_dir / "playlists"
+    playlists_dir.mkdir(exist_ok=True)
+    keep = set()
+    for persistent_id, playlist in library.playlists.items():
+        save_playlist(playlist, data_dir)
+        keep.add(f"{persistent_id}.json")
+    # Remove files for playlists that no longer exist
+    for playlist_file in playlists_dir.glob("*.json"):
+        if playlist_file.name not in keep:
+            playlist_file.unlink()
+
+
+def save_tracks(library: Library, data_dir: Path):
+    data_dir.mkdir(parents=True, exist_ok=True)
+    tracks_dict = {str(tid): track.to_dict() for tid, track in library.tracks.items()}
+    _write_json(data_dir / "library.json", tracks_dict)
+
+
+def save_metadata(library: Library, data_dir: Path):
+    data_dir.mkdir(parents=True, exist_ok=True)
+    metadata = {
+        "music_folder": library.music_folder,
+        "import_date": library.import_date,
+        "track_count": len(library.tracks),
+        "playlist_count": len(library.playlists),
+        "library_settings": library.library_settings.to_dict(),
+    }
+    _write_json(data_dir / "library_metadata.json", metadata)
+
+
+def save_playlist(playlist: Playlist, data_dir: Path):
+    playlists_dir = data_dir / "playlists"
+    playlists_dir.mkdir(parents=True, exist_ok=True)
+    _write_json(playlists_dir / f"{playlist.persistent_id}.json", playlist.to_dict())
+
+
+def delete_playlist_file(persistent_id: str, data_dir: Path):
+    path = data_dir / "playlists" / f"{persistent_id}.json"
+    if path.exists():
+        path.unlink()
+
+
+def load_library(data_dir: Path) -> Library:
+    if not data_dir.exists():
+        return Library()
+
+    # Load tracks
+    tracks = {}
+    library_path = data_dir / "library.json"
+    if library_path.exists():
+        tracks_dict = _read_json(library_path)
+        for tid_str, track_data in tracks_dict.items():
+            track = Track.from_dict(track_data)
+            tracks[track.track_id] = track
+
+    # Load metadata
+    music_folder = ""
+    import_date = None
+    library_settings = PlaylistSettings()
+    metadata_path = data_dir / "library_metadata.json"
+    if metadata_path.exists():
+        metadata = _read_json(metadata_path)
+        music_folder = metadata.get("music_folder", "")
+        import_date = metadata.get("import_date")
+        if "library_settings" in metadata:
+            library_settings = PlaylistSettings.from_dict(metadata["library_settings"])
+
+    # Load playlists
+    playlists = {}
+    playlists_dir = data_dir / "playlists"
+    if playlists_dir.exists():
+        for playlist_file in playlists_dir.glob("*.json"):
+            playlist_data = _read_json(playlist_file)
+            playlist = Playlist.from_dict(playlist_data)
+            playlists[playlist.persistent_id] = playlist
+
+    return Library(
+        tracks=tracks,
+        playlists=playlists,
+        music_folder=music_folder,
+        import_date=import_date,
+        library_settings=library_settings,
+    )
+
+
+def _write_json(path: Path, data: dict):
+    tmp = path.with_suffix(path.suffix + ".tmp")
+    with open(tmp, "w", encoding="utf-8") as f:
+        json.dump(data, f, indent=2, ensure_ascii=False)
+    tmp.replace(path)
+
+
+def _read_json(path: Path) -> dict:
+    with open(path, "r", encoding="utf-8") as f:
+        return json.load(f)