Update app.py

app.py

@@ -1,206 +1,44 @@
-import os
-import json
-import glob
-import shutil
-import subprocess
-import time
-import re
-import requests
-from pathlib import Path
 import gradio as gr
-from ytmusicapi import YTMusic
-from mutagen.id3 import ID3, TIT2, TPE1, TALB, TCON, TYER, TRCK, TPE2, USLT, APIC, COMM
-from mutagen.mp3 import MP3
-
-TMP      = Path("/tmp")
-WGCF_BIN = TMP / "wgcf"
-WARP_CONF= TMP / "wgcf-account.toml"
-WG_CONF  = TMP / "wgcf-profile.conf"
-ytm      = YTMusic()
-
-def safe(s):
-    return re.sub(r'[\\/:"*?<>|]+', "", s).strip()
-
-def install_tools():
-    import urllib.request
-    if not WGCF_BIN.exists():
-        url = "https://github.com/ViRb3/wgcf/releases/download/v2.2.29/wgcf_2.2.29_linux_amd64"
-        urllib.request.urlretrieve(url, WGCF_BIN)
-        os.chmod(WGCF_BIN, 0o755)
-
-def start_warp():
-    import urllib.request, gzip, shutil, os, subprocess, time
-
-    wgcf = TMP / "wgcf"
-    if not wgcf.exists():
-        url = "https://github.com/ViRb3/wgcf/releases/download/v2.2.20/wgcf_2.2.20_linux_amd64"
-        urllib.request.urlretrieve(url, wgcf)
-        os.chmod(wgcf, 0o755)
-
-    toml = TMP / "wgcf-account.toml"
-    conf = TMP / "wgcf-profile.conf"
-    
-    # Skip registration if toml already exists
-    if not toml.exists():
-        max_attempts = 3
-        for attempt in range(max_attempts):
-            try:
-                env = os.environ.copy()
-                env["WGCF_ENDPOINT"] = "162.159.193.1"  # Alternative endpoint
-                result = subprocess.run(
-                    [wgcf, "register", "--accept-tos"],
-                    cwd=TMP,
-                    capture_output=True,
-                    text=True,
-                    env=env
-                )
-                if result.returncode == 0:
-                    break
-                else:
-                    print(f"Attempt {attempt + 1} failed: {result.stderr}")
-                    # Check if account file was created despite non-zero exit code
-                    if toml.exists():
-                        print("Account file created despite error, proceeding...")
-                        break
-                    if attempt < max_attempts - 1:
-                        time.sleep(5)  # Wait before retrying
-            except subprocess.CalledProcessError as e:
-                print(f"Attempt {attempt + 1} failed with error: {e.stderr}")
-                # Check if account file was created despite exception
-                if toml.exists():
-                    print("Account file created despite exception, proceeding...")
-                    break
-                if attempt == max_attempts - 1:
-                    if not conf.exists():
-                        raise RuntimeError("Failed to register wgcf after multiple attempts and no existing profile found")
-                time.sleep(5)
-        else:
-            # If all attempts failed but profile exists, proceed
-            if not conf.exists():
-                raise RuntimeError("Failed to register wgcf and no existing profile found")
-    else:
-        # Account file already exists, update it if needed
-        try:
-            subprocess.run([wgcf, "update"], cwd=TMP, check=True)
-        except subprocess.CalledProcessError as e:
-            print(f"Warning: Failed to update existing account: {e}")
-    
-    # Generate profile if not exists
-    if not conf.exists():
-        subprocess.run([wgcf, "generate"], cwd=TMP, check=True)
-
-    wggo = TMP / "wireguard-go"
-    if not wggo.exists():
-        gz = TMP / "wggo.gz"
-        urllib.request.urlretrieve(
-            "https://github.com/WireGuard/wireguard-go/releases/download/0.0.20230223/wireguard-go-linux-amd64.tar.gz", gz
-        )
-        with gzip.open(gz, "rb") as f_in:
-            with open(wggo, "wb") as f_out:
-                shutil.copyfileobj(f_in, f_out)
-        os.chmod(wggo, 0o755)
-        gz.unlink(missing_ok=True)
-
-    # Start wireguard-go with the profile
-    subprocess.Popen([wggo, "wgcf0"], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-    time.sleep(3)
-    
-def ytdlp_fast(url, out):
-    subprocess.run([
-        "yt-dlp", "-f", "bestaudio/best", "--extract-audio",
-        "--audio-format", "mp3", "--audio-quality", "320K",
-        "--concurrent-fragments", "16", "--external-downloader", "aria2c",
-        "--external-downloader-args", "aria2c:-x16 -s16 -j16 -k1M",
-        "--fragment-retries", "infinite", "--skip-unavailable-fragments",
-        "--no-part", "--no-cache-dir", "--force-ipv4",
-        "-o", out, url
-    ], check=True)
-
-def write_tags(fn, title, artist, album, year, cover, lyrics, genre="", track="", album_artist="", description=""):
-    audio = MP3(fn, ID3=ID3)
-    if audio.tags is None:
-        audio.add_tags()
-    tags = audio.tags
-    tags.add(TIT2(encoding=3, text=title))
-    tags.add(TPE1(encoding=3, text=artist))
-    tags.add(TALB(encoding=3, text=album))
-    for v, cls, val in [(genre, TCON, genre), (year, TYER, year), (track, TRCK, track), (album_artist, TPE2, album_artist)]:
-        if val:
-            tags.add(cls(encoding=3, text=val))
-    if lyrics:
-        tags.add(USLT(encoding=3, lang="eng", desc="", text=lyrics))
-    if description:
-        tags.add(COMM(encoding=3, lang="eng", desc="desc", text=description))
-    if cover:
-        try:
-            img = requests.get(cover, timeout=15).content
-            tags.add(APIC(encoding=3, mime="image/jpeg", type=3, desc="Cover", data=img))
-        except Exception:
-            pass
-    audio.save(v2_version=3)
-
-def download_yt(url):
-    url = url.strip()
-    if not url.startswith("http"):
-        url = "https://" + url
-    subprocess.run([
-        "yt-dlp", "--skip-download", "--write-info-json", "--write-thumbnail",
-        "-o", str(TMP / "tmp"), url
-    ], check=True)
-    info_file = next(TMP.glob("tmp*.json"), None)
-    if not info_file:
-        raise RuntimeError("info missing")
-    info = json.loads(info_file.read_text(encoding="utf8"))
-    vid, title, author = info["id"], info.get("title") or info.get("fulltitle") or "Unknown", info.get("artist") or info.get("uploader") or "Unknown"
-    upload_date, extractor, description = info.get("upload_date", ""), info.get("extractor", ""), info.get("description", "")
-    album, album_artist, genre, track, year, lyrics = "Single", "", "", "0", "", ""
-    if "music.youtube" in url or extractor.startswith("youtube:music"):
-        album = info.get("album") or "Single"
-        album_artist = info.get("album_artist", "")
-        track = str(info.get("track_number", 0))
-        year = str(info.get("release_date") or upload_date)
-        genre = info.get("genre", "") or info.get("category", "")
-        try:
-            pl = ytm.get_watch_playlist(videoId=vid)
-            if pl.get("lyrics"):
-                lyr = ytm.get_lyrics(pl["lyrics"])
-                if lyr and lyr.get("lyrics"):
-                    lyrics = lyr["lyrics"]
-        except Exception:
-            pass
-    else:
-        year = upload_date
-    base_name = safe(f"{title} [{vid}]")
-    out_mp3 = TMP / f"{base_name}.mp3"
-    ytdlp_fast(url, str(TMP / f"{base_name}.%(ext)s"))
-    candidates = list(TMP.glob(f"{base_name}*.mp3"))
-    if not candidates:
-        raise RuntimeError("mp3 missing")
-    shutil.move(str(candidates[0]), str(out_mp3))
-    cover = None
-    for ext in ("jpg", "jpeg", "webp", "png"):
-        thumb = TMP / f"tmp.{ext}"
-        if thumb.exists():
-            cover = str(thumb)
-            break
-    write_tags(str(out_mp3), title, author, album, year, cover, lyrics, genre, track, album_artist, description)
-    return out_mp3
+import requests
+import io
+from urllib.parse import urlparse, unquote
 
 def gradio_fn(url):
+    if not url:
+        return None
     try:
-        path = download_yt(url)
-        return str(path), path.name
-    except Exception as e:
-        return None, f"Error: {e}"
+        api_url = f"https://eolithic-daniele-nonchivalrously.ngrok-free.app/dl?url={url}"
+        response = requests.get(api_url)
+        response.raise_for_status()
+        data = response.json()
+        
+        mp3_url = data.get('download_link')
+        if not mp3_url:
+            return None
+            
+        mp3_response = requests.get(mp3_url)
+        mp3_response.raise_for_status()
+        
+        parsed_url = urlparse(mp3_url)
+        filename = unquote(os.path.basename(parsed_url.path))
+        if not filename or '.' not in filename:
+            filename = "audio.mp3"  # fallback
+        
+        audio_bytes = io.BytesIO(mp3_response.content)
+        audio_bytes.seek(0)
+        
+        return audio_bytes, filename
 
-start_warp()
+    except Exception as e:
+        print(f"Error: {e}")
+        return None
 
 with gr.Blocks(title="YT/YT-Music Downloader") as demo:
     gr.Markdown("### Paste YouTube / YouTube-Music link → 320 k MP3 with tags")
     with gr.Row():
-        inp = gr.Textbox(label="URL", placeholder="https://www.youtube.com/watch?v=...")
+        inp = gr.Textbox(label="URL", placeholder=" https://www.youtube.com/watch?v=...")
         btn = gr.Button("Download", variant="primary")
-    out = gr.File(label="MP3")
-    btn.click(gradio_fn, inputs=inp, outputs=[out, out])
+        out = gr.File(label="MP3")
+    btn.click(gradio_fn, inputs=inp, outputs=out)
 
-demo.launch(server_name="0.0.0.0", server_port=7860)
+demo.launch()