import os import time import hashlib import difflib from datetime import datetime, timedelta from typing import List, Optional from telebot import types from sqlalchemy import func from config import Config from db import db, ChatSetting, AdminPermission, Warning, Sponsor, RecentMessage, RateEvent from utils import TTLCache, normalize_text, t import keyboards ALL_PERMISSIONS = [ "ban", "unban", "mute", "unmute", "warn", "delete", "lock", "unlock", "echo", "settings", "grant", "sponsor", "panel", "close", "open", ] LOCK_TYPES = [ "all", "links", "media", "sticker", "fosh", "flood", "duplicate", "username", "forward", ] class SettingsService: def get(self, chat_id: int, key: str, default: Optional[str] = None) -> Optional[str]: row = db.query(ChatSetting).filter_by(chat_id=chat_id, key=key).first() return row.value if row else default def set(self, chat_id: int, key: str, value): row = db.query(ChatSetting).filter_by(chat_id=chat_id, key=key).first() if row: row.value = str(value) else: row = ChatSetting(chat_id=chat_id, key=key, value=str(value)) db.add(row) db.commit() def get_bool(self, chat_id: int, key: str, default: bool = False) -> bool: value = self.get(chat_id, key) if value is None: return default return str(value).strip().lower() in ("1", "true", "yes", "on") def get_int(self, chat_id: int, key: str, default: int = 0) -> int: value = self.get(chat_id, key) try: return int(value) except Exception: return default def get_float(self, chat_id: int, key: str, default: float = 0.0) -> float: value = self.get(chat_id, key) try: return float(value) except Exception: return default def set_bool(self, chat_id: int, key: str, value: bool): self.set(chat_id, key, "1" if value else "0") def toggle(self, chat_id: int, key: str, default: bool = False) -> bool: current = self.get_bool(chat_id, key, default) self.set_bool(chat_id, key, not current) return not current class PermissionService: def __init__(self, bot, settings: SettingsService): self.bot = bot self.settings = settings self.cache = TTLCache(default_ttl=120) def is_bot_owner(self, user_id: int) -> bool: return user_id in Config.BOT_OWNER_IDS def get_member(self, chat_id: int, user_id: int): key = f"{chat_id}:{user_id}" cached = self.cache.get(key) if cached is not None: return cached try: member = self.bot.get_chat_member(chat_id, user_id) except Exception: member = None self.cache.set(key, member, 120) return member def is_owner(self, chat_id: int, user_id: int) -> bool: member = self.get_member(chat_id, user_id) return bool(member and member.status == "creator") def is_admin(self, chat_id: int, user_id: int) -> bool: member = self.get_member(chat_id, user_id) return bool(member and member.status in ("creator", "administrator")) def get_role(self, chat_id: int, user_id: int) -> str: if self.is_bot_owner(user_id): return "bot_owner" if self.is_owner(chat_id, user_id): return "owner" if self.is_admin(chat_id, user_id): return "admin" return "user" def has_permission(self, chat_id: int, user_id: int, permission: str) -> bool: if self.is_bot_owner(user_id): return True member = self.get_member(chat_id, user_id) if not member: return False if member.status == "creator": return True row = db.query(AdminPermission).filter_by( chat_id=chat_id, user_id=user_id, permission=permission, ).first() if row: return True # Optional default Telegram-admin permission toggles: # default_admin_ban=1, default_admin_echo=1, etc. if member.status == "administrator": if self.settings.get_bool(chat_id, f"default_admin_{permission}", False): return True return False def grant(self, chat_id: int, user_id: int, permission: str) -> bool: if permission not in ALL_PERMISSIONS: return False row = db.query(AdminPermission).filter_by( chat_id=chat_id, user_id=user_id, permission=permission, ).first() if not row: row = AdminPermission(chat_id=chat_id, user_id=user_id, permission=permission) db.add(row) db.commit() return True def revoke(self, chat_id: int, user_id: int, permission: str) -> bool: deleted = db.query(AdminPermission).filter_by( chat_id=chat_id, user_id=user_id, permission=permission, ).delete(synchronize_session=False) db.commit() return bool(deleted) def list_permissions(self, chat_id: int, user_id: int) -> List[str]: rows = db.query(AdminPermission).filter_by(chat_id=chat_id, user_id=user_id).all() return [row.permission for row in rows] class SponsorService: def __init__(self, bot, settings: SettingsService, permissions: PermissionService): self.bot = bot self.settings = settings self.permissions = permissions def active_sponsors(self) -> List[Sponsor]: return db.query(Sponsor).filter(Sponsor.active == True).all() def add(self, sponsor_chat_id: str, link: Optional[str], added_by: int): try: chat = self.bot.get_chat(sponsor_chat_id) except Exception: return False, "invalid_sponsor" sponsor_id = chat.id title = getattr(chat, "title", None) or getattr(chat, "username", None) or str(sponsor_id) if not link: username = getattr(chat, "username", None) invite_link = getattr(chat, "invite_link", None) if username: link = f"https://t.me/{username}" elif invite_link: link = invite_link else: return False, "invalid_sponsor" row = db.query(Sponsor).filter_by(sponsor_chat_id=sponsor_id).first() if row: row.title = title row.link = link row.active = True row.added_by = added_by else: row = Sponsor( sponsor_chat_id=sponsor_id, title=title, link=link, active=True, added_by=added_by, ) db.add(row) db.commit() return True, "sponsor_added" def remove(self, sponsor_chat_id: str) -> bool: try: sponsor_id = int(sponsor_chat_id) except Exception: sponsor_id = None query = db.query(Sponsor) if sponsor_id is not None: row = query.filter_by(sponsor_chat_id=sponsor_id).first() else: row = query.filter(Sponsor.title == sponsor_chat_id).first() if not row: return False row.active = False db.commit() return True def is_member(self, sponsor_chat_id: int, user_id: int) -> bool: member = self.permissions.get_member(sponsor_chat_id, user_id) if not member: return False if member.status in ("creator", "administrator", "member"): return True if member.status == "restricted": return bool(getattr(member, "is_member", False)) return False def missing_sponsors(self, user_id: int) -> List[Sponsor]: result = [] for sponsor in self.active_sponsors(): if not self.is_member(sponsor.sponsor_chat_id, user_id): result.append(sponsor) return result def ensure(self, message, user_id: int, chat_id: int, lang: str) -> bool: if message.chat.type == "private": return True if self.permissions.is_bot_owner(user_id): return True if Config.SPONSOR_EXEMPT_ADMINS and self.permissions.is_admin(chat_id, user_id): return True missing = self.missing_sponsors(user_id) if not missing: return True kb = keyboards.sponsor_keyboard(missing, lang) self.bot.send_message(chat_id, t("sponsor_required", lang), reply_markup=kb) return False class ProfanityFilter: def __init__(self, fa_path="data/bad_words_fa.txt", en_path="data/bad_words_en.txt"): self.words = set() self._load_file(fa_path) self._load_file(en_path) def _load_file(self, path: str): directory = os.path.dirname(path) if directory: os.makedirs(directory, exist_ok=True) if not os.path.exists(path): with open(path, "w", encoding="utf-8") as f: f.write("# Add bad words here, one per line.\n") with open(path, "r", encoding="utf-8") as f: for line in f: line = line.strip() if not line or line.startswith("#"): continue normalized = normalize_text(line) if normalized: self.words.add(normalized) def contains(self, text: str) -> bool: if not text: return False normalized = normalize_text(text) if not normalized: return False for word in self.words: if word and word in normalized: return True return False class SpamService: def __init__(self, settings: SettingsService): self.settings = settings def record_and_check_flood(self, chat_id: int, user_id: int) -> bool: if not self.settings.get_bool(chat_id, "flood_check", True): return False window = self.settings.get_int(chat_id, "flood_window", 10) threshold = self.settings.get_int(chat_id, "flood_threshold", 5) now = datetime.utcnow() cutoff = now - timedelta(seconds=window) db.query(RateEvent).filter( RateEvent.chat_id == chat_id, RateEvent.created_at < cutoff, ).delete(synchronize_session=False) db.add(RateEvent(chat_id=chat_id, user_id=user_id, event="message")) db.commit() count = db.query(func.count(RateEvent.id)).filter( RateEvent.chat_id == chat_id, RateEvent.user_id == user_id, RateEvent.created_at >= cutoff, ).scalar() return count > threshold def is_duplicate(self, chat_id: int, user_id: int, text: str) -> bool: if not self.settings.get_bool(chat_id, "duplicate_check", True): return False normalized = normalize_text(text) if len(normalized) < 15: return False threshold = self.settings.get_float(chat_id, "duplicate_threshold", 0.85) window = self.settings.get_int(chat_id, "duplicate_window", 120) now = datetime.utcnow() cutoff = now - timedelta(seconds=window) text_hash = hashlib.md5(normalized.encode("utf-8")).hexdigest() db.query(RecentMessage).filter( RecentMessage.chat_id == chat_id, RecentMessage.created_at < cutoff, ).delete(synchronize_session=False) recent_rows = db.query(RecentMessage).filter( RecentMessage.chat_id == chat_id, RecentMessage.user_id == user_id, RecentMessage.created_at >= cutoff, ).order_by(RecentMessage.id.desc()).limit(5).all() for row in recent_rows: if row.text_hash == text_hash: return True if row.text: ratio = difflib.SequenceMatcher(None, normalized, row.text).ratio() if ratio >= threshold: return True db.add( RecentMessage( chat_id=chat_id, user_id=user_id, text_hash=text_hash, text=normalized, created_at=now, ) ) db.commit() return False class ModerationService: def __init__(self, bot, settings: SettingsService, permissions: PermissionService): self.bot = bot self.settings = settings self.permissions = permissions try: self.bot_id = bot.get_me().id except Exception: self.bot_id = None def can_moderate(self, actor_id: int, target_id: int, chat_id: int) -> bool: if self.bot_id and target_id == self.bot_id: return False if self.permissions.is_bot_owner(target_id): return False target_member = self.permissions.get_member(chat_id, target_id) if not target_member: return True if target_member.status == "creator": return False if target_member.status == "administrator": if self.permissions.is_bot_owner(actor_id): return True actor_member = self.permissions.get_member(chat_id, actor_id) if actor_member and actor_member.status == "creator": return True return False return True def delete_message(self, chat_id: int, message_id: int) -> bool: try: self.bot.delete_message(chat_id, message_id) return True except Exception: return False def warn(self, chat_id: int, user_id: int, admin_id: int, reason: str, lang: str): db.add(Warning(chat_id=chat_id, user_id=user_id, admin_id=admin_id, reason=reason)) db.commit() count = db.query(func.count(Warning.id)).filter_by( chat_id=chat_id, user_id=user_id, ).scalar() max_warn = self.settings.get_int(chat_id, "max_warn", 3) if count >= max_warn: action = self.settings.get(chat_id, "warn_action", "mute") if action == "ban": try: self.ban_user(chat_id, user_id) except Exception: pass else: try: self.mute_user(chat_id, user_id, 3600) except Exception: pass db.query(Warning).filter_by(chat_id=chat_id, user_id=user_id).delete( synchronize_session=False ) db.commit() return count, True return count, False def clear_warnings(self, chat_id: int, user_id: int): db.query(Warning).filter_by(chat_id=chat_id, user_id=user_id).delete( synchronize_session=False ) db.commit() def count_warnings(self, chat_id: int, user_id: int) -> int: return db.query(func.count(Warning.id)).filter_by( chat_id=chat_id, user_id=user_id, ).scalar() def mute_user(self, chat_id: int, user_id: int, seconds: int = 3600): until = int(time.time()) + seconds permissions = types.ChatPermissions( can_send_messages=False, can_send_media_messages=False, can_send_polls=False, can_send_other_messages=False, can_add_web_page_previews=False, can_change_info=False, can_invite_users=False, can_pin_messages=False, ) self.bot.restrict_chat_member( chat_id, user_id, permissions=permissions, until_date=until, ) def unmute_user(self, chat_id: int, user_id: int): permissions = types.ChatPermissions( can_send_messages=True, can_send_media_messages=True, can_send_polls=True, can_send_other_messages=True, can_add_web_page_previews=True, can_change_info=True, can_invite_users=True, can_pin_messages=True, ) self.bot.restrict_chat_member(chat_id, user_id, permissions=permissions) def ban_user(self, chat_id: int, user_id: int): self.bot.ban_chat_member(chat_id, user_id) def unban_user(self, chat_id: int, user_id: int): try: self.bot.unban_chat_member(chat_id, user_id, only_if_banned=True) except Exception: self.bot.unban_chat_member(chat_id, user_id) def close_group(self, chat_id: int): permissions = types.ChatPermissions( can_send_messages=False, can_send_media_messages=False, can_send_polls=False, can_send_other_messages=False, can_add_web_page_previews=False, can_change_info=False, can_invite_users=False, can_pin_messages=False, ) self.bot.set_chat_permissions(chat_id, permissions) def open_group(self, chat_id: int): permissions = types.ChatPermissions( can_send_messages=True, can_send_media_messages=True, can_send_polls=True, can_send_other_messages=True, can_add_web_page_previews=True, can_change_info=True, can_invite_users=True, can_pin_messages=True, ) self.bot.set_chat_permissions(chat_id, permissions) class Services: def __init__(self, bot): self.bot = bot self.settings = SettingsService() self.permissions = PermissionService(bot, self.settings) self.sponsors = SponsorService(bot, self.settings, self.permissions) self.profanity = ProfanityFilter() self.spam = SpamService(self.settings) self.moderation = ModerationService(bot, self.settings, self.permissions)