# -*- coding: utf-8 -*- # SoundCloud IRAN Bot | SoundCloud + YouTube # Host: Debian 12 / Python 3.13 / yt-dlp + ffmpeg # Results printed reversed: 20 on top ... 1 (best) at the bottom import os import re import sys import json import time import html import math import shutil import hashlib import sqlite3 import difflib import logging import threading import subprocess import queue import requests try: import mutagen from mutagen.id3 import ID3, TIT2, TPE1, TALB, APIC, ID3NoHeaderError from mutagen.mp4 import MP4, MP4Cover HAS_TAGS = True except Exception: mutagen = None HAS_TAGS = False # ====================================================================== # CONFIG # ====================================================================== BOT_TOKEN = "8465773942:AAEich-2f9yijv8PL9C8HDE905YU_iM1wbs" BOT_NAME = u"ساوند کلاد ایران" ADMINS = [5466341850] # --- YouTube audio --- YTDLP_BIN = "/home/private/.local/bin/yt-dlp" FFMPEG_BIN = "/home/private/bin/ffmpeg" YT_COOKIES = "" # اگر لازم شد: "/home/private/scbot/cookies.txt" YT_CLIENTS = "tv,web_safari,mweb,android" # "m4a" → بهترین صدا، خروجی m4a/AAC (سازگارترین با تلگرام) ← پیشنهادی # "mp3" → خروجی MP3 320 (فقط برچسب؛ کیفیت اضافه نمی‌شود، حجم ۳ برابر) # "raw" → فایل اصلی دست‌نخورده (ممکن است webm/opus باشد) YT_MODE = "m4a" YT_API_KEY = "" # اختیاری SEARCH_LIMIT = 20 SC_FETCH = 30 YT_FETCH = 25 CACHE_TTL = 43200 CID_TTL = 21600 MAX_SIZE = 49 * 1024 * 1024 MAX_DURATION = 25 * 60 WORKERS = 2 USER_COOLDOWN = 1 CHAT_COOLDOWN = 1 PAGE_LEN = 3600 NET_TIMEOUT = 25 POLL_TIMEOUT = 25 DL_TIMEOUT = 420 DEBUG_UPDATES = True ADMIN_RAW_ERR = True MATCH_THRESHOLD = 0.62 MIN_RELEVANCE = 0.34 MIN_RESULTS = 5 MIN_YT_VISIBLE = 3 BALANCE_SOURCES = True CACHE_VERSION = "v7" MANUAL_CLIENT_IDS = [] BASE = os.path.dirname(os.path.abspath(__file__)) DL_DIR = os.path.join(BASE, "downloads") DB_PATH = os.path.join(BASE, "bot.db") LOCK_PATH = os.path.join(BASE, "bot.lock") UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 " "(KHTML, like Gecko) Chrome/124.0 Safari/537.36") logging.basicConfig(level=logging.INFO, format="%(asctime)s %(levelname)s %(message)s", handlers=[logging.StreamHandler(sys.stdout)]) log = logging.getLogger("scbot") def ensure_dl_dir(): try: if not os.path.isdir(DL_DIR): os.makedirs(DL_DIR) except Exception as e: log.warning("cannot create DL_DIR: %s", e) ensure_dl_dir() # ====================================================================== # BINARIES # ====================================================================== def find_bin(explicit, names): if explicit and os.path.isfile(explicit): if not os.access(explicit, os.X_OK): try: os.chmod(explicit, 0o755) except Exception: pass if os.access(explicit, os.X_OK): return explicit for n in names: for p in (os.path.join(BASE, n), os.path.expanduser("~/bin/" + n), os.path.expanduser("~/.local/bin/" + n), "/usr/local/bin/" + n, "/usr/bin/" + n): if os.path.isfile(p) and os.access(p, os.X_OK): return p w = shutil.which(n) if w: return w return None YTDLP_PATH = find_bin(YTDLP_BIN, ["yt-dlp", "yt-dlp_linux"]) FFMPEG_PATH = find_bin(FFMPEG_BIN, ["ffmpeg"]) if not FFMPEG_PATH: try: import imageio_ffmpeg FFMPEG_PATH = imageio_ffmpeg.get_ffmpeg_exe() except Exception: FFMPEG_PATH = None HAS_FFMPEG = bool(FFMPEG_PATH) HAS_YT_AUDIO = bool(YTDLP_PATH) YT_LAST_ERR = [u""] # ====================================================================== # TEXTS # ====================================================================== T_START_PV = ( u"🎧 {name}\n" u"──────────────\n" u"سلام {user} 👋\n\n" u"جستجوی همزمان در ساوندکلاد و یوتیوب.\n\n" u"🔎 جستجو\n" u"• پیوی: اسم آهنگ رو بفرست\n" u"• گروه: $اسم آهنگ یا /s اسم آهنگ\n\n" u"🔗 لینک\n" u"لینک ساوندکلاد یا یوتیوب بفرست، خودش دانلود می‌کنه.\n\n" u"📥 دانلود\n" u"زیر هر نتیجه دستوری مثل /dl_abc1234 هست، روش بزن.\n\n" u"🔢 ترتیب\n" u"لیست از ۲۰ شروع می‌شه تا ۱؛\n" u"بهترین نتیجه پایین‌ترین مورده 👇\n" u"──────────────\n" u"⚡️ سریع، رایگان، بدون تبلیغ" ) T_START_GP = (u"🎧 {name} فعاله!\n" u"جستجو: $اسم آهنگ\n" u"یا لینک ساوندکلاد / یوتیوب بفرست.") T_ADDED = (u"🎉 {name} به گروه اضافه شد!\n" u"──────────────\n" u"🔎 $اسم آهنگ\n" u"🔗 لینک ساوندکلاد یا یوتیوب\n" u"📥 روی دستور زیر هر نتیجه بزنید\n" u"🔢 بهترین نتیجه پایین لیسته (شماره ۱)\n" u"⚠️ اگه جواب نداد، ربات رو ادمین کنید.") T_HELP = ( u"📖 راهنما\n" u"──────────────\n" u"جستجو\n" u"پیوی: اسم آهنگ رو مستقیم بفرست\n" u"گروه: $اسم ، !اسم ، /s اسم\n\n" u"لینک\n" u"تک‌آهنگ ساوندکلاد یا ویدیوی یوتیوب.\n" u"آلبوم، پلی‌لیست و کانال پشتیبانی نمی‌شه.\n\n" u"دانلود\n" u"دستور زیر هر نتیجه، مثل /dl_abc1234\n\n" u"ترتیب\n" u"از ۲۰ (بالا) تا ۱ (پایین) — شماره ۱ بهترینه.\n\n" u"برچسب‌ها\n" u"⭐ در هر دو سرویس بود (فایل از ساوندکلاد)\n" u"🟠 ساوندکلاد | 🔴 یوتیوب\n" u"💿 نسخه اصلی با کیفیت بالا\n\n" u"کیفیت\n" u"ساوندکلاد: ۱۲۸ (و تا ۳۲۰ برای 💿)\n" u"یوتیوب: حداکثر ۱۶۰\n\n" u"محدودیت\n" u"حداکثر {dur} دقیقه و {sz} مگابایت" ) T_SEARCHING = u"🔍 در حال جستجو در ساوندکلاد و یوتیوب…" T_BUSY = u"⏳ صف پره، چند لحظه بعد دوباره بزن." T_SHORT = u"✍️ حداقل ۲ حرف بنویس." T_CMD_EMPTY = (u"✍️ بعد از دستور اسم آهنگ رو بنویس.\n" u"مثال: /s Tataloo Didi") T_COOLDOWN = u"⏱ یکم آروم‌تر! {s} ثانیه دیگه." T_ERR_NET = u"🌐 اتصال برقرار نشد.\nچند دقیقه بعد امتحان کن." T_ERR_CID = u"🔑 کلید ساوندکلاد گرفته نشد.\nموقتیه." T_ERR_BLOCK = u"🚫 درخواست رد شد.\nآی‌پی سرور محدود شده." T_ERR_HTTP = u"⚠️ پاسخ نامعتبر (کد {code})." T_ERR_EMPTY = (u"❌ برای «{q}» نتیجه‌ای پیدا نشد.\n" u"اسم رو انگلیسی یا ساده‌تر بنویس.") T_ERR_GEN = u"⚠️ خطای داخلی ربات\nدوباره امتحان کن." T_EXPIRED = u"⌛️ این نتیجه منقضی شده. دوباره جستجو کن." T_PREPARING = u"⏳ در حال آماده‌سازی {t}…" T_DL_YT = u"⬇️ در حال دریافت از یوتیوب…" T_UPLOADING = u"📤 در حال ارسال…" T_TOOBIG = u"📦 حجم فایل بیشتر از {sz} مگابایته." T_NOAUDIO = (u"🔴 دریافت صدای این ترک ممکن نشد.\n\n" u"🔗 باز کردن در یوتیوب") T_DLFAIL = u"❌ دانلود این ترک ممکن نشد. یکی دیگه رو امتحان کن." T_LINK_WORK = u"🔗 در حال بررسی لینک…" T_LINK_BAD = u"⚠️ این لینک معتبر نیست یا ترک حذف شده." T_LINK_PRIV = u"🔒 این ترک خصوصیه." T_LINK_SET = (u"📂 آلبوم، پلی‌لیست، کانال یا پروفایل پشتیبانی نمی‌شه.\n" u"لینک یه آهنگ بفرست.") T_LINK_LONG = u"⏱ این ترک از {dur} دقیقه بلندتره." T_HDR = (u"🎧 نتایج برای: {q}\n" u"📊 {n} نتیجه | از {n} تا ۱\n" u"⬇️ بهترین نتیجه پایین‌ترینه (شماره ۱)\n" u"──────────────") T_FTR = (u"──────────────\n" u"⬆️ شماره ۱ = بهترین نتیجه\n" u"💡 برای دانلود، دستور زیر هر آهنگ رو بزن.") # ====================================================================== # UTIL # ====================================================================== def esc(s): try: return html.escape(s or u"") except Exception: return u"" def fmt_dur(sec): try: sec = int(sec or 0) except Exception: return u"?" h, m, s = sec // 3600, (sec % 3600) // 60, sec % 60 if h: return u"%d:%02d:%02d" % (h, m, s) return u"%d:%02d" % (m, s) def fmt_size(b): if not b: return u"?" mb = float(b) / (1024.0 * 1024.0) if mb < 1: return u"%d KB" % int(float(b) / 1024.0) return u"%.1f MB" % mb def est_size(duration_sec, bitrate_kbps): if not duration_sec or not bitrate_kbps: return 0 return int(duration_sec * bitrate_kbps * 1000 / 8) _PUNCT = re.compile(u"[^\\w\\s\u0600-\u06FF]", re.UNICODE) _SPACE = re.compile(u"\\s+", re.UNICODE) _NOISE = re.compile(u"(official|video|audio|lyrics?|prod\\b|feat\\.?|ft\\.?|" u"hd|hq|full|music|song|version|topic)", re.I) def norm(s): if not s: return u"" s = s.lower() s = _NOISE.sub(u" ", s) s = _PUNCT.sub(u" ", s) return _SPACE.sub(u" ", s).strip() def sim(a, b): a, b = norm(a), norm(b) if not a or not b: return 0.0 return difflib.SequenceMatcher(None, a, b).ratio() def relevance(item, query): q = norm(query) if not q: return 0.0 title = norm(item.get("title") or u"") full = norm(u"%s %s" % (item.get("artist") or u"", item.get("title") or u"")) best = max(sim(q, title), sim(q, full)) qt = [w for w in q.split() if len(w) > 1] if qt: words = full.split() hit = 0 for w in qt: if w in words: hit += 1 continue for t in words: if len(t) > 2 and (t.startswith(w) or w.startswith(t)): hit += 1 break best = max(best, (float(hit) / len(qt)) * 0.95) return best ALPHABET = "23456789abcdefghijkmnpqrstuvwxyzABCDEFGHJKLMNPQRSTUVWXYZ" def make_code(key): h = hashlib.sha1(key.encode("utf-8")).digest() n = int.from_bytes(h[:8], "big") out = "" for _ in range(7): out += ALPHABET[n % len(ALPHABET)] n //= len(ALPHABET) return out def safe_name(s, maxlen=60): s = re.sub(u"[\\\\/:*?\"<>|\\n\\r\\t]", u" ", s or u"track") s = _SPACE.sub(u" ", s).strip() return s[:maxlen] or u"track" def probe_audio(path): """(bitrate_kbps, duration_sec) واقعی فایل.""" br, dur = 0, 0 if HAS_TAGS and mutagen is not None: try: f = mutagen.File(path) if f is not None and getattr(f, "info", None) is not None: br = int(round((getattr(f.info, "bitrate", 0) or 0) / 1000.0)) dur = int(getattr(f.info, "length", 0) or 0) except Exception: pass if (not br or not dur) and HAS_FFMPEG: probe = FFMPEG_PATH.replace("ffmpeg", "ffprobe") if os.path.isfile(probe): try: out = subprocess.run( [probe, "-v", "error", "-show_entries", "format=bit_rate,duration", "-of", "default=nw=1:nk=1", path], capture_output=True, text=True, timeout=30).stdout.split() if len(out) >= 1 and not br: br = int(round(float(out[0]) / 1000.0)) if len(out) >= 2 and not dur: dur = int(float(out[1])) except Exception: pass if not br and dur: try: br = int(round(os.path.getsize(path) * 8.0 / dur / 1000.0)) except Exception: pass return br, dur # ====================================================================== # DATABASE # ====================================================================== class DB(object): def __init__(self, path): self.path = path self.local = threading.local() self.lock = threading.Lock() self._init() def conn(self): c = getattr(self.local, "c", None) if c is None: c = sqlite3.connect(self.path, timeout=20) c.row_factory = sqlite3.Row c.execute("PRAGMA journal_mode=WAL") c.execute("PRAGMA synchronous=NORMAL") c.execute("PRAGMA busy_timeout=15000") self.local.c = c return c def _init(self): c = self.conn() c.executescript(""" CREATE TABLE IF NOT EXISTS tracks ( code TEXT PRIMARY KEY, source TEXT, payload TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS files ( code TEXT PRIMARY KEY, file_id TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS searches ( q TEXT PRIMARY KEY, payload TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS kv ( k TEXT PRIMARY KEY, v TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS users ( uid INTEGER PRIMARY KEY, name TEXT, ts INTEGER); CREATE TABLE IF NOT EXISTS chats ( cid INTEGER PRIMARY KEY, title TEXT, ts INTEGER); """) c.commit() def kv_get(self, k, ttl): r = self.conn().execute("SELECT v, ts FROM kv WHERE k=?", (k,)).fetchone() if not r: return None if ttl and time.time() - r["ts"] > ttl: return None return r["v"] def kv_set(self, k, v): with self.lock: c = self.conn() c.execute("REPLACE INTO kv (k,v,ts) VALUES (?,?,?)", (k, v, int(time.time()))) c.commit() def put_track(self, code, source, payload): with self.lock: try: c = self.conn() c.execute("REPLACE INTO tracks (code,source,payload,ts) " "VALUES (?,?,?,?)", (code, source, json.dumps(payload), int(time.time()))) c.commit() except Exception as e: log.warning("put_track %s: %s", code, e) def get_track(self, code): c = self.conn() r = c.execute("SELECT payload FROM tracks WHERE code=?", (code,)).fetchone() if not r: r = c.execute("SELECT payload FROM tracks WHERE code=? COLLATE NOCASE", (code,)).fetchone() if not r: return None try: return json.loads(r["payload"]) except Exception: return None def put_file(self, code, file_id): with self.lock: c = self.conn() c.execute("REPLACE INTO files (code,file_id,ts) VALUES (?,?,?)", (code, file_id, int(time.time()))) c.commit() def get_file(self, code): c = self.conn() r = c.execute("SELECT file_id FROM files WHERE code=?", (code,)).fetchone() if not r: r = c.execute("SELECT file_id FROM files WHERE code=? COLLATE NOCASE", (code,)).fetchone() return r["file_id"] if r else None def put_search(self, q, payload): with self.lock: c = self.conn() c.execute("REPLACE INTO searches (q,payload,ts) VALUES (?,?,?)", (q, json.dumps(payload), int(time.time()))) c.commit() def get_search(self, q): r = self.conn().execute("SELECT payload, ts FROM searches WHERE q=?", (q,)).fetchone() if not r or time.time() - r["ts"] > CACHE_TTL: return None try: return json.loads(r["payload"]) except Exception: return None def touch_user(self, uid, name): with self.lock: c = self.conn() c.execute("REPLACE INTO users (uid,name,ts) VALUES (?,?,?)", (uid, name, int(time.time()))) c.commit() def touch_chat(self, cid, title): with self.lock: c = self.conn() c.execute("REPLACE INTO chats (cid,title,ts) VALUES (?,?,?)", (cid, title, int(time.time()))) c.commit() def stats(self): c = self.conn() def one(sql): try: return c.execute(sql).fetchone()[0] except Exception: return 0 return {"users": one("SELECT COUNT(*) FROM users"), "chats": one("SELECT COUNT(*) FROM chats"), "tracks": one("SELECT COUNT(*) FROM tracks"), "files": one("SELECT COUNT(*) FROM files"), "searches": one("SELECT COUNT(*) FROM searches")} def cleanup(self): with self.lock: c = self.conn() now = int(time.time()) c.execute("DELETE FROM searches WHERE ts]+>", "", text)[:4090] r = self.api("sendMessage", d) return r def edit(self, chat_id, message_id, text, kb=None): d = {"chat_id": chat_id, "message_id": message_id, "text": text[:4090], "parse_mode": "HTML", "disable_web_page_preview": "true"} if kb: d["reply_markup"] = json.dumps(kb) return self.api("editMessageText", d) def delete(self, chat_id, message_id): return self.api("deleteMessage", {"chat_id": chat_id, "message_id": message_id}) def action(self, chat_id, act="upload_document"): return self.api("sendChatAction", {"chat_id": chat_id, "action": act}) def answer_cb(self, cb_id, text=None, alert=False): d = {"callback_query_id": cb_id} if text: d["text"] = text d["show_alert"] = "true" if alert else "false" return self.api("answerCallbackQuery", d) def send_audio(self, chat_id, path_or_id, title=None, performer=None, duration=None, caption=None, reply_to=None, thumb_path=None, thread_id=None): d = {"chat_id": chat_id} if title: d["title"] = title[:64] if performer: d["performer"] = performer[:64] if duration: d["duration"] = int(duration) if caption: d["caption"] = caption[:1000] d["parse_mode"] = "HTML" if reply_to: d["reply_to_message_id"] = reply_to d["allow_sending_without_reply"] = "true" if thread_id: d["message_thread_id"] = thread_id if not os.path.exists(str(path_or_id)): d["audio"] = path_or_id return self.api("sendAudio", d, timeout=120) fh = open(path_or_id, "rb") th = None try: files = {"audio": (os.path.basename(path_or_id), fh)} if thumb_path and os.path.exists(thumb_path): th = open(thumb_path, "rb") files["thumbnail"] = ("cover.jpg", th) r = self.api("sendAudio", d, files=files, timeout=600) if r is None and th is not None: th.close() th = None fh.seek(0) r = self.api("sendAudio", d, files={"audio": (os.path.basename(path_or_id), fh)}, timeout=600) return r finally: try: fh.close() except Exception: pass if th: try: th.close() except Exception: pass tg = TG(BOT_TOKEN) # ====================================================================== # SOUNDCLOUD # ====================================================================== CID_RE = re.compile(r'client_id\s*[:=]\s*["\']([0-9a-zA-Z]{28,40})["\']') SCRIPT_RE = re.compile(r']+src="([^"]+)"') class SoundCloud(object): API = "https://api-v2.soundcloud.com" def __init__(self): self.s = requests.Session() self.s.headers.update({ "User-Agent": UA, "Accept": "application/json, text/javascript, */*; q=0.1"}) self._cid = None self._cid_ts = 0 self.lock = threading.Lock() self.last_error = None def _validate(self, cid): try: r = self.s.get(self.API + "/search/tracks", params={"q": "test", "limit": 1, "client_id": cid}, timeout=NET_TIMEOUT) return r.status_code == 200 except Exception: return False def client_id(self, force=False): with self.lock: now = time.time() if not force and self._cid and now - self._cid_ts < CID_TTL: return self._cid if not force: cached = db.kv_get("sc_client_id", CID_TTL) if cached and self._validate(cached): self._cid, self._cid_ts = cached, now return cached for cid in MANUAL_CLIENT_IDS: if self._validate(cid): self._cid, self._cid_ts = cid, now db.kv_set("sc_client_id", cid) return cid try: r = self.s.get("https://soundcloud.com/", timeout=NET_TIMEOUT) if r.status_code != 200: self.last_error = "home status %s" % r.status_code return None body = r.text except Exception as e: self.last_error = "home error %s" % e return None m = CID_RE.search(body) if m and self._validate(m.group(1)): cid = m.group(1) self._cid, self._cid_ts = cid, now db.kv_set("sc_client_id", cid) return cid for url in reversed([u for u in SCRIPT_RE.findall(body) if u.startswith("http")]): try: rr = self.s.get(url, timeout=NET_TIMEOUT) if rr.status_code != 200: continue mm = CID_RE.search(rr.text) if mm and self._validate(mm.group(1)): cid = mm.group(1) self._cid, self._cid_ts = cid, now db.kv_set("sc_client_id", cid) return cid except Exception: continue self.last_error = "client_id not found" return None def _get(self, path, params, retry=True): cid = self.client_id() if not cid: self.last_error = self.last_error or "no client_id" return None p = dict(params) p["client_id"] = cid try: r = self.s.get(self.API + path, params=p, timeout=NET_TIMEOUT) except Exception as e: self.last_error = "net %s" % e return None if r.status_code in (401, 403) and retry: self.client_id(force=True) return self._get(path, params, retry=False) if r.status_code == 429: self.last_error = "rate limited (429)" return None if r.status_code != 200: self.last_error = "http %s" % r.status_code return None try: return r.json() except Exception as e: self.last_error = "bad json %s" % e return None def search(self, query, limit=SC_FETCH): self.last_error = None j = self._get("/search/tracks", {"q": query, "limit": limit, "offset": 0}) if j is None: return None out = [] for it in (j.get("collection") or []): t = self._parse(it) if t: out.append(t) return out def resolve(self, url): self.last_error = None j = self._get("/resolve", {"url": url}) if j is None: return None if isinstance(j, dict) and j.get("kind") == "track": t = self._parse(j) if t: return t self.last_error = "unparsable track" return None self.last_error = "not a track (%s)" % ( j.get("kind") if isinstance(j, dict) else "?") return None def _parse(self, it): if not isinstance(it, dict) or it.get("kind") != "track": return None dur = int((it.get("full_duration") or it.get("duration") or 0) / 1000) if dur <= 0: return None user = it.get("user") or {} art = it.get("artwork_url") or user.get("avatar_url") or "" if art: art = art.replace("-large.", "-t500x500.") trans = (it.get("media") or {}).get("transcodings") or [] bitrate, quality = self._quality(trans, it) return { "src": "sc", "id": it.get("id"), "title": it.get("title") or u"Unknown", "artist": user.get("username") or u"Unknown", "album": u"", "duration": dur, "artwork": art, "permalink": it.get("permalink_url") or "", "plays": int(it.get("playback_count") or 0), "likes": int(it.get("likes_count") or 0), "popularity": 0, "downloadable": bool(it.get("downloadable") and it.get("has_downloads_left")), "bitrate": bitrate, "quality": quality, "auth": it.get("track_authorization") or "", "transcodings": [ {"url": t.get("url"), "preset": t.get("preset") or "", "protocol": ((t.get("format") or {}).get("protocol") or ""), "mime": ((t.get("format") or {}).get("mime_type") or "")} for t in trans if t.get("url")], } @staticmethod def _quality(trans, it): if it.get("downloadable") and it.get("has_downloads_left"): return 320, "HQ" best = 0 for t in trans: preset = (t.get("preset") or "").lower() mime = ((t.get("format") or {}).get("mime_type") or "").lower() if "mp3" in preset or "mpeg" in mime: best = max(best, 128) elif "aac" in preset or "aac" in mime: best = max(best, 96) elif "opus" in preset or "opus" in mime: best = max(best, 64) return (best or 128), ("STD" if best >= 128 else "LOW") def pick_stream(self, track): for want_codec, want_proto in [("mp3", "progressive"), ("mp3", "hls"), ("aac", "hls"), ("opus", "hls")]: for t in track.get("transcodings", []): preset = (t.get("preset") or "").lower() mime = (t.get("mime") or "").lower() proto = (t.get("protocol") or "").lower() if "mp3" in preset or "mpeg" in mime: codec = "mp3" elif "opus" in preset or "opus" in mime: codec = "opus" elif "aac" in preset or "aac" in mime: codec = "aac" else: codec = "?" if codec == want_codec and proto == want_proto: return t tl = track.get("transcodings") or [] return tl[0] if tl else None def resolve_url(self, transcoding, auth): cid = self.client_id() if not cid: return None params = {"client_id": cid} if auth: params["track_authorization"] = auth try: r = self.s.get(transcoding["url"], params=params, timeout=NET_TIMEOUT) if r.status_code != 200: self.last_error = "stream http %s" % r.status_code return None return (r.json() or {}).get("url") except Exception as e: self.last_error = "stream %s" % e return None def original_url(self, track_id): cid = self.client_id() if not cid: return None try: r = self.s.get(self.API + "/tracks/%s/download" % track_id, params={"client_id": cid}, timeout=NET_TIMEOUT) if r.status_code != 200: return None return (r.json() or {}).get("redirectUri") except Exception: return None def download(self, track, dest): ensure_dl_dir() if track.get("downloadable"): url = self.original_url(track.get("id")) if url and self._fetch_plain(url, dest): return dest tc = self.pick_stream(track) if not tc: self.last_error = "no transcoding" return None url = self.resolve_url(tc, track.get("auth")) if not url: return None proto = (tc.get("protocol") or "").lower() if proto == "hls" or ".m3u8" in url: return dest if self._fetch_hls(url, dest) else None return dest if self._fetch_plain(url, dest) else None def _fetch_plain(self, url, dest): ensure_dl_dir() try: r = self.s.get(url, stream=True, timeout=NET_TIMEOUT) if r.status_code != 200: self.last_error = "dl http %s" % r.status_code return False total = 0 with open(dest, "wb") as f: for chunk in r.iter_content(65536): if not chunk: continue total += len(chunk) if total > MAX_SIZE: self.last_error = "too big" break f.write(chunk) if total > MAX_SIZE or total <= 1024: try: os.remove(dest) except Exception: pass return False return True except Exception as e: self.last_error = "dl %s" % e return False def _fetch_hls(self, m3u8_url, dest): ensure_dl_dir() try: r = self.s.get(m3u8_url, timeout=NET_TIMEOUT) if r.status_code != 200: self.last_error = "m3u8 http %s" % r.status_code return False segs = [ln.strip() for ln in r.text.splitlines() if ln.strip() and not ln.strip().startswith("#")] if not segs: self.last_error = "empty playlist" return False total = 0 with open(dest, "wb") as f: for seg in segs: try: rr = self.s.get(seg, timeout=NET_TIMEOUT) if rr.status_code != 200: continue total += len(rr.content) if total > MAX_SIZE: self.last_error = "too big" break f.write(rr.content) except Exception: continue if total <= 1024 or total > MAX_SIZE: try: os.remove(dest) except Exception: pass return False return True except Exception as e: self.last_error = "hls %s" % e return False def artwork(self, url): if not url: return None try: r = self.s.get(url, timeout=15) if r.status_code == 200 and len(r.content) < 900000: return r.content except Exception: pass return None sc = SoundCloud() # ====================================================================== # YOUTUBE # ====================================================================== _YT_BRACKET = re.compile( u"[\\(\\[][^\\)\\]]*(official|video|audio|lyric|hd|4k|mv|visualizer|" u"explicit|full|music|آهنگ|موزیک|ویدیو)[^\\)\\]]*[\\)\\]]", re.I) _YT_CH_MUSIC = re.compile(u"(- topic|vevo|music|records|band|dj|beat|sound|media)", re.I) _YT_T_MUSIC = re.compile( u"(official audio|official video|official music|lyric|audio|song|music|" u"remix|cover|album|feat\\.?|ft\\.?|آهنگ|موزیک|ریمیکس|ترک)", re.I) _YT_NOT_MUSIC = re.compile( u"(podcast|interview|tutorial|trailer|full episode|reaction|review|" u"highlights|news|vlog|gameplay|unboxing|مصاحبه|آموزش|مستند|سریال|" u"فیلم کامل|قسمت)", re.I) _ISO_DUR = re.compile(r"PT(?:(\d+)H)?(?:(\d+)M)?(?:(\d+)S)?") def clean_yt_title(t): t = _YT_BRACKET.sub(u" ", t or u"") t = re.sub(u"\\s*[\\|｜]\\s*.*$", u"", t) t = _SPACE.sub(u" ", t).strip(u" -–—|_") return t or u"Unknown" class YouTube(object): WEB_SEARCH = "https://www.youtube.com/results" WATCH = "https://www.youtube.com/watch" API = "https://www.googleapis.com/youtube/v3" SP_VIDEO = "EgIQAQ==" def __init__(self, api_key=""): self.api_key = api_key or "" self.s = requests.Session() self.s.headers.update({ "User-Agent": UA, "Accept-Language": "en-US,en;q=0.9", "Accept": "text/html,application/xhtml+xml,*/*;q=0.8"}) try: self.s.cookies.set("CONSENT", "YES+cb", domain=".youtube.com") self.s.cookies.set("SOCS", "CAI", domain=".youtube.com") except Exception: pass self.last_error = None @property def mode(self): return "api" if self.api_key else "web" @staticmethod def _extract_json(body, marker): i = body.find(marker) if i < 0: return None j = body.find("{", i) if j < 0: return None try: obj, _ = json.JSONDecoder().raw_decode(body[j:]) return obj except Exception: return None @staticmethod def _walk(node, key, out, limit=200): if len(out) >= limit: return if isinstance(node, dict): for k, v in node.items(): if k == key and isinstance(v, dict): out.append(v) if len(out) >= limit: return else: YouTube._walk(v, key, out, limit) elif isinstance(node, list): for v in node: YouTube._walk(v, key, out, limit) @staticmethod def _dur_to_sec(txt): if not txt: return 0 try: parts = [int(p) for p in txt.strip().split(":")] except Exception: return 0 s = 0 for p in parts: s = s * 60 + p return s @staticmethod def _iso_to_sec(txt): m = _ISO_DUR.match(txt or "") if not m: return 0 return (int(m.group(1) or 0) * 3600 + int(m.group(2) or 0) * 60 + int(m.group(3) or 0)) @staticmethod def _views_to_pop(views): if not views: return 0 return int(min(100.0, math.log10(views + 1) * 13.5)) @staticmethod def _mk_item(vid, title, channel, dur, views, art=None): ch = (channel or u"Unknown").strip() if ch.lower().endswith("- topic"): ch = ch[:-7].strip() return { "src": "yt", "id": vid, "title": clean_yt_title(title), "raw_title": title or u"", "artist": ch, "album": u"", "duration": int(dur or 0), "artwork": art or ("https://i.ytimg.com/vi/%s/hqdefault.jpg" % vid), "permalink": "https://www.youtube.com/watch?v=%s" % vid, "plays": int(views or 0), "likes": 0, "popularity": YouTube._views_to_pop(views), "bitrate": 160, "quality": "YT", "downloadable": False, } @staticmethod def is_music(item): d = item.get("duration") or 0 if d < 40 or d > MAX_DURATION: return False blob = u"%s %s" % (item.get("raw_title") or item.get("title") or u"", item.get("artist") or u"") return not _YT_NOT_MUSIC.search(blob) @staticmethod def music_bonus(item): b = 0.0 if _YT_CH_MUSIC.search(item.get("artist") or u""): b += 8.0 if _YT_T_MUSIC.search(item.get("raw_title") or u""): b += 5.0 d = item.get("duration") or 0 if 90 <= d <= 480: b += 4.0 return b def search(self, query, limit=YT_FETCH): self.last_error = None if self.api_key: res = self._api_search(query, limit) if res is not None: return res log.warning("YT api failed (%s), fallback to web", self.last_error) return self._web_search(query, limit) def _web_search(self, query, limit): q = query if not re.search(u"(music|song|آهنگ|موزیک)", q, re.I): q = q + u" music" try: r = self.s.get(self.WEB_SEARCH, params={"search_query": q, "sp": self.SP_VIDEO, "hl": "en", "gl": "US"}, timeout=NET_TIMEOUT) except Exception as e: self.last_error = "net %s" % e return None if r.status_code != 200: self.last_error = "http %s" % r.status_code return None data = self._extract_json(r.text, "ytInitialData") if not data: self.last_error = "ytInitialData not found" return None raw = [] self._walk(data, "videoRenderer", raw, limit=limit * 3) out, seen = [], set() for v in raw: try: vid = v.get("videoId") if not vid or vid in seen: continue if v.get("badges") and "LIVE" in json.dumps(v.get("badges")): continue ttl = u"" t = v.get("title") or {} if t.get("runs"): ttl = u"".join(x.get("text", u"") for x in t["runs"]) elif t.get("simpleText"): ttl = t["simpleText"] if not ttl: continue ch = u"" for key in ("ownerText", "longBylineText", "shortBylineText"): node = v.get(key) or {} if node.get("runs"): ch = node["runs"][0].get("text", u"") break dur = self._dur_to_sec((v.get("lengthText") or {}).get("simpleText")) if dur <= 0: continue vc = (v.get("viewCountText") or {}).get("simpleText") or u"" digits = re.sub(u"[^0-9]", u"", vc) views = int(digits) if digits else 0 item = self._mk_item(vid, ttl, ch, dur, views) if not self.is_music(item): continue seen.add(vid) out.append(item) if len(out) >= limit: break except Exception: continue return out def _api_get(self, path, params): p = dict(params) p["key"] = self.api_key try: r = self.s.get(self.API + path, params=p, timeout=NET_TIMEOUT) except Exception as e: self.last_error = "net %s" % e return None if r.status_code != 200: self.last_error = "http %s: %s" % (r.status_code, r.text[:160]) return None try: return r.json() except Exception as e: self.last_error = "bad json %s" % e return None def _api_search(self, query, limit): j = self._api_get("/search", {"part": "snippet", "q": query, "type": "video", "videoCategoryId": "10", "maxResults": min(limit, 50)}) if j is None: return None ids = [((it.get("id") or {}).get("videoId")) for it in (j.get("items") or [])] ids = [i for i in ids if i] if not ids: return [] j2 = self._api_get("/videos", { "part": "snippet,contentDetails,statistics", "id": ",".join(ids[:50])}) if j2 is None: return None out = [] for it in (j2.get("items") or []): sn = it.get("snippet") or {} cd = it.get("contentDetails") or {} st = it.get("statistics") or {} dur = self._iso_to_sec(cd.get("duration")) if dur <= 0: continue th = (sn.get("thumbnails") or {}) art = ((th.get("high") or th.get("medium") or th.get("default") or {}).get("url")) item = self._mk_item(it.get("id"), sn.get("title"), sn.get("channelTitle"), dur, int(st.get("viewCount") or 0), art) if self.is_music(item): out.append(item) return out def video(self, vid): self.last_error = None if self.api_key: j = self._api_get("/videos", { "part": "snippet,contentDetails,statistics", "id": vid}) if j and (j.get("items") or []): it = j["items"][0] sn = it.get("snippet") or {} cd = it.get("contentDetails") or {} st = it.get("statistics") or {} th = (sn.get("thumbnails") or {}) art = ((th.get("high") or th.get("default") or {}).get("url")) return self._mk_item(vid, sn.get("title"), sn.get("channelTitle"), self._iso_to_sec(cd.get("duration")), int(st.get("viewCount") or 0), art) try: r = self.s.get(self.WATCH, params={"v": vid, "hl": "en"}, timeout=NET_TIMEOUT) except Exception as e: self.last_error = "net %s" % e return None if r.status_code != 200: self.last_error = "http %s" % r.status_code return None data = self._extract_json(r.text, "ytInitialPlayerResponse") if not data: self.last_error = "playerResponse not found" return None vd = data.get("videoDetails") or {} if not vd.get("videoId"): self.last_error = "no videoDetails" return None try: views = int(vd.get("viewCount") or 0) except Exception: views = 0 thumbs = ((vd.get("thumbnail") or {}).get("thumbnails") or []) art = thumbs[-1].get("url") if thumbs else None return self._mk_item(vd.get("videoId"), vd.get("title"), vd.get("author"), int(vd.get("lengthSeconds") or 0), views, art) yt = YouTube(YT_API_KEY) def _find_out(prefix, want_ext=None): best = None try: for f in sorted(os.listdir(DL_DIR)): if not f.startswith(prefix): continue if f.endswith((".part", ".ytdl", ".tmp")): continue p = os.path.join(DL_DIR, f) try: sz = os.path.getsize(p) except Exception: continue if sz <= 1024 or sz > MAX_SIZE: try: os.remove(p) except Exception: pass continue if want_ext and f.lower().endswith("." + want_ext): return p if best is None: best = p except Exception: pass return best def yt_fetch_audio(video_id, code): """دانلود صدای یوتیوب. خروجی: مسیر فایل یا None.""" YT_LAST_ERR[0] = u"" if not video_id: YT_LAST_ERR[0] = u"no video id" return None if not YTDLP_PATH: YT_LAST_ERR[0] = u"yt-dlp پیدا نشد" return None ensure_dl_dir() prefix = "%s_yt." % code for f in os.listdir(DL_DIR): if f.startswith(prefix): try: os.remove(os.path.join(DL_DIR, f)) except Exception: pass url = "https://www.youtube.com/watch?v=" + video_id outtmpl = os.path.join(DL_DIR, prefix + "%(ext)s") mode = YT_MODE if mode in ("m4a", "mp3") and not HAS_FFMPEG: mode = "raw" cmd = [YTDLP_PATH, "-o", outtmpl, "--no-playlist", "--no-progress", "--no-warnings", "--no-part", "--no-mtime", "--no-cache-dir", "--retries", "3", "--socket-timeout", "30", "--geo-bypass", "--max-filesize", str(MAX_SIZE), "--extractor-args", "youtube:player_client=" + YT_CLIENTS, # هرگز ویدیو نگیر: فقط استریم صوتی خالص "-f", "ba[vcodec=none]/bestaudio/best", "-S", "abr,asr"] want_ext = None if mode == "m4a": cmd += ["-x", "--audio-format", "m4a", "--audio-quality", "0", "--ffmpeg-location", FFMPEG_PATH] want_ext = "m4a" elif mode == "mp3": cmd += ["-x", "--audio-format", "mp3", "--audio-quality", "0", "--postprocessor-args", "ffmpeg:-b:a 320k -ar 44100 -ac 2", "--ffmpeg-location", FFMPEG_PATH] want_ext = "mp3" elif HAS_FFMPEG: cmd += ["--ffmpeg-location", FFMPEG_PATH] if YT_COOKIES and os.path.isfile(YT_COOKIES): cmd += ["--cookies", YT_COOKIES] cmd.append(url) p = None try: p = subprocess.Popen(cmd, stdout=subprocess.PIPE, stderr=subprocess.STDOUT) out, _ = p.communicate(timeout=DL_TIMEOUT) txt = (out or b"").decode("utf-8", "replace") if p.returncode != 0: tail = u" | ".join([l for l in txt.splitlines() if l.strip()][-3:]) YT_LAST_ERR[0] = u"rc=%s %s" % (p.returncode, tail[:300]) log.warning("yt-dlp %s: %s", video_id, YT_LAST_ERR[0]) except subprocess.TimeoutExpired: try: p.kill() except Exception: pass YT_LAST_ERR[0] = u"timeout" except Exception as e: YT_LAST_ERR[0] = u"exec %s" % e got = _find_out(prefix, want_ext) # اگر پس‌پردازش انجام نشد، دستی به m4a تبدیل کن تا تلگرام پخشش کند if got and want_ext and not got.lower().endswith("." + want_ext) and HAS_FFMPEG: dst = os.path.join(DL_DIR, prefix + want_ext) args = [FFMPEG_PATH, "-y", "-loglevel", "error", "-i", got, "-vn"] if want_ext == "m4a": args += ["-c:a", "aac", "-b:a", "192k"] else: args += ["-c:a", "libmp3lame", "-b:a", "320k"] args += ["-ar", "44100", "-ac", "2", dst] try: subprocess.call(args, timeout=DL_TIMEOUT) if os.path.isfile(dst) and 1024 < os.path.getsize(dst) <= MAX_SIZE: try: os.remove(got) except Exception: pass got = dst except Exception as e: log.warning("ffmpeg convert: %s", e) if not got: if not YT_LAST_ERR[0]: YT_LAST_ERR[0] = u"فایلی تولید نشد" return None br, dur = probe_audio(got) log.info("yt audio: %s (%s, %s kbps, %s)", os.path.basename(got), mode, br or "?", fmt_size(os.path.getsize(got))) return got # ====================================================================== # MERGE / RANK # ====================================================================== def score_item(item, query): full = u"%s %s" % (item.get("artist") or u"", item.get("title") or u"") s = sim(query, full) * 100.0 s += sim(query, item.get("title") or u"") * 40.0 if item.get("both"): s += 45.0 s += (item.get("popularity") or 0) * 0.30 plays = item.get("plays") or 0 if plays > 0: s += min(math.log10(plays + 1) * 6.0, 30.0) if item.get("downloadable"): s += 12.0 if item.get("src") == "yt": s += YouTube.music_bonus(item) d = item.get("duration") or 0 if 60 <= d <= 600: s += 6.0 if d > MAX_DURATION: s -= 100.0 t = (item.get("title") or u"").lower() ql = norm(query) for bad in ("live", "demo", "snippet", "teaser", "remix", "cover"): if bad in t and bad not in ql: s -= 10.0 return s def balance_sources(items, limit=SEARCH_LIMIT): if not BALANCE_SOURCES: return items[:limit] both = [i for i in items if i.get("both")] scs = [i for i in items if not i.get("both") and i.get("src") == "sc"] yts = [i for i in items if not i.get("both") and i.get("src") == "yt"] yt_need = limit // 2 sc_need = limit - yt_need out = [] for it in both: if len(out) >= limit or (sc_need <= 0 and yt_need <= 0): break out.append(it) sc_need -= 1 yt_need -= 1 si = yi = 0 while len(out) < limit and (si < len(scs) or yi < len(yts)): took = False if sc_need > 0 and si < len(scs): out.append(scs[si]); si += 1; sc_need -= 1; took = True if len(out) >= limit: break if yt_need > 0 and yi < len(yts): out.append(yts[yi]); yi += 1; yt_need -= 1; took = True if not took: break if len(out) < limit: rest = scs[si:] + yts[yi:] rest.sort(key=lambda x: x.get("score", 0), reverse=True) for it in rest: if len(out) >= limit: break out.append(it) out.sort(key=lambda x: x.get("score", 0), reverse=True) return out[:limit] def merge_results(query, sc_list, yt_list): sc_list = sc_list or [] yt_list = yt_list or [] used_sc = set() merged = [] for ytt in yt_list: yt_key = u"%s %s" % (ytt["artist"], ytt["title"]) best_i, best_r = -1, 0.0 for i, sct in enumerate(sc_list): if i in used_sc: continue r = max(sim(yt_key, u"%s %s" % (sct["artist"], sct["title"])), sim(ytt["title"], sct["title"])) if ytt["duration"] and sct["duration"] and \ abs(ytt["duration"] - sct["duration"]) <= 6: r += 0.12 if r > best_r: best_r, best_i = r, i if best_i >= 0 and best_r >= MATCH_THRESHOLD: used_sc.add(best_i) base = dict(sc_list[best_i]) base["both"] = True base["popularity"] = ytt.get("popularity") or 0 base["yt_id"] = ytt.get("id") base["yt_url"] = ytt.get("permalink") or "" base["raw_title"] = ytt.get("raw_title") or u"" if not base.get("artwork") and ytt.get("artwork"): base["artwork"] = ytt["artwork"] merged.append(base) else: it = dict(ytt) it["both"] = False merged.append(it) for i, sct in enumerate(sc_list): if i in used_sc: continue it = dict(sct) it["both"] = False merged.append(it) for it in merged: it["rel"] = relevance(it, query) it["score"] = score_item(it, query) + it["rel"] * 60.0 merged = [m for m in merged if (m.get("duration") or 0) <= MAX_DURATION] merged.sort(key=lambda x: x["score"], reverse=True) strong = [m for m in merged if m["rel"] >= MIN_RELEVANCE] weak = [m for m in merged if m["rel"] < MIN_RELEVANCE] if len(strong) >= MIN_RESULTS: result = strong weak_yt = [m for m in weak if m.get("src") == "yt" and not m.get("both")] already = sum(1 for m in result if m.get("src") == "yt" and not m.get("both")) need = max(0, MIN_YT_VISIBLE - already) if need and weak_yt: result = result + weak_yt[:need] elif strong: result = strong + weak[:max(0, MIN_RESULTS - len(strong))] else: result = merged result.sort(key=lambda x: x["score"], reverse=True) result = balance_sources(result, SEARCH_LIMIT) result.sort(key=lambda x: x.get("score", 0), reverse=True) return result def badge(item): if item.get("both"): return u"⭐" if item.get("src") == "yt": return u"🔴" return u"🟠" def source_line(item): if item.get("src") == "sc": if item.get("both"): return u"ساوندکلاد (در یوتیوب هم بود)" return u"ساوندکلاد" if HAS_YT_AUDIO: return u"یوتیوب" return u"یوتیوب (صدا از ساوندکلاد)" def used_source(item, via_ytdl): if via_ytdl: return u"یوتیوب" if item.get("src") == "yt": return u"یوتیوب (صدا از ساوندکلاد)" return u"ساوندکلاد" # ====================================================================== # FORMAT (reversed: 20 top ... 1 bottom) # ====================================================================== def render_block(rank, it, in_group): code = it["code"] dl = (u"/dl_%s@%s" % (code, tg.username)) if (in_group and tg.username) \ else (u"/dl_%s" % code) br = it.get("bitrate") or (160 if it.get("src") == "yt" else 128) qtxt = (u"%d kbps" % br) + (u" 💿" if it.get("downloadable") else u"") sztxt = fmt_size(est_size(it.get("duration"), br)) star = u" 🥇" if rank == 1 else u"" return (u"\n\n{n}. {b} {title}{star}\n" u"├ 👤 {artist}\n" u"├ ⏱ {dur} 📦 ~{size} 🎚 {q}\n" u"├ 🎯 {src}\n" u"└ 📥 {dl}").format( n=rank, b=badge(it), title=esc(it.get("title")), star=star, artist=esc(it.get("artist")), dur=fmt_dur(it.get("duration")), size=sztxt, q=qtxt, src=source_line(it), dl=dl) def format_results(query, items, in_group): blocks = [render_block(rank, it, in_group) for rank, it in enumerate(items, 1)] blocks.reverse() pages = [] head = T_HDR.format(q=esc(query), n=len(items)) buf = [head] cur = len(head) for block in blocks: if cur + len(block) > PAGE_LEN: pages.append(u"".join(buf)) buf, cur = [], 0 buf.append(block) cur += len(block) buf.append(u"\n\n" + T_FTR) pages.append(u"".join(buf)) return pages # ====================================================================== # KEYBOARDS / THROTTLE # ====================================================================== def kb_start(): uname = tg.username or "bot" return {"inline_keyboard": [ [{"text": u"➕ افزودن به گروه", "url": "https://t.me/%s?startgroup=true" % uname}], [{"text": u"📖 راهنما", "callback_data": "help"}, {"text": u"🎧 نمونه جستجو", "callback_data": "demo"}]]} def kb_help(): return {"inline_keyboard": [[{"text": u"🔙 بازگشت", "callback_data": "back"}]]} def kb_youtube(url): if not url: return None return {"inline_keyboard": [[{"text": u"🔴 باز کردن در یوتیوب", "url": url}]]} class Throttle(object): def __init__(self): self.u, self.c = {}, {} self.lock = threading.Lock() def check(self, uid, cid): now = time.time() with self.lock: lu = self.u.get(uid, 0) if now - lu < USER_COOLDOWN: return int(USER_COOLDOWN - (now - lu)) + 1 lc = self.c.get(cid, 0) if now - lc < CHAT_COOLDOWN: return int(CHAT_COOLDOWN - (now - lc)) + 1 self.u[uid] = now self.c[cid] = now if len(self.u) > 4000: self.u = {k: v for k, v in self.u.items() if now - v < 600} if len(self.c) > 2000: self.c = {k: v for k, v in self.c.items() if now - v < 600} return 0 throttle = Throttle() chat_locks = set() chat_locks_guard = threading.Lock() # ====================================================================== # LINK DETECTION # ====================================================================== RE_YT_WATCH = re.compile( r"https?://(?:www\.|m\.|music\.)?youtube\.com/watch\?[^\s]*?v=([A-Za-z0-9_\-]{11})", re.I) RE_YT_BE = re.compile(r"https?://youtu\.be/([A-Za-z0-9_\-]{11})", re.I) RE_YT_SHORTS = re.compile( r"https?://(?:www\.|m\.)?youtube\.com/shorts/([A-Za-z0-9_\-]{11})", re.I) RE_YT_SET = re.compile( r"https?://(?:www\.|m\.|music\.)?youtube\.com/(playlist|channel|c/|@|user/|results)", re.I) RE_SC_LINK = re.compile(r"https?://(?:www\.|m\.)?soundcloud\.com/[^\s<>\"]+", re.I) RE_SC_SHORT = re.compile(r"https?://(?:on\.)?soundcloud\.com/[A-Za-z0-9]+", re.I) RE_ANY_URL = re.compile(r"https?://\S+", re.I) SC_SET_HINT = re.compile(r"/sets/|/discover|/you/|/stations/|/tags/", re.I) def detect_link(text): m = RE_YT_WATCH.search(text) or RE_YT_BE.search(text) or \ RE_YT_SHORTS.search(text) if m: return ("yt", m.group(1)) if RE_YT_SET.search(text): return ("set", text) m = RE_SC_LINK.search(text) or RE_SC_SHORT.search(text) if m: url = m.group(0).rstrip(u".,;:)»،") if SC_SET_HINT.search(url): return ("set", url) path = url.split("soundcloud.com/", 1)[-1].strip("/") if path and "/" not in path and "on.soundcloud" not in url: return ("set", url) return ("sc", url) if RE_ANY_URL.search(text): return ("other", None) return None # ====================================================================== # TAGGING # ====================================================================== def tag_file(path, title, artist, album, art_bytes): if not HAS_TAGS: return low = path.lower() try: if low.endswith(".mp3"): try: tags = ID3(path) except ID3NoHeaderError: tags = ID3() for k in ("TIT2", "TPE1", "TALB"): tags.delall(k) tags.add(TIT2(encoding=3, text=title or u"")) tags.add(TPE1(encoding=3, text=artist or u"")) if album: tags.add(TALB(encoding=3, text=album)) if art_bytes: tags.delall("APIC") tags.add(APIC(encoding=3, mime="image/jpeg", type=3, desc="Cover", data=art_bytes)) tags.save(path, v2_version=3) elif low.endswith((".m4a", ".mp4", ".aac")): mp4 = MP4(path) mp4["\xa9nam"] = [title or u""] mp4["\xa9ART"] = [artist or u""] if album: mp4["\xa9alb"] = [album] if art_bytes: mp4["covr"] = [MP4Cover(art_bytes, imageformat=MP4Cover.FORMAT_JPEG)] mp4.save() except Exception as e: log.warning("tag error: %s", e) # ====================================================================== # JOBS # ====================================================================== jobs = queue.Queue(maxsize=40) active = [0] active_lock = threading.Lock() def job_search(chat_id, msg_id, uid, query, in_group, thread_id): wait = tg.send(chat_id, T_SEARCHING, reply_to=msg_id, thread_id=thread_id) wait_id = wait.get("message_id") if wait else None def out(txt, kb=None): if wait_id: tg.edit(chat_id, wait_id, txt, kb=kb) else: tg.send(chat_id, txt, reply_to=msg_id, kb=kb, thread_id=thread_id) cache_key = u"%s:%s" % (CACHE_VERSION, norm(query)) items = db.get_search(cache_key) if items is None: sc_res = sc.search(query, SC_FETCH) yt_res = yt.search(query, YT_FETCH) if yt_res is None: log.warning("YT search failed %r: %s", query, yt.last_error) yt_res = [] else: log.info("YT search ok (%s) %r: %d", yt.mode, query, len(yt_res)) if sc_res is None and not yt_res: err = sc.last_error or "" if "client_id" in err: txt = T_ERR_CID elif "429" in err or "403" in err: txt = T_ERR_BLOCK elif err.startswith("net") or err.startswith("home"): txt = T_ERR_NET elif "http" in err: txt = T_ERR_HTTP.format(code=err.split()[-1]) else: txt = T_ERR_NET if ADMIN_RAW_ERR and uid in ADMINS: if err: txt += u"\n\nSC: %s" % esc(err) if yt.last_error: txt += u"\nYT: %s" % esc(yt.last_error) out(txt) return items = merge_results(query, sc_res or [], yt_res or []) for it in items: it["code"] = make_code(u"%s:%s" % (it.get("src"), it.get("id"))) db.put_track(it["code"], it.get("src"), it) if items: db.put_search(cache_key, items) else: for it in items: if it.get("code"): db.put_track(it["code"], it.get("src"), it) if not items: out(T_ERR_EMPTY.format(q=esc(query))) return items = sorted(items, key=lambda x: x.get("score", 0), reverse=True) pages = format_results(query, items, in_group) out(pages[0]) for p in pages[1:]: tg.send(chat_id, p, thread_id=thread_id) def find_sc_match(item): q = (u"%s %s" % (item.get("artist") or u"", item.get("title") or u"")).strip() res = sc.search(q, 12) if not res: return None best, best_r = None, 0.0 for c in res: r = max(sim(q, u"%s %s" % (c["artist"], c["title"])), sim(item.get("title") or u"", c["title"])) if item.get("duration") and c.get("duration") and \ abs(item["duration"] - c["duration"]) <= 8: r += 0.15 if r > best_r: best_r, best = r, c return best if (best and best_r >= MATCH_THRESHOLD) else None def job_download(chat_id, msg_id, uid, code, thread_id): item = db.get_track(code) if not item: log.warning("track not found: %r", code) tg.send(chat_id, T_EXPIRED, reply_to=msg_id, thread_id=thread_id) return cached = db.get_file(code) if cached: tg.action(chat_id, "upload_document") r = tg.send_audio(chat_id, cached, title=item.get("title"), performer=item.get("artist"), duration=item.get("duration"), reply_to=msg_id, thread_id=thread_id) if r: return status = tg.send(chat_id, T_PREPARING.format(t=esc(item.get("title"))), reply_to=msg_id, thread_id=thread_id) status_id = status.get("message_id") if status else None def out(txt, kb=None): if status_id and not kb: tg.edit(chat_id, status_id, txt) else: if status_id: tg.delete(chat_id, status_id) tg.send(chat_id, txt, reply_to=msg_id, kb=kb, thread_id=thread_id) path = None thumb_path = None audio_item = item via_ytdl = False yt_only = (item.get("src") == "yt" and not item.get("both")) try: tg.action(chat_id, "upload_document") ok = False if yt_only: if HAS_YT_AUDIO: if status_id: tg.edit(chat_id, status_id, T_DL_YT) p = yt_fetch_audio(item.get("id"), code) if p: path = p audio_item = item via_ytdl = True ok = True if not ok: match = find_sc_match(item) if not match: txt = T_NOAUDIO.format(url=item.get("permalink") or "") if ADMIN_RAW_ERR and uid in ADMINS and YT_LAST_ERR[0]: txt += u"\n\n%s" % esc(YT_LAST_ERR[0]) out(txt, kb=kb_youtube(item.get("permalink"))) return audio_item = dict(match) audio_item["title"] = item.get("title") audio_item["artist"] = item.get("artist") audio_item["artwork"] = item.get("artwork") or match.get("artwork") ensure_dl_dir() fname = u"%s - %s" % (safe_name(audio_item.get("artist"), 30), safe_name(audio_item.get("title"), 40)) path = os.path.join(DL_DIR, "%s_%s.mp3" % (code, safe_name(fname, 50))) ok = bool(sc.download(audio_item, path)) else: ensure_dl_dir() fname = u"%s - %s" % (safe_name(audio_item.get("artist"), 30), safe_name(audio_item.get("title"), 40)) path = os.path.join(DL_DIR, "%s_%s.mp3" % (code, safe_name(fname, 50))) ok = bool(sc.download(audio_item, path)) if not ok or not path or not os.path.exists(path): if yt_only: txt = T_NOAUDIO.format(url=item.get("permalink") or "") if ADMIN_RAW_ERR and uid in ADMINS and YT_LAST_ERR[0]: txt += u"\n\n%s" % esc(YT_LAST_ERR[0]) out(txt, kb=kb_youtube(item.get("permalink"))) else: txt = T_DLFAIL if ADMIN_RAW_ERR and uid in ADMINS and sc.last_error: txt += u"\n%s" % esc(sc.last_error) out(txt) return if os.path.getsize(path) > MAX_SIZE: out(T_TOOBIG.format(sz=int(MAX_SIZE / (1024 * 1024)))) return art = sc.artwork(audio_item.get("artwork")) tag_file(path, audio_item.get("title"), audio_item.get("artist"), item.get("album") or u"", art) if art: thumb_path = path + ".jpg" try: with open(thumb_path, "wb") as f: f.write(art) except Exception: thumb_path = None br, dur = probe_audio(path) duration = dur or audio_item.get("duration") or item.get("duration") if status_id: tg.edit(chat_id, status_id, T_UPLOADING) qline = (u"%d kbps" % br) if br else u"—" cap = u"🎧 %s\n👤 %s\n🎚 %s\n🎯 %s" % ( esc(audio_item.get("title")), esc(audio_item.get("artist")), qline, used_source(item, via_ytdl)) if tg.username: cap += u"\n\n🤖 @%s" % tg.username tg.action(chat_id, "upload_document") r = tg.send_audio(chat_id, path, title=audio_item.get("title"), performer=audio_item.get("artist"), duration=duration, caption=cap, reply_to=msg_id, thumb_path=thumb_path, thread_id=thread_id) if r and (r.get("audio") or {}).get("file_id"): db.put_file(code, r["audio"]["file_id"]) if status_id: tg.delete(chat_id, status_id) elif not r: out(T_DLFAIL) elif status_id: tg.delete(chat_id, status_id) except Exception as e: log.exception("download job error") txt = T_ERR_GEN if ADMIN_RAW_ERR and uid in ADMINS: txt += u"\n%s: %s" % (type(e).__name__, esc(str(e))[:200]) out(txt) finally: for p in (path, thumb_path): if p and os.path.exists(p): try: os.remove(p) except Exception: pass def job_link(chat_id, msg_id, uid, kind, value, thread_id): wait = tg.send(chat_id, T_LINK_WORK, reply_to=msg_id, thread_id=thread_id) wait_id = wait.get("message_id") if wait else None def fail(txt): if wait_id: tg.edit(chat_id, wait_id, txt) else: tg.send(chat_id, txt, reply_to=msg_id, thread_id=thread_id) if kind == "sc": item = sc.resolve(value) if not item: err = sc.last_error or "" if "403" in err or "401" in err: fail(T_LINK_PRIV) elif "not a track" in err: fail(T_LINK_SET) elif "404" in err: fail(T_LINK_BAD) else: txt = T_LINK_BAD if ADMIN_RAW_ERR and uid in ADMINS and err: txt += u"\n%s" % esc(err) fail(txt) return item["both"] = False elif kind == "yt": item = yt.video(value) if not item: txt = T_LINK_BAD if ADMIN_RAW_ERR and uid in ADMINS and yt.last_error: txt += u"\n%s" % esc(yt.last_error) fail(txt) return item["both"] = False else: fail(T_LINK_SET) return if (item.get("duration") or 0) > MAX_DURATION: fail(T_LINK_LONG.format(dur=int(MAX_DURATION / 60))) return code = make_code(u"%s:%s" % (item.get("src"), item.get("id"))) item["code"] = code db.put_track(code, item.get("src"), item) if wait_id: tg.delete(chat_id, wait_id) job_download(chat_id, msg_id, uid, code, thread_id) def worker(idx): while True: job = jobs.get() if job is None: jobs.task_done() break fn, args = job with active_lock: active[0] += 1 try: fn(*args) except Exception as e: log.exception("worker %s: %s", idx, e) try: txt = T_ERR_GEN if ADMIN_RAW_ERR: txt += u"\n%s" % esc(str(e))[:200] tg.send(args[0], txt, reply_to=args[1]) except Exception: pass finally: with active_lock: active[0] -= 1 with chat_locks_guard: chat_locks.discard(args[0]) jobs.task_done() def enqueue(fn, args): try: jobs.put_nowait((fn, args)) return True except queue.Full: return False # ====================================================================== # HANDLERS # ====================================================================== PREFIXES = (u"$", u"!", u".", u"#", u"＄") SEARCH_CMDS = ("s", "search", "music", "ahang", "sc", "find", "m", "j") def handle_message(msg): chat = msg.get("chat") or {} chat_id = chat.get("id") chat_type = chat.get("type") or "private" msg_id = msg.get("message_id") thread_id = msg.get("message_thread_id") frm = msg.get("from") or {} uid = frm.get("id") uname = frm.get("first_name") or frm.get("username") or u"دوست من" text = (msg.get("text") or msg.get("caption") or u"").strip() if not chat_id or not uid: return if DEBUG_UPDATES: log.info("MSG chat=%s type=%s uid=%s text=%r", chat_id, chat_type, uid, text[:80]) in_group = chat_type in ("group", "supergroup") if in_group: db.touch_chat(chat_id, chat.get("title") or u"") else: db.touch_user(uid, uname) if msg.get("new_chat_members"): for m in msg["new_chat_members"]: if tg.bot_id and m.get("id") == tg.bot_id: tg.send(chat_id, T_ADDED.format(name=BOT_NAME), thread_id=thread_id) return if not text: return cmd, cmd_raw, arg = None, None, u"" if text.startswith("/"): parts = text.split(None, 1) head = parts[0][1:] arg = parts[1].strip() if len(parts) > 1 else u"" if "@" in head: head, target = head.split("@", 1) if tg.username and target.lower() != tg.username.lower(): return cmd_raw = head cmd = head.lower() # ---- download (case sensitive code) ---- if cmd_raw and cmd_raw.lower().startswith("dl_"): code = cmd_raw[3:].strip() if not code: return w = throttle.check(uid, chat_id) if w: tg.send(chat_id, T_COOLDOWN.format(s=w), reply_to=msg_id, thread_id=thread_id) return if not enqueue(job_download, (chat_id, msg_id, uid, code, thread_id)): tg.send(chat_id, T_BUSY, reply_to=msg_id, thread_id=thread_id) return if cmd == "start": if in_group: tg.send(chat_id, T_START_GP.format(name=BOT_NAME), reply_to=msg_id, thread_id=thread_id) else: tg.send(chat_id, T_START_PV.format(name=BOT_NAME, user=esc(uname)), kb=kb_start()) return if cmd == "help": tg.send(chat_id, T_HELP.format(dur=int(MAX_DURATION / 60), sz=int(MAX_SIZE / (1024 * 1024))), reply_to=msg_id, thread_id=thread_id) return if cmd == "ping": tg.send(chat_id, u"🏓 زنده‌ام.\nصف: %d | در حال کار: %d" % (jobs.qsize(), active[0]), reply_to=msg_id, thread_id=thread_id) return if cmd == "diag" and uid in ADMINS: lines = [] ensure_dl_dir() lines.append(u"پوشه دانلود: %s" % (u"✅" if os.path.isdir(DL_DIR) else u"❌")) cid = sc.client_id(force=True) lines.append(u"SoundCloud client_id: %s" % ((u"✅ " + cid[:8] + u"…") if cid else u"❌")) if not cid and sc.last_error: lines.append(u"علت: %s" % esc(sc.last_error)) r1 = sc.search("tataloo", 5) if cid else None lines.append(u"جستجوی SC: %s" % (u"✅ %d" % len(r1) if r1 is not None else u"❌ %s" % esc(sc.last_error or ""))) r2 = yt.search("tataloo", 5) lines.append(u"جستجوی YouTube (%s): %s" % (yt.mode, u"✅ %d" % len(r2) if r2 else u"❌ %s" % esc(yt.last_error or "0"))) lines.append(u"yt-dlp: %s" % esc(YTDLP_PATH or u"نیست")) lines.append(u"ffmpeg: %s" % esc(FFMPEG_PATH or u"نیست")) lines.append(u"حالت خروجی یوتیوب: %s" % YT_MODE) st = db.stats() lines.append(u"صف: %d | فعال: %d | تگ: %s" % (jobs.qsize(), active[0], u"✅" if HAS_TAGS else u"❌")) lines.append(u"ترک‌های ذخیره‌شده: %d" % st["tracks"]) tg.send(chat_id, u"🔧 تشخیص\n" + u"\n".join(lines), reply_to=msg_id, thread_id=thread_id) return if cmd == "stats" and uid in ADMINS: st = db.stats() tg.send(chat_id, u"📊 آمار\nکاربران: %d\nگروه‌ها: %d\n" u"ترک‌ها: %d\nفایل کش‌شده: %d\n" u"جستجوی کش‌شده: %d" % (st["users"], st["chats"], st["tracks"], st["files"], st["searches"]), reply_to=msg_id, thread_id=thread_id) return # ---- links ---- link = detect_link(text) if link: kind, value = link if kind == "other": return if kind == "set": tg.send(chat_id, T_LINK_SET, reply_to=msg_id, thread_id=thread_id) return w = throttle.check(uid, chat_id) if w: tg.send(chat_id, T_COOLDOWN.format(s=w), reply_to=msg_id, thread_id=thread_id) return if not enqueue(job_link, (chat_id, msg_id, uid, kind, value, thread_id)): tg.send(chat_id, T_BUSY, reply_to=msg_id, thread_id=thread_id) return # ---- search ---- query = None if cmd in SEARCH_CMDS: if len(arg) < 2: tg.send(chat_id, T_CMD_EMPTY, reply_to=msg_id, thread_id=thread_id) return query = arg elif text[:1] in PREFIXES: query = text[1:].strip() elif not in_group and not text.startswith("/"): query = text if query is None: return if len(query) < 2: tg.send(chat_id, T_SHORT, reply_to=msg_id, thread_id=thread_id) return w = throttle.check(uid, chat_id) if w: tg.send(chat_id, T_COOLDOWN.format(s=w), reply_to=msg_id, thread_id=thread_id) return with chat_locks_guard: if chat_id in chat_locks: tg.send(chat_id, T_BUSY, reply_to=msg_id, thread_id=thread_id) return chat_locks.add(chat_id) if not enqueue(job_search, (chat_id, msg_id, uid, query, in_group, thread_id)): with chat_locks_guard: chat_locks.discard(chat_id) tg.send(chat_id, T_BUSY, reply_to=msg_id, thread_id=thread_id) def handle_callback(cb): data = cb.get("data") or "" msg = cb.get("message") or {} chat_id = (msg.get("chat") or {}).get("id") msg_id = msg.get("message_id") uname = (cb.get("from") or {}).get("first_name") or u"دوست من" if data == "help": tg.edit(chat_id, msg_id, T_HELP.format(dur=int(MAX_DURATION / 60), sz=int(MAX_SIZE / (1024 * 1024))), kb=kb_help()) elif data == "back": tg.edit(chat_id, msg_id, T_START_PV.format(name=BOT_NAME, user=esc(uname)), kb=kb_start()) elif data == "demo": tg.answer_cb(cb.get("id"), u"مثلاً بنویس: Tataloo Didi", alert=True) return tg.answer_cb(cb.get("id")) def dispatch(update): try: if "message" in update: handle_message(update["message"]) elif "callback_query" in update: handle_callback(update["callback_query"]) elif "my_chat_member" in update: m = update["my_chat_member"] st = (m.get("new_chat_member") or {}).get("status") chat = m.get("chat") or {} if st in ("member", "administrator") and \ chat.get("type") in ("group", "supergroup"): db.touch_chat(chat.get("id"), chat.get("title") or u"") except Exception: log.exception("dispatch error") # ====================================================================== # HOUSEKEEPING / MAIN # ====================================================================== def housekeeping(): while True: try: ensure_dl_dir() db.cleanup() now = time.time() for f in os.listdir(DL_DIR): p = os.path.join(DL_DIR, f) try: if os.path.isfile(p) and now - os.path.getmtime(p) > 1800: os.remove(p) except Exception: pass except Exception: log.exception("housekeeping error") time.sleep(1800) def acquire_lock(): try: import fcntl f = open(LOCK_PATH, "w") fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB) f.write(str(os.getpid())) f.flush() return f except Exception: log.error("Another instance is already running. Exiting.") sys.exit(0) def main(): lock = acquire_lock() ensure_dl_dir() me = tg.me() if not me: log.error("getMe failed. Check BOT_TOKEN and network.") sys.exit(1) log.info("Bot @%s (id=%s) started. can_read_all_group_messages=%s", me.get("username"), me.get("id"), me.get("can_read_all_group_messages")) if not me.get("can_read_all_group_messages"): log.warning("Privacy mode is ON. Disable it in BotFather, then " "remove and re-add the bot to the group.") tg.api("deleteWebhook", {"drop_pending_updates": "true"}) log.info("yt-dlp: %s", YTDLP_PATH or "NOT FOUND") log.info("ffmpeg: %s", FFMPEG_PATH or "NOT FOUND") log.info("YouTube output mode: %s", YT_MODE) if not YTDLP_PATH: log.warning("YouTube-only tracks will fall back to SoundCloud match.") test = yt.search("shadmehr", 5) if test: log.info("YouTube search (%s): OK, %d results", yt.mode, len(test)) else: log.warning("YouTube search (%s): FAILED: %s", yt.mode, yt.last_error) for i in range(WORKERS): t = threading.Thread(target=worker, args=(i,)) t.daemon = True t.start() hk = threading.Thread(target=housekeeping) hk.daemon = True hk.start() offset = None errors = 0 while True: try: params = {"timeout": POLL_TIMEOUT, "allowed_updates": json.dumps( ["message", "callback_query", "my_chat_member"])} if offset is not None: params["offset"] = offset updates = tg.api("getUpdates", params, timeout=POLL_TIMEOUT + 10) if updates is None: errors += 1 time.sleep(min(errors * 2, 20)) continue errors = 0 for up in updates: offset = up["update_id"] + 1 dispatch(up) except KeyboardInterrupt: log.info("Stopping...") break except Exception: log.exception("main loop error") time.sleep(5) for _ in range(WORKERS): jobs.put(None) try: lock.close() except Exception: pass if __name__ == "__main__": main()