321 lines
10 KiB
Python
321 lines
10 KiB
Python
#!/usr/bin/python3
|
|
|
|
import os
|
|
from typing import Dict, Any, Optional, Union
|
|
from deezspot.libutils.utils import request
|
|
from deezspot.libutils.logging_utils import logger
|
|
from deezspot.__taggers__ import write_tags
|
|
from deezspot.models.download import Track, Episode
|
|
|
|
|
|
def fetch_and_process_image(image_url_or_bytes: Union[str, bytes, None]) -> Optional[bytes]:
|
|
"""
|
|
Fetch and process image data from URL or return bytes directly.
|
|
|
|
Args:
|
|
image_url_or_bytes: Image URL string, bytes, or None
|
|
|
|
Returns:
|
|
Image bytes or None if failed/not available
|
|
"""
|
|
if not image_url_or_bytes:
|
|
return None
|
|
|
|
if isinstance(image_url_or_bytes, bytes):
|
|
return image_url_or_bytes
|
|
|
|
if isinstance(image_url_or_bytes, str):
|
|
try:
|
|
response = request(image_url_or_bytes)
|
|
return response.content
|
|
except Exception as e:
|
|
logger.warning(f"Failed to fetch image from URL {image_url_or_bytes}: {e}")
|
|
return None
|
|
|
|
return None
|
|
|
|
|
|
def enhance_metadata_with_image(metadata_dict: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Enhance metadata dictionary by fetching image data if image URL is present.
|
|
|
|
Args:
|
|
metadata_dict: Metadata dictionary potentially containing image URL
|
|
|
|
Returns:
|
|
Enhanced metadata dictionary with image bytes
|
|
"""
|
|
image_url = metadata_dict.get('image')
|
|
if image_url:
|
|
image_bytes = fetch_and_process_image(image_url)
|
|
if image_bytes:
|
|
metadata_dict['image'] = image_bytes
|
|
|
|
return metadata_dict
|
|
|
|
|
|
def add_deezer_enhanced_metadata(
|
|
metadata_dict: Dict[str, Any],
|
|
infos_dw: Dict[str, Any],
|
|
track_ids: str,
|
|
api_gw_instance: Any = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Add Deezer-specific enhanced metadata including contributors, lyrics, and version info.
|
|
|
|
Args:
|
|
metadata_dict: Base metadata dictionary
|
|
infos_dw: Deezer track information dictionary
|
|
track_ids: Track IDs for lyrics fetching
|
|
api_gw_instance: API gateway instance for lyrics retrieval
|
|
|
|
Returns:
|
|
Enhanced metadata dictionary with Deezer-specific data
|
|
"""
|
|
# Add contributor information
|
|
contributors = infos_dw.get('SNG_CONTRIBUTORS', {})
|
|
metadata_dict['author'] = "; ".join(contributors.get('author', []))
|
|
metadata_dict['composer'] = "; ".join(contributors.get('composer', []))
|
|
metadata_dict['lyricist'] = "; ".join(contributors.get('lyricist', []))
|
|
|
|
# Handle composer+lyricist combination
|
|
if contributors.get('composerlyricist'):
|
|
metadata_dict['composer'] = "; ".join(contributors.get('composerlyricist', []))
|
|
|
|
# Add version information
|
|
metadata_dict['version'] = infos_dw.get('VERSION', '')
|
|
|
|
# Initialize lyric fields
|
|
metadata_dict['lyric'] = ""
|
|
metadata_dict['copyright'] = ""
|
|
metadata_dict['lyric_sync'] = []
|
|
|
|
# Add lyrics if available and API instance provided
|
|
if api_gw_instance and infos_dw.get('LYRICS_ID', 0) != 0:
|
|
try:
|
|
lyrics_data = api_gw_instance.get_lyric(track_ids)
|
|
|
|
if lyrics_data and "LYRICS_TEXT" in lyrics_data:
|
|
metadata_dict['lyric'] = lyrics_data["LYRICS_TEXT"]
|
|
|
|
if lyrics_data and "LYRICS_SYNC_JSON" in lyrics_data:
|
|
# Import here to avoid circular imports
|
|
from deezspot.libutils.utils import trasform_sync_lyric
|
|
metadata_dict['lyric_sync'] = trasform_sync_lyric(
|
|
lyrics_data['LYRICS_SYNC_JSON']
|
|
)
|
|
except Exception as e:
|
|
logger.warning(f"Failed to retrieve lyrics: {str(e)}")
|
|
|
|
# Extract album artist from contributors with 'Main' role
|
|
if 'contributors' in infos_dw:
|
|
main_artists = [c['name'] for c in infos_dw['contributors'] if c.get('role') == 'Main']
|
|
if main_artists:
|
|
metadata_dict['album_artist'] = "; ".join(main_artists)
|
|
|
|
return metadata_dict
|
|
|
|
|
|
def add_spotify_enhanced_metadata(metadata_dict: Dict[str, Any], track_obj: Any) -> Dict[str, Any]:
|
|
"""
|
|
Add Spotify-specific enhanced metadata.
|
|
|
|
Args:
|
|
metadata_dict: Base metadata dictionary
|
|
track_obj: Spotify track object
|
|
|
|
Returns:
|
|
Enhanced metadata dictionary with Spotify-specific data
|
|
"""
|
|
# Spotify tracks already have most metadata from the unified converter
|
|
# Add any Spotify-specific enhancements here if needed in the future
|
|
|
|
# Ensure image is processed
|
|
return enhance_metadata_with_image(metadata_dict)
|
|
|
|
|
|
def prepare_track_metadata(
|
|
metadata_dict: Dict[str, Any],
|
|
source_type: str = 'unknown',
|
|
enhanced_data: Optional[Dict[str, Any]] = None,
|
|
api_instance: Any = None,
|
|
track_ids: Optional[str] = None
|
|
) -> Dict[str, Any]:
|
|
"""
|
|
Prepare and enhance track metadata for tagging based on source type.
|
|
|
|
Args:
|
|
metadata_dict: Base metadata dictionary
|
|
source_type: Source type ('spotify', 'deezer', or 'unknown')
|
|
enhanced_data: Additional source-specific data (infos_dw for Deezer)
|
|
api_instance: API instance for additional data fetching
|
|
track_ids: Track IDs for API calls
|
|
|
|
Returns:
|
|
Fully prepared metadata dictionary
|
|
"""
|
|
# Always process images first
|
|
metadata_dict = enhance_metadata_with_image(metadata_dict)
|
|
|
|
if source_type == 'deezer' and enhanced_data:
|
|
metadata_dict = add_deezer_enhanced_metadata(
|
|
metadata_dict,
|
|
enhanced_data,
|
|
track_ids or '',
|
|
api_instance
|
|
)
|
|
elif source_type == 'spotify':
|
|
metadata_dict = add_spotify_enhanced_metadata(metadata_dict, enhanced_data)
|
|
|
|
return metadata_dict
|
|
|
|
|
|
def apply_tags_to_track(track: Track, metadata_dict: Dict[str, Any]) -> None:
|
|
"""
|
|
Apply metadata tags to a track object and write them to the file.
|
|
|
|
Args:
|
|
track: Track object to tag
|
|
metadata_dict: Metadata dictionary containing tag information
|
|
"""
|
|
if not track or not metadata_dict:
|
|
return
|
|
|
|
try:
|
|
track.tags = metadata_dict
|
|
write_tags(track)
|
|
logger.debug(f"Successfully applied tags to track: {metadata_dict.get('music', 'Unknown')}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to apply tags to track: {e}")
|
|
|
|
|
|
def apply_tags_to_episode(episode: Episode, metadata_dict: Dict[str, Any]) -> None:
|
|
"""
|
|
Apply metadata tags to an episode object and write them to the file.
|
|
|
|
Args:
|
|
episode: Episode object to tag
|
|
metadata_dict: Metadata dictionary containing tag information
|
|
"""
|
|
if not episode or not metadata_dict:
|
|
return
|
|
|
|
try:
|
|
episode.tags = metadata_dict
|
|
write_tags(episode)
|
|
logger.debug(f"Successfully applied tags to episode: {metadata_dict.get('music', 'Unknown')}")
|
|
except Exception as e:
|
|
logger.error(f"Failed to apply tags to episode: {e}")
|
|
|
|
|
|
def save_cover_image_for_track(
|
|
metadata_dict: Dict[str, Any],
|
|
track_path: str,
|
|
save_cover: bool = False,
|
|
cover_filename: str = "cover.jpg"
|
|
) -> None:
|
|
"""
|
|
Save cover image for a track if requested and image data is available.
|
|
|
|
Args:
|
|
metadata_dict: Metadata dictionary potentially containing image data
|
|
track_path: Path to the track file
|
|
save_cover: Whether to save cover image
|
|
cover_filename: Filename for the cover image
|
|
"""
|
|
if not save_cover or not metadata_dict.get('image'):
|
|
return
|
|
|
|
try:
|
|
from deezspot.libutils.utils import save_cover_image
|
|
track_directory = os.path.dirname(track_path)
|
|
|
|
# Handle both URL and bytes
|
|
image_data = metadata_dict['image']
|
|
if isinstance(image_data, str):
|
|
image_bytes = fetch_and_process_image(image_data)
|
|
else:
|
|
image_bytes = image_data
|
|
|
|
if image_bytes:
|
|
save_cover_image(image_bytes, track_directory, cover_filename)
|
|
logger.info(f"Saved cover image for track in {track_directory}")
|
|
except Exception as e:
|
|
logger.warning(f"Failed to save cover image for track: {e}")
|
|
|
|
|
|
# Convenience function that combines metadata preparation and tagging
|
|
def process_and_tag_track(
|
|
track: Track,
|
|
metadata_dict: Dict[str, Any],
|
|
source_type: str = 'unknown',
|
|
enhanced_data: Optional[Dict[str, Any]] = None,
|
|
api_instance: Any = None,
|
|
track_ids: Optional[str] = None,
|
|
save_cover: bool = False
|
|
) -> None:
|
|
"""
|
|
Complete metadata processing and tagging workflow for a track.
|
|
|
|
Args:
|
|
track: Track object to process
|
|
metadata_dict: Base metadata dictionary
|
|
source_type: Source type ('spotify', 'deezer', or 'unknown')
|
|
enhanced_data: Additional source-specific data
|
|
api_instance: API instance for additional data fetching
|
|
track_ids: Track IDs for API calls
|
|
save_cover: Whether to save cover image
|
|
"""
|
|
# Prepare enhanced metadata
|
|
prepared_metadata = prepare_track_metadata(
|
|
metadata_dict,
|
|
source_type,
|
|
enhanced_data,
|
|
api_instance,
|
|
track_ids
|
|
)
|
|
|
|
# Apply tags to track
|
|
apply_tags_to_track(track, prepared_metadata)
|
|
|
|
# Save cover image if requested
|
|
if hasattr(track, 'song_path') and track.song_path:
|
|
save_cover_image_for_track(prepared_metadata, track.song_path, save_cover)
|
|
|
|
|
|
def process_and_tag_episode(
|
|
episode: Episode,
|
|
metadata_dict: Dict[str, Any],
|
|
source_type: str = 'unknown',
|
|
enhanced_data: Optional[Dict[str, Any]] = None,
|
|
api_instance: Any = None,
|
|
track_ids: Optional[str] = None,
|
|
save_cover: bool = False
|
|
) -> None:
|
|
"""
|
|
Complete metadata processing and tagging workflow for an episode.
|
|
|
|
Args:
|
|
episode: Episode object to process
|
|
metadata_dict: Base metadata dictionary
|
|
source_type: Source type ('spotify', 'deezer', or 'unknown')
|
|
enhanced_data: Additional source-specific data
|
|
api_instance: API instance for additional data fetching
|
|
track_ids: Track IDs for API calls
|
|
save_cover: Whether to save cover image
|
|
"""
|
|
# Prepare enhanced metadata
|
|
prepared_metadata = prepare_track_metadata(
|
|
metadata_dict,
|
|
source_type,
|
|
enhanced_data,
|
|
api_instance,
|
|
track_ids
|
|
)
|
|
|
|
# Apply tags to episode
|
|
apply_tags_to_episode(episode, prepared_metadata)
|
|
|
|
# Save cover image if requested
|
|
if hasattr(episode, 'episode_path') and episode.episode_path:
|
|
save_cover_image_for_track(prepared_metadata, episode.episode_path, save_cover) |