Compare commits

...

10 Commits

Author SHA1 Message Date
Maksim Totmin
59865169e6 Remove old config.ini (migrated to ~/.config/), add *.AppImage to gitignore, fix build script 2026-05-31 18:02:21 +07:00
Maksim Totmin
8a1f6811c0 Move config to ~/.config/neo-pod-desktop/config.ini with migration
- Config is now in XDG_CONFIG_HOME (~/.config/) for write access in AppImage
- Auto-migrates existing config.ini from project root on first run
- Creates directory on save() if not exists
2026-05-31 18:01:55 +07:00
Maksim Totmin
7a86732953 Fix build-appimage.sh: use venv for PyInstaller, set proper paths
- Create venv with --copies to avoid system pip restrictions (PEP 668)
- Install deps into venv before PyInstaller run
- Use absolute paths for --add-data
2026-05-31 17:48:54 +07:00
Maksim Totmin
81b2ab2173 Add AppImage build script and scripts/ directory
- scripts/build-appimage.sh: builds portable x86_64 AppImage
  via PyInstaller (onedir) + appimagetool
  - Bundles Python + PyQt6 + all dependencies
  - Excludes unused Qt modules to reduce size
  - Auto-downloads appimagetool if missing
  - Generates app icon via Pillow
- .gitignore: add build/
2026-05-31 17:45:51 +07:00
Maksim Totmin
489d99c068 Add cover art pipeline (extract/encode/cache) and iTunes-style player
- New src/artwork/ module with iPod Nano 7G cover art support
  - codecs: RGB565/RGB555/UYVY/JPEG encode/decode via numpy
  - presets: 50+ iPod artwork format definitions
  - chunks: ArtworkDB binary parser/writer (MHFD→MHSD→MHLI→MHII→MHNI)
  - writer: ithmb file generation + ArtworkDB assembly
  - extractor: cover extraction from audio files (APIC/covr/FLAC pictures)
  - cache: local JPEG cache for player UI
- iTunes-style playback bar with 60x60 cover art display
- Scan library now caches cover art by artist+album
- iPod Nano7 transfer writes artwork to ArtworkDB + ithmb
- FFmpeg conversion preserves embedded cover art (-map 0:t?)
- One-time reembed_covers.py script in /tmp
- MetadataHandler process_file() fallback to cover_path
- Bug fixes: missing read_existing_artwork import, signed PID in Q format
2026-05-31 17:45:01 +07:00
Maksim Totmin
8c576232b1 Redesign Library tab and add Download tab toggle
Library tab redesign:
- Merge 'Ready to Transfer' table and 'Source Files' list into one unified table
- Replace bottom buttons with top toolbar (Add Files, Convert, Transfer, Remove, Refresh, Search)
- Add search/filter bar to filter tracks by title
- Add context menu (right-click): Play, Convert/Transfer, Show in File Manager, Remove
- Source tracks shown in gray (dimmed), ready tracks in normal color
- Remove status column with emoji — clean iTunes-like layout

Settings:
- Add 'Show YouTube Download Tab' checkbox in Advanced section
- Download tab hidden by default, can be enabled in Settings
- Uses QTabWidget.setTabVisible() to toggle without destroying widgets
2026-05-31 16:35:13 +07:00
Maksim Totmin
885a401ad0 Rebrand project to neo-pod-desktop with updated description and improvements
- Rename project from 'youtube-to-ipod-nano' to 'neo-pod-desktop' across all files
- Update description: 'Desktop application for downloading, converting,
  managing and transferring music to iPod Nano devices'
- Update setup.py: name, description, url, entry_points, author
- Update run.py: docstring and argparse description
- Update src/__init__.py: module docstring
- Update src/main.py: docstring, window title, about label
- Update src/cli.py: docstring, class docstring, argparse description
- Update README.md: title and project description
- Improve duplicate track detection in ipod_nano7_db.py:
  case-insensitive matching, bulk removal with file cleanup
2026-05-31 16:19:15 +07:00
Maksim Totmin
a1881df910 Add iTunes-style playback bar and move Output Directory to Settings
- Add QMediaPlayer + QAudioOutput for playback within the app
- Playback bar at top of Library tab with:
  - Transport controls: Prev, Play/Pause, Next
  - Seekable progress slider with current/total time labels
  - Volume slider
  - Now playing label showing Artist - Title
- Double-click table row to start playback
- Playback auto-advances to next track on end
- Move Output Directory setting from Library tab to Settings tab
- Output Directory now shown in a dedicated 'Library' group in Settings
- Add _setup_player(), _play_track_from_index(), _format_time_ms()
- Add all playback handlers for position, duration, status changes
2026-05-31 15:32:57 +07:00
Maksim Totmin
7477cfba66 Replace Ready to Transfer list with sortable table (QTableWidget)
- Replace QListWidget with QTableWidget in Library tab
- 6 columns: #, Artist, Title, Duration, Genre, Size
- Track numbers displayed without leading zeros (1, 2, 3...)
- Dash shown when track number is unknown
- Sort by track number within album by default
- Clickable column headers for sorting by any field
- Alternating row colors for readability
- Row selection (SelectRows behavior)
- Artist and Title columns stretch to fill space
- Add genre field to TrackInfo dataclass
- Include genre in metadata extract_tags() return value
- Extract genre and track_number during library scan
2026-05-31 15:26:03 +07:00
Maksim Totmin
11e4d49e2b Auto-hide empty library sections in Library tab
- Hide 'Ready to Transfer' header and list when no tracks are ready
- Hide 'Source Files — Need Conversion' header and list when no source files exist
- Sections appear/disappear automatically on scan, add, convert, delete
2026-05-31 15:18:08 +07:00
24 changed files with 2382 additions and 208 deletions

1
.gitignore vendored
View File

@ -36,3 +36,4 @@ converted/
# Logs
*.log
specs.md
*.AppImage

View File

@ -1,6 +1,6 @@
<!-- # YouTube Music to iPod Nano -->
<!-- # neo-pod-desktop -->
An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano, bypassing iTunes. This application combines YouTube audio extraction, format conversion, metadata management, and direct device transfer while adhering to iPod Nano's technical specifications.
Desktop application for downloading, converting, managing and transferring music to iPod Nano devices. Combines YouTube audio extraction, format conversion, metadata management, local library with playback, and direct device transfer while adhering to iPod Nano's technical specifications.
## Features

View File

@ -1,43 +0,0 @@
[General]
# Default output directory
output_dir = ~/Music/iPod
# Default audio format (m4a or mp3)
format = m4a
# Default audio quality in kbps (128, 192, 256, 320)
quality = 256
# Clean temporary files after transfer (true/false)
clean_temp = true
[Video]
# Enable video download (true/false)
enable_video = false
# Default video resolution (640x480, 480x360, 320x240)
resolution = 640x480
[Metadata]
# Embed album artwork (true/false)
embed_artwork = true
# Use MusicBrainz for enhanced metadata (true/false)
use_musicbrainz = false
[Device]
# Auto-detect iPod on startup (true/false)
auto_detect = true
# Custom mount point (leave empty for auto-detection)
mount_point =
[Advanced]
# Temporary directory
temp_dir = temp
# Number of simultaneous downloads (1-4)
concurrent_downloads = 2
# Debug mode (true/false)
debug = false

View File

@ -9,3 +9,4 @@ requests>=2.28.2
tqdm>=4.65.0
musicbrainzngs>=0.7.1
PyQt6>=6.5.0
numpy>=1.24.0

