#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Telegram Upload Bot - نسخه پایتون برای سرور اجرا با: python3 bot.py """ import os import time import json import sqlite3 import requests from datetime import datetime from threading import Lock # کتابخانه تلگرام try: from telegram import Update, InlineKeyboardButton, InlineKeyboardMarkup from telegram.ext import Application, CommandHandler, CallbackQueryHandler, MessageHandler, filters, ContextTypes except ImportError: print("لطفا کتابخانه را نصب کنید: pip install python-telegram-bot") exit(1) # ============ تنظیمات ============ BOT_TOKEN = "1646603492:AAFDa_iH5cB7rxGgHO7gj4b_BqOuwyx29HI" UPLOAD_DIR = "uploads" MAX_FILE_SIZE = 1.8 * 1024 * 1024 * 1024 # ایجاد پوشه آپلود if not os.path.exists(UPLOAD_DIR): os.makedirs(UPLOAD_DIR) # ============ دیتابیس SQLite (بدون نیاز به MySQL) ============ def init_db(): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute('''CREATE TABLE IF NOT EXISTS users (telegram_id INTEGER PRIMARY KEY, username TEXT, first_name TEXT, last_name TEXT, created_at TEXT)''') c.execute('''CREATE TABLE IF NOT EXISTS files (id INTEGER PRIMARY KEY AUTOINCREMENT, user_id INTEGER, original_name TEXT, stored_name TEXT, download_url TEXT, size INTEGER, size_formatted TEXT, extension TEXT, created_at TEXT)''') conn.commit() conn.close() init_db() # ============ توابع ============ def format_size(size): for unit in ['B', 'KB', 'MB', 'GB']: if size < 1024.0: return f"{size:.2f} {unit}" size /= 1024.0 return f"{size:.2f} TB" def get_extension(filename): ext = os.path.splitext(filename)[1].lower() return ext if ext else ".unknown" def get_icon(ext): icons = {'.mp4': '🎬', '.mp3': '🎵', '.jpg': '🖼️', '.pdf': '📄', '.zip': '🗂️'} return icons.get(ext, '📄') def save_user(telegram_id, username, first_name, last_name): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute("INSERT OR IGNORE INTO users (telegram_id, username, first_name, last_name, created_at) VALUES (?, ?, ?, ?, ?)", (telegram_id, username, first_name, last_name, datetime.now().isoformat())) conn.commit() conn.close() def save_file(user_id, original_name, stored_name, download_url, size, extension): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute("INSERT INTO files (user_id, original_name, stored_name, download_url, size, size_formatted, extension, created_at) VALUES (?, ?, ?, ?, ?, ?, ?, ?)", (user_id, original_name, stored_name, download_url, size, format_size(size), extension, datetime.now().isoformat())) conn.commit() conn.close() def is_duplicate(user_id, file_name, file_size): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute("SELECT * FROM files WHERE user_id = ? AND original_name = ? AND size = ?", (user_id, file_name, file_size)) result = c.fetchone() conn.close() return result def get_user_files(user_id): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute("SELECT * FROM files WHERE user_id = ? ORDER BY created_at DESC LIMIT 10", (user_id,)) result = c.fetchall() conn.close() return result def get_user_stats(user_id): conn = sqlite3.connect('bot.db') c = conn.cursor() c.execute("SELECT COUNT(*), COALESCE(SUM(size), 0) FROM files WHERE user_id = ?", (user_id,)) result = c.fetchone() conn.close() return result # ============ منوها ============ def get_main_menu(): keyboard = [ [InlineKeyboardButton("📤 فایل", callback_data="file"), InlineKeyboardButton("🔗 لینک", callback_data="link")], [InlineKeyboardButton("📁 فایل‌های من", callback_data="my_files"), InlineKeyboardButton("📊 آمار من", callback_data="my_stats")], [InlineKeyboardButton("❓ راهنما", callback_data="help")] ] return InlineKeyboardMarkup(keyboard) def get_back_menu(): keyboard = [[InlineKeyboardButton("🔙 بازگشت", callback_data="back")]] return InlineKeyboardMarkup(keyboard) # ============ هندلرها ============ async def start(update: Update, context: ContextTypes.DEFAULT_TYPE): user = update.effective_user save_user(user.id, user.username, user.first_name, user.last_name) welcome = f"✨ سلام {user.first_name}!\n\nبه ربات آپلود خوش آمدی.\n\n📌 فایل خود را ارسال کن.\n\n📦 فایل‌ها روی سرور ذخیره می‌شوند." await update.message.reply_text(welcome, reply_markup=get_main_menu()) async def handle_callback(update: Update, context: ContextTypes.DEFAULT_TYPE): query = update.callback_query await query.answer() if query.data == "file": await query.edit_message_text("📤 فایل خود را ارسال کن\n\nحداکثر 1.8 GB", reply_markup=get_back_menu()) elif query.data == "link": await query.edit_message_text("🔗 لینک خود را ارسال کن\n\nحداکثر 3 GB", reply_markup=get_back_menu()) elif query.data == "my_files": user_id = query.from_user.id files = get_user_files(user_id) if not files: await query.edit_message_text("📂 هیچ فایلی وجود ندارد.", reply_markup=get_main_menu()) else: msg = "📁 فایل‌های شما\n\n" for i, f in enumerate(files[:10], 1): icon = get_icon(f[6]) # extension msg += f"{i}. {icon} {f[3][:30]}\n 💾 {f[7]}\n 🔗 {f[4][:50]}...\n\n" await query.edit_message_text(msg, reply_markup=get_main_menu()) elif query.data == "my_stats": user_id = query.from_user.id total, total_size = get_user_stats(user_id) await query.edit_message_text(f"📊 آمار شما\n\n📁 تعداد: {total}\n💾 فضا: {format_size(total_size)}", reply_markup=get_main_menu()) elif query.data == "help": await query.edit_message_text("📚 راهنما\n\n📤 فایل بفرست\n📦 محدودیت: 1.8 GB", reply_markup=get_main_menu()) elif query.data == "back": await query.edit_message_text("🔙 بازگشت به منو", reply_markup=get_main_menu()) async def handle_file(update: Update, context: ContextTypes.DEFAULT_TYPE): message = update.message user_id = message.from_user.id if message.document: file = message.document file_name = file.file_name file_size = file.file_size file_id = file.file_id elif message.video: file = message.video file_name = file.file_name or f"video_{int(time.time())}.mp4" file_size = file.file_size file_id = file.file_id elif message.audio: file = message.audio file_name = file.file_name or f"audio_{int(time.time())}.mp3" file_size = file.file_size file_id = file.file_id else: return if file_size > MAX_FILE_SIZE: await message.reply_text(f"❌ حجم فایل بیشتر از 1.8 GB است.\n\nحجم: {format_size(file_size)}") return if is_duplicate(user_id, file_name, file_size): await message.reply_text(f"⚠️ فایل تکراری!\n\n📄 {file_name}") return status = await message.reply_text("📥 در حال دریافت فایل...") new_file = await context.bot.get_file(file_id) temp_path = os.path.join(UPLOAD_DIR, f"{int(time.time())}_{file_name}") await new_file.download_to_drive(temp_path) download_url = f"https://{os.environ.get('HTTP_HOST', 'dl23.hostdl.com')}/uploads/{os.path.basename(temp_path)}" extension = get_extension(file_name) save_file(user_id, file_name, os.path.basename(temp_path), download_url, file_size, extension) icon = get_icon(extension) await status.edit_text( f"✅ آپلود کامل شد\n\n" f"{icon} {file_name}\n" f"📁 {extension}\n" f"💾 {format_size(file_size)}\n\n" f"🔗 لینک دانلود:\n{download_url}" ) # ============ راه‌اندازی ============ def main(): # برای رفع مشکل فیلتر، از polling استفاده می‌کنیم application = Application.builder().token(BOT_TOKEN).build() application.add_handler(CommandHandler("start", start)) application.add_handler(CallbackQueryHandler(handle_callback)) application.add_handler(MessageHandler(filters.Document.ALL, handle_file)) application.add_handler(MessageHandler(filters.VIDEO, handle_file)) application.add_handler(MessageHandler(filters.AUDIO, handle_file)) print("🚀 ربات راه‌اندازی شد...") print("📁 فایل‌ها در پوشه uploads ذخیره می‌شوند") print("💾 دیتابیس: bot.db") # استفاده از polling (نیاز به webhook نیست) application.run_polling(allowed_updates=Update.ALL_TYPES) if __name__ == "__main__": main()