updated slugification

akarpov/music/services/info.py

@@ -21,30 +21,33 @@
 from akarpov.utils.text import is_similar_artist, normalize_text
 
 
-def generate_readable_slug(name: str, model) -> str:
+def generate_readable_slug(name: str, model: Model) -> str:
     # Translate and slugify the name
-    slug = str(
-        slugify(
-            GoogleTranslator(source="auto", target="en").translate(
-                name,
-                target_language="en",
-            )
-        )
-    )
+    slug = safe_translate(name)
 
+    # 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(slug) > 14:
-            slug = slug[:14]
-            last_dash = slug.rfind("-")
+        if len(original_slug) > 14:
+            truncated_slug = original_slug[:14]
+            last_dash = truncated_slug.rfind("-")
             if last_dash != -1:
-                slug = slug[:last_dash]
-        slug = slug + "_" + generate_charset(5)
+                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
 
     return slug
 
@@ -217,16 +220,12 @@ def update_album_info(album: AlbumModel, author_name: str = None) -> None:
     client = yandex_login()
     spotify_session = create_spotify_session()
 
-    if author_name:
-        yandex_album_info = get_yandex_album_info(
-            album.name + " - " + author_name, client
+    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
     )
-        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 = {}
@@ -236,14 +235,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["external_urls"]["spotify"],
+            "link": spotify_album_info.get("external_urls", {}).get("spotify", ""),
             "genre": spotify_album_info.get("genres", []),
         }
     if yandex_album_info:
         album_data.update(
             {
-                "name": album_data.get("name", yandex_album_info.title),
-                "genre": album_data.get("genre", yandex_album_info.genre),
+                "name": album_data.get("name") or yandex_album_info.title,
+                "genre": album_data.get("genre") or yandex_album_info.genre,
                 "description": yandex_album_info.description,
                 "type": yandex_album_info.type,
             }
@@ -253,102 +252,120 @@ def update_album_info(album: AlbumModel, author_name: str = None) -> None:
     album.save()
 
     # Handle Album Image - Prefer Spotify, fallback to Yandex
-    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",
-        )
-    )
+    image_path = get_album_image(spotify_album_info, yandex_album_info)
 
     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)}")
 
-    # Update Album Authors from Spotify data if available
-    if spotify_album_info and "artists" in spotify_album_info:
+
+def update_album_authors(album, artists):
     album_authors = []
-        for artist in spotify_album_info["artists"]:
+    for artist in 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()
 
-    # 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)
-
-    # Combine and prioritize Spotify data
-    author_data = {}
-    if spotify_artist_info:
-        author_data = {
-            "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_artist_info:
-        author_data.update(
-            {
-                "name": author_data.get("name", yandex_artist_info.name),
-                "genres": author_data.get("genres", yandex_artist_info.genres),
-                "description": yandex_artist_info.description,
-            }
+    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
     )
 
+    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()
 
-    # 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)
 
-    generated_name = slugify(
-        GoogleTranslator(source="auto", target="en").translate(
-            author.name,
-            target_language="en",
+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):
+    author_data = {}
+    if spotify_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", ""),
+        }
+    if yandex_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,
+            }
         )
-    )
-    if image_path:
+    return author_data
+
+
+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
+
+
+def save_author_image(author, image_path):
+    if not image_path:
+        return
+
+    try:
+        generated_name = safe_translate(author.name)
         with open(image_path, "rb") as f:
             author.image.save(
                 generated_name + ".png",
@@ -357,10 +374,17 @@ def update_author_info(author: Author) -> None:
             )
         os.remove(image_path)
         author.save()
+    except Exception as e:
+        print(f"Error saving author image: {str(e)}")
 
-    author.slug = generate_readable_slug(author.name, Author)
-    with transaction.atomic():
-        author.save()
+
+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)
 
 
 def search_all_platforms(track_name: str) -> dict:

akarpov/music/tasks.py

@@ -31,16 +31,49 @@
 
 @shared_task(soft_time_limit=60 * 60, time_limit=60 * 120)
 def list_tracks(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:
+    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):
     spotify.download_url(url, user_id)
-    elif "music.yandex.ru" in url:
+    return url
+
+
+def handle_yandex(url, user_id):
     yandex.load_url(url, user_id)
-    if "youtube.com" in url:
+    return url
+
+
+def handle_youtube(url, user_id):
     if "channel" in url or "/c/" in url:
-            ytmusic = ytmusicapi.YTMusic()
+        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()
     channel_id = url.split("/")[-1]
     channel_songs = ytmusic.get_artist(channel_id)["songs"]["results"]
 
@@ -51,22 +84,20 @@ def list_tracks(url, user_id):
                 "user_id": user_id,
             }
         )
+    return url
 
-        elif "playlist" in url or "&list=" in url:
-            ytmusic = ytmusicapi.YTMusic()
 
-            # Parse the URL and the query string
+def handle_youtube_playlist(url, user_id):
+    ytmusic = YTMusic()
     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 playlist_id:
+    if not playlist_id:
+        raise ValueError("No playlist ID found in the URL.")
+
     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={
@@ -74,9 +105,10 @@ def list_tracks(url, user_id):
                 "user_id": user_id,
             }
         )
-        else:
-            process_yb.apply_async(kwargs={"url": url, "user_id": user_id})
-    else:
+    return url
+
+
+def fallback_search(url, user_id):
     spotify_manager = SpotifyClientCredentials(
         client_id=settings.MUSIC_SPOTIFY_ID,
         client_secret=settings.MUSIC_SPOTIFY_SECRET,
@@ -84,13 +116,12 @@ def list_tracks(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.download_url(top_track["external_urls"]["spotify"], user_id)
-            url = top_track["external_urls"]["spotify"]
+        spotify_url = top_track["external_urls"]["spotify"]
+        spotify.download_url(spotify_url, user_id)
+        return spotify_url
 
     return url