159 lines
4.9 KiB
Python
159 lines
4.9 KiB
Python
"""
|
|
Song validation utilities for checking if songs should be downloaded.
|
|
Centralizes song validation logic to eliminate code duplication.
|
|
"""
|
|
|
|
from pathlib import Path
|
|
from typing import Any, Dict, List, Optional, Tuple
|
|
|
|
from karaoke_downloader.file_utils import check_file_exists_with_patterns
|
|
from karaoke_downloader.tracking_manager import TrackingManager
|
|
|
|
|
|
class SongValidator:
|
|
"""
|
|
Centralized song validation logic for checking if songs should be downloaded.
|
|
"""
|
|
|
|
def __init__(self, tracker: TrackingManager, downloads_dir: Path):
|
|
"""
|
|
Initialize the song validator.
|
|
|
|
Args:
|
|
tracker: Tracking manager instance
|
|
downloads_dir: Base downloads directory
|
|
"""
|
|
self.tracker = tracker
|
|
self.downloads_dir = downloads_dir
|
|
|
|
def should_skip_song(
|
|
self,
|
|
artist: str,
|
|
title: str,
|
|
channel_name: str,
|
|
video_id: Optional[str] = None,
|
|
video_title: Optional[str] = None,
|
|
server_songs: Optional[Dict[str, Any]] = None,
|
|
server_duplicates_tracking: Optional[Dict[str, Any]] = None,
|
|
) -> Tuple[bool, Optional[str], int]:
|
|
"""
|
|
Check if a song should be skipped based on multiple criteria.
|
|
|
|
Performs checks in order:
|
|
1. Already downloaded (tracking)
|
|
2. File exists on filesystem
|
|
3. Already on server
|
|
4. Previously failed download (bad file)
|
|
|
|
Args:
|
|
artist: Song artist name
|
|
title: Song title
|
|
channel_name: Channel name
|
|
video_id: YouTube video ID (optional)
|
|
video_title: YouTube video title (optional)
|
|
server_songs: Server songs data (optional)
|
|
server_duplicates_tracking: Server duplicates tracking (optional)
|
|
|
|
Returns:
|
|
Tuple of (should_skip, reason, total_filtered)
|
|
"""
|
|
total_filtered = 0
|
|
|
|
# Check 1: Already downloaded by this system
|
|
if self.tracker.is_song_downloaded(artist, title, channel_name, video_id):
|
|
return True, "already downloaded", total_filtered
|
|
|
|
# Check 2: File already exists on filesystem
|
|
file_exists, _ = check_file_exists_with_patterns(
|
|
self.downloads_dir, channel_name, artist, title
|
|
)
|
|
if file_exists:
|
|
return True, "file exists", total_filtered
|
|
|
|
# Check 3: Already on server (if server data provided)
|
|
if server_songs is not None and server_duplicates_tracking is not None:
|
|
from karaoke_downloader.server_manager import (
|
|
check_and_mark_server_duplicate,
|
|
)
|
|
|
|
if check_and_mark_server_duplicate(
|
|
server_songs,
|
|
server_duplicates_tracking,
|
|
artist,
|
|
title,
|
|
video_title,
|
|
channel_name,
|
|
):
|
|
total_filtered += 1
|
|
return True, "on server", total_filtered
|
|
|
|
# Check 4: Previously failed download (bad file)
|
|
if self.tracker.is_song_failed(artist, title, channel_name, video_id):
|
|
return True, "previously failed", total_filtered
|
|
|
|
return False, None, total_filtered
|
|
|
|
def mark_song_failed(
|
|
self,
|
|
artist: str,
|
|
title: str,
|
|
video_id: Optional[str],
|
|
channel_name: str,
|
|
error_message: str,
|
|
) -> None:
|
|
"""
|
|
Mark a song as failed in tracking.
|
|
|
|
Args:
|
|
artist: Song artist name
|
|
title: Song title
|
|
video_id: YouTube video ID (optional)
|
|
channel_name: Channel name
|
|
error_message: Error message to record
|
|
"""
|
|
self.tracker.mark_song_failed(
|
|
artist, title, video_id, channel_name, error_message
|
|
)
|
|
print(f"🏷️ Marked song as failed: {artist} - {title}")
|
|
|
|
def handle_download_failure(
|
|
self,
|
|
artist: str,
|
|
title: str,
|
|
video_id: Optional[str],
|
|
channel_name: str,
|
|
error_type: str,
|
|
error_details: str = "",
|
|
) -> None:
|
|
"""
|
|
Handle download failures with consistent error formatting.
|
|
|
|
Args:
|
|
artist: Song artist name
|
|
title: Song title
|
|
video_id: YouTube video ID (optional)
|
|
channel_name: Channel name
|
|
error_type: Type of error (e.g., "yt-dlp failed", "file verification failed")
|
|
error_details: Additional error details
|
|
"""
|
|
error_msg = f"{error_type}"
|
|
if error_details:
|
|
error_msg += f": {error_details}"
|
|
self.mark_song_failed(artist, title, video_id, channel_name, error_msg)
|
|
|
|
|
|
def create_song_validator(
|
|
tracker: TrackingManager, downloads_dir: Path
|
|
) -> SongValidator:
|
|
"""
|
|
Factory function to create a song validator instance.
|
|
|
|
Args:
|
|
tracker: Tracking manager instance
|
|
downloads_dir: Base downloads directory
|
|
|
|
Returns:
|
|
SongValidator instance
|
|
"""
|
|
return SongValidator(tracker, downloads_dir)
|