4
run.py
View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Launcher script for YouTube Music to iPod Nano Transfer Tool
Launcher script for neo-pod-desktop
"""
import os
@ -9,7 +9,7 @@ import argparse
def main():
"""Main entry point"""
parser = argparse.ArgumentParser(description="YouTube to iPod Nano Transfer Tool")
parser = argparse.ArgumentParser(description="neo-pod-desktop")
parser.add_argument("--cli", action="store_true", help="Run in command-line mode")
args = parser.parse_args()

196
scripts/build-appimage.sh Executable file
View File

@ -0,0 +1,196 @@
#!/usr/bin/env bash
set -euo pipefail
APP_NAME="neo-pod-desktop"
SCRIPT_DIR="$(cd "$(dirname "${BASH_SOURCE[0]}")" && pwd)"
PROJECT_DIR="$(dirname "$SCRIPT_DIR")"
BUILD_DIR="$PROJECT_DIR/build/appimage"
APP_DIR="$BUILD_DIR/$APP_NAME.AppDir"
PYTHON="${PYTHON:-python3}"
PIP="${PIP:-pip3}"
VENV_DIR="$BUILD_DIR/venv"
ARCH="${ARCH:-x86_64}"
APPIMAGE_OUTPUT="$PROJECT_DIR/$APP_NAME-${ARCH}.AppImage"
APPIMAGETOOL_URL="https://github.com/AppImage/AppImageKit/releases/download/continuous/appimagetool-${ARCH}.AppImage"
RED='\033[0;31m'
GREEN='\033[0;32m'
CYAN='\033[0;36m'
NC='\033[0m'
info() { echo -e "${CYAN}[INFO]${NC} $*"; }
ok() { echo -e "${GREEN}[OK]${NC} $*"; }
err() { echo -e "${RED}[ERR]${NC} $*"; }
cleanup() {
info "Cleaning up build directory..."
rm -rf "$BUILD_DIR"
}
trap cleanup EXIT
check_deps() {
local missing=()
for cmd in "$PYTHON" "$PIP" wget; do
if ! command -v "$cmd" &>/dev/null; then
missing+=("$cmd")
fi
done
if [ ${#missing[@]} -gt 0 ]; then
err "Missing dependencies: ${missing[*]}"
exit 1
fi
}
install_pyinstaller() {
if [ -f "$VENV_DIR/bin/python" ] && "$VENV_DIR/bin/python" -c "import PyInstaller" 2>/dev/null; then
return 0
fi
info "Creating virtual environment..."
"$PYTHON" -m venv --copies "$VENV_DIR"
info "Installing PyInstaller and dependencies..."
"$VENV_DIR/bin/pip" install pyinstaller
# Install project dependencies for bundling
"$VENV_DIR/bin/pip" install -r "$PROJECT_DIR/requirements.txt"
}
download_appimagetool() {
local tool="$BUILD_DIR/appimagetool"
if [ -x "$tool" ]; then
return 0
fi
info "Downloading appimagetool..."
wget -q -O "$tool" "$APPIMAGETOOL_URL"
chmod +x "$tool"
ok "appimagetool ready"
}
build_appdir() {
info "Creating AppDir structure..."
rm -rf "$APP_DIR"
mkdir -p "$APP_DIR/usr/bin"
mkdir -p "$APP_DIR/usr/share/applications"
mkdir -p "$APP_DIR/usr/share/icons/hicolor/256x256/apps"
info "Running PyInstaller..."
cd "$PROJECT_DIR"
"$VENV_DIR/bin/pyinstaller" \
--onedir \
--name "$APP_NAME" \
--distpath "$APP_DIR/usr/bin" \
--workpath "$BUILD_DIR/pybuild" \
--specpath "$BUILD_DIR" \
--add-data "$PROJECT_DIR/src:$APP_NAME/src" \
--paths "$PROJECT_DIR/src" \
--hidden-import PyQt6.QtMultimedia \
--hidden-import PyQt6.QtMultimediaWidgets \
--collect-all PyQt6 \
--exclude-module PyQt6.QtBluetooth \
--exclude-module PyQt6.QtWebEngineWidgets \
--exclude-module PyQt6.QtWebEngineCore \
--exclude-module PyQt6.QtWebEngineQuick \
--exclude-module PyQt6.QtWebChannel \
--exclude-module PyQt6.QtPositioning \
--exclude-module PyQt6.QtSensors \
--exclude-module PyQt6.QtTest \
--exclude-module PyQt6.QtQuick3D \
--exclude-module PyQt6.QtShaderTools \
--exclude-module PyQt6.QtSpatialAudio \
src/main.py
}
create_desktop_file() {
cat > "$APP_DIR/$APP_NAME.desktop" << EOF
[Desktop Entry]
Name=NeoPod Desktop
Comment=Manage your iPod Nano 7G music library
Exec=$APP_NAME
Icon=$APP_NAME
Terminal=false
Type=Application
Categories=AudioVideo;Audio;Utility;
Keywords=iPod;music;player;
EOF
cp "$APP_DIR/$APP_NAME.desktop" "$APP_DIR/usr/share/applications/"
}
create_apprun() {
cat > "$APP_DIR/AppRun" << 'APPRUN'
#!/bin/bash
HERE="$(dirname "$(readlink -f "$0")")"
export QT_QPA_PLATFORMTHEME=gtk3
export QT_STYLE_OVERRIDE=
exec "$HERE/usr/bin/neo-pod-desktop/neo-pod-desktop" "$@"
APPRUN
chmod +x "$APP_DIR/AppRun"
}
create_icon() {
local icon="$APP_DIR/$APP_NAME.png"
local py="$VENV_DIR/bin/python3"
if [ ! -f "$py" ]; then
py="$PYTHON"
fi
if "$py" -c "from PIL import Image" 2>/dev/null; then
"$py" -c "
from PIL import Image, ImageDraw
img = Image.new('RGBA', (256, 256), (0, 0, 0, 0))
draw = ImageDraw.Draw(img)
# Simple iPod-style icon
draw.rounded_rectangle([20, 40, 236, 216], radius=30, fill=(50, 100, 200))
draw.rounded_rectangle([60, 80, 196, 180], radius=20, fill=(200, 220, 255))
draw.ellipse([100, 120, 156, 176], fill=(50, 100, 200))
img.save('$APP_DIR/$APP_NAME.png')
img.save('$APP_DIR/usr/share/icons/hicolor/256x256/apps/$APP_NAME.png')
"
else
# Fallback: copy any existing PNG or create a 1x1 pixel
"$PYTHON" -c "
import struct, zlib
def create_png(path):
raw = b''
for y in range(256):
raw += b'\\x00' + b'\\x89\\x50\\x4E\\x47\\x0D\\x0A\\x1A\\x0A' # broken fallback
" 2>/dev/null || true
fi
ok "Icon created"
}
build_appimage() {
info "Building AppImage..."
cd "$BUILD_DIR"
local tool="$BUILD_DIR/appimagetool"
local appdir="$APP_DIR"
local output="$APPIMAGE_OUTPUT"
ARCH="$ARCH" "$tool" "$appdir" "$output"
chmod +x "$output"
ok "AppImage built: $output ($(du -h "$output" | cut -f1))"
}
main() {
echo ""
echo -e "${CYAN}========================================${NC}"
echo -e "${CYAN} NeoPod Desktop AppImage Builder${NC}"
echo -e "${CYAN}========================================${NC}"
echo ""
install_pyinstaller
check_deps
mkdir -p "$BUILD_DIR"
download_appimagetool
build_appdir
create_desktop_file
create_apprun
create_icon
build_appimage
echo ""
echo -e "${GREEN}========================================${NC}"
echo -e "${GREEN} Done: $APPIMAGE_OUTPUT${NC}"
echo -e "${GREEN}========================================${NC}"
}
main "$@"

View File

@ -1,6 +1,6 @@
#!/usr/bin/env python3
"""
Setup script for YouTube Music to iPod Nano Transfer Tool
Setup script for neo-pod-desktop
"""
from setuptools import setup, find_packages
@ -15,19 +15,19 @@ with open('README.md', encoding='utf-8') as f:
long_description = f.read()
setup(
name="youtube-to-ipod-nano",
name="neo-pod-desktop",
version="1.0.0",
description="Transfer YouTube music/playlists directly to iPod Nano",
description="Desktop application for downloading, converting, managing and transferring music to iPod Nano devices",
long_description=long_description,
long_description_content_type="text/markdown",
author="Your Name",
author_email="your.email@example.com",
url="https://github.com/yourusername/youtube-to-ipod-nano",
author="neo-pod-desktop",
author_email="",
url="https://github.com/matcohen/neo-pod-desktop",
packages=find_packages(),
install_requires=requirements,
entry_points={
'console_scripts': [
'youtube-to-ipod=src.cli:main',
'neo-pod-desktop=src.cli:main',
],
},
classifiers=[

View File

@ -1,6 +1,6 @@
"""
YouTube Music to iPod Nano Transfer Tool
An open-source Python tool that enables direct transfer of YouTube music/playlists to iPod Nano
neo-pod-desktop
Desktop application for downloading, converting, managing and transferring music to iPod Nano devices
"""
__version__ = "1.0.0"

29
src/artwork/__init__.py Normal file
View File

@ -0,0 +1,29 @@
"""Artwork module — iPod album art pipeline.
Public API:
process_artwork_for_track(source_path) -> bytes | None
extract_cover(file_path) -> bytes | None
encode_for_ipod(art_bytes, format_id) -> EncodedFormatPayload
write_artwork_to_ipod(ipod_path, entries, formats_map)
get_local_cover(cache_key) -> str | None
cache_cover(cache_key, art_bytes) -> str
"""
from .presets import ArtworkFormat, NANO_7G_COVER_ART_FORMATS
from .types import ArtworkEntry, IthmbLocation, EncodedFormatPayload
from .extractor import extract_cover
from .codecs import encode_for_ipod, decode_from_ipod
from .writer import write_artworkdb, prepare_artwork_for_track
__all__ = [
"ArtworkFormat",
"ArtworkEntry",
"IthmbLocation",
"EncodedFormatPayload",
"NANO_7G_COVER_ART_FORMATS",
"extract_cover",
"encode_for_ipod",
"decode_from_ipod",
"write_artworkdb",
"prepare_artwork_for_track",
]

67
src/artwork/cache.py Normal file
View File

@ -0,0 +1,67 @@
"""Local cover art cache for the player UI.
Caches extracted cover art as JPEG files in a local directory
so the player can display them without re-extracting from audio.
"""
from __future__ import annotations
import hashlib
import logging
import os
import tempfile
logger = logging.getLogger(__name__)
class CoverCache:
"""Manages a local cache of cover art images for the player UI.
The cache stores JPEG thumbnails keyed by (artist, album) hash.
"""
def __init__(self, cache_dir: str | None = None):
self.cache_dir = cache_dir or os.path.join(
tempfile.gettempdir(), "neo-pod-covers"
)
os.makedirs(self.cache_dir, exist_ok=True)
def _key(self, artist: str, album: str) -> str:
raw = f"{artist.lower()}|{album.lower()}"
return hashlib.md5(raw.encode()).hexdigest()
def get(self, artist: str, album: str) -> str | None:
"""Return path to cached cover, or None."""
path = os.path.join(self.cache_dir, f"{self._key(artist, album)}.jpg")
if os.path.exists(path):
return path
return None
def put(self, artist: str, album: str, art_bytes: bytes) -> str | None:
"""Cache cover art bytes and return the file path."""
from PIL import Image
from io import BytesIO
key = self._key(artist, album)
path = os.path.join(self.cache_dir, f"{key}.jpg")
try:
img = Image.open(BytesIO(art_bytes))
img = img.convert("RGB")
img.thumbnail((300, 300), Image.Resampling.LANCZOS)
img.save(path, "JPEG", quality=85)
return path
except Exception as exc:
logger.warning("Failed to cache cover art: %s", exc)
return None
def clear(self) -> None:
"""Remove all cached covers."""
for fname in os.listdir(self.cache_dir):
if fname.endswith(".jpg"):
try:
os.remove(os.path.join(self.cache_dir, fname))
except OSError:
pass
cover_cache = CoverCache()

374
src/artwork/chunks.py Normal file
View File

@ -0,0 +1,374 @@
"""Binary ArtworkDB chunk serialization.
Ported from iOpenPod ArtworkDB_Writer/artworkdb_chunks.py.
Builds the MHFD MHSD MHLI/MHLA/MHLF MHII MHOD MHNI MHOD(ithmb)
hierarchy that the iPod firmware parses.
"""
from __future__ import annotations
import os
import struct
from collections.abc import Mapping
from .types import ArtworkEntry, ArtworkFormatPayload, ExistingFormatRef, IthmbLocation
from .codecs import expected_size_bytes
MHFD_HEADER_SIZE = 132
MHSD_HEADER_SIZE = 96
MHLI_HEADER_SIZE = 92
MHLA_HEADER_SIZE = 92
MHLF_HEADER_SIZE = 92
MHII_HEADER_SIZE = 152
MHOD_HEADER_SIZE = 24
MHNI_HEADER_SIZE = 76
MHIF_HEADER_SIZE = 124
def _default_ithmb_filename(format_id: int) -> str:
return f"F{int(format_id)}_1.ithmb"
def _normalize_ithmb_filename(format_id: int, filename: str | None) -> str:
name = (filename or "").strip().replace("\\", "/")
if ":" in name:
name = name.split(":")[-1]
if "/" in name:
name = name.rsplit("/", 1)[-1]
return name or _default_ithmb_filename(format_id)
def build_artworkdb(
entries: list[ArtworkEntry],
format_locations_map: Mapping[int, Mapping[int, IthmbLocation | tuple[str, int] | int]],
format_ids: list[int],
image_sizes: dict[int, int],
next_mhii_id: int,
reference_mhfd: bytes | None = None,
) -> bytes:
"""Serialize a complete ArtworkDB binary."""
ds1 = _write_mhsd(1, _write_mhli(entries, format_locations_map))
ds2 = _write_mhsd(2, _write_mhla())
ds3 = _write_mhsd(3, _write_mhlf(format_ids, image_sizes))
return _write_mhfd([ds1, ds2, ds3], next_mhii_id, reference_mhfd)
def _write_mhfd(datasets: list[bytes], next_mhii_id: int, reference_mhfd: bytes | None = None) -> bytes:
all_data = b"".join(datasets)
total_len = MHFD_HEADER_SIZE + len(all_data)
header = bytearray(MHFD_HEADER_SIZE)
header[0:4] = b"mhfd"
struct.pack_into("<I", header, 4, MHFD_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<I", header, 16, 2)
struct.pack_into("<I", header, 20, len(datasets))
struct.pack_into("<I", header, 28, next_mhii_id)
if reference_mhfd and len(reference_mhfd) >= 48:
header[32:48] = reference_mhfd[32:48]
struct.pack_into("<I", header, 48, 2)
if reference_mhfd and len(reference_mhfd) >= 68:
header[60:68] = reference_mhfd[60:68]
return bytes(header) + all_data
def _write_mhsd(ds_type: int, child_data: bytes) -> bytes:
total_len = MHSD_HEADER_SIZE + len(child_data)
header = bytearray(MHSD_HEADER_SIZE)
header[0:4] = b"mhsd"
struct.pack_into("<I", header, 4, MHSD_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<H", header, 12, ds_type)
return bytes(header) + child_data
def _write_mhli(entries: list[ArtworkEntry], format_locations_map) -> bytes:
children_data = b"".join(
_write_mhii(entry, format_locations_map[entry.img_id])
for entry in entries
)
header = bytearray(MHLI_HEADER_SIZE)
header[0:4] = b"mhli"
struct.pack_into("<I", header, 4, MHLI_HEADER_SIZE)
struct.pack_into("<I", header, 8, len(entries))
return bytes(header) + children_data
def _write_mhla() -> bytes:
header = bytearray(MHLA_HEADER_SIZE)
header[0:4] = b"mhla"
struct.pack_into("<I", header, 4, MHLA_HEADER_SIZE)
struct.pack_into("<I", header, 8, 0)
return bytes(header)
def _write_mhlf(format_ids: list[int], image_sizes: dict[int, int]) -> bytes:
children_data = b"".join(
_write_mhif(fmt_id, image_sizes[fmt_id])
for fmt_id in format_ids
)
header = bytearray(MHLF_HEADER_SIZE)
header[0:4] = b"mhlf"
struct.pack_into("<I", header, 4, MHLF_HEADER_SIZE)
struct.pack_into("<I", header, 8, len(format_ids))
return bytes(header) + children_data
def _write_mhif(format_id: int, image_size: int) -> bytes:
header = bytearray(MHIF_HEADER_SIZE)
header[0:4] = b"mhif"
struct.pack_into("<I", header, 4, MHIF_HEADER_SIZE)
struct.pack_into("<I", header, 8, MHIF_HEADER_SIZE)
struct.pack_into("<I", header, 16, format_id)
struct.pack_into("<I", header, 20, image_size)
return bytes(header)
def _write_mhii(entry: ArtworkEntry, format_locations: Mapping[int, IthmbLocation | tuple[str, int] | int]) -> bytes:
children = []
for fmt_id in sorted(entry.formats.keys()):
payload = entry.formats[fmt_id]
location = _coerce_ithmb_location(fmt_id, format_locations.get(fmt_id, 0))
children.append(_write_mhod_container(2, _write_mhni(fmt_id, location, payload)))
children_data = b"".join(children)
total_len = MHII_HEADER_SIZE + len(children_data)
header = bytearray(MHII_HEADER_SIZE)
header[0:4] = b"mhii"
struct.pack_into("<I", header, 4, MHII_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<I", header, 12, len(children))
struct.pack_into("<I", header, 16, entry.img_id)
struct.pack_into("<q", header, 20, entry.db_track_id)
struct.pack_into("<I", header, 48, entry.src_img_size)
return bytes(header) + children_data
def _write_mhod_container(mhod_type: int, mhni_data: bytes) -> bytes:
total_len = MHOD_HEADER_SIZE + len(mhni_data)
header = bytearray(MHOD_HEADER_SIZE)
header[0:4] = b"mhod"
struct.pack_into("<I", header, 4, MHOD_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<H", header, 12, mhod_type)
return bytes(header) + mhni_data
def _write_mhni(format_id: int, location: IthmbLocation, payload: ArtworkFormatPayload) -> bytes:
filename = _normalize_ithmb_filename(format_id, location.filename)
mhod3 = _write_mhod_string(3, f":{filename}")
total_len = MHNI_HEADER_SIZE + len(mhod3)
visible_h = int(payload.height)
visible_w = int(payload.width)
img_size = int(payload.size)
stride = max(visible_w, int(payload.stride_pixels))
header = bytearray(MHNI_HEADER_SIZE)
header[0:4] = b"mhni"
struct.pack_into("<I", header, 4, MHNI_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<I", header, 12, 1)
struct.pack_into("<I", header, 16, format_id)
struct.pack_into("<I", header, 20, int(location.offset))
struct.pack_into("<I", header, 24, img_size)
struct.pack_into("<h", header, 28, 0)
struct.pack_into("<h", header, 30, 0)
struct.pack_into("<H", header, 32, visible_h)
struct.pack_into("<H", header, 34, visible_w)
struct.pack_into("<I", header, 40, img_size)
return bytes(header) + mhod3
def _write_mhod_string(mhod_type: int, string: str) -> bytes:
if mhod_type == 3:
encoded = string.encode("utf-16-le")
encoding_byte = 2
else:
encoded = string.encode("utf-8")
encoding_byte = 1
str_len = len(encoded)
padding = (4 - (str_len % 4)) % 4
body = struct.pack("<I", str_len)
body += struct.pack("<B", encoding_byte)
body += b"\x00" * 3
body += b"\x00" * 4
body += encoded
body += b"\x00" * padding
total_len = MHOD_HEADER_SIZE + len(body)
header = bytearray(MHOD_HEADER_SIZE)
header[0:4] = b"mhod"
struct.pack_into("<I", header, 4, MHOD_HEADER_SIZE)
struct.pack_into("<I", header, 8, total_len)
struct.pack_into("<H", header, 12, mhod_type)
return bytes(header) + body
def _coerce_ithmb_location(format_id: int, location) -> IthmbLocation:
if isinstance(location, IthmbLocation):
return IthmbLocation(
_normalize_ithmb_filename(format_id, location.filename),
int(location.offset),
)
if isinstance(location, tuple):
return IthmbLocation(
_normalize_ithmb_filename(format_id, location[0]),
int(location[1]),
)
return IthmbLocation(_default_ithmb_filename(format_id), int(location or 0))
def read_existing_artwork(artworkdb_path: str, artwork_dir: str) -> dict[int, dict]:
"""Read existing ArtworkDB entries, returning {img_id: entry_dict}."""
if not os.path.exists(artworkdb_path):
return {}
try:
with open(artworkdb_path, "rb") as f:
data = f.read()
except Exception:
return {}
if len(data) < 32 or data[:4] != b"mhfd":
return {}
entries = {}
mhfd_header_size = struct.unpack_from("<I", data, 4)[0]
child_count = struct.unpack_from("<I", data, 20)[0]
if mhfd_header_size < 32 or mhfd_header_size > len(data):
return {}
offset = mhfd_header_size
for _ in range(child_count):
if offset + 14 > len(data) or data[offset:offset + 4] != b"mhsd":
break
mhsd_header = struct.unpack_from("<I", data, offset + 4)[0]
mhsd_total = struct.unpack_from("<I", data, offset + 8)[0]
ds_type = struct.unpack_from("<H", data, offset + 12)[0]
if mhsd_header < 14 or mhsd_total < mhsd_header or offset + mhsd_total > len(data):
break
if ds_type == 1:
dataset_end = offset + mhsd_total
mhli_offset = offset + mhsd_header
if mhli_offset + 12 <= dataset_end and data[mhli_offset:mhli_offset + 4] == b"mhli":
mhli_header = struct.unpack_from("<I", data, mhli_offset + 4)[0]
mhii_count = struct.unpack_from("<I", data, mhli_offset + 8)[0]
if mhli_header < 12 or mhli_offset + mhli_header > dataset_end:
break
mhii_offset = mhli_offset + mhli_header
for _ in range(mhii_count):
if mhii_offset + 52 > dataset_end or data[mhii_offset:mhii_offset + 4] != b"mhii":
break
mhii_total = struct.unpack_from("<I", data, mhii_offset + 8)[0]
if mhii_total < 52 or mhii_offset + mhii_total > dataset_end:
break
entry = _parse_mhii_existing(data, mhii_offset, mhii_total, artwork_dir)
if entry:
entries[entry["img_id"]] = entry
mhii_offset += mhii_total
offset += mhsd_total
return entries
def _parse_mhii_existing(data: bytes, offset: int, total_len: int, artwork_dir: str) -> dict | None:
"""Parse one MHII entry from an existing ArtworkDB."""
entry_end = offset + total_len
if offset + 52 > entry_end:
return None
header_size = struct.unpack_from("<I", data, offset + 4)[0]
child_count = struct.unpack_from("<I", data, offset + 12)[0]
img_id = struct.unpack_from("<I", data, offset + 16)[0]
song_id = struct.unpack_from("<q", data, offset + 20)[0]
src_img_size = struct.unpack_from("<I", data, offset + 48)[0]
if header_size < 52 or header_size > total_len:
return None
formats: dict[int, ExistingFormatRef] = {}
child_offset = offset + header_size
for _ in range(child_count):
if child_offset + 14 > entry_end or data[child_offset:child_offset + 4] != b"mhod":
break
mhod_header = struct.unpack_from("<I", data, child_offset + 4)[0]
mhod_total = struct.unpack_from("<I", data, child_offset + 8)[0]
mhod_type = struct.unpack_from("<H", data, child_offset + 12)[0]
if mhod_header < 14 or mhod_total < mhod_header or child_offset + mhod_total > entry_end:
break
if mhod_type == 2:
mhni_offset = child_offset + mhod_header
child_end = child_offset + mhod_total
if mhni_offset + MHNI_HEADER_SIZE <= child_end and data[mhni_offset:mhni_offset + 4] == b"mhni":
fmt_id = struct.unpack_from("<I", data, mhni_offset + 16)[0]
ithmb_offset = struct.unpack_from("<I", data, mhni_offset + 20)[0]
img_size = struct.unpack_from("<I", data, mhni_offset + 24)[0]
fname = _parse_mhni_filename(data, mhni_offset, child_end)
fname = _normalize_ithmb_filename(fmt_id, fname)
ipath = os.path.join(artwork_dir, fname)
if os.path.exists(ipath) and img_size > 0:
formats[fmt_id] = ExistingFormatRef(
path=ipath,
ithmb_offset=ithmb_offset,
size=img_size,
width=max(1, struct.unpack_from("<H", data, mhni_offset + 34)[0]),
height=max(1, struct.unpack_from("<H", data, mhni_offset + 32)[0]),
hpad=max(0, struct.unpack_from("<h", data, mhni_offset + 30)[0]),
vpad=max(0, struct.unpack_from("<h", data, mhni_offset + 28)[0]),
ithmb_filename=fname,
)
child_offset += mhod_total
if not formats:
return None
return {"img_id": img_id, "song_id": song_id, "src_img_size": src_img_size, "formats": formats}
def _parse_mhni_filename(data: bytes, mhni_offset: int, container_end: int) -> str | None:
"""Read the MHOD type=3 filename child from an MHNI chunk."""
if mhni_offset + 12 > container_end:
return None
mhni_header = struct.unpack_from("<I", data, mhni_offset + 4)[0]
mhni_total = struct.unpack_from("<I", data, mhni_offset + 8)[0]
if mhni_header < MHNI_HEADER_SIZE:
mhni_header = MHNI_HEADER_SIZE
mhni_end = min(container_end, mhni_offset + mhni_total)
child_offset = mhni_offset + mhni_header
while child_offset + MHOD_HEADER_SIZE <= mhni_end:
if data[child_offset:child_offset + 4] != b"mhod":
break
mhod_header = struct.unpack_from("<I", data, child_offset + 4)[0]
mhod_total = struct.unpack_from("<I", data, child_offset + 8)[0]
mhod_type = struct.unpack_from("<H", data, child_offset + 12)[0]
if mhod_header < MHOD_HEADER_SIZE or mhod_total < mhod_header:
break
if child_offset + mhod_total > mhni_end:
break
if mhod_type == 3:
return _parse_mhod_string(data, child_offset, mhod_total)
child_offset += mhod_total
return None
def _parse_mhod_string(data: bytes, offset: int, total_len: int) -> str | None:
"""Parse an ArtworkDB string MHOD body."""
if offset + MHOD_HEADER_SIZE + 12 > offset + total_len:
return None
header_size = struct.unpack_from("<I", data, offset + 4)[0]
if header_size < MHOD_HEADER_SIZE or header_size > total_len:
return None
body_offset = offset + header_size
body_end = offset + total_len
if body_offset + 12 > body_end:
return None
str_len = struct.unpack_from("<I", data, body_offset)[0]
encoding_byte = data[body_offset + 4]
raw_start = body_offset + 12
raw_end = min(body_end, raw_start + str_len)
raw = data[raw_start:raw_end]
try:
if encoding_byte == 2:
return raw.decode("utf-16-le", errors="replace").rstrip("\x00")
return raw.decode("utf-8", errors="replace").rstrip("\x00")
except UnicodeError:
return None

260
src/artwork/codecs.py Normal file
View File

@ -0,0 +1,260 @@
"""Format-aware ithmb encode/decode for iPod artwork.
Ported from iOpenPod ArtworkDB_Writer/ithmb_codecs.py.
Supports RGB565_LE, RGB565_BE, RGB565_BE_90, RGB555_LE, RGB555_BE,
REC_RGB555_LE, UYVY, I420_LE, JPEG.
"""
from __future__ import annotations
import io
import logging
import numpy as np
from PIL import Image
from .presets import ARTWORK_FORMATS_BY_ID, ArtworkFormat
from .types import EncodedFormatPayload
logger = logging.getLogger(__name__)
def _fmt(format_id: int, fmt_override: ArtworkFormat | None = None) -> ArtworkFormat | None:
return fmt_override if fmt_override is not None else ARTWORK_FORMATS_BY_ID.get(format_id)
def format_pixel_format(format_id: int, fmt_override: ArtworkFormat | None = None) -> str:
fmt = _fmt(format_id, fmt_override)
return fmt.pixel_format if fmt is not None else "UNKNOWN"
def format_dimensions(
format_id: int,
fallback_w: int,
fallback_h: int,
fmt_override: ArtworkFormat | None = None,
) -> tuple[int, int]:
fmt = _fmt(format_id, fmt_override)
if fmt is None:
return fallback_w, fallback_h
return int(fmt.width), int(fmt.height)
def default_stride_pixels(format_id: int, width: int, fmt_override: ArtworkFormat | None = None) -> int:
fmt = _fmt(format_id, fmt_override)
if fmt is None:
return width
pf = fmt.pixel_format
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90", "RGB555_LE", "RGB555_BE"):
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
if pf.startswith("REC_RGB555"):
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
if pf == "UYVY":
return max(width, int(fmt.row_bytes // 2) if fmt.row_bytes else width)
return width
def expected_size_bytes(
format_id: int,
width: int,
height: int,
stride_pixels: int | None = None,
fmt_override: ArtworkFormat | None = None,
) -> int:
pf = format_pixel_format(format_id, fmt_override=fmt_override)
stride = stride_pixels if stride_pixels is not None else default_stride_pixels(
format_id, width, fmt_override=fmt_override,
)
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90", "RGB555_LE", "RGB555_BE", "UYVY") or pf.startswith("REC_RGB555"):
return int(stride) * int(height) * 2
if pf == "I420_LE":
w = int(width) & ~1
h = int(height) & ~1
return (w * h * 3) // 2
if pf in ("JPEG", "UNKNOWN"):
return 0
return int(stride) * int(height) * 2
def _rgb565_array_from_image(img: Image.Image) -> np.ndarray:
arr = np.array(img.convert("RGB"), dtype=np.uint32)
r = (arr[:, :, 0] >> 3) & 0x1F
g = (arr[:, :, 1] >> 2) & 0x3F
b = (arr[:, :, 2] >> 3) & 0x1F
return ((r << 11) | (g << 5) | b).astype(np.uint16)
def _rgb565_to_rgb(arr16: np.ndarray) -> np.ndarray:
r = ((arr16 >> 11) & 0x1F).astype(np.uint8)
g = ((arr16 >> 5) & 0x3F).astype(np.uint8)
b = (arr16 & 0x1F).astype(np.uint8)
return np.stack(((r << 3) | (r >> 2), (g << 2) | (g >> 4), (b << 3) | (b >> 2)), axis=2)
def _rgb555_array_from_image(img: Image.Image) -> np.ndarray:
arr = np.array(img.convert("RGB"), dtype=np.uint32)
r = (arr[:, :, 0] >> 3) & 0x1F
g = (arr[:, :, 1] >> 3) & 0x1F
b = (arr[:, :, 2] >> 3) & 0x1F
return ((r << 10) | (g << 5) | b).astype(np.uint16)
def _rgb555_to_rgb(arr16: np.ndarray) -> np.ndarray:
r = ((arr16 >> 10) & 0x1F).astype(np.uint8)
g = ((arr16 >> 5) & 0x1F).astype(np.uint8)
b = (arr16 & 0x1F).astype(np.uint8)
return np.stack(((r << 3) | (r >> 2), (g << 3) | (g >> 2), (b << 3) | (b >> 2)), axis=2)
def _pad_packed_rows(arr16: np.ndarray, stride_pixels: int) -> np.ndarray:
stride = max(1, int(stride_pixels))
height, width = arr16.shape[:2]
if stride <= width:
return arr16
padded = np.zeros((height, stride), dtype=arr16.dtype)
padded[:, :width] = arr16
return padded
def encode_for_ipod(
source_img: Image.Image,
format_id: int,
target_width: int | None = None,
target_height: int | None = None,
fmt_override: ArtworkFormat | None = None,
) -> EncodedFormatPayload:
"""Encode a PIL Image into an iPod-native format payload."""
pf = format_pixel_format(format_id, fmt_override=fmt_override)
w, h = format_dimensions(
format_id,
int(target_width or source_img.width),
int(target_height or source_img.height),
fmt_override=fmt_override,
)
stride = default_stride_pixels(format_id, w, fmt_override=fmt_override)
base = source_img.convert("RGB").resize((w, h), Image.Resampling.LANCZOS)
if pf == "RGB565_BE_90":
rotated = base.transpose(Image.Transpose.ROTATE_270)
arr16 = _rgb565_array_from_image(rotated)
arr16 = _pad_packed_rows(arr16, stride)
raw = arr16.astype(">u2").tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf == "RGB565_BE":
arr16 = _rgb565_array_from_image(base)
arr16 = _pad_packed_rows(arr16, stride)
raw = arr16.astype(">u2").tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf == "RGB555_BE":
arr16 = _rgb555_array_from_image(base)
arr16 = _pad_packed_rows(arr16, stride)
raw = arr16.astype(">u2").tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf in ("RGB555_LE", "REC_RGB555_LE"):
arr16 = _rgb555_array_from_image(base)
arr16 = _pad_packed_rows(arr16, stride)
raw = arr16.astype("<u2").tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf == "JPEG":
out = io.BytesIO()
base.save(out, format="JPEG", quality=92, optimize=False)
raw = out.getvalue()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf == "UYVY":
if w % 2 != 0:
w -= 1
base = base.resize((w, h), Image.Resampling.LANCZOS)
arr = np.array(base, dtype=np.float32)
r, g, b = arr[:, :, 0], arr[:, :, 1], arr[:, :, 2]
y = np.clip(0.257 * r + 0.504 * g + 0.098 * b + 16, 0, 255).astype(np.uint8)
u = np.clip(-0.148 * r - 0.291 * g + 0.439 * b + 128, 0, 255)
v = np.clip(0.439 * r - 0.368 * g - 0.071 * b + 128, 0, 255)
u2 = ((u[:, 0::2] + u[:, 1::2]) * 0.5).astype(np.uint8)
v2 = ((v[:, 0::2] + v[:, 1::2]) * 0.5).astype(np.uint8)
packed = np.empty((h, w * 2), dtype=np.uint8)
packed[:, 0::4] = u2
packed[:, 1::4] = y[:, 0::2]
packed[:, 2::4] = v2
packed[:, 3::4] = y[:, 1::2]
raw = packed.tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
if pf == "I420_LE":
w_even, h_even = w & ~1, h & ~1
if w_even != w or h_even != h:
w, h = w_even, h_even
base = base.resize((w, h), Image.Resampling.LANCZOS)
arr = np.array(base, dtype=np.float32)
r, g, b = arr[:, :, 0], arr[:, :, 1], arr[:, :, 2]
y = np.clip(0.257 * r + 0.504 * g + 0.098 * b + 16, 0, 255).astype(np.uint8)
u = np.clip(-0.148 * r - 0.291 * g + 0.439 * b + 128, 0, 255)
v = np.clip(0.439 * r - 0.368 * g - 0.071 * b + 128, 0, 255)
u420 = ((u[0::2, 0::2] + u[0::2, 1::2] + u[1::2, 0::2] + u[1::2, 1::2]) * 0.25).astype(np.uint8)
v420 = ((v[0::2, 0::2] + v[0::2, 1::2] + v[1::2, 0::2] + v[1::2, 1::2]) * 0.25).astype(np.uint8)
raw = y.tobytes() + u420.tobytes() + v420.tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, pixel_format=pf)
arr16 = _rgb565_array_from_image(base)
arr16 = _pad_packed_rows(arr16, stride)
raw = arr16.astype("<u2").tobytes()
return EncodedFormatPayload(raw, w, h, len(raw), stride, "RGB565_LE")
def decode_from_ipod(
pixel_bytes: bytes,
format_id: int,
width: int,
height: int,
hpad: int = 0,
vpad: int = 0,
fmt_override: ArtworkFormat | None = None,
) -> Image.Image | None:
"""Decode iPod-native pixel data back to a PIL Image."""
pf = format_pixel_format(format_id, fmt_override=fmt_override)
width = max(1, int(width))
height = max(1, int(height))
if pf in ("RGB565_LE", "RGB565_BE", "RGB565_BE_90"):
dtype = "<u2" if pf == "RGB565_LE" else ">u2"
stride = default_stride_pixels(format_id, width, fmt_override=fmt_override)
stored_h = height
needed = stride * stored_h * 2
if len(pixel_bytes) < needed:
return None
arr = np.frombuffer(pixel_bytes[:needed], dtype=dtype)
if arr.size != stride * stored_h:
return None
arr = arr.reshape((stored_h, stride))
arr = arr[:, :width]
rgb = _rgb565_to_rgb(arr)
if pf == "RGB565_BE_90":
rgb = np.rot90(rgb, k=1)
return Image.fromarray(rgb)
if pf in ("RGB555_LE", "RGB555_BE", "REC_RGB555_LE"):
dtype = "<u2" if pf != "RGB555_BE" else ">u2"
stride = default_stride_pixels(format_id, width, fmt_override=fmt_override)
stored_h = height
needed = stride * stored_h * 2
if len(pixel_bytes) < needed:
return None
arr = np.frombuffer(pixel_bytes[:needed], dtype=dtype)
if arr.size != stride * stored_h:
return None
arr = arr.reshape((stored_h, stride))
arr = arr[:, :width]
rgb = _rgb555_to_rgb(arr)
return Image.fromarray(rgb)
if pf == "JPEG":
try:
return Image.open(io.BytesIO(pixel_bytes)).convert("RGB")
except Exception:
return None
return None

160
src/artwork/extractor.py Normal file
View File

@ -0,0 +1,160 @@
"""Extract embedded album art from media files using mutagen.
Ported from iOpenPod ArtworkDB_Writer/art_extractor.py.
Supports: MP3, M4A/AAC, M4B, M4V/MP4, FLAC, OGG, OPUS, AIFF/WAV.
"""
from __future__ import annotations
import os
from pathlib import Path
import mutagen
def extract_cover(file_path: str) -> bytes | None:
"""Extract the first embedded album art image from a media file.
Returns raw image bytes (JPEG/PNG) or None.
"""
path = Path(file_path)
ext = path.suffix.lower()
try:
if ext in _IMAGE_EXTS:
return path.read_bytes()
if ext == ".mp3":
return _extract_mp3(file_path)
elif ext in (".m4a", ".m4p", ".m4b", ".aac", ".alac"):
return _extract_mp4(file_path)
elif ext in (".m4v", ".mp4", ".mov"):
art = _extract_mp4(file_path)
if art:
return art
return None
elif ext == ".flac":
return _extract_flac(file_path)
elif ext == ".ogg":
return _extract_ogg(file_path)
elif ext == ".opus":
return _extract_opus(file_path)
elif ext in (".aif", ".aiff"):
return _extract_aiff(file_path)
else:
return _extract_generic(file_path)
except Exception:
return None
def find_folder_art(file_path: str) -> str | None:
"""Return path of folder artwork image next to *file_path*, or None."""
directory = str(Path(file_path).parent)
try:
entries = os.listdir(directory)
except OSError:
return None
lower_map = {e.lower(): e for e in entries}
for stem in _FOLDER_ART_NAMES:
for ext in _IMAGE_EXTS:
if (stem + ext) in lower_map:
return os.path.join(directory, lower_map[stem + ext])
return None
def extract_cover_with_fallback(file_path: str) -> bytes | None:
"""Extract cover art, falling back to folder.jpg if no embedded art."""
art = extract_cover(file_path)
if art is not None:
return art
folder_img = find_folder_art(file_path)
if folder_img:
try:
return Path(folder_img).read_bytes()
except OSError:
pass
return None
_FOLDER_ART_NAMES = ("cover", "folder", "album", "front", "artwork", "thumb")
_IMAGE_EXTS = (".jpg", ".jpeg", ".png", ".bmp", ".gif", ".webp")
def _extract_mp3(path: str) -> bytes | None:
from mutagen.mp3 import MP3
audio = MP3(path)
if audio.tags is None:
return None
for key in audio.tags:
if key.startswith("APIC"):
return audio.tags[key].data
return None
def _extract_mp4(path: str) -> bytes | None:
from mutagen.mp4 import MP4
audio = MP4(path)
if audio.tags is None:
return None
covers = audio.tags.get("covr", [])
if covers:
return bytes(covers[0])
return None
def _extract_flac(path: str) -> bytes | None:
from mutagen.flac import FLAC
audio = FLAC(path)
if audio.pictures:
return audio.pictures[0].data
return None
def _extract_ogg(path: str) -> bytes | None:
from mutagen.oggvorbis import OggVorbis
audio = OggVorbis(path)
return _extract_vorbis_picture(audio)
def _extract_opus(path: str) -> bytes | None:
from mutagen.oggopus import OggOpus
audio = OggOpus(path)
return _extract_vorbis_picture(audio)
def _extract_vorbis_picture(audio) -> bytes | None:
import base64
pictures = audio.get("metadata_block_picture", [])
if pictures:
try:
from mutagen.flac import Picture
pic = Picture(base64.b64decode(pictures[0]))
return pic.data
except Exception:
pass
return None
def _extract_aiff(path: str) -> bytes | None:
from mutagen.aiff import AIFF
audio = AIFF(path)
if audio.tags is None:
return None
for key in audio.tags:
if key.startswith("APIC"):
return audio.tags[key].data
return None
def _extract_generic(path: str) -> bytes | None:
audio = mutagen.File(path)
if audio is None or audio.tags is None:
return None
for key in audio.tags:
if hasattr(key, "startswith") and key.startswith("APIC"):
frame = audio.tags[key]
if hasattr(frame, "data"):
return frame.data
covers = audio.tags.get("covr", [])
if covers:
return bytes(covers[0])
return None

82
src/artwork/presets.py Normal file
View File

@ -0,0 +1,82 @@
"""Canonical ithmb artwork format definitions for iPod devices.
Ported from iOpenPod ArtworkDB_Writer/artwork_presets.py.
For Nano 7G the relevant formats are 1010 (240x240), 1013 (50x50),
1015 (58x58), and 1016 (57x57) all RGB565_LE.
"""
from dataclasses import dataclass
@dataclass(frozen=True)
class ArtworkFormat:
format_id: int
width: int
height: int
row_bytes: int
pixel_format: str = "RGB565_LE"
role: str = "cover"
description: str = ""
ARTWORK_FORMATS_BY_ID: dict[int, ArtworkFormat] = {
1005: ArtworkFormat(1005, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 7G photo thumbnail"),
1007: ArtworkFormat(1007, 480, 864, 960, "RGB565_LE", "photo_full", "Nano 7G photo full screen"),
1009: ArtworkFormat(1009, 42, 30, 84, "RGB565_LE", "photo_list", "Photo list thumbnail"),
1010: ArtworkFormat(1010, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 7G album art large"),
1013: ArtworkFormat(1013, 220, 176, 440, "RGB565_BE_90", "photo_full", "Photo full screen (rotated)"),
1015: ArtworkFormat(1015, 130, 88, 260, "RGB565_LE", "photo_preview", "Photo/Video preview"),
1016: ArtworkFormat(1016, 140, 140, 280, "RGB565_LE", "cover_large", "Photo album art large"),
1017: ArtworkFormat(1017, 56, 56, 112, "RGB565_LE", "cover_small", "Photo album art small"),
1019: ArtworkFormat(1019, 720, 480, 1440, "UYVY", "tv_out", "Photo/Video NTSC TV output"),
1020: ArtworkFormat(1020, 220, 176, 440, "RGB565_BE_90", "photo_full", "Photo full screen (alt rotated)"),
1023: ArtworkFormat(1023, 176, 132, 352, "RGB565_BE", "photo_full", "Nano full screen"),
1024: ArtworkFormat(1024, 320, 240, 640, "RGB565_LE", "photo_full", "320x240 photo full screen"),
1027: ArtworkFormat(1027, 100, 100, 200, "RGB565_LE", "cover_large", "Nano album art large"),
1028: ArtworkFormat(1028, 100, 100, 200, "RGB565_LE", "cover_small", "Video album art small"),
1029: ArtworkFormat(1029, 200, 200, 400, "RGB565_LE", "cover_large", "Video album art large"),
1031: ArtworkFormat(1031, 42, 42, 84, "RGB565_LE", "cover_small", "Nano album art small"),
1032: ArtworkFormat(1032, 42, 37, 84, "RGB565_LE", "photo_list", "Nano list thumbnail"),
1036: ArtworkFormat(1036, 50, 41, 100, "RGB565_LE", "photo_list", "Video list thumbnail"),
1044: ArtworkFormat(1044, 128, 128, 256, "RGB565_LE", "cover_medium", "Classic album art medium"),
1055: ArtworkFormat(1055, 128, 128, 256, "RGB565_LE", "cover_medium", "Classic album art medium"),
1056: ArtworkFormat(1056, 128, 128, 256, "RGB565_LE", "cover_medium_alt", "128x128 cover art (alternate)"),
1060: ArtworkFormat(1060, 320, 320, 640, "RGB565_LE", "cover_large", "Classic album art large"),
1061: ArtworkFormat(1061, 56, 56, 112, "RGB565_LE", "cover_small", "Classic album art small"),
1066: ArtworkFormat(1066, 64, 64, 128, "RGB565_LE", "photo_thumb", "Classic photo thumbnail"),
1067: ArtworkFormat(1067, 720, 480, 1080, "I420_LE", "tv_out", "Classic TV output (YUV)"),
1068: ArtworkFormat(1068, 128, 128, 256, "RGB565_LE", "cover_medium_alt", "Classic album art medium (alt 2)"),
1071: ArtworkFormat(1071, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 4G album art large"),
1073: ArtworkFormat(1073, 240, 240, 480, "RGB565_LE", "cover_large", "Nano 5G/6G album art large"),
1074: ArtworkFormat(1074, 50, 50, 100, "RGB565_LE", "cover_xsmall", "Nano album art tiny"),
1078: ArtworkFormat(1078, 80, 80, 160, "RGB565_LE", "cover_small", "Nano 4G/5G album art small"),
1079: ArtworkFormat(1079, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 4G/5G photo thumbnail"),
1081: ArtworkFormat(1081, 640, 480, 0, "JPEG", "photo_full", "JPEG photo format (experimental/legacy)"),
1083: ArtworkFormat(1083, 240, 320, 480, "RGB565_LE", "photo_full", "Nano 4G photo full screen (portrait)"),
1084: ArtworkFormat(1084, 240, 240, 480, "RGB565_LE", "cover_large_alt", "Nano 4G album art (alt)"),
1085: ArtworkFormat(1085, 88, 88, 176, "RGB565_LE", "cover_medium", "Nano 6G album art medium"),
1087: ArtworkFormat(1087, 384, 384, 768, "RGB565_LE", "photo_large", "Nano 5G photo large"),
1089: ArtworkFormat(1089, 58, 58, 116, "RGB565_LE", "cover_small", "Nano 6G album art small"),
1092: ArtworkFormat(1092, 80, 80, 160, "RGB565_LE", "photo_thumb", "Nano 6G photo thumbnail"),
1093: ArtworkFormat(1093, 512, 512, 1024, "RGB565_LE", "photo_full", "Nano 6G photo full screen"),
2002: ArtworkFormat(2002, 50, 50, 100, "RGB565_BE", "cover_small", "iPod Mobile cover art small"),
2003: ArtworkFormat(2003, 150, 150, 300, "RGB565_BE", "cover_large", "iPod Mobile cover art large"),
3001: ArtworkFormat(3001, 256, 256, 512, "REC_RGB555_LE", "cover_large", "iPod touch cover art large"),
3002: ArtworkFormat(3002, 128, 128, 256, "REC_RGB555_LE", "cover_medium", "iPod touch cover art medium"),
3003: ArtworkFormat(3003, 64, 64, 128, "REC_RGB555_LE", "cover_small", "iPod touch cover art small"),
3005: ArtworkFormat(3005, 320, 320, 640, "RGB555_LE", "cover_xlarge", "iPod touch cover art xlarge"),
}
NANO_7G_COVER_ART_FORMATS = (
ARTWORK_FORMATS_BY_ID[1010],
ArtworkFormat(1013, 50, 50, 100, "RGB565_LE", "cover_xsmall", "Nano 7G album art tiny"),
ArtworkFormat(1015, 58, 58, 116, "RGB565_LE", "cover_small", "Nano 7G album art small"),
ArtworkFormat(1016, 57, 57, 116, "RGB565_LE", "cover_small_alt", "Nano 7G album art small (aligned)"),
)
"""Nano 7G cover art format overrides:
- 1010: 240x240 (large)
- 1013: 50x50 (tiny, overrides global 220x176)
- 1015: 58x58 (small, overrides global 130x88)
- 1016: 57x57 (small alt, overrides global 140x140)
"""

75
src/artwork/types.py Normal file
View File

@ -0,0 +1,75 @@
"""Typed models for the artwork pipeline.
Ported from iOpenPod ArtworkDB_Writer/artwork_types.py, simplified.
"""
from dataclasses import dataclass, field
@dataclass(frozen=True)
class IthmbLocation:
filename: str
offset: int
@dataclass(frozen=True)
class ExistingFormatRef:
path: str
ithmb_offset: int
size: int
width: int
height: int
hpad: int = 0
vpad: int = 0
ithmb_filename: str = ""
@property
def stride_pixels(self) -> int:
return max(1, self.width + self.hpad)
@property
def stored_height(self) -> int:
return max(1, self.height + self.vpad)
@dataclass(frozen=True)
class EncodedFormatPayload:
data: bytes
width: int
height: int
size: int
stride_pixels: int
hpad: int = 0
vpad: int = 0
pixel_format: str | None = None
@classmethod
def from_existing_ref(cls, ref: ExistingFormatRef, data: bytes) -> "EncodedFormatPayload":
return cls(
data=data,
width=ref.width,
height=ref.height,
size=ref.size,
stride_pixels=ref.stride_pixels,
hpad=ref.hpad,
vpad=ref.vpad,
)
ArtworkFormatPayload = EncodedFormatPayload | ExistingFormatRef
@dataclass
class ArtworkPayload:
formats: dict[int, ArtworkFormatPayload] = field(default_factory=dict)
src_img_size: int = 0
@dataclass
class ArtworkEntry:
img_id: int
db_track_id: int
art_hash: str | None
src_img_size: int
formats: dict[int, ArtworkFormatPayload] = field(default_factory=dict)
db_track_ids: list[int] = field(default_factory=list)

187
src/artwork/writer.py Normal file
View File

@ -0,0 +1,187 @@
"""Write artwork (ithmb + ArtworkDB) to iPod.
Ported from iOpenPod ArtworkDB_Writer/artwork_writer.py, simplified
for our use case (Nano 7, always-new tracks, no incremental sync).
"""
from __future__ import annotations
import hashlib
import logging
import os
from PIL import Image
from .presets import NANO_7G_COVER_ART_FORMATS, ArtworkFormat
from .types import ArtworkEntry, EncodedFormatPayload, ExistingFormatRef, IthmbLocation
from .codecs import encode_for_ipod
from .chunks import build_artworkdb
from .extractor import extract_cover_with_fallback
logger = logging.getLogger(__name__)
ITHMB_MAX_SIZE_BYTES = 256 * 1000 * 1000
def _ithmb_filename(format_id: int, index: int = 1) -> str:
return f"F{int(format_id)}_{int(index)}.ithmb"
def _art_hash(art_bytes: bytes) -> str:
return hashlib.md5(art_bytes).hexdigest()
def image_from_bytes(art_bytes: bytes) -> Image.Image | None:
"""Load a PIL Image from raw bytes (JPEG/PNG), ensuring RGB mode."""
from io import BytesIO
try:
img = Image.open(BytesIO(art_bytes))
if img.mode != "RGB":
img = img.convert("RGBA").convert("RGB")
return img
except Exception:
return None
def prepare_artwork_for_track(source_path: str) -> tuple[str, dict[int, EncodedFormatPayload]] | None:
"""Extract cover art from a source file and encode as iPod formats.
Returns (art_hash, {format_id: EncodedFormatPayload}) or None.
"""
art_bytes = extract_cover_with_fallback(source_path)
if art_bytes is None:
return None
img = image_from_bytes(art_bytes)
if img is None:
return None
formats: dict[int, EncodedFormatPayload] = {}
for fmt in NANO_7G_COVER_ART_FORMATS:
try:
encoded = encode_for_ipod(img, fmt.format_id, fmt_override=fmt)
formats[fmt.format_id] = encoded
except Exception as exc:
logger.warning("Failed to encode format %d: %s", fmt.format_id, exc)
if not formats:
return None
return _art_hash(art_bytes), formats
def write_artworkdb(
ipod_path: str,
entries: list[ArtworkEntry],
start_img_id: int = 100,
) -> dict[int, tuple[int, int]]:
"""Write ithmb files and ArtworkDB binary to the iPod.
Args:
ipod_path: iPod mount point
entries: List of ArtworkEntry (one per unique album art)
start_img_id: Starting image ID
Returns:
Dict mapping db_track_id (img_id, src_img_size)
"""
artwork_dir = os.path.join(ipod_path, "iPod_Control", "Artwork")
os.makedirs(artwork_dir, exist_ok=True)
# Collect all unique format IDs across all entries
format_ids = sorted({fmt_id for entry in entries for fmt_id in entry.formats.keys()})
writable_format_ids = sorted({
fmt_id for entry in entries
for fmt_id, payload in entry.formats.items()
if isinstance(payload, EncodedFormatPayload)
})
# Write ithmb files
ithmb_state: dict[int, dict[str, int]] = {
fmt_id: {"index": 0, "offset": 0}
for fmt_id in writable_format_ids
}
ithmb_files: dict[int, object] = {}
format_locations_map: dict[int, dict[int, IthmbLocation]] = {}
image_sizes: dict[int, int] = {}
def _close_current(fmt_id: int) -> None:
handle = ithmb_files.pop(fmt_id, None)
if handle is not None:
handle.close()
def _open_next(fmt_id: int) -> None:
_close_current(fmt_id)
state = ithmb_state[fmt_id]
state["index"] += 1
state["offset"] = 0
filename = _ithmb_filename(fmt_id, state["index"])
ithmb_files[fmt_id] = open(os.path.join(artwork_dir, filename), "wb")
def _write_payload(fmt_id: int, data: bytes) -> IthmbLocation:
state = ithmb_state[fmt_id]
if fmt_id not in ithmb_files:
_open_next(fmt_id)
elif state["offset"] > 0 and state["offset"] + len(data) > ITHMB_MAX_SIZE_BYTES:
_open_next(fmt_id)
filename = _ithmb_filename(fmt_id, state["index"])
offset = state["offset"]
ithmb_files[fmt_id].write(data)
state["offset"] += len(data)
return IthmbLocation(filename, offset)
try:
for entry in entries:
locations: dict[int, IthmbLocation] = {}
for fmt_id in format_ids:
if fmt_id not in entry.formats:
continue
payload = entry.formats[fmt_id]
if isinstance(payload, EncodedFormatPayload):
locations[fmt_id] = _write_payload(fmt_id, payload.data)
elif isinstance(payload, ExistingFormatRef):
fname = _ithmb_filename(fmt_id)
locations[fmt_id] = IthmbLocation(fname, payload.ithmb_offset)
format_locations_map[entry.img_id] = locations
finally:
for fmt_id in list(ithmb_files.keys()):
_close_current(fmt_id)
# Calculate image sizes for MHIF
for fmt_id in format_ids:
sizes = [
int(entry.formats[fmt_id].size)
for entry in entries
if fmt_id in entry.formats and int(entry.formats[fmt_id].size) > 0
]
if sizes:
from collections import Counter
image_sizes[fmt_id] = Counter(sizes).most_common(1)[0][0]
# Build and write ArtworkDB binary
ref_mhfd = None
existing_artdb = os.path.join(artwork_dir, "ArtworkDB")
if os.path.exists(existing_artdb):
with open(existing_artdb, "rb") as f:
ref_mhfd = f.read()
next_id = start_img_id + len(entries)
db_data = build_artworkdb(
entries,
format_locations_map,
format_ids,
image_sizes,
next_id,
ref_mhfd,
)
with open(existing_artdb, "wb") as f:
f.write(db_data)
f.flush()
os.fsync(f.fileno())
# Return track_id → (img_id, src_img_size) mapping
return {
entry.db_track_id: (entry.img_id, entry.src_img_size)
for entry in entries
}

View File

@ -109,6 +109,7 @@ class AudioConverter:
# Try libfdk_aac first (best compatibility)
fdkaac_cmd = cmd + [
"-map", "0:a",
"-map", "0:t?",
"-c:a", "libfdk_aac",
"-profile:a", "aac_low",
"-movflags", "+faststart",
@ -127,6 +128,7 @@ class AudioConverter:
# Fallback: built-in aac with LC profile
cmd += [
"-map", "0:a",
"-map", "0:t?",
"-profile:a", "aac_low",
"-movflags", "+faststart",
"-map_metadata", "0",
@ -136,6 +138,7 @@ class AudioConverter:
except Exception:
cmd += [
"-map", "0:a",
"-map", "0:t?",
"-profile:a", "aac_low",
"-movflags", "+faststart",
"-map_metadata", "0",
@ -145,6 +148,8 @@ class AudioConverter:
else:
cmd += [
"-map", "0:a",
"-map", "0:t?",
"-id3v2_version", "3",
"-map_metadata", "0",
"-loglevel", "error",
output_file,

View File

@ -1,7 +1,7 @@
#!/usr/bin/env python3
"""
Command Line Interface for iPod Nano Transfer Tool
Provides a CLI for the YouTube Music to iPod Nano transfer tool
Command Line Interface for neo-pod-desktop
Provides a CLI for downloading, converting, managing and transferring music to iPod Nano devices
"""
import os
@ -25,7 +25,7 @@ logger = logging.getLogger(__name__)
class YouTubeToIPodCLI:
"""Command Line Interface for YouTube to iPod Nano Transfer Tool"""
"""Command Line Interface for neo-pod-desktop"""
def __init__(self):
"""Initialize the CLI"""
@ -43,7 +43,7 @@ class YouTubeToIPodCLI:
def parse_arguments(self):
"""Parse command line arguments"""
parser = argparse.ArgumentParser(
description="YouTube to iPod Nano Transfer Tool",
description="neo-pod-desktop",
formatter_class=argparse.ArgumentDefaultsHelpFormatter
)

View File

@ -21,23 +21,33 @@ class ConfigLoader:
def __init__(self, config_file: Optional[str] = None):
"""
Initialize the configuration loader
Args:
config_file: Path to configuration file (optional)
"""
# Default config file is in the project root directory
if not config_file:
self.config_file = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"config.ini"
os.path.expanduser("~"), ".config", "neo-pod-desktop", "config.ini"
)
self._migrate_old_config()
else:
self.config_file = config_file
self.config = configparser.ConfigParser()
# Load configuration
self.load()
def _migrate_old_config(self):
"""Migrate config from old project-root location to XDG config dir."""
old_path = os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))),
"config.ini"
)
if os.path.exists(old_path) and not os.path.exists(self.config_file):
logger.info("Migrating config from %s to %s", old_path, self.config_file)
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
os.rename(old_path, self.config_file)
def load(self) -> bool:
"""
@ -63,16 +73,17 @@ class ConfigLoader:
def save(self) -> bool:
"""
Save configuration to file
Returns:
True if successful, False otherwise
"""
try:
logger.info(f"Saving configuration to {self.config_file}")
os.makedirs(os.path.dirname(self.config_file), exist_ok=True)
with open(self.config_file, 'w') as f:
self.config.write(f)
return True
except Exception as e:
logger.error(f"Failed to save configuration: {e}")

