import asyncio import hashlib import json import os import re import shutil import tempfile from typing import Dict, Tuple, List from telegram import Update from telegram.ext import ApplicationBuilder, CommandHandler, MessageHandler, ContextTypes, filters # ---------- Config ---------- HTML_ROOT = "/var/www/html" VALID_DIR_RE = re.compile(r"^[A-Za-z0-9._-]+$") TELEGRAM_BOT_TOKEN = "7718088623:AAG1jXolMhGf0-dLjah-5YQ9P3CSXiI1WqI" ADMIN_ID = 7883601886 ALLOWED_IDS_JSON = os.path.join(HTML_ROOT, "allowed_ids.json") # نگه‌داری پوشه مقصد هر کاربر در حافظه user_dirs: Dict[int, str] = {} # ---------- Utils for IDs JSON ---------- def load_allowed_ids() -> List[int]: try: if not os.path.exists(ALLOWED_IDS_JSON): # اگر نبود، فقط ادمین داخلش باشه save_allowed_ids([ADMIN_ID]) return [ADMIN_ID] with open(ALLOWED_IDS_JSON, "r", encoding="utf-8") as f: data = json.load(f) # تضمین نوع int return [int(x) for x in data if isinstance(x, (int, str)) and str(x).isdigit()] except Exception: # در صورت خرابی فایل، حداقل اجازهٔ ادمین return [ADMIN_ID] def save_allowed_ids(ids: List[int]) -> None: try: # حذف Duplicate و مرتب‌سازی uniq = sorted(set(int(x) for x in ids)) with open(ALLOWED_IDS_JSON, "w", encoding="utf-8") as f: json.dump(uniq, f, ensure_ascii=False, indent=2) try: os.chmod(ALLOWED_IDS_JSON, 0o644) except Exception: pass except Exception as e: # در بدترین حالت نادیده بگیر pass def ensure_id_in_db(uid: int) -> None: ids = load_allowed_ids() if uid not in ids: ids.append(uid) save_allowed_ids(ids) # ---------- File helpers ---------- def sha256_and_md5(path: str) -> Tuple[str, str]: sha256 = hashlib.sha256() md5 = hashlib.md5() with open(path, 'rb') as f: for chunk in iter(lambda: f.read(1024 * 1024), b''): sha256.update(chunk) md5.update(chunk) return sha256.hexdigest(), md5.hexdigest() def ensure_within_root(path: str) -> None: real_root = os.path.realpath(HTML_ROOT) real_path = os.path.realpath(path) if not real_path.startswith(real_root + os.sep) and real_path != real_root: raise PermissionError("Path escapes HTML_ROOT") def fix_permissions(dir_path: str, file_path: str) -> None: try: os.chmod(dir_path, 0o755) except Exception: pass try: os.chmod(file_path, 0o644) except Exception: pass # ---------- Access checks ---------- async def is_allowed_user(update: Update) -> bool: uid = update.effective_user.id return uid in load_allowed_ids() async def require_allowed(update: Update) -> bool: if not await is_allowed_user(update): await update.effective_chat.send_message("⛔️ دسترسی غیرمجاز.") return False return True def is_admin(update: Update) -> bool: return update.effective_user.id == ADMIN_ID async def require_admin(update: Update) -> bool: if not is_admin(update): await update.effective_chat.send_message("⛔️ فقط ادمین اجازه دارد.") return False return True # ---------- Admin: /ls ---------- async def cmd_ls(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await require_admin(update): return try: entries = sorted(e.name for e in os.scandir(HTML_ROOT) if e.is_dir()) if entries: msg = "📂 پوشه‌های موجود:\n" + "\n".join(f"▫️ {name}" for name in entries) await update.effective_chat.send_message(msg) else: await update.effective_chat.send_message("⚠️ هیچ پوشه‌ای یافت نشد.") except Exception as e: await update.effective_chat.send_message(f"❌ خطا در لیست پوشه‌ها: {e}") # ---------- Admin: /set__ ---------- SET_CMD_REGEX = re.compile(r"^/set_(\d+)_([A-Za-z0-9._-]+)$") async def admin_set_inline(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: """ هندلر برای دستوری مثل: /set_755387189_lrwNVC570 """ if not await require_admin(update): return text = (update.message.text or "").strip() m = SET_CMD_REGEX.match(text) if not m: await update.effective_chat.send_message("❌ فرمت نامعتبر. مثال:\n/set_755387189_lrwNVC570") return uid_str, dirname = m.group(1), m.group(2) try: uid = int(uid_str) except ValueError: await update.effective_chat.send_message("❌ USER_ID نامعتبر است.") return if not VALID_DIR_RE.match(dirname): await update.effective_chat.send_message("❌ نام پوشه نامعتبر است.") return dir_path = os.path.join(HTML_ROOT, dirname) try: ensure_within_root(dir_path) if not os.path.isdir(dir_path): await update.effective_chat.send_message("❌ پوشه وجود ندارد.") return # 1) اضافه‌کردن کاربر به دیتابیس JSON اگر نبود ensure_id_in_db(uid) # 2) تنظیم مقصد فایل برای این کاربر (فقط در حافظه) user_dirs[uid] = dirname await update.effective_chat.send_message( f"✅ کاربر {uid} به مجازها اضافه شد و پوشهٔ مقصد روی '{dirname}' تنظیم شد." ) except Exception as e: await update.effective_chat.send_message(f"❌ خطا: {e}") # ---------- Document Handler (users & admin) ---------- async def on_document(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: if not await require_allowed(update): return doc = update.message.document if not doc or not doc.file_name.lower().endswith('.apk'): await update.effective_chat.send_message("❌ فقط فایل .apk پذیرفته می‌شود.") return user_id = update.effective_user.id dirname = user_dirs.get(user_id) # اگر برای کاربر مقصد ست نشده باشد (ادمین هم اگر برای خودش ست نکرده باشد) if not dirname: await update.effective_chat.send_message("⚠️ ابتدا ادمین باید با دستور /set__ پوشه مقصد شما را تنظیم کند.") return target_dir = os.path.join(HTML_ROOT, dirname) target_file = os.path.join(target_dir, "base.apk") try: ensure_within_root(target_dir) # دانلود موقت tf = tempfile.NamedTemporaryFile(delete=False) tmp_path = tf.name tf.close() tg_file = await context.bot.get_file(doc.file_id) await tg_file.download_to_drive(tmp_path) new_sha256, new_md5 = sha256_and_md5(tmp_path) existed = os.path.exists(target_file) old_sha256 = old_md5 = None if existed: try: old_sha256, old_md5 = sha256_and_md5(target_file) except Exception: existed = False if existed and old_sha256 and old_md5: equal = (new_sha256 == old_sha256) and (new_md5 == old_md5) if equal: await update.effective_chat.send_message("⚠️ فایل جدید با فایل فعلی یکسان است. آپلود لغو شد.") os.unlink(tmp_path) return else: os.replace(tmp_path, target_file) fix_permissions(target_dir, target_file) await update.effective_chat.send_message(f"✅ فایل جایگزین شد در {target_dir}/base.apk") else: os.makedirs(target_dir, exist_ok=True) os.replace(tmp_path, target_file) fix_permissions(target_dir, target_file) await update.effective_chat.send_message(f"✅ فایل ذخیره شد در {target_dir}/base.apk") except Exception as e: try: os.unlink(tmp_path) except Exception: pass await update.effective_chat.send_message(f"❌ خطا: {e}") # ---------- Start / Help ---------- async def cmd_start(update: Update, context: ContextTypes.DEFAULT_TYPE) -> None: uid = update.effective_user.id # اگر کاربر در دیتابیس نیست، دسترسی ندارد if not await is_allowed_user(update): await update.effective_chat.send_message("⛔️ دسترسی غیرمجاز.") return if is_admin(update): await update.effective_chat.send_message( "👋 سلام ادمین!\n" "فرمان‌ها:\n" "• /ls — لیست پوشه‌های موجود در HTML_ROOT\n" "• /set__ — افزودن کاربر به دیتابیس و تنظیم پوشهٔ مقصد\n" "مثال: /set_755387189_lrwNVC570\n" "برای آپلود، یک فایل .apk ارسال کن." ) else: await update.effective_chat.send_message( "👋 سلام! شما مجاز هستید. لطفاً فقط فایل .apk را ارسال کنید تا در پوشهٔ تعریف‌شده جایگزین شود." ) # ---------- Wire up & run ---------- def main() -> None: if not TELEGRAM_BOT_TOKEN: raise RuntimeError("توکن ربات را تنظیم کنید") # اطمینان از وجود فایل JSON با حداقل ادمین ensure_id_in_db(ADMIN_ID) app = ApplicationBuilder().token(TELEGRAM_BOT_TOKEN).build() # فقط برای ادمین فعال هستند app.add_handler(CommandHandler("ls", cmd_ls)) # الگوی /set__ برای ادمین app.add_handler(MessageHandler(filters.TEXT & filters.Regex(SET_CMD_REGEX), admin_set_inline)) # استارت و راهنما app.add_handler(CommandHandler(["start", "help"], cmd_start)) # دریافت فایل‌ها app.add_handler(MessageHandler(filters.Document.ALL, on_document)) app.run_polling(close_loop=False) if __name__ == "__main__": main()