#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
═══════════════════════════════════════════════════════════════
Spotify → Telegram Bot · v3.0 FULL
يرسل رابط Spotify → يرجّع MP3 + تفاصيل + غلاف
بحث متعدد المصادر · yt-dlp بأوضاع متعددة · SoundCloud احتياط
═══════════════════════════════════════════════════════════════
"""
import asyncio
import io
import os
import re
import time
import shutil
import subprocess
import logging
from typing import Optional
from html import escape
import requests
from telegram import Update, InputFile, BotCommand
from telegram.constants import ParseMode, ChatAction
from telegram.ext import (
Application, CommandHandler, MessageHandler,
ContextTypes, filters as tg_filters
)
# ═══════════════════════════════════════════════════════════════
# CONFIG
# ═══════════════════════════════════════════════════════════════
BOT_TOKEN = "8977155307:AAEpTfSP_IapYP-hn8Gg-HnclvxWD_Cpghw"
TMP_DIR = os.path.join(os.path.expanduser("~"), "spotify_dl")
os.makedirs(TMP_DIR, exist_ok=True)
HEADERS = {
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/120.0.0.0 Safari/537.36",
"Accept-Language": "en-US,en;q=0.9",
}
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s | %(levelname)s | %(message)s"
)
log = logging.getLogger("spotify-bot")
SPOTIFY_RE = re.compile(
r"https?://open\.spotify\.com/(?:intl-[a-z]{2}/)?(track|album|playlist|artist)/([A-Za-z0-9]+)"
)
# ═══════════════════════════════════════════════════════════════
# HELPERS
# ═══════════════════════════════════════════════════════════════
def esc(s):
return escape(str(s)) if s is not None else ""
def safe_name(s: str, max_len: int = 80) -> str:
s = re.sub(r'[<>:"/\\|?*\x00-\x1f]', "_", str(s))
s = re.sub(r"\s+", " ", s).strip(" .")
return s[:max_len] or "track"
def cleanup_old(out_dir: str, max_age: int = 3600):
try:
now = time.time()
for f in os.listdir(out_dir):
p = os.path.join(out_dir, f)
if os.path.isfile(p) and now - os.path.getmtime(p) > max_age:
try:
os.remove(p)
except Exception:
pass
except Exception:
pass
def ffmpeg_available() -> bool:
return shutil.which("ffmpeg") is not None
def ytdlp_available() -> bool:
return shutil.which("yt-dlp") is not None
# ═══════════════════════════════════════════════════════════════
# SPOTIFY METADATA (oEmbed)
# ═══════════════════════════════════════════════════════════════
def spotify_oembed(url: str) -> dict:
r = requests.get(
"https://open.spotify.com/oembed",
params={"url": url},
headers=HEADERS,
timeout=20
)
r.raise_for_status()
return r.json()
# ═══════════════════════════════════════════════════════════════
# بحث متعدد المصادر
# ═══════════════════════════════════════════════════════════════
def yt_search_direct(query: str) -> Optional[str]:
"""YouTube scraping مباشر — video filter"""
try:
r = requests.get(
"https://www.youtube.com/results",
params={"search_query": query, "sp": "EgIQAQ%3D%3D"},
headers=HEADERS, timeout=20
)
vids = re.findall(r'"videoId":"([^"]{11})"', r.text)
if vids:
return f"https://www.youtube.com/watch?v={vids[0]}"
except Exception as e:
log.warning(f"yt direct: {e}")
return None
def yt_search_music(query: str) -> Optional[str]:
"""YouTube Music scraping"""
try:
r = requests.get(
"https://music.youtube.com/search",
params={"q": query},
headers=HEADERS, timeout=20
)
vids = re.findall(r'"videoId":"([^"]{11})"', r.text)
if vids:
return f"https://www.youtube.com/watch?v={vids[0]}"
except Exception as e:
log.warning(f"yt music: {e}")
return None
def yt_search_invidious(query: str) -> Optional[str]:
"""Invidious instances متعددة"""
instances = [
"https://yewtu.be",
"https://inv.nadeko.net",
"https://invidious.nerdvpn.de",
"https://invidious.privacyredirect.com",
"https://iv.melmac.space",
"https://invidious.f5.si",
"https://invidious.reallyaweso.me",
]
for base in instances:
try:
r = requests.get(
f"{base}/api/v1/search",
params={"q": query, "type": "video"},
headers=HEADERS, timeout=12
)
data = r.json()
if isinstance(data, list) and data:
vid = data[0].get("videoId")
if vid:
return f"https://www.youtube.com/watch?v={vid}"
except Exception:
continue
return None
def soundcloud_search(query: str) -> Optional[str]:
"""SoundCloud بحث مباشر"""
try:
r = requests.get(
"https://soundcloud.com/search/sounds",
params={"q": query},
headers=HEADERS, timeout=20
)
m = re.findall(r'"permalink_url":"(https://soundcloud\.com/[^"]+)"', r.text)
if m:
return m[0]
except Exception:
pass
return None
def youtube_search(query: str) -> Optional[str]:
"""يجرب كل المصادر بالترتيب حتى ينجح"""
for fn in (yt_search_direct, yt_search_music, yt_search_invidious, soundcloud_search):
try:
url = fn(query)
if url:
log.info(f"Found via {fn.__name__}: {url}")
return url
except Exception as e:
log.warning(f"{fn.__name__}: {e}")
return None
# ═══════════════════════════════════════════════════════════════
# yt-dlp متعدد الأوضاع
# ═══════════════════════════════════════════════════════════════
def _run_ytdlp(url: str, out_dir: str, extra_args: list) -> Optional[str]:
try:
out_tpl = os.path.join(out_dir, "dl_%(id)s.%(ext)s")
base = [
"yt-dlp",
"--no-playlist",
"--no-warnings",
"--quiet",
"--max-filesize", "48M",
"--no-check-certificate",
"-o", out_tpl
]
cmd = base + extra_args + [url]
r = subprocess.run(cmd, capture_output=True, timeout=280, text=True)
if r.returncode != 0:
log.warning(f"yt-dlp: {r.stderr[:200]}")
return None
files = [
os.path.join(out_dir, f)
for f in os.listdir(out_dir)
if f.startswith("dl_") and f.endswith((".mp3", ".m4a", ".webm", ".opus", ".ogg", ".aac"))
]
if not files:
return None
newest = max(files, key=os.path.getmtime)
# تحويل إلى mp3
if not newest.endswith(".mp3") and ffmpeg_available():
mp3_out = newest.rsplit(".", 1)[0] + ".mp3"
try:
conv = subprocess.run(
["ffmpeg", "-y", "-i", newest, "-vn",
"-acodec", "libmp3lame", "-q:a", "2", mp3_out],
capture_output=True, timeout=120, text=True
)
if conv.returncode == 0 and os.path.exists(mp3_out):
try:
os.remove(newest)
except Exception:
pass
return mp3_out
except Exception:
pass
return newest
except Exception as e:
log.warning(f"_run_ytdlp: {e}")
return None
def ytdlp_download(url: str, out_dir: str) -> Optional[str]:
"""يجرب 7 أوضاع حتى ينجح"""
if not ytdlp_available():
return None
cleanup_old(out_dir)
attempts = [
["-x", "--audio-format", "mp3", "--audio-quality", "0"],
["-f", "bestaudio[ext=m4a]/bestaudio/best"],
["-x", "--audio-format", "mp3",
"--extractor-args", "youtube:player_client=android"],
["-x", "--audio-format", "mp3",
"--extractor-args", "youtube:player_client=ios"],
["-x", "--audio-format", "mp3",
"--extractor-args", "youtube:player_client=web_safari"],
["-x", "--audio-format", "mp3",
"--extractor-args", "youtube:player_client=tv_embedded"],
["-x", "--audio-format", "mp3",
"--user-agent", HEADERS["User-Agent"]],
]
for i, args in enumerate(attempts, 1):
log.info(f"yt-dlp attempt {i}/{len(attempts)}")
result = _run_ytdlp(url, out_dir, args)
if result and os.path.exists(result):
log.info(f"✅ Success with attempt {i}")
return result
return None
# ═══════════════════════════════════════════════════════════════
# COMMANDS
# ═══════════════════════════════════════════════════════════════
async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
"🎵 <b>Spotify → MP3 Downloader</b>\n\n"
"أرسل رابط Spotify وأنا أرجّعلك الأغنية:\n"
"• 🎵 track — أغنية\n"
"• 💿 album — ألبوم\n"
"• 📂 playlist — قائمة\n\n"
"مثال:\n"
"<code>https://open.spotify.com/track/xxxxx</code>",
parse_mode=ParseMode.HTML
)
async def cmd_help(update: Update, context: ContextTypes.DEFAULT_TYPE):
await cmd_start(update, context)
async def cmd_id(update: Update, context: ContextTypes.DEFAULT_TYPE):
await update.message.reply_text(
f"🆔 <code>{update.effective_user.id}</code>",
parse_mode=ParseMode.HTML
)
async def cmd_status(update: Update, context: ContextTypes.DEFAULT_TYPE):
ytdlp = "✅" if ytdlp_available() else "❌"
ffmpeg = "✅" if ffmpeg_available() else "❌"
await update.message.reply_text(
f"📊 <b>حالة الأدوات</b>\n\n"
f"yt-dlp: {ytdlp}\n"
f"ffmpeg: {ffmpeg}",
parse_mode=ParseMode.HTML
)
# ═══════════════════════════════════════════════════════════════
# TRACK FLOW
# ═══════════════════════════════════════════════════════════════
async def send_single(update: Update, context: ContextTypes.DEFAULT_TYPE, spotify_url: str):
msg = update.message
m = await msg.reply_text("🔎 جاري تحليل الرابط...")
# 1) oembed
try:
info = spotify_oembed(spotify_url)
except Exception as e:
await m.edit_text(
f"❌ <b>فشل قراءة الرابط</b>\n<code>{esc(str(e)[:150])}</code>",
parse_mode=ParseMode.HTML
)
return
title = (info.get("title") or "").strip()
artist = (info.get("author_name") or "").strip()
cover = info.get("thumbnail_url") or ""
if not title:
await m.edit_text("❌ لم أستطع قراءة اسم الأغنية")
return
query = f"{artist} {title}".strip() if artist else title
# 2) عرض التقدم
await m.edit_text(
f"🎵 <b>{esc(title)}</b>\n"
+ (f"👤 <b>{esc(artist)}</b>\n" if artist else "")
+ "\n🔎 البحث على YouTube / SoundCloud...",
parse_mode=ParseMode.HTML
)
# 3) بحث متعدد
yt_url = youtube_search(query)
if not yt_url:
# محاولة أخيرة: scsearch (SoundCloud search داخل yt-dlp)
yt_url = f"scsearch1:{query}"
await m.edit_text("📥 جاري التحميل...")
# 4) yt-dlp
loop = asyncio.get_event_loop()
path = await loop.run_in_executor(None, ytdlp_download, yt_url, TMP_DIR)
if not path or not os.path.exists(path):
await m.edit_text(
"❌ <b>فشل التحميل</b>\n\n"
"تأكد من:\n"
"• <code>pip install --upgrade yt-dlp</code>\n"
"• <code>pkg install ffmpeg</code>\n\n"
"أو الأغنية محذوفة من كل المصادر.",
parse_mode=ParseMode.HTML
)
return
size = os.path.getsize(path)
if size > 50 * 1024 * 1024:
await m.edit_text(f"⚠️ الملف كبير ({size // 1024 // 1024}MB)")
try:
os.remove(path)
except Exception:
pass
return
caption = (
f"🎵 <b>{esc(title)}</b>\n"
+ (f"👤 <b>{esc(artist)}</b>\n" if artist else "")
+ f"🔗 <a href='{spotify_url}'>Spotify</a>"
)
# 5) أرسل
try:
with open(path, "rb") as f:
await msg.reply_audio(
audio=InputFile(f, filename=safe_name(f"{artist} - {title}") + ".mp3"),
title=title,
performer=artist or "Unknown",
caption=caption,
parse_mode=ParseMode.HTML
)
try:
os.remove(path)
except Exception:
pass
# 6) الغلاف
if cover:
try:
img = requests.get(cover, headers=HEADERS, timeout=15).content
ib = io.BytesIO(img)
ib.name = "cover.jpg"
await msg.reply_photo(
ib,
caption=f"💿 <b>{esc(title)}</b>",
parse_mode=ParseMode.HTML
)
except Exception:
pass
try:
await m.delete()
except Exception:
pass
except Exception as e:
log.exception("send audio")
await m.edit_text(
f"❌ فشل الإرسال: <code>{esc(str(e)[:180])}</code>",
parse_mode=ParseMode.HTML
)
# ═══════════════════════════════════════════════════════════════
# TEXT HANDLER
# ═══════════════════════════════════════════════════════════════
async def on_text(update: Update, context: ContextTypes.DEFAULT_TYPE):
if not update.message or not update.message.text:
return
text = update.message.text.strip()
m = SPOTIFY_RE.search(text)
if not m:
await update.message.reply_text("❌ أرسل رابط Spotify صالح")
return
kind, sid = m.group(1), m.group(2)
url = f"https://open.spotify.com/{kind}/{sid}"
await send_single(update, context, url)
# ═══════════════════════════════════════════════════════════════
# ERROR HANDLER
# ═══════════════════════════════════════════════════════════════
async def on_error(update, context):
log.exception("error", exc_info=context.error)
async def post_init(app: Application):
try:
await app.bot.set_my_commands([
BotCommand("start", "بدء"),
BotCommand("help", "مساعدة"),
BotCommand("status", "حالة الأدوات"),
BotCommand("id", "ايدي حسابك"),
])
except Exception:
pass
# ═══════════════════════════════════════════════════════════════
# MAIN
# ═══════════════════════════════════════════════════════════════
def main():
if not BOT_TOKEN or "PUT" in BOT_TOKEN:
print("❌ ضع التوكن في BOT_TOKEN")
return
print("═" * 55)
print("🎵 Spotify → Telegram Bot · v3.0")
print("═" * 55)
if not ytdlp_available():
print("⚠️ yt-dlp غير مثبت → pip install --upgrade yt-dlp")
if not ffmpeg_available():
print("⚠️ ffmpeg غير مثبت → pkg install ffmpeg")
print("═" * 55)
app = (
Application.builder()
.token(BOT_TOKEN)
.post_init(post_init)
.build()
)
app.add_handler(CommandHandler("start", cmd_start))
app.add_handler(CommandHandler("help", cmd_help))
app.add_handler(CommandHandler("status", cmd_status))
app.add_handler(CommandHandler("id", cmd_id))
app.add_handler(MessageHandler(tg_filters.TEXT & ~tg_filters.COMMAND, on_text))
app.add_error_handler(on_error)
print("✅ البوت يعمل. Ctrl+C للإيقاف.")
print("═" * 55)
app.run_polling(allowed_updates=Update.ALL_TYPES, drop_pending_updates=True)
if __name__ == "__main__":
try:
main()
except KeyboardInterrupt:
print("\n⛔ تم الإيقاف")
