#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ Persian Telegram Group Manager Bot - MVP Features: - Telegram webhook via Flask - No web panel; Telegram inline/text management - Sponsor / forced membership system - Creator, owner, admin roles - Persian text commands - Close/open group - Ban/unban/promote/demote/mute/unmute - Echo - Simple entertainment - Basic text/link/file moderation """ import os import re import json import time import random import sqlite3 import logging import sys from datetime import datetime import requests from flask import Flask, request, abort from dotenv import load_dotenv # ============================================================ # CONFIG # ============================================================ BASE_DIR = os.path.dirname(os.path.abspath(__file__)) load_dotenv(os.path.join(BASE_DIR, ".env")) DATA_DIR = os.getenv("DATA_DIR", os.path.join(BASE_DIR, "data")) LOG_DIR = os.getenv("LOG_DIR", os.path.join(BASE_DIR, "logs")) TMP_DIR = os.getenv("TMP_DIR", os.path.join(BASE_DIR, "tmp")) for d in (DATA_DIR, LOG_DIR, TMP_DIR): os.makedirs(d, exist_ok=True) DB_PATH = os.getenv("DATABASE_PATH", os.path.join(DATA_DIR, "bot.db")) BOT_TOKEN = os.getenv("BOT_TOKEN", "").strip() WEBHOOK_SECRET = os.getenv("WEBHOOK_SECRET", "change_me_secret").strip() PUBLIC_BASE_URL = os.getenv("PUBLIC_BASE_URL", "").strip().rstrip("/") WEBHOOK_PATH = os.getenv("WEBHOOK_PATH", "/tgbot/webhook").strip() if not WEBHOOK_PATH.startswith("/"): WEBHOOK_PATH = "/" + WEBHOOK_PATH REQUEST_TIMEOUT = int(os.getenv("REQUEST_TIMEOUT", "15")) DELETE_SCORE = int(os.getenv("DELETE_SCORE", "80")) LINK_LOCK = os.getenv("LINK_LOCK", "1").strip().lower() in ("1", "true", "on", "yes") BAD_WORDS_ENV = os.getenv("BAD_WORDS", "") BAD_WORDS = [w.strip() for w in BAD_WORDS_ENV.split(",") if w.strip()] AD_WORDS_ENV = os.getenv("AD_WORDS", None) if AD_WORDS_ENV is None: AD_WORDS = ["عضویت", "چنل", "کانال", "تبلیغ", "فالو", "بازدید"] else: AD_WORDS = [w.strip() for w in AD_WORDS_ENV.split(",") if w.strip()] OWNER_IDS = set() for part in os.getenv("OWNER_IDS", "").split(","): part = part.strip() if part.isdigit(): OWNER_IDS.add(int(part)) logging.basicConfig( level=logging.INFO, format="%(asctime)s %(levelname)s %(name)s %(message)s", handlers=[ logging.FileHandler(os.path.join(LOG_DIR, "bot.log"), encoding="utf-8"), logging.StreamHandler() ] ) logger = logging.getLogger("tgbot") application = Flask(__name__) # ============================================================ # CONSTANTS # ============================================================ ALL_PERMS = [ "ban", "unban", "promote", "demote", "mute", "unmute", "echo", "open_close", "delete", "warn", "settings", "manage_admins", "entertainment" ] DEFAULT_CHAT_PERMISSIONS = { "can_send_messages": True, "can_send_audios": True, "can_send_documents": True, "can_send_photos": True, "can_send_videos": True, "can_send_video_notes": True, "can_send_voice_notes": True, "can_send_polls": True, "can_send_other_messages": True, "can_add_web_page_previews": True, "can_change_info": False, "can_invite_users": True, "can_pin_messages": False, "can_manage_topics": False } MUTE_PERMISSIONS = { "can_send_messages": False, "can_send_audios": False, "can_send_documents": False, "can_send_photos": False, "can_send_videos": False, "can_send_video_notes": False, "can_send_voice_notes": False, "can_send_polls": False, "can_send_other_messages": False, "can_add_web_page_previews": False } UNMUTE_PERMISSIONS = { "can_send_messages": True, "can_send_audios": True, "can_send_documents": True, "can_send_photos": True, "can_send_videos": True, "can_send_video_notes": True, "can_send_voice_notes": True, "can_send_polls": True, "can_send_other_messages": True, "can_add_web_page_previews": True } DANGEROUS_EXTS = ( ".exe", ".scr", ".bat", ".cmd", ".ps1", ".vbs", ".js", ".jar", ".apk", ".msi", ".dll", ".com", ".pif" ) LINK_RE = re.compile(r"(https?://|t\.me/|www\.|@[A-Za-z0-9_]{3,})", re.I) ADMIN_CACHE = {} ADMIN_CACHE_TTL = 60 START_CMDS = {"شروع", "استارت", "/start", "start", "/شروع"} HELP_CMDS = {"راهنما", "کمک", "/help", "help", "/راهنما"} MENU_CMDS = {"منو", "/منو", "/menu", "menu"} PROFILE_CMDS = {"پروفایل", "پروفایل من", "ایدی", "/id", "id", "/ایدی"} JOKE_CMDS = {"جوک", "جوک بگو", "لطیفه", "/جوک"} FACT_CMDS = {"دانستنی", "دانستنی بگو", "/دانستنی"} DICE_CMDS = {"تاس", "تاس بنداز", "/تاس"} COIN_CMDS = {"شیر یا خط", "پول بنداز", "شیر خط", "/شیرخط"} RPS_CMDS = {"سنگ کاغذ قیچی", "بازی سنگ کاغذ قیچی", "/سنگ"} NUMBER_CMDS = {"عدد شانسی", "شانس", "/عدد"} CREATOR_PANEL_CMDS = {"پنل سازنده", "/پنل_سازنده", "سازنده"} OWNER_PANEL_CMDS = {"پنل مالک", "/پنل_مالک", "مالک"} ADMIN_PANEL_CMDS = {"پنل ادمین", "/پنل_ادمین", "ادمین"} CLOSE_CMDS = {"بستن گروه", "بستن چت", "قفل کردن گروه", "گروه رو ببند", "تعطیل کردن گروه"} OPEN_CMDS = {"باز کردن گروه", "باز کردن چت", "گروه رو باز کن", "آزاد کردن گروه", "باز کن گروه"} BAN_CMDS = {"بن", "بلاک", "مسدود کن", "بنداز بیرون"} UNBAN_CMDS = {"آنبان", "رفع بن", "از بن در بیار", "آن بن"} UNMUTE_CMDS = {"رفع سکوت", "باز کردن سکوت", "سکوت را بردار"} PROMOTE_CMDS = {"ادمین کن", "ادمین گروه کن", "پرموت"} DEMOTE_CMDS = {"ادمین بگیر", "ادمین حذف کن", "دموت"} DELETE_CMDS = {"حذف پیام", "پاک کن", "حذف کن"} WARN_CMDS = {"اخطار", "ورن"} PERM_NAMES = { "بن": "ban", "آنبان": "unban", "رفع بن": "unban", "سکوت": "mute", "رفع سکوت": "unmute", "اکو": "echo", "باز و بسته": "open_close", "بستن": "open_close", "باز کردن": "open_close", "حذف": "delete", "پرموت": "promote", "ادمین کردن": "promote", "ادمین": "promote", "دموت": "demote", "ادمین گرفتن": "demote", "تنظیمات": "settings", "مدیریت ادمین": "manage_admins", "سرگرمی": "entertainment", "اخطار": "warn" } SPONSOR_DEFAULT_MESSAGE = ( "⚠️ برای استفاده از ربات باید ابتدا عضو کانال‌ها/گروه‌های زیر شوید.\n" "پس از عضویت روی دکمه «بررسی عضویت» بزنید." ) # ============================================================ # DATABASE # ============================================================ def db(): conn = sqlite3.connect(DB_PATH) conn.row_factory = sqlite3.Row return conn def query(sql, args=(), one=False): conn = db() try: cur = conn.execute(sql, args) rows = cur.fetchall() return rows[0] if rows and one else rows finally: conn.close() def execute(sql, args=()): conn = db() try: conn.execute(sql, args) conn.commit() finally: conn.close() def execute_return(sql, args=()): conn = db() try: cur = conn.execute(sql, args) conn.commit() return cur.lastrowid finally: conn.close() def init_db(): execute( """ CREATE TABLE IF NOT EXISTS users ( user_id INTEGER PRIMARY KEY, username TEXT, first_name TEXT, last_name TEXT, created_at TEXT, last_seen TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS chats ( chat_id INTEGER PRIMARY KEY, title TEXT, type TEXT, username TEXT, owner_id INTEGER, active INTEGER DEFAULT 1, is_closed INTEGER DEFAULT 0, saved_permissions TEXT, created_at TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS chat_admins ( chat_id INTEGER, user_id INTEGER, role TEXT DEFAULT 'admin', permissions TEXT DEFAULT '{}', active INTEGER DEFAULT 1, PRIMARY KEY(chat_id, user_id) ) """ ) execute( """ CREATE TABLE IF NOT EXISTS sponsors ( chat_id INTEGER PRIMARY KEY, title TEXT, type TEXT, username TEXT, invite_link TEXT, active INTEGER DEFAULT 1, required INTEGER DEFAULT 1, added_by INTEGER, created_at TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS sponsor_settings ( key TEXT PRIMARY KEY, value TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS pending_verifications ( user_id INTEGER, chat_id INTEGER, command TEXT, payload TEXT, last_check_at INTEGER DEFAULT 0, PRIMARY KEY(user_id, chat_id) ) """ ) execute( """ CREATE TABLE IF NOT EXISTS user_states ( user_id INTEGER PRIMARY KEY, chat_id INTEGER, state TEXT, data TEXT, expires_at INTEGER ) """ ) execute( """ CREATE TABLE IF NOT EXISTS entertainment ( id INTEGER PRIMARY KEY AUTOINCREMENT, category TEXT, text TEXT, file_id TEXT, file_type TEXT, active INTEGER DEFAULT 1, added_by INTEGER, created_at TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS echo_logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER, admin_id INTEGER, text TEXT, created_at TEXT ) """ ) execute( """ CREATE TABLE IF NOT EXISTS logs ( id INTEGER PRIMARY KEY AUTOINCREMENT, chat_id INTEGER, user_id INTEGER, action TEXT, reason TEXT, created_at TEXT ) """ ) # default settings defaults = [ ("enabled", "1"), ("message", SPONSOR_DEFAULT_MESSAGE), ("bypass_creator", "1"), ("bypass_owners", "1"), ("bypass_admins", "1") ] for k, v in defaults: execute("INSERT OR IGNORE INTO sponsor_settings(key, value) VALUES (?, ?)", (k, v)) # default entertainment row = query("SELECT COUNT(*) AS cnt FROM entertainment", one=True) if row and row["cnt"] == 0: items = [ ("joke", "چرا برنامه‌نویس‌ها عینکی هستن؟ چون نمی‌تونن این رو ببینن! 😄"), ("joke", "به یه سرور گفتن چرا همیشه مشغولی؟ گفت: چون همه از من درخواست دارن!"), ("fact", "مغز انسان حدود ۲۰ درصد انرژی بدن را مصرف می‌کند."), ("fact", "اولین پیامک دنیا در سال ۱۹۹۲ ارسال شد.") ] for category, text in items: execute( "INSERT INTO entertainment(category, text, active, created_at) VALUES (?, ?, 1, ?)", (category, text, now_iso()) ) logger.info("Database initialized at %s", DB_PATH) # ============================================================ # UTILS # ============================================================ def now_iso(): return datetime.utcnow().isoformat() def now_ts(): return int(time.time()) def parse_ids(s): ids = set() for part in (s or "").split(","): part = part.strip() if part.isdigit(): ids.add(int(part)) return ids def is_creator(user_id): return bool(user_id) and user_id in OWNER_IDS def normalize_fa(text): if not text: return "" text = str(text).lower() replacements = { "ي": "ی", "ئ": "ی", "ؤ": "و", "إ": "ا", "أ": "ا", "آ": "ا", "ة": "ه", "ك": "ک", "٠": "0", "١": "1", "٢": "2", "٣": "3", "٤": "4", "٥": "5", "٦": "6", "٧": "7", "٨": "8", "٩": "9", "۰": "0", "۱": "1", "۲": "2", "۳": "3", "۴": "4", "۵": "5", "۶": "6", "۷": "7", "۸": "8", "۹": "9", } for k, v in replacements.items(): text = text.replace(k, v) text = text.replace("\u200c", " ") text = text.replace("\u200e", " ") text = text.replace("\u200f", " ") text = re.sub(r"\s+", " ", text).strip() return text def log_action(chat_id, user_id, action, reason=""): try: execute( "INSERT INTO logs(chat_id, user_id, action, reason, created_at) VALUES (?, ?, ?, ?, ?)", (chat_id, user_id, action, reason, now_iso()) ) except Exception as e: logger.exception("log_action error: %s", e) # ============================================================ # TELEGRAM API # ============================================================ def tg(method, **params): if not BOT_TOKEN: logger.error("BOT_TOKEN is not set") return None url = f"https://api.telegram.org/bot{BOT_TOKEN}/{method}" try: r = requests.post(url, json=params, timeout=REQUEST_TIMEOUT) data = r.json() if not data.get("ok"): logger.warning("Telegram API error %s: %s", method, data.get("description")) return None return data.get("result") except Exception as e: logger.exception("Telegram API exception %s: %s", method, e) return None def send_message(chat_id, text, reply_markup=None, reply_to_message_id=None): params = { "chat_id": chat_id, "text": text } if reply_markup: params["reply_markup"] = reply_markup if reply_to_message_id: params["reply_to_message_id"] = reply_to_message_id return tg("sendMessage", **params) def edit_message_text(chat_id, message_id, text, reply_markup=None): params = { "chat_id": chat_id, "message_id": message_id, "text": text } if reply_markup: params["reply_markup"] = reply_markup return tg("editMessageText", **params) def answer_callback(callback_id, text="", show_alert=False): return tg("answerCallbackQuery", callback_query_id=callback_id, text=text, show_alert=show_alert) def delete_message(chat_id, message_id): return tg("deleteMessage", chat_id=chat_id, message_id=message_id) def inline_keyboard(rows): return {"inline_keyboard": rows} def url_button(text, url): return {"text": text, "url": url} def callback_button(text, data): return {"text": text, "callback_data": data} # ============================================================ # USERS / CHATS / ROLES # ============================================================ def ensure_user(user): if not user or not user.get("id"): return user_id = user.get("id") username = user.get("username") first_name = user.get("first_name") last_name = user.get("last_name") existing = query("SELECT user_id FROM users WHERE user_id = ?", (user_id,), one=True) if existing: execute( "UPDATE users SET username=?, first_name=?, last_name=?, last_seen=? WHERE user_id=?", (username, first_name, last_name, now_iso(), user_id) ) else: execute( "INSERT INTO users(user_id, username, first_name, last_name, created_at, last_seen) VALUES (?, ?, ?, ?, ?, ?)", (user_id, username, first_name, last_name, now_iso(), now_iso()) ) def get_user_id_by_username(username): if not username: return None username = username.lstrip("@").lower() row = query("SELECT user_id FROM users WHERE LOWER(username)=? LIMIT 1", (username,), one=True) return row["user_id"] if row else None def get_chat_admins(chat_id): now = now_ts() cached = ADMIN_CACHE.get(chat_id) if cached and now - cached[0] < ADMIN_CACHE_TTL: return cached[1] admins = tg("getChatAdministrators", chat_id=chat_id) or [] ADMIN_CACHE[chat_id] = (now, admins) return admins def refresh_chat_owner(chat_id): admins = get_chat_admins(chat_id) for a in admins: if a.get("status") == "creator": owner_id = (a.get("user") or {}).get("id") if owner_id: execute("UPDATE chats SET owner_id=? WHERE chat_id=?", (owner_id, chat_id)) return owner_id return None def ensure_chat(chat): if not chat or not chat.get("id"): return chat_id = chat.get("id") title = chat.get("title") ctype = chat.get("type") username = chat.get("username") existing = query("SELECT chat_id FROM chats WHERE chat_id=?", (chat_id,), one=True) if not existing: execute( "INSERT INTO chats(chat_id, title, type, username, owner_id, active, created_at) VALUES (?, ?, ?, ?, NULL, 1, ?)", (chat_id, title, ctype, username, now_iso()) ) refresh_chat_owner(chat_id) else: execute( "UPDATE chats SET title=?, type=?, username=? WHERE chat_id=?", (title, ctype, username, chat_id) ) def get_owner_id(chat_id): row = query("SELECT owner_id FROM chats WHERE chat_id=?", (chat_id,), one=True) if row and row["owner_id"]: return row["owner_id"] return refresh_chat_owner(chat_id) def is_telegram_admin(user_id, chat_id): if not user_id: return False admins = get_chat_admins(chat_id) for a in admins: if (a.get("user") or {}).get("id") == user_id: return True return False def get_custom_admin(chat_id, user_id): return query( "SELECT permissions FROM chat_admins WHERE chat_id=? AND user_id=? AND active=1", (chat_id, user_id), one=True ) def can_any_admin(user_id, chat_id): if not user_id: return False if is_creator(user_id): return True owner_id = get_owner_id(chat_id) if owner_id == user_id: return True row = get_custom_admin(chat_id, user_id) if row: try: perms = json.loads(row["permissions"] or "{}") return any(bool(v) for v in perms.values()) except Exception: return False return False def can(user_id, chat_id, perm): if not user_id: return False if is_creator(user_id): return True owner_id = get_owner_id(chat_id) if owner_id == user_id: return True row = get_custom_admin(chat_id, user_id) if row: try: perms = json.loads(row["permissions"] or "{}") return bool(perms.get(perm)) except Exception: return False return False # ============================================================ # SPONSOR SYSTEM # ============================================================ def get_setting(key, default=None): row = query("SELECT value FROM sponsor_settings WHERE key=?", (key,), one=True) return row["value"] if row else default def set_setting(key, value): execute("INSERT OR REPLACE INTO sponsor_settings(key, value) VALUES (?, ?)", (key, str(value))) def setting_bool(key, default=False): val = get_setting(key) if val is None: return default return str(val).strip().lower() in ("1", "true", "on", "yes", "بله", "فعال") def get_active_sponsors(): return query("SELECT * FROM sponsors WHERE active=1 AND required=1 ORDER BY created_at ASC") def check_membership(user_id, sponsor_chat_id): res = tg("getChatMember", chat_id=sponsor_chat_id, user_id=user_id) if not res: return False status = res.get("status") if status in ("creator", "administrator", "member"): return True if status == "restricted": return bool(res.get("is_member")) return False def missing_sponsors(user_id): sponsors = get_active_sponsors() missing = [] for s in sponsors: if not check_membership(user_id, s["chat_id"]): missing.append(s) return missing def sponsor_keyboard(): rows = [] sponsors = get_active_sponsors() for s in sponsors: title = s["title"] or str(s["chat_id"]) link = s["invite_link"] if not link and s["username"]: link = f"https://t.me/{s['username']}" if link: rows.append([url_button(f"عضویت در {title}", link)]) rows.append([callback_button("✅ بررسی عضویت", "sp:chk")]) return inline_keyboard(rows) def save_pending(user_id, chat_id, command, payload=None): execute( "INSERT OR REPLACE INTO pending_verifications(user_id, chat_id, command, payload, last_check_at) VALUES (?, ?, ?, ?, ?)", (user_id, chat_id, command, payload, now_ts()) ) def send_sponsor_required(message, missing, command=""): chat_id = message.get("chat", {}).get("id") user_id = message.get("from", {}).get("id") message_id = message.get("message_id") base = get_setting("message", SPONSOR_DEFAULT_MESSAGE) titles = "\n".join([f"- {s['title'] or s['chat_id']}" for s in missing]) text = f"{base}\n\n{titles}\n\nپس از عضویت روی دکمه «بررسی عضویت» بزنید." send_message(chat_id, text, reply_markup=sponsor_keyboard(), reply_to_message_id=message_id) save_pending(user_id, chat_id, command, json.dumps({"message_id": message_id}, ensure_ascii=False)) def sponsor_gate(message, command=""): if not setting_bool("enabled", True): return True user_id = (message.get("from") or {}).get("id") chat_id = (message.get("chat") or {}).get("id") if not user_id or not chat_id: return True if setting_bool("bypass_creator", True) and is_creator(user_id): return True if setting_bool("bypass_owners", True) and get_owner_id(chat_id) == user_id: return True if setting_bool("bypass_admins", True): if can_any_admin(user_id, chat_id) or is_telegram_admin(user_id, chat_id): return True missing = missing_sponsors(user_id) if not missing: return True send_sponsor_required(message, missing, command) return False def add_sponsor_from_text(text, user_id, message): text = text.strip() chat_id_input = None if text.startswith("@"): chat_id_input = text.strip() else: m = re.search(r"t\.me/([A-Za-z0-9_]+)", text) if m: chat_id_input = "@" + m.group(1) elif re.fullmatch(r"-?\d+", text): chat_id_input = int(text) if not chat_id_input: send_message(message["chat"]["id"], "فرمت نامعتبر است. لطفاً @username، لینک عمومی یا آیدی عددی کانال/گروه را بفرستید.") return False chat = tg("getChat", chat_id=chat_id_input) if not chat: send_message(message["chat"]["id"], "چت پیدا نشد. مطمئن شوید ربات داخل آن کانال/گروه ادمین است یا چت عمومی است.") return False real_id = chat.get("id") title = chat.get("title") or chat.get("username") or str(real_id) username = chat.get("username") invite_link = "" if not username: link = tg("exportChatInviteLink", chat_id=real_id) if isinstance(link, str): invite_link = link execute( """ INSERT OR REPLACE INTO sponsors(chat_id, title, type, username, invite_link, active, required, added_by, created_at) VALUES (?, ?, ?, ?, ?, 1, 1, ?, ?) """, (real_id, title, chat.get("type"), username, invite_link, user_id, now_iso()) ) send_message(message["chat"]["id"], f"✅ اسپانسر اضافه شد: {title}") log_action(message["chat"]["id"], user_id, "add_sponsor", str(real_id)) return True # ============================================================ # STATES # ============================================================ def set_state(user_id, chat_id, state, data=None, ttl=300): execute( "INSERT OR REPLACE INTO user_states(user_id, chat_id, state, data, expires_at) VALUES (?, ?, ?, ?, ?)", (user_id, chat_id, state, json.dumps(data or {}, ensure_ascii=False), now_ts() + ttl) ) def get_state(user_id): return query( "SELECT * FROM user_states WHERE user_id=? AND expires_at > ?", (user_id, now_ts()), one=True ) def clear_state(user_id): execute("DELETE FROM user_states WHERE user_id=?", (user_id,)) def process_state(message): user_id = (message.get("from") or {}).get("id") text = message.get("text") or "" if not user_id or not text: return False state_row = get_state(user_id) if not state_row: return False state = state_row["state"] chat_id = state_row["chat_id"] or message.get("chat", {}).get("id") if state == "add_sponsor": if not is_creator(user_id): clear_state(user_id) return True add_sponsor_from_text(text, user_id, message) clear_state(user_id) return True if state == "add_joke": if not is_creator(user_id): clear_state(user_id) return True execute( "INSERT INTO entertainment(category, text, active, added_by, created_at) VALUES ('joke', ?, 1, ?, ?)", (text.strip(), user_id, now_iso()) ) send_message(chat_id, "✅ جوک اضافه شد.") clear_state(user_id) return True if state == "add_fact": if not is_creator(user_id): clear_state(user_id) return True execute( "INSERT INTO entertainment(category, text, active, added_by, created_at) VALUES ('fact', ?, 1, ?, ?)", (text.strip(), user_id, now_iso()) ) send_message(chat_id, "✅ دانستنی اضافه شد.") clear_state(user_id) return True return False # ============================================================ # GROUP ACTIONS # ============================================================ def close_group(chat_id): chat = tg("getChat", chat_id=chat_id) or {} saved = chat.get("permissions") or {} execute( "UPDATE chats SET is_closed=1, saved_permissions=? WHERE chat_id=?", (json.dumps(saved, ensure_ascii=False), chat_id) ) perms = {k: False for k in DEFAULT_CHAT_PERMISSIONS.keys()} res = tg("setChatPermissions", chat_id=chat_id, permissions=perms) return res is not None def open_group(chat_id): row = query("SELECT saved_permissions FROM chats WHERE chat_id=?", (chat_id,), one=True) saved = {} if row and row["saved_permissions"]: try: saved = json.loads(row["saved_permissions"]) except Exception: saved = {} perms = {} for k, default_val in DEFAULT_CHAT_PERMISSIONS.items(): perms[k] = bool(saved.get(k, default_val)) res = tg("setChatPermissions", chat_id=chat_id, permissions=perms) execute("UPDATE chats SET is_closed=0, saved_permissions=NULL WHERE chat_id=?", (chat_id,)) return res is not None def ban_user(chat_id, user_id): return tg("banChatMember", chat_id=chat_id, user_id=user_id) is not None def unban_user(chat_id, user_id): return tg("unbanChatMember", chat_id=chat_id, user_id=user_id, only_if_banned=True) is not None def mute_user(chat_id, user_id, until_date=0): params = { "chat_id": chat_id, "user_id": user_id, "permissions": MUTE_PERMISSIONS } if until_date and until_date > 0: params["until_date"] = until_date return tg("restrictChatMember", **params) is not None def unmute_user(chat_id, user_id): return tg("restrictChatMember", chat_id=chat_id, user_id=user_id, permissions=UNMUTE_PERMISSIONS) is not None def promote_user(chat_id, user_id): rights = { "can_delete_messages": True, "can_invite_users": True, "can_restrict_members": True, "can_pin_messages": True } return tg("promoteChatMember", chat_id=chat_id, user_id=user_id, **rights) is not None def demote_user(chat_id, user_id): rights = { "can_delete_messages": False, "can_invite_users": False, "can_restrict_members": False, "can_pin_messages": False, "can_change_info": False, "can_promote_members": False, "can_manage_video_chats": False, "can_manage_topics": False } return tg("promoteChatMember", chat_id=chat_id, user_id=user_id, **rights) is not None # ============================================================ # COMMAND HELPERS # ============================================================ def extract_target(message, args_text=""): reply = message.get("reply_to_message") if reply and reply.get("from") and reply["from"].get("id"): return reply["from"].get("id"), reply["from"].get("username") if args_text: m = re.search(r"@([A-Za-z0-9_]+)", args_text) if m: username = m.group(1) uid = get_user_id_by_username(username) return uid, username m = re.search(r"\b(\d{5,})\b", args_text) if m: return int(m.group(1)), None return None, None def parse_duration(text): if not text: return 3600 norm = normalize_fa(text) if "دائم" in norm: return 0 units = { "ثانیه": 1, "دقیقه": 60, "ساعت": 3600, "روز": 86400, "هفته": 604800, "ماه": 2592000 } total = 0 for m in re.finditer(r"(\d+)\s*(ثانیه|دقیقه|ساعت|روز|هفته|ماه)", norm): total += int(m.group(1)) * units[m.group(2)] if total == 0: m = re.search(r"(\d+)", norm) if m: total = int(m.group(1)) * 60 return total if total > 0 else 3600 def get_random_entertainment(category): row = query( "SELECT text FROM entertainment WHERE category=? AND active=1 ORDER BY RANDOM() LIMIT 1", (category,), one=True ) if row and row["text"]: return row["text"] if category == "joke": return "فعلاً جوکی ثبت نشده است. 😅" if category == "fact": return "فعلاً دانستنی‌ای ثبت نشده است." return "محتوایی یافت نشد." # ============================================================ # PERMISSION MANAGEMENT # ============================================================ def add_bot_admin(chat_id, target_id): default_perms = {"delete": True, "warn": True} execute( "INSERT OR REPLACE INTO chat_admins(chat_id, user_id, role, permissions, active) VALUES (?, ?, 'admin', ?, 1)", (chat_id, target_id, json.dumps(default_perms, ensure_ascii=False)) ) def remove_bot_admin(chat_id, target_id): execute("DELETE FROM chat_admins WHERE chat_id=? AND user_id=?", (chat_id, target_id)) def set_admin_permission(chat_id, target_id, perm, enabled): row = query("SELECT permissions FROM chat_admins WHERE chat_id=? AND user_id=?", (chat_id, target_id), one=True) if not row: return False try: perms = json.loads(row["permissions"] or "{}") except Exception: perms = {} perms[perm] = bool(enabled) execute( "UPDATE chat_admins SET permissions=? WHERE chat_id=? AND user_id=?", (json.dumps(perms, ensure_ascii=False), chat_id, target_id) ) return True # ============================================================ # MODERATION # ============================================================ def moderate_message(message): if not message: return from_user = message.get("from") or {} chat = message.get("chat") or {} user_id = from_user.get("id") chat_id = chat.get("id") message_id = message.get("message_id") if not user_id or not chat_id: return if is_creator(user_id): return owner_id = get_owner_id(chat_id) if owner_id == user_id: return if can_any_admin(user_id, chat_id) or is_telegram_admin(user_id, chat_id): return text = message.get("text") or message.get("caption") or "" norm = normalize_fa(text) score = 0 reasons = [] for w in BAD_WORDS: if w and w in norm: score += 60 reasons.append("کلمه ممنوع") for w in AD_WORDS: if w and w in norm: score += 25 reasons.append("کلمه تبلیغاتی") if LINK_LOCK and LINK_RE.search(text): score += 55 reasons.append("لینک/آیدی") doc = message.get("document") if doc: fname = (doc.get("file_name") or "").lower() if any(fname.endswith(ext) for ext in DANGEROUS_EXTS): score += 90 reasons.append("فایل خطرناک") if score >= DELETE_SCORE: delete_message(chat_id, message_id) log_action(chat_id, user_id, "moderate_delete", ",".join(reasons)) # ============================================================ # COMMAND HANDLERS # ============================================================ def cmd_start(message): chat_id = message["chat"]["id"] user = message.get("from") or {} name = user.get("first_name") or "کاربر" text = ( f"سلام {name} 👋\n" "من ربات مدیریت گروه هستم.\n\n" "دستورات عمومی:\n" "جوک | دانستنی | تاس | شیر یا خط | سنگ کاغذ قیچی | عدد شانسی | پروفایل من | راهنما\n" ) send_message(chat_id, text, reply_to_message_id=message.get("message_id")) def cmd_help(message): chat_id = message["chat"]["id"] text = ( "📖 راهنمای سریع:\n" "عمومی: جوک، دانستنی، تاس، شیر یا خط، سنگ کاغذ قیچی، عدد شانسی، پروفایل من، منو\n" "ادمین: بستن گروه، باز کردن گروه، بن، آنبان، سکوت 1 ساعت، رفع سکوت، ادمین کن، ادمین بگیر، اکو سلام، حذف پیام، اخطار، دسترسی بن @username روشن/خاموش، پنل ادمین، پنل مالک، پنل سازنده" ) send_message(chat_id, text, reply_to_message_id=message.get("message_id")) def cmd_menu(message): chat_id = message["chat"]["id"] keyboard = inline_keyboard([ [callback_button("🎭 جوک", "ent:joke"), callback_button("🧩 دانستنی", "ent:fact")], [callback_button("🎲 تاس", "ent:dice"), callback_button("🪙 شیر یا خط", "ent:coin")], [callback_button("📊 پروفایل", "usr:profile")] ]) send_message(chat_id, "منوی سرگرمی و کاربری:", reply_markup=keyboard, reply_to_message_id=message.get("message_id")) def cmd_profile(message): chat_id = message["chat"]["id"] user = message.get("from") or {} user_id = user.get("id") username = user.get("username") or "—" first_name = user.get("first_name") or "—" last_name = user.get("last_name") or "" text = ( "👤 پروفایل شما:\n" f"آیدی: {user_id}\n" f"یوزرنیم: @{username}\n" f"نام: {first_name} {last_name}".strip() ) send_message(chat_id, text, reply_to_message_id=message.get("message_id")) def cmd_joke(message): send_message(message["chat"]["id"], get_random_entertainment("joke"), reply_to_message_id=message.get("message_id")) def cmd_fact(message): send_message(message["chat"]["id"], get_random_entertainment("fact"), reply_to_message_id=message.get("message_id")) def cmd_dice(message): n = random.randint(1, 6) send_message(message["chat"]["id"], f"🎲 تاس آمد: {n}", reply_to_message_id=message.get("message_id")) def cmd_coin(message): result = random.choice(["شیر", "خط"]) send_message(message["chat"]["id"], f"🪙 نتیجه: {result}", reply_to_message_id=message.get("message_id")) def cmd_rps(message): norm = normalize_fa(message.get("text") or "") choices = ["سنگ", "کاغذ", "قیچی"] user_choice = None for c in choices: if c in norm: user_choice = c break if not user_choice: send_message(message["chat"]["id"], "یکی از این‌ها را بفرست: سنگ کاغذ قیچی", reply_to_message_id=message.get("message_id")) return bot_choice = random.choice(choices) if user_choice == bot_choice: result = "مساوی شد. 🤝" elif ( (user_choice == "سنگ" and bot_choice == "قیچی") or (user_choice == "کاغذ" and bot_choice == "سنگ") or (user_choice == "قیچی" and bot_choice == "کاغذ") ): result = "شما بردید. 🎉" else: result = "ربات برد. 🤖" send_message( message["chat"]["id"], f"انتخاب شما: {user_choice}\nانتخاب ربات: {bot_choice}\n{result}", reply_to_message_id=message.get("message_id") ) def cmd_number(message): n = random.randint(1, 100) send_message(message["chat"]["id"], f"🔢 عدد شانسی: {n}", reply_to_message_id=message.get("message_id")) def send_creator_panel(message): chat_id = message["chat"]["id"] keyboard = inline_keyboard([ [callback_button("➕ افزودن اسپانسر", "sp:add"), callback_button("📋 لیست اسپانسر", "sp:list")], [callback_button("🎭 افزودن جوک", "ent:add_joke"), callback_button("🧩 افزودن دانستنی", "ent:add_fact")], [callback_button("✅ بررسی عضویت", "sp:chk")] ]) text = ( "🛠 پنل سازنده:\n" "برای افزودن اسپانسر روی دکمه بزنید یا دستور «افزودن اسپانسر» را بفرستید.\n" "برای افزودن جوک/دانستنی: افزودن جوک [متن] یا افزودن دانستنی [متن]" ) send_message(chat_id, text, reply_markup=keyboard, reply_to_message_id=message.get("message_id")) def send_owner_panel(message): chat_id = message["chat"]["id"] keyboard = inline_keyboard([ [callback_button("🔐 بستن گروه", "grp:close"), callback_button("🔓 باز کردن گروه", "grp:open")] ]) text = ( "👑 پنل مالک:\n" "برای مدیریت ادمین‌ها از دستورات زیر استفاده کنید:\n" "ادمین کن | ادمین بگیر | دسترسی بن @username روشن/خاموش | دسترسی سکوت @username روشن/خاموش | دسترسی اکو @username روشن/خاموش" ) send_message(chat_id, text, reply_markup=keyboard, reply_to_message_id=message.get("message_id")) def send_admin_panel(message): chat_id = message["chat"]["id"] keyboard = inline_keyboard([ [callback_button("🔐 بستن گروه", "grp:close"), callback_button("🔓 باز کردن گروه", "grp:open")] ]) text = ( "🛡 پنل ادمین:\n" "بستن گروه | باز کردن گروه | بن | آنبان | سکوت 1 ساعت | رفع سکوت | اکو سلام | حذف پیام | اخطار" ) send_message(chat_id, text, reply_markup=keyboard, reply_to_message_id=message.get("message_id")) def cmd_ban(message, args_text): chat_id = message["chat"]["id"] user_id = message["from"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.", reply_to_message_id=message["message_id"]) return owner_id = get_owner_id(chat_id) if target_id == owner_id or is_creator(target_id): send_message(chat_id, "نمی‌توانم مالک یا سازنده را بن کنم.", reply_to_message_id=message["message_id"]) return if is_telegram_admin(target_id, chat_id) and not (is_creator(user_id) or user_id == owner_id): send_message(chat_id, "فقط مالک یا سازنده می‌تواند ادمین تلگرامی را بن کند.", reply_to_message_id=message["message_id"]) return ok = ban_user(chat_id, target_id) if ok: log_action(chat_id, user_id, "ban", str(target_id)) send_message(chat_id, "🔨 کاربر بن شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "بن انجام نشد. احتمالاً ربات دسترسی ندارد.", reply_to_message_id=message["message_id"]) def cmd_unban(message, args_text): chat_id = message["chat"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.", reply_to_message_id=message["message_id"]) return ok = unban_user(chat_id, target_id) if ok: log_action(chat_id, message["from"]["id"], "unban", str(target_id)) send_message(chat_id, "✅ کاربر از بن خارج شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "آنبان انجام نشد.", reply_to_message_id=message["message_id"]) def cmd_mute(message, args_text): chat_id = message["chat"]["id"] user_id = message["from"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.\nمثال: سکوت @username 1 ساعت", reply_to_message_id=message["message_id"]) return owner_id = get_owner_id(chat_id) if target_id == owner_id or is_creator(target_id): send_message(chat_id, "نمی‌توانم مالک یا سازنده را سکوت کنم.", reply_to_message_id=message["message_id"]) return duration = parse_duration(args_text) until_date = int(time.time()) + duration if duration > 0 else 0 ok = mute_user(chat_id, target_id, until_date) if ok: log_action(chat_id, user_id, "mute", f"{target_id}:{duration}") if duration == 0: send_message(chat_id, "🔇 کاربر دائم سکوت شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, f"🔇 کاربر برای {duration} ثانیه سکوت شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "سکوت انجام نشد. احتمالاً ربات دسترسی ندارد.", reply_to_message_id=message["message_id"]) def cmd_unmute(message, args_text): chat_id = message["chat"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.", reply_to_message_id=message["message_id"]) return ok = unmute_user(chat_id, target_id) if ok: log_action(chat_id, message["from"]["id"], "unmute", str(target_id)) send_message(chat_id, "🔊 رفع سکوت شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "رفع سکوت انجام نشد.", reply_to_message_id=message["message_id"]) def cmd_promote(message, args_text): chat_id = message["chat"]["id"] user_id = message["from"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.", reply_to_message_id=message["message_id"]) return if target_id == user_id: send_message(chat_id, "نمی‌توانید خودتان را ادمین کنید.", reply_to_message_id=message["message_id"]) return owner_id = get_owner_id(chat_id) if target_id == owner_id or is_creator(target_id): send_message(chat_id, "مالک/سازنده نیازی به ادمین شدن ندارد.", reply_to_message_id=message["message_id"]) return promoted = promote_user(chat_id, target_id) add_bot_admin(chat_id, target_id) log_action(chat_id, user_id, "promote", str(target_id)) if promoted: send_message(chat_id, "✅ کاربر ادمین شد و دسترسی‌های رباتی پیش‌فرض حذف/اخطار را دارد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "کاربر به‌عنوان ادمین ربات ثبت شد، اما ارتقای تلگرامی انجام نشد؛ احتمالاً ربات دسترسی پرموت ندارد.", reply_to_message_id=message["message_id"]) def cmd_demote(message, args_text): chat_id = message["chat"]["id"] user_id = message["from"]["id"] target_id, username = extract_target(message, args_text) if not target_id: send_message(chat_id, "کاربر را ریپلای کنید یا @username/آیدی بدهید.", reply_to_message_id=message["message_id"]) return owner_id = get_owner_id(chat_id) if target_id == owner_id or is_creator(target_id): send_message(chat_id, "نمی‌توانم مالک یا سازنده را دموت کنم.", reply_to_message_id=message["message_id"]) return demote_user(chat_id, target_id) remove_bot_admin(chat_id, target_id) log_action(chat_id, user_id, "demote", str(target_id)) send_message(chat_id, "✅ کاربر از ادمینی خارج شد.", reply_to_message_id=message["message_id"]) def cmd_echo(message): chat_id = message["chat"]["id"] user_id = message["from"]["id"] text = message.get("text") or "" norm = normalize_fa(text) if not can(user_id, chat_id, "echo"): send_message(chat_id, "شما دسترسی اکو ندارید.", reply_to_message_id=message["message_id"]) return args = "" parts = text.strip().split(maxsplit=1) if norm in ("اکو", "بگو"): args = "" elif len(parts) > 1 and normalize_fa(parts[0]) in ("اکو", "بگو"): args = parts[1] if not args: reply = message.get("reply_to_message") if reply: args = reply.get("text") or reply.get("caption") or "" if not args: send_message(chat_id, "مثال: اکو سلام", reply_to_message_id=message["message_id"]) return if len(args) > 3500: args = args[:3500] send_message(chat_id, args) execute( "INSERT INTO echo_logs(chat_id, admin_id, text, created_at) VALUES (?, ?, ?, ?)", (chat_id, user_id, args, now_iso()) ) log_action(chat_id, user_id, "echo", args[:100]) def cmd_delete(message): chat_id = message["chat"]["id"] user_id = message["from"]["id"] reply = message.get("reply_to_message") if not reply: send_message(chat_id, "روی پیام مورد نظر ریپلای کنید و دوباره این دستور را بفرستید.", reply_to_message_id=message["message_id"]) return ok = delete_message(chat_id, reply.get("message_id")) if ok: log_action(chat_id, user_id, "delete", str(reply.get("message_id"))) else: send_message(chat_id, "حذف پیام انجام نشد.", reply_to_message_id=message["message_id"]) def cmd_warn(message): chat_id = message["chat"]["id"] user_id = message["from"]["id"] reply = message.get("reply_to_message") if not reply or not reply.get("from"): send_message(chat_id, "روی پیام کاربر ریپلای کنید.", reply_to_message_id=message["message_id"]) return target_id = reply["from"].get("id") log_action(chat_id, user_id, "warn", str(target_id)) send_message(chat_id, "⚠️ اخطار ثبت شد.", reply_to_message_id=message["message_id"]) def cmd_set_permission(message, text): chat_id = message["chat"]["id"] user_id = message["from"]["id"] if not can(user_id, chat_id, "manage_admins"): send_message(chat_id, "فقط مالک یا سازنده می‌تواند دسترسی‌ها را تغییر دهد.", reply_to_message_id=message["message_id"]) return norm = normalize_fa(text) m = re.search(r"دسترسی\s+(.+?)\s+(@[A-Za-z0-9_]+|\d{5,})\s+(روشن|خاموش|فعال|غیرفعال)", norm) if not m: send_message( chat_id, "فرمت: دسترسی بن @username روشن\nیا: دسترسی سکوت 123456789 خاموش", reply_to_message_id=message["message_id"] ) return perm_name = m.group(1).strip() target_token = m.group(2).strip() state_word = m.group(3).strip() enabled = state_word in ("روشن", "فعال") perm = PERM_NAMES.get(perm_name) if not perm: send_message(chat_id, f"دسترسی «{perm_name}» شناخته نشد.", reply_to_message_id=message["message_id"]) return if target_token.startswith("@"): target_id = get_user_id_by_username(target_token) else: target_id = int(target_token) if not target_id: send_message(chat_id, "کاربر پیدا نشد. اول باید ربات او را دیده باشد یا آیدی عددی بدهید.", reply_to_message_id=message["message_id"]) return ok = set_admin_permission(chat_id, target_id, perm, enabled) if ok: log_action(chat_id, user_id, "set_permission", f"{target_id}:{perm}:{enabled}") send_message(chat_id, f"✅ دسترسی «{perm_name}» {'فعال' if enabled else 'غیرفعال'} شد.", reply_to_message_id=message["message_id"]) else: send_message(chat_id, "این کاربر هنوز ادمین ربات نیست. ابتدا با «ادمین کن» اضافه‌اش کنید.", reply_to_message_id=message["message_id"]) # ============================================================ # CALLBACK HANDLER # ============================================================ def handle_callback(cq): data = cq.get("data") or "" user = cq.get("from") or {} user_id = user.get("id") message = cq.get("message") or {} chat_id = message.get("chat", {}).get("id") message_id = message.get("message_id") if not user_id: return if data == "sp:chk": missing = missing_sponsors(user_id) if missing: answer_callback(cq["id"], "هنوز عضو همه کانال‌ها/گروه‌ها نشده‌اید.", show_alert=True) if chat_id and message_id: titles = "\n".join([f"- {s['title'] or s['chat_id']}" for s in missing]) edit_message_text(chat_id, message_id, f"هنوز عضو این موارد نشده‌اید:\n{titles}", reply_markup=sponsor_keyboard()) else: answer_callback(cq["id"], "عضویت شما تایید شد.", show_alert=True) if chat_id and message_id: edit_message_text(chat_id, message_id, "✅ عضویت شما تایید شد. حالا می‌توانید دستور قبلی را دوباره بفرستید.") return if data == "sp:add": if is_creator(user_id): set_state(user_id, chat_id, "add_sponsor") answer_callback(cq["id"], "حالا @username یا آیدی کانال/گروه را بفرستید.") if chat_id and message_id: edit_message_text(chat_id, message_id, "لطفاً @username، لینک عمومی یا آیدی عددی کانال/گروه اسپانسر را بفرستید.") else: answer_callback(cq["id"], "فقط سازنده می‌تواند اسپانسر اضافه کند.", show_alert=True) return if data == "sp:list": sponsors = get_active_sponsors() if not sponsors: text = "اسپانسر فعالی ثبت نشده است." else: lines = [] for s in sponsors: lines.append(f"- {s['title'] or s['chat_id']} | آیدی: {s['chat_id']}") text = "📋 اسپانسرهای فعال:\n" + "\n".join(lines) answer_callback(cq["id"]) if chat_id and message_id: edit_message_text(chat_id, message_id, text) return if data == "ent:add_joke": if is_creator(user_id): set_state(user_id, chat_id, "add_joke") answer_callback(cq["id"], "حالا متن جوک را بفرستید.") if chat_id and message_id: edit_message_text(chat_id, message_id, "لطفاً متن جوک را بفرستید.") else: answer_callback(cq["id"], "فقط سازنده می‌تواند.", show_alert=True) return if data == "ent:add_fact": if is_creator(user_id): set_state(user_id, chat_id, "add_fact") answer_callback(cq["id"], "حالا متن دانستنی را بفرستید.") if chat_id and message_id: edit_message_text(chat_id, message_id, "لطفاً متن دانستنی را بفرستید.") else: answer_callback(cq["id"], "فقط سازنده می‌تواند.", show_alert=True) return if data == "ent:joke": answer_callback(cq["id"]) send_message(chat_id, get_random_entertainment("joke")) return if data == "ent:fact": answer_callback(cq["id"]) send_message(chat_id, get_random_entertainment("fact")) return if data == "ent:dice": answer_callback(cq["id"]) send_message(chat_id, f"🎲 تاس آمد: {random.randint(1, 6)}") return if data == "ent:coin": answer_callback(cq["id"]) send_message(chat_id, f"🪙 نتیجه: {random.choice(['شیر', 'خط'])}") return if data == "usr:profile": answer_callback(cq["id"]) username = user.get("username") or "—" first_name = user.get("first_name") or "—" send_message(chat_id, f"👤 آیدی: {user_id}\nیوزرنیم: @{username}\nنام: {first_name}") return if data == "grp:close": if can(user_id, chat_id, "open_close"): ok = close_group(chat_id) answer_callback(cq["id"], "درخواست بستن گروه ثبت شد.") if chat_id and message_id: edit_message_text(chat_id, message_id, "🔐 گروه بسته شد." if ok else "بستن گروه انجام نشد. ربات دسترسی ندارد.") log_action(chat_id, user_id, "close_group", "") else: answer_callback(cq["id"], "شما دسترسی بستن گروه را ندارید.", show_alert=True) return if data == "grp:open": if can(user_id, chat_id, "open_close"): ok = open_group(chat_id) answer_callback(cq["id"], "درخواست باز کردن گروه ثبت شد.") if chat_id and message_id: edit_message_text(chat_id, message_id, "🔓 گروه باز شد." if ok else "باز کردن گروه انجام نشد. ربات دسترسی ندارد.") log_action(chat_id, user_id, "open_group", "") else: answer_callback(cq["id"], "شما دسترسی باز کردن گروه را ندارید.", show_alert=True) return answer_callback(cq["id"]) # ============================================================ # MESSAGE HANDLER # ============================================================ def handle_my_chat_member(update): chat = update.get("chat") or {} new_status = (update.get("new_chat_member") or {}).get("status") chat_id = chat.get("id") if not chat_id: return ensure_chat(chat) if new_status in ("left", "kicked"): execute("UPDATE chats SET active=0 WHERE chat_id=?", (chat_id,)) else: execute("UPDATE chats SET active=1 WHERE chat_id=?", (chat_id,)) refresh_chat_owner(chat_id) def handle_message(message, edited=False): if not message: return from_user = message.get("from") if not from_user: return user_id = from_user.get("id") chat = message.get("chat") or {} chat_id = chat.get("id") if not user_id or not chat_id: return ensure_user(from_user) ensure_chat(chat) if edited: moderate_message(message) return owner_id = get_owner_id(chat_id) privileged = is_creator(user_id) or owner_id == user_id or can_any_admin(user_id, chat_id) or is_telegram_admin(user_id, chat_id) # closed group fallback chat_row = query("SELECT is_closed FROM chats WHERE chat_id=?", (chat_id,), one=True) if chat_row and chat_row["is_closed"] and not privileged: delete_message(chat_id, message.get("message_id")) return if not privileged: moderate_message(message) if process_state(message): return text = message.get("text") or "" if not text: return norm = normalize_fa(text) parts = text.strip().split(maxsplit=1) args_text = parts[1] if len(parts) > 1 else "" # public commands with sponsor gate if norm in START_CMDS or norm in HELP_CMDS or norm in MENU_CMDS or norm in PROFILE_CMDS: if not sponsor_gate(message, "public"): return if norm in START_CMDS: cmd_start(message) elif norm in HELP_CMDS: cmd_help(message) elif norm in MENU_CMDS: cmd_menu(message) elif norm in PROFILE_CMDS: cmd_profile(message) return if norm in JOKE_CMDS: if not sponsor_gate(message, "joke"): return cmd_joke(message) return if norm in FACT_CMDS: if not sponsor_gate(message, "fact"): return cmd_fact(message) return if norm in DICE_CMDS: if not sponsor_gate(message, "dice"): return cmd_dice(message) return if norm in COIN_CMDS: if not sponsor_gate(message, "coin"): return cmd_coin(message) return if norm in RPS_CMDS: if not sponsor_gate(message, "rps"): return cmd_rps(message) return if norm in NUMBER_CMDS: if not sponsor_gate(message, "number"): return cmd_number(message) return # panels if norm in CREATOR_PANEL_CMDS: if is_creator(user_id): send_creator_panel(message) else: send_message(chat_id, "شما سازنده نیستید.", reply_to_message_id=message.get("message_id")) return if norm in OWNER_PANEL_CMDS: if chat.get("type") not in ("group", "supergroup"): send_message(chat_id, "پنل مالک فقط در گروه کار می‌کند.", reply_to_message_id=message.get("message_id")) return if is_creator(user_id) or get_owner_id(chat_id) == user_id: send_owner_panel(message) else: send_message(chat_id, "فقط مالک گروه می‌تواند پنل مالک را ببیند.", reply_to_message_id=message.get("message_id")) return if norm in ADMIN_PANEL_CMDS: if chat.get("type") not in ("group", "supergroup"): send_message(chat_id, "پنل ادمین فقط در گروه کار می‌کند.", reply_to_message_id=message.get("message_id")) return if can_any_admin(user_id, chat_id): send_admin_panel(message) else: send_message(chat_id, "شما ادمین نیستید.", reply_to_message_id=message.get("message_id")) return # direct creator content commands if norm.startswith("افزودن جوک"): if not is_creator(user_id): send_message(chat_id, "فقط سازنده می‌تواند.", reply_to_message_id=message.get("message_id")) return if len(parts) > 1 and parts[1].strip(): execute( "INSERT INTO entertainment(category, text, active, added_by, created_at) VALUES ('joke', ?, 1, ?, ?)", (parts[1].strip(), user_id, now_iso()) ) send_message(chat_id, "✅ جوک اضافه شد.", reply_to_message_id=message.get("message_id")) else: set_state(user_id, chat_id, "add_joke") send_message(chat_id, "لطفاً متن جوک را بفرستید.", reply_to_message_id=message.get("message_id")) return if norm.startswith("افزودن دانستنی"): if not is_creator(user_id): send_message(chat_id, "فقط سازنده می‌تواند.", reply_to_message_id=message.get("message_id")) return if len(parts) > 1 and parts[1].strip(): execute( "INSERT INTO entertainment(category, text, active, added_by, created_at) VALUES ('fact', ?, 1, ?, ?)", (parts[1].strip(), user_id, now_iso()) ) send_message(chat_id, "✅ دانستنی اضافه شد.", reply_to_message_id=message.get("message_id")) else: set_state(user_id, chat_id, "add_fact") send_message(chat_id, "لطفاً متن دانستنی را بفرستید.", reply_to_message_id=message.get("message_id")) return if norm.startswith("افزودن اسپانسر"): if not is_creator(user_id): send_message(chat_id, "فقط سازنده می‌تواند.", reply_to_message_id=message.get("message_id")) return if len(parts) > 1 and parts[1].strip(): add_sponsor_from_text(parts[1], user_id, message) else: set_state(user_id, chat_id, "add_sponsor") send_message(chat_id, "لطفاً @username، لینک عمومی یا آیدی عددی کانال/گروه را بفرستید.", reply_to_message_id=message.get("message_id")) return if norm.startswith("دسترسی"): cmd_set_permission(message, text) return # group admin commands if norm in CLOSE_CMDS: if can(user_id, chat_id, "open_close"): ok = close_group(chat_id) log_action(chat_id, user_id, "close_group", "") send_message(chat_id, "🔐 گروه بسته شد." if ok else "بستن گروه ناموفق بود. ربات دسترسی ندارد.", reply_to_message_id=message.get("message_id")) else: send_message(chat_id, "شما دسترسی بستن گروه را ندارید.", reply_to_message_id=message.get("message_id")) return if norm in OPEN_CMDS: if can(user_id, chat_id, "open_close"): ok = open_group(chat_id) log_action(chat_id, user_id, "open_group", "") send_message(chat_id, "🔓 گروه باز شد." if ok else "باز کردن گروه ناموفق بود. ربات دسترسی ندارد.", reply_to_message_id=message.get("message_id")) else: send_message(chat_id, "شما دسترسی باز کردن گروه را ندارید.", reply_to_message_id=message.get("message_id")) return if norm in BAN_CMDS: if can(user_id, chat_id, "ban"): cmd_ban(message, args_text) else: send_message(chat_id, "شما دسترسی بن ندارید.", reply_to_message_id=message.get("message_id")) return if norm in UNBAN_CMDS: if can(user_id, chat_id, "unban"): cmd_unban(message, args_text) else: send_message(chat_id, "شما دسترسی آنبان ندارید.", reply_to_message_id=message.get("message_id")) return if norm.startswith("سکوت") or norm.startswith("میوت"): if can(user_id, chat_id, "mute"): cmd_mute(message, args_text) else: send_message(chat_id, "شما دسترسی سکوت ندارید.", reply_to_message_id=message.get("message_id")) return if norm in UNMUTE_CMDS: if can(user_id, chat_id, "unmute"): cmd_unmute(message, args_text) else: send_message(chat_id, "شما دسترسی رفع سکوت ندارید.", reply_to_message_id=message.get("message_id")) return if norm in PROMOTE_CMDS: if can(user_id, chat_id, "promote"): cmd_promote(message, args_text) else: send_message(chat_id, "شما دسترسی ادمین کردن ندارید.", reply_to_message_id=message.get("message_id")) return if norm in DEMOTE_CMDS: if can(user_id, chat_id, "demote"): cmd_demote(message, args_text) else: send_message(chat_id, "شما دسترسی ادمین گرفتن ندارید.", reply_to_message_id=message.get("message_id")) return if norm.startswith("اکو") or norm.startswith("بگو"): cmd_echo(message) return if norm in DELETE_CMDS: if can(user_id, chat_id, "delete"): cmd_delete(message) else: send_message(chat_id, "شما دسترسی حذف پیام ندارید.", reply_to_message_id=message.get("message_id")) return if norm in WARN_CMDS: if can(user_id, chat_id, "warn"): cmd_warn(message) else: send_message(chat_id, "شما دسترسی اخطار ندارید.", reply_to_message_id=message.get("message_id")) return def handle_update(update): try: if "message" in update: handle_message(update["message"], edited=False) elif "edited_message" in update: handle_message(update["edited_message"], edited=True) elif "callback_query" in update: handle_callback(update["callback_query"]) elif "my_chat_member" in update: handle_my_chat_member(update["my_chat_member"]) except Exception as e: logger.exception("handle_update error: %s", e) # ============================================================ # FLASK / WEBHOOK # ============================================================ @application.route("/", methods=["GET"]) def index(): return "TG Bot OK", 200 @application.route(WEBHOOK_PATH, methods=["POST"]) def webhook(): secret = request.headers.get("X-Telegram-Bot-Api-Secret-Token", "") if secret != WEBHOOK_SECRET: abort(403) update = request.get_json(silent=True) if update: handle_update(update) return "OK", 200 # ============================================================ # CLI # ============================================================ def cli(): if len(sys.argv) < 2: print("Usage: python bot.py initdb | set-webhook | delete-webhook | webhook-info | run | check") return cmd = sys.argv[1] if cmd == "initdb": init_db() print("Database initialized.") elif cmd == "set-webhook": if not BOT_TOKEN: print("BOT_TOKEN is empty.") return if not PUBLIC_BASE_URL: print("PUBLIC_BASE_URL is empty.") return url = PUBLIC_BASE_URL + WEBHOOK_PATH allowed_updates = ["message", "edited_message", "callback_query", "my_chat_member"] res = tg( "setWebhook", url=url, secret_token=WEBHOOK_SECRET, allowed_updates=allowed_updates, drop_pending_updates=False ) print("setWebhook result:", res) print("Webhook URL:", url) elif cmd == "delete-webhook": res = tg("deleteWebhook") print("deleteWebhook result:", res) elif cmd == "webhook-info": res = tg("getWebhookInfo") print(json.dumps(res, ensure_ascii=False, indent=2)) elif cmd == "run": init_db() port = int(os.getenv("PORT", "8000")) application.run(host="0.0.0.0", port=port) elif cmd == "check": print("Python:", sys.version) print("BASE_DIR:", BASE_DIR) print("DB_PATH:", DB_PATH) print("BOT_TOKEN set:", bool(BOT_TOKEN)) print("PUBLIC_BASE_URL:", PUBLIC_BASE_URL or "(empty)") print("WEBHOOK_PATH:", WEBHOOK_PATH) print("OWNER_IDS:", OWNER_IDS) init_db() print("DB init OK") else: print("Unknown command:", cmd) if __name__ == "__main__": cli()