import os import re import time import threading import logging from datetime import datetime, timedelta from dotenv import load_dotenv import telebot from telebot import types load_dotenv() logging.basicConfig( format="%(asctime)s - %(levelname)s - %(message)s", level=logging.INFO, ) logger = logging.getLogger(__name__) BOT_TOKEN = os.getenv("BOT_TOKEN") PLATFORM = os.getenv("PLATFORM", "telegram") if not BOT_TOKEN: raise ValueError("BOT_TOKEN is not set") bot = telebot.TeleBot(BOT_TOKEN, parse_mode=None) if PLATFORM == "bale": telebot.apihelper.API_URL = "https://tapi.bale.ai/bot{0}/{1}" LINKS = [ r'https?://[^\s]+|www\.[^\s]+|[a-zA-Z0-9-]+\.[a-zA-Z]{2,}(?:/\S*)?' ] GROUP_ID_PATTERNS = [ r"@[a-zA-Z0-9_]{5,}", # @username با حداقل ۵ کاراکتر r"id:\s*\d{5,}", # id: 123456789 r"آیدی:\s*\d{5,}", # آیدی: 123456789 r"شناسه:\s*\d{5,}", # شناسه: 123456789 ] muted_users = {} def bot_id(): try: return bot.get_me().id except Exception: return 0 def chat_admins(chat_id): try: return bot.get_chat_administrators(chat_id) or [] except Exception: return [] def is_admin(chat_id, user_id): for a in chat_admins(chat_id): if a.user.id == user_id: return True return False def is_bot_admin(chat_id): return is_admin(chat_id, bot_id()) def find_user_by_username(chat_id, username): name = username[1:].lower() if username.startswith("@") else username.lower() for a in chat_admins(chat_id): if a.user.username and a.user.username.lower() == name: return a.user.id return None def find_user(chat_id, text): if not text: return None if text.startswith("@"): return find_user_by_username(chat_id, text) try: return int(text) except ValueError: return None def extract_target(message): if message.reply_to_message: return message.reply_to_message.from_user.id args = message.text.strip().split() return find_user(message.chat.id, args[1]) if len(args) > 1 else None def extract_minutes(message): args = message.text.strip().split() idx = 1 if message.reply_to_message else 2 if len(args) > idx and args[idx].isdigit(): return int(args[idx]) return None def is_group(message): return message.chat.type in ["group", "supergroup"] def guard(message): if not is_group(message): bot.reply_to(message, "این دستور فقط در گروه قابل استفاده است") return False if not is_admin(message.chat.id, message.from_user.id): bot.reply_to(message, "⛔ فقط ادمین‌ها مجاز هستند") return False if not is_bot_admin(message.chat.id): bot.reply_to(message, "⛔ ربات باید ادمین گروه باشد") return False return True def is_promotional(message): text = message.text.lower() if message.text else "" # ۱. بررسی لینک‌ها if any(re.search(p, text) for p in LINKS): return True # ۲. بررسی آیدی گروه if any(re.search(p, text) for p in GROUP_ID_PATTERNS): return True # ۳. بررسی پیام فورواردی if message.forward_from or message.forward_from_chat: return True # ۴. بررسی فوروارد از کانال (با لینک) if message.forward_from_chat and message.forward_from_chat.username: return True return False # ── Start ────────────────────────────────────── @bot.message_handler(commands=["start"]) def start(message): if message.chat.type == "private": bot.send_message(message.chat.id, f"👋 سلام {message.from_user.first_name} عزیز\n\n" "🤖 ربات مدیریت گروه\n\n" "🎯 قابلیت‌ها:\n" "• خوش‌آمدگویی خودکار\n" "• حذف لینک‌های تبلیغاتی\n" "• اخراج کاربر\n" "• سکوت کاربر\n" "• لغو سکوت کاربر\n" "• پشتیبانی از ریپلای و @\n\n" "💡 ربات را به گروه اضافه و ادمین کنید" ) else: bot.send_message(message.chat.id, "✅ ربات مدیریت گروه فعال شد") # ── Welcome ──────────────────────────────────── @bot.message_handler(content_types=["new_chat_members"]) def welcome(message): for m in message.new_chat_members: if m.is_bot: continue name = m.first_name or "کاربر" title = message.chat.title or "گروه" bot.send_message(message.chat.id, f"👋 سلام {name} عزیز\nبه گروه {title} خوش آمدید\nلطفاً قوانین را رعایت کنید" ) # ── Anti‑spam + Mute ─────────────────────────── @bot.message_handler(func=lambda m: m.text and not m.text.startswith("/") and m.chat.type in ["group", "supergroup"]) def handle_msg(message): chat_id = message.chat.id user_id = message.from_user.id if not is_bot_admin(chat_id): return # بررسی سکوت group = str(chat_id) if group in muted_users and user_id in muted_users[group]: try: bot.delete_message(chat_id, message.message_id) except Exception: pass return # بررسی تبلیغات if is_promotional(message): try: bot.delete_message(chat_id, message.message_id) bot.send_message( chat_id, f"⚠️ پیام تبلیغاتی از {message.from_user.first_name} حذف شد" ) logger.info(f"🚫 تبلیغ حذف شد از {user_id} در گروه {chat_id}") except Exception as e: logger.error(f"❌ خطا در حذف تبلیغ: {e}") # ── Kick ─────────────────────────────────────── @bot.message_handler(commands=["kick"]) def kick(message): if not guard(message): return target = extract_target(message) if not target: bot.reply_to(message, "📝 روی پیام کاربر ریپلای کنید یا\n/kick @username") return if target == bot_id(): bot.reply_to(message, "❌ نمی‌توانید ربات را اخراج کنید") return if is_admin(message.chat.id, target): bot.reply_to(message, "❌ نمی‌توانید ادمین را اخراج کنید") return try: bot.ban_chat_member(message.chat.id, target) bot.reply_to(message, "✅ کاربر از گروه اخراج شد") logger.info(f"Kicked {target}") except Exception: try: bot.unban_chat_member(message.chat.id, target) bot.reply_to(message, "✅ کاربر از گروه اخراج شد") except Exception: bot.reply_to(message, "⚠️ خطا در اخراج") # ── Mute ─────────────────────────────────────── @bot.message_handler(commands=["mute"]) def mute(message): if not guard(message): return target = extract_target(message) minutes = extract_minutes(message) if not target: bot.reply_to(message, "📝 روی پیام کاربر ریپلای کنید یا\n/mute @username [دقیقه]") return if target == bot_id(): bot.reply_to(message, "❌ نمی‌توانید ربات را سکوت کنید") return if is_admin(message.chat.id, target): bot.reply_to(message, "❌ نمی‌توانید ادمین را سکوت کنید") return group = str(message.chat.id) muted_users.setdefault(group, {})[target] = { "at": datetime.now().isoformat(), "until": (datetime.now() + timedelta(minutes=minutes)).isoformat() if minutes else None, } label = f"برای {minutes} دقیقه" if minutes else "دائم" bot.reply_to(message, f"✅ کاربر {label} سکوت شد") if minutes: def auto_unmute(): time.sleep(minutes * 60) g = muted_users.get(group, {}) if target in g: del g[target] if not g: muted_users.pop(group, None) threading.Thread(target=auto_unmute, daemon=True).start() # ── Unmute ───────────────────────────────────── @bot.message_handler(commands=["unmute"]) def unmute(message): if not guard(message): return target = extract_target(message) if not target: bot.reply_to(message, "📝 روی پیام کاربر ریپلای کنید یا\n/unmute @username") return group = str(message.chat.id) if group in muted_users and target in muted_users[group]: del muted_users[group][target] if not muted_users[group]: del muted_users[group] bot.reply_to(message, "✅ سکوت کاربر لغو شد") else: bot.reply_to(message, "ℹ️ این کاربر سکوت نیست") # ── Mutelist ─────────────────────────────────── @bot.message_handler(commands=["mutelist"]) def mutelist(message): if not guard(message): return group = str(message.chat.id) if group not in muted_users or not muted_users[group]: bot.reply_to(message, "📋 هیچ کاربری سکوت نیست") return lines = [] for uid, d in muted_users[group].items(): until = d.get("until") if until: try: rem = int((datetime.fromisoformat(until) - datetime.now()).total_seconds() / 60) lines.append(f"• {uid} — {rem} دقیقه مانده") except Exception: lines.append(f"• {uid}") else: lines.append(f"• {uid} — دائم") bot.reply_to(message, f"📋 کاربران سکوت‌شده:\n" + "\n".join(lines)) # ── Help ─────────────────────────────────────── @bot.message_handler(commands=["help"]) def help_cmd(message): bot.send_message(message.chat.id, "🤖 راهنمای ربات مدیریت گروه\n\n" "🔹 /start — فعال‌سازی\n" "🔹 /kick — اخراج کاربر\n" "🔹 /mute — سکوت کاربر\n" "🔹 /unmute — لغو سکوت\n" "🔹 /mutelist — لیست سکوت‌شده‌ها\n" "🔹 /help — راهنما\n\n" "📝 با ریپلای هم کار می‌کنند\n\n" "⚡ خودکار:\n" "• خوش‌آمدگویی\n" "• حذف لینک تبلیغاتی\n" "• حذف پیام کاربران سکوت‌شده\n\n" "⚠️ ربات باید ادمین باشد" ) # ── Run ──────────────────────────────────────── if __name__ == "__main__": print("=" * 40) print("🤖 Group Manager Bot") print(f"📡 {PLATFORM}") print("=" * 40) bot.remove_webhook() bot.infinity_polling()