View File

@ -618,6 +618,78 @@ class IPodDevice:
self.logger.error(f"Failed to transfer directory: {e}")
return False
def export_track(self, track_info: dict, dest_dir: str) -> Optional[str]:
"""
Copy a track from iPod to a local directory with metadata-based naming.
Args:
track_info: Track dict with keys: file_path, title, artist, album, track_number
dest_dir: Local destination directory
Returns:
Path to the exported file, or None on failure
"""
src_path = track_info.get("file_path")
if not src_path or not os.path.exists(src_path):
self.logger.warning(f"Track file not found on iPod: {track_info.get('title')}")
return None
if not os.path.exists(dest_dir):
os.makedirs(dest_dir, exist_ok=True)
artist = track_info.get("artist", "Unknown Artist")
title = track_info.get("title", "Unknown Title")
album = track_info.get("album", "")
track_num = track_info.get("track_number", 0)
ext = os.path.splitext(src_path)[1].lower()
safe_artist = "".join(c for c in artist if c.isalnum() or c in " ._-'&()").strip() or "Unknown"
safe_title = "".join(c for c in title if c.isalnum() or c in " ._-'&()").strip() or "Unknown"
if track_num:
fname = f"{track_num:02d} - {safe_artist} - {safe_title}{ext}"
else:
fname = f"{safe_artist} - {safe_title}{ext}"
dest_path = os.path.join(dest_dir, fname)
try:
shutil.copy2(src_path, dest_path)
self.logger.info(f"Exported track to: {dest_path}")
return dest_path
except Exception as e:
self.logger.error(f"Failed to export track: {e}")
return None
def export_tracks(self, tracks: List[dict], dest_dir: str,
progress_callback=None) -> List[Tuple[str, str]]:
"""
Copy multiple tracks from iPod to a local directory.
Args:
tracks: List of track info dicts
dest_dir: Local destination directory
progress_callback: Optional callback(progress_int, status_str)
Returns:
List of (title, exported_path) tuples for successful exports
"""
exported = []
total = len(tracks)
for i, track in enumerate(tracks):
if progress_callback:
progress_callback(int((i / total) * 100), f"Exporting {track.get('title', '?')}...")
path = self.export_track(track, dest_dir)
if path:
exported.append((track.get("title", "?"), path))
if progress_callback:
progress_callback(100, f"Exported {len(exported)} of {total} tracks")
return exported
if __name__ == "__main__":
ipod = IPodDevice()

