akarpov/music/services/info.py

@@ -21,33 +21,30 @@
 from akarpov.utils.text import is_similar_artist, normalize_text
 
 
-def generate_readable_slug(name: str, model: Model) -> str:
+def generate_readable_slug(name: str, model) -> str:
     # Translate and slugify the name
-    slug = safe_translate(name)
+    slug = str(
+        slugify(
+            GoogleTranslator(source="auto", target="en").translate(
+                name,
+                target_language="en",
+            )
+        )
+    )
 
-    # Truncate slug if it's too long
     if len(slug) > 20:
         slug = slug[:20]
         last_dash = slug.rfind("-")
         if last_dash != -1:
             slug = slug[:last_dash]
 
-    original_slug = slug
-
-    # Ensure uniqueness
-    counter = 1
     while model.objects.filter(slug=slug).exists():
-        if len(original_slug) > 14:
-            truncated_slug = original_slug[:14]
-            last_dash = truncated_slug.rfind("-")
+        if len(slug) > 14:
+            slug = slug[:14]
+            last_dash = slug.rfind("-")
             if last_dash != -1:
-                truncated_slug = truncated_slug[:last_dash]
-        else:
-            truncated_slug = original_slug
-
-        suffix = f"_{generate_charset(5)}" if counter == 1 else f"_{counter}"
-        slug = f"{truncated_slug}{suffix}"
-        counter += 1
+                slug = slug[:last_dash]
+        slug = slug + "_" + generate_charset(5)
 
     return slug
 
@@ -220,12 +217,16 @@ def update_album_info(album: AlbumModel, author_name: str = None) -> None:
     client = yandex_login()
     spotify_session = create_spotify_session()
 
-    search_term = f"{album.name} - {author_name}" if author_name else album.name
-
-    yandex_album_info = get_api_info(get_yandex_album_info, search_term, client)
-    spotify_album_info = get_api_info(
-        get_spotify_album_info, search_term, spotify_session
+    if author_name:
+        yandex_album_info = get_yandex_album_info(
+            album.name + " - " + author_name, client
         )
+        spotify_album_info = get_spotify_album_info(
+            album.name + " - " + author_name, spotify_session
+        )
+    else:
+        yandex_album_info = get_yandex_album_info(album.name, client)
+        spotify_album_info = get_spotify_album_info(album.name, spotify_session)
 
     # Combine and prioritize Spotify data
     album_data = {}
@@ -235,14 +236,14 @@ def update_album_info(album: AlbumModel, author_name: str = None) -> None:
             "name": spotify_album_info.get("name", album.name),
             "release_date": spotify_album_info.get("release_date", ""),
             "total_tracks": spotify_album_info.get("total_tracks", ""),
-            "link": spotify_album_info.get("external_urls", {}).get("spotify", ""),
+            "link": spotify_album_info["external_urls"]["spotify"],
             "genre": spotify_album_info.get("genres", []),
         }
     if yandex_album_info:
         album_data.update(
             {
-                "name": album_data.get("name") or yandex_album_info.title,
-                "genre": album_data.get("genre") or yandex_album_info.genre,
+                "name": album_data.get("name", yandex_album_info.title),
+                "genre": album_data.get("genre", yandex_album_info.genre),
                 "description": yandex_album_info.description,
                 "type": yandex_album_info.type,
             }
@@ -252,120 +253,102 @@ def update_album_info(album: AlbumModel, author_name: str = None) -> None:
     album.save()
 
     # Handle Album Image - Prefer Spotify, fallback to Yandex
-    image_path = get_album_image(spotify_album_info, yandex_album_info)
+    image_path = None
+    if (
+        spotify_album_info
+        and "images" in spotify_album_info
+        and spotify_album_info["images"]
+    ):
+        image_path = download_image(
+            spotify_album_info["images"][0]["url"], settings.MEDIA_ROOT
+        )
+    elif yandex_album_info and yandex_album_info.cover_uri:
+        image_path = download_image(
+            "https://" + yandex_album_info.cover_uri, settings.MEDIA_ROOT
+        )
+
+    generated_name = slugify(
+        GoogleTranslator(source="auto", target="en").translate(
+            album.name,
+            target_language="en",
+        )
+    )
 
     if image_path:
-        save_album_image(album, image_path)
-
-    # Update Album Authors from Spotify data if available
-    if spotify_album_info and "artists" in spotify_album_info:
-        update_album_authors(album, spotify_album_info["artists"])
-
-    album.slug = generate_readable_slug(album.name, AlbumModel)
-    album.save()
-
-
-def get_album_image(spotify_info, yandex_info):
-    if spotify_info and "images" in spotify_info and spotify_info["images"]:
-        return download_image(spotify_info["images"][0]["url"], settings.MEDIA_ROOT)
-    elif yandex_info and yandex_info.cover_uri:
-        return download_image("https://" + yandex_info.cover_uri, settings.MEDIA_ROOT)
-    return None
-
-
-def save_album_image(album, image_path):
-    if not image_path:
-        return
-
-    try:
-        generated_name = safe_translate(album.name)
         with open(image_path, "rb") as f:
             album.image.save(
                 generated_name + ".png",
-                File(f, name=generated_name + ".png"),
+                File(
+                    f,
+                    name=generated_name + ".png",
+                ),
                 save=True,
             )
         os.remove(image_path)
         album.save()
-    except Exception as e:
-        print(f"Error saving album image: {str(e)}")
 
-
-def update_album_authors(album, artists):
+    # Update Album Authors from Spotify data if available
+    if spotify_album_info and "artists" in spotify_album_info:
         album_authors = []
-    for artist in artists:
+        for artist in spotify_album_info["artists"]:
             author, created = Author.objects.get_or_create(name=artist["name"])
             album_authors.append(author)
         album.authors.set(album_authors)
 
+    album.slug = generate_readable_slug(album.name, AlbumModel)
+    album.save()
+
 
 def update_author_info(author: Author) -> None:
     client = yandex_login()
     spotify_session = create_spotify_session()
 
-    yandex_artist_info = get_api_info(get_yandex_artist_info, author.name, client)
-    spotify_artist_info = get_api_info(
-        get_spotify_artist_info, author.name, spotify_session
-    )
+    # Retrieve info from both services
+    yandex_artist_info = get_yandex_artist_info(author.name, client)
+    spotify_artist_info = get_spotify_artist_info(author.name, spotify_session)
 
-    author_data = combine_artist_data(author, spotify_artist_info, yandex_artist_info)
-
-    with transaction.atomic():
-        author.meta = author_data
-        author.save()
-
-    image_path = get_author_image(spotify_artist_info, yandex_artist_info)
-
-    if image_path:
-        save_author_image(author, image_path)
-
-    author.slug = generate_readable_slug(author.name, Author)
-    with transaction.atomic():
-        author.save()
-
-
-def get_api_info(api_func, search_term, session):
-    try:
-        return api_func(search_term, session)
-    except Exception as e:
-        print(f"Error fetching info from {api_func.__name__}: {str(e)}")
-        return None
-
-
-def combine_artist_data(author, spotify_info, yandex_info):
+    # Combine and prioritize Spotify data
     author_data = {}
-    if spotify_info:
+    if spotify_artist_info:
         author_data = {
-            "name": spotify_info.get("name", author.name),
-            "genres": spotify_info.get("genres", []),
-            "popularity": spotify_info.get("popularity", 0),
-            "link": spotify_info.get("external_urls", {}).get("spotify", ""),
+            "name": spotify_artist_info.get("name", author.name),
+            "genres": spotify_artist_info.get("genres", []),
+            "popularity": spotify_artist_info.get("popularity", 0),
+            "link": spotify_artist_info["external_urls"]["spotify"],
         }
-    if yandex_info:
+    if yandex_artist_info:
         author_data.update(
             {
-                "name": author_data.get("name") or yandex_info.name,
-                "genres": author_data.get("genres") or yandex_info.genres,
-                "description": yandex_info.description,
+                "name": author_data.get("name", yandex_artist_info.name),
+                "genres": author_data.get("genres", yandex_artist_info.genres),
+                "description": yandex_artist_info.description,
             }
         )
-    return author_data
 
+    author.meta = author_data
+    with transaction.atomic():
+        author.save()
 
-def get_author_image(spotify_info, yandex_info):
-    if spotify_info and "images" in spotify_info and spotify_info["images"]:
-        return download_image(spotify_info["images"][0]["url"], settings.MEDIA_ROOT)
-    elif yandex_info and yandex_info.cover:
-        return download_image(yandex_info.cover, settings.MEDIA_ROOT)
-    return None
+    # Handle Author Image - Prefer Spotify, fallback to Yandex
+    image_path = None
+    if (
+        spotify_artist_info
+        and "images" in spotify_artist_info
+        and spotify_artist_info["images"]
+    ):
+        image_path = download_image(
+            spotify_artist_info["images"][0]["url"], settings.MEDIA_ROOT
+        )
+    elif yandex_artist_info and yandex_artist_info.cover:
+        image_path = download_image(yandex_artist_info.cover, settings.MEDIA_ROOT)
 
-
-def save_author_image(author, image_path):
-    if not image_path:
-        return
-
-    try:
-        generated_name = safe_translate(author.name)
+    generated_name = slugify(
+        GoogleTranslator(source="auto", target="en").translate(
+            author.name,
+            target_language="en",
+        )
+    )
+    if image_path:
         with open(image_path, "rb") as f:
             author.image.save(
                 generated_name + ".png",
@@ -374,17 +357,10 @@ def save_author_image(author, image_path):
             )
         os.remove(image_path)
         author.save()
-    except Exception as e:
-        print(f"Error saving author image: {str(e)}")
 
-
-def safe_translate(text):
-    try:
-        translated = GoogleTranslator(source="auto", target="en").translate(text)
-        return slugify(translated)
-    except Exception as e:
-        print(f"Error translating text: {str(e)}")
-        return slugify(text)
+    author.slug = generate_readable_slug(author.name, Author)
+    with transaction.atomic():
+        author.save()
 
 
 def search_all_platforms(track_name: str) -> dict:

akarpov/music/tasks.py

@@ -31,49 +31,16 @@
 
 @shared_task(soft_time_limit=60 * 60, time_limit=60 * 120)
 def list_tracks(url, user_id):
-    url = normalize_url(url)
-
-    handlers = {
-        "spotify.com": handle_spotify,
-        "music.yandex.ru": handle_yandex,
-        "youtube.com": handle_youtube,
-    }
-
-    for domain, handler in handlers.items():
-        if domain in url:
-            return handler(url, user_id)
-    print("failed to find handler, falling back to search")
-    return fallback_search(url, user_id)
-
-
-def normalize_url(url):
-    return url.replace("music.youtube.com", "youtube.com").replace(
-        "youtu.be", "youtube.com"
-    )
-
-
-def handle_spotify(url, user_id):
+    if "music.youtube.com" in url or "youtu.be" in url:
+        url = url.replace("music.youtube.com", "youtube.com")
+        url = url.replace("youtu.be", "youtube.com")
+    if "spotify.com" in url:
         spotify.download_url(url, user_id)
-    return url
-
-
-def handle_yandex(url, user_id):
+    elif "music.yandex.ru" in url:
         yandex.load_url(url, user_id)
-    return url
-
-
-def handle_youtube(url, user_id):
+    if "youtube.com" in url:
         if "channel" in url or "/c/" in url:
-        return handle_youtube_channel(url, user_id)
-    elif "playlist" in url or "&list=" in url:
-        return handle_youtube_playlist(url, user_id)
-    else:
-        process_yb.apply_async(kwargs={"url": url, "user_id": user_id})
-        return url
-
-
-def handle_youtube_channel(url, user_id):
-    ytmusic = YTMusic()
+            ytmusic = ytmusicapi.YTMusic()
             channel_id = url.split("/")[-1]
             channel_songs = ytmusic.get_artist(channel_id)["songs"]["results"]
 
@@ -84,20 +51,22 @@ def handle_youtube_channel(url, user_id):
                         "user_id": user_id,
                     }
                 )
-    return url
 
+        elif "playlist" in url or "&list=" in url:
+            ytmusic = ytmusicapi.YTMusic()
 
-def handle_youtube_playlist(url, user_id):
-    ytmusic = YTMusic()
+            # Parse the URL and the query string
             parsed_url = urlparse(url)
             parsed_qs = parse_qs(parsed_url.query)
+
+            # Get the playlist ID from the parsed query string
             playlist_id = parsed_qs.get("list", [None])[0]
 
-    if not playlist_id:
-        raise ValueError("No playlist ID found in the URL.")
-
+            if playlist_id:
                 playlist_songs = ytmusic.get_playlist(playlist_id)["tracks"]
 
+            else:
+                raise ValueError("No playlist ID found in the URL.")
             for song in playlist_songs:
                 process_yb.apply_async(
                     kwargs={
@@ -105,10 +74,9 @@ def handle_youtube_playlist(url, user_id):
                         "user_id": user_id,
                     }
                 )
-    return url
-
-
-def fallback_search(url, user_id):
+        else:
+            process_yb.apply_async(kwargs={"url": url, "user_id": user_id})
+    else:
         spotify_manager = SpotifyClientCredentials(
             client_id=settings.MUSIC_SPOTIFY_ID,
             client_secret=settings.MUSIC_SPOTIFY_SECRET,
@@ -116,12 +84,13 @@ def fallback_search(url, user_id):
         spotify_search = spotipy.Spotify(client_credentials_manager=spotify_manager)
 
         results = spotify_search.search(q=url, type="track", limit=1)
-    top_track = results["tracks"]["items"][0] if results["tracks"]["items"] else None
+        top_track = (
+            results["tracks"]["items"][0] if results["tracks"]["items"] else None
+        )
 
         if top_track:
-        spotify_url = top_track["external_urls"]["spotify"]
-        spotify.download_url(spotify_url, user_id)
-        return spotify_url
+            spotify.download_url(top_track["external_urls"]["spotify"], user_id)
+            url = top_track["external_urls"]["spotify"]
 
     return url