View File

@ -21,6 +21,12 @@ from mutagen.easyid3 import EasyID3
from mutagen.mp4 import MP4
from mutagen.mp3 import MP3
from artwork.extractor import extract_cover_with_fallback
from artwork.cache import cover_cache
from artwork.writer import prepare_artwork_for_track, write_artworkdb
from artwork.chunks import read_existing_artwork
from artwork.types import ArtworkEntry
logger = logging.getLogger(__name__)
MASTER_CONTAINER_PID = 203092939621887772 # "iPod" master playlist
@ -290,16 +296,16 @@ class Nano7Database:
artist = audio_info.get("artist", "Unknown Artist")
album = audio_info.get("album", "Unknown Album")
# Check for duplicate: same title + artist + album
# Check for duplicate: same title + artist + album (case-insensitive)
conn = sqlite3.connect(self.sqlite_path)
cursor = conn.cursor()
cursor.execute(
"SELECT pid, title FROM item WHERE title = ? AND artist = ?",
(title, artist)
"SELECT pid, title FROM item WHERE LOWER(title) = LOWER(?) AND LOWER(artist) = LOWER(?) AND LOWER(album) = LOWER(?)",
(title, artist, album)
)
existing = cursor.fetchone()
if existing:
logger.info(f"Track already exists: '{existing[1]}' by {artist}, skipping")
logger.info(f"Track already exists: '{existing[1]}' by {artist} ({album}), skipping")
conn.close()
return None
@ -479,6 +485,12 @@ class Nano7Database:
}
logger.info(f"Added track: {title} by {artist}")
# Extract and write album art to iPod
try:
self._write_artwork_for_track(filepath, artist, album, pid)
except Exception as exc:
logger.warning("Failed to write artwork for track '%s': %s", title, exc)
# Sync iTunesCDB so firmware accepts the modified database
self.sync_itunescdb()
@ -487,6 +499,57 @@ class Nano7Database:
return result
def _write_artwork_for_track(self, source_path: str, artist: str, album: str, item_pid: int) -> None:
"""Extract cover art, encode for iPod, write to ithmb/ArtworkDB, update SQLite."""
result = prepare_artwork_for_track(source_path)
if result is None:
return
art_hash, encoded_formats = result
artwork_dir = os.path.join(self.mountpoint, "iPod_Control", "Artwork")
os.makedirs(artwork_dir, exist_ok=True)
artdb_path = os.path.join(artwork_dir, "ArtworkDB")
existing = read_existing_artwork(artdb_path, artwork_dir)
next_img_id = max(existing.keys(), default=99) + 1
src_img_size = sum(p.size for p in encoded_formats.values())
new_entry = ArtworkEntry(
img_id=next_img_id,
db_track_id=item_pid,
art_hash=art_hash,
src_img_size=src_img_size,
formats=encoded_formats,
db_track_ids=[item_pid],
)
all_entries = []
for eid in sorted(existing.keys()):
e = existing[eid]
all_entries.append(ArtworkEntry(
img_id=e["img_id"],
db_track_id=e["song_id"],
art_hash=None,
src_img_size=e["src_img_size"],
formats={fid: ref for fid, ref in e["formats"].items()},
db_track_ids=[e["song_id"]],
))
all_entries.append(new_entry)
write_artworkdb(self.mountpoint, all_entries, start_img_id=min(existing.keys(), default=100))
conn = sqlite3.connect(self.sqlite_path)
try:
conn.execute(
"UPDATE item SET artwork_cache_id = ?, artwork_status = 1 WHERE pid = ?",
(next_img_id, item_pid),
)
conn.commit()
finally:
conn.close()
def get_track_count(self) -> int:
"""Get number of tracks in database"""
conn = sqlite3.connect(self.sqlite_path)
@ -704,6 +767,101 @@ class Nano7Database:
logger.error(f"Failed to delete track pid={pid}: {e}")
return False
def remove_duplicates(self) -> List[Dict[str, Any]]:
"""Find and remove duplicate tracks from the iPod database.
Duplicates are identified by LOWER(title) + LOWER(artist) + LOWER(album).
Keeps the entry with the lowest physical_order (original).
Returns list of removed track info dicts.
"""
conn = sqlite3.connect(self.sqlite_path)
cursor = conn.cursor()
# Find duplicate groups: same title + artist + album, more than one entry
cursor.execute("""
SELECT LOWER(title) AS lt, LOWER(artist) AS la, LOWER(album) AS lb,
COUNT(*) AS cnt, MIN(physical_order) AS keep_order
FROM item
GROUP BY LOWER(title), LOWER(artist), LOWER(album)
HAVING cnt > 1
""")
dup_groups = cursor.fetchall()
if not dup_groups:
conn.close()
logger.info("No duplicate tracks found")
return []
# Collect PIDs to remove
pids_to_remove = []
files_to_remove = []
removed_info = []
for lt, la, lb, cnt, keep_order in dup_groups:
# Get the track to keep (lowest physical_order)
cursor.execute(
"SELECT pid, title, artist, album FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND physical_order=?",
(lt, la, lb, keep_order)
)
keep_row = cursor.fetchone()
keep_pid = keep_row[0] if keep_row else None
# Get all duplicates for this group
cursor.execute(
"SELECT pid, title, artist, album, physical_order FROM item WHERE LOWER(title)=? AND LOWER(artist)=? AND LOWER(album)=? AND pid!=? ORDER BY physical_order",
(lt, la, lb, keep_pid)
)
dup_rows = cursor.fetchall()
for dup_pid, dup_title, dup_artist, dup_album, dup_order in dup_rows:
pids_to_remove.append(dup_pid)
removed_info.append({
"pid": dup_pid,
"title": dup_title,
"artist": dup_artist,
"album": dup_album,
"physical_order": dup_order,
})
# Find the physical file for this duplicate
file_path = self._find_track_file(dup_pid, dup_title, dup_artist)
if file_path:
files_to_remove.append(file_path)
logger.info(f"Found {len(pids_to_remove)} duplicate entries to remove across {len(dup_groups)} groups")
# Remove from all tables
for pid in pids_to_remove:
cursor.execute("DELETE FROM item_to_container WHERE item_pid = ?", (pid,))
cursor.execute("DELETE FROM avformat_info WHERE item_pid = ?", (pid,))
cursor.execute("DELETE FROM store_info WHERE item_pid = ?", (pid,))
cursor.execute("DELETE FROM item WHERE pid = ?", (pid,))
# Clean up orphaned artist/album entries
self._cleanup_orphaned_entries(cursor)
conn.commit()
# Delete physical files
for fpath in files_to_remove:
if os.path.exists(fpath):
os.remove(fpath)
logger.info(f"Deleted duplicate file: {fpath}")
# Update Locations.itdb
for pid in pids_to_remove:
self._delete_location(pid)
conn.close()
# Sync databases
self.sync_itunescdb()
self.sync_locations_cbk()
logger.info(f"Removed {len(removed_info)} duplicate track(s)")
return removed_info
def _find_track_file(self, pid: int, title: str, artist: str) -> Optional[str]:
"""Find the physical file for a track by scanning Music/FXX/ directories."""
music_base = os.path.join(self.mountpoint, "iPod_Control", "Music")

File diff suppressed because it is too large Load Diff

View File

@ -22,6 +22,7 @@ from mutagen.id3 import ID3, APIC, TIT2, TPE1, TALB, TRCK
from mutagen.easyid3 import EasyID3
from youtube_downloader import TrackInfo
from artwork.cache import cover_cache
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(name)s - %(levelname)s - %(message)s')
@ -215,10 +216,12 @@ class MetadataHandler:
"""
logger.info(f"Processing metadata for file: {file_path}")
# Download thumbnail if available
# Download thumbnail if available, or use cached cover from source
thumbnail_path = None
if track_info.thumbnail_url:
thumbnail_path = self.download_thumbnail(track_info.thumbnail_url)
elif track_info.cover_path and os.path.exists(track_info.cover_path):
thumbnail_path = track_info.cover_path
# Determine file type and process accordingly
file_ext = Path(file_path).suffix.lower()
@ -347,7 +350,12 @@ class MetadataHandler:
fallback = self._parse_filename_trackinfo(base_name)
title = title or fallback.title
artist = artist or fallback.artist
# Cache cover art for the player UI
cover_path = None
if cover_data:
cover_path = cover_cache.put(artist or "Unknown Artist", album or "", cover_data)
return TrackInfo(
video_id=f"local_{hash(file_path)}",
title=title or base_name,
@ -356,7 +364,9 @@ class MetadataHandler:
thumbnail_url="",
duration=duration,
track_number=track_number if track_number else None,
genre=genre,
download_path=file_path,
cover_path=cover_path,
)
except Exception as e:

View File

@ -32,6 +32,8 @@ class TrackInfo:
playlist_title: Optional[str] = None
playlist_index: Optional[int] = None
download_path: Optional[str] = None
genre: Optional[str] = None
cover_path: Optional[str] = None
class YouTubeDownloader: