""" Анти-спам сервис — проверка ссылок и доменов """ import json import logging import re from datetime import datetime from typing import List, Dict, Tuple, Optional from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession logger = logging.getLogger(__name__) class AntiSpamService: """Сервис проверки сообщений на спам""" def __init__(self, session: AsyncSession): self.session = session self.config = self._load_config() def _load_config(self) -> Dict: """Загрузка конфигурации из JSON""" try: with open('data/bad_domains.json', 'r', encoding='utf-8') as f: return json.load(f) except Exception as e: logger.error(f'Не удалось загрузить bad_domains.json: {e}') return self._get_default_config() def _get_default_config(self) -> Dict: """Конфигурация по умолчанию""" return { 'domains': [], 'patterns': [], 'keywords': [], 'settings': { 'delete_message': True, 'warn_user': True, 'notify_admin': True, 'ban_after_warnings': 3, 'exclude_verified': False, 'exclude_admins': True } } def reload_config(self): """Перезагрузить конфигурацию из файла""" self.config = self._load_config() logger.info('Конфигурация анти-спама перезаряжена') def extract_links(self, text: str) -> List[str]: """Извлечь все ссылки из текста""" if not text: return [] # Паттерн для поиска URL url_pattern = r'https?://[^\s<>"{}|\\^`\[\]]+|www\.[^\s<>"{}|\\^`\[\]]+' links = re.findall(url_pattern, text, re.IGNORECASE) # Также ищем t.me ссылки telegram_pattern = r't\.me/[^\s<>"{}|\\^`\[\]]+' telegram_links = re.findall(telegram_pattern, text, re.IGNORECASE) return links + telegram_links def extract_domains(self, links: List[str]) -> List[str]: """Извлечь домены из ссылок""" domains = [] for link in links: try: # Удаляем протокол link = link.replace('https://', '').replace('http://', '') # Берём домен domain = link.split('/')[0].lower() # Удаляем www. domain = domain.replace('www.', '') if domain: domains.append(domain) except Exception: continue return domains def check_message(self, text: str) -> Tuple[bool, List[str], str]: """ Проверить сообщение на наличие запрещённых ссылок Returns: (is_spam, found_items, reason) """ if not text: return False, [], '' found_spam = [] reasons = [] # 1. Проверка доменов из чёрного списка links = self.extract_links(text) domains = self.extract_domains(links) bad_domains = set(d.lower() for d in self.config.get('domains', [])) for domain in domains: # Проверяем точное совпадение и поддомены for bad_domain in bad_domains: if domain == bad_domain or domain.endswith(f'.{bad_domain}'): found_spam.append(domain) reasons.append(f'Запрещённый домен: {bad_domain}') # 2. Проверка паттернов patterns = self.config.get('patterns', []) for pattern in patterns: if pattern.lower() in text.lower(): found_spam.append(pattern) reasons.append(f'Запрещённый паттерн: {pattern}') # 3. Проверка ключевых слов keywords = self.config.get('keywords', []) text_lower = text.lower() for keyword in keywords: if keyword.lower() in text_lower: found_spam.append(keyword) reasons.append(f'Спам-ключ: {keyword}') is_spam = len(found_spam) > 0 reason = '; '.join(reasons) if reasons else '' return is_spam, found_spam, reason def get_settings(self) -> Dict: """Получить настройки""" return self.config.get('settings', {}) def add_domain(self, domain: str) -> bool: """Добавить домен в чёрный список""" domain = domain.lower().strip() if domain not in self.config['domains']: self.config['domains'].append(domain) self._save_config() logger.info(f'Добавлен домен в ЧС: {domain}') return True return False def remove_domain(self, domain: str) -> bool: """Удалить домен из чёрного списка""" domain = domain.lower().strip() if domain in self.config['domains']: self.config['domains'].remove(domain) self._save_config() logger.info(f'Удалён домен из ЧС: {domain}') return True return False def add_pattern(self, pattern: str) -> bool: """Добавить паттерн в чёрный список""" pattern = pattern.strip() if pattern not in self.config['patterns']: self.config['patterns'].append(pattern) self._save_config() logger.info(f'Добавлен паттерн в ЧС: {pattern}') return True return False def remove_pattern(self, pattern: str) -> bool: """Удалить паттерн из чёрного списка""" pattern = pattern.strip() if pattern in self.config['patterns']: self.config['patterns'].remove(pattern) self._save_config() logger.info(f'Удалён паттерн из ЧС: {pattern}') return True return False def add_keyword(self, keyword: str) -> bool: """Добавить ключевое слово в чёрный список""" keyword = keyword.strip() if keyword not in self.config['keywords']: self.config['keywords'].append(keyword) self._save_config() logger.info(f'Добавлено ключевое слово в ЧС: {keyword}') return True return False def remove_keyword(self, keyword: str) -> bool: """Удалить ключевое слово из чёрного списка""" keyword = keyword.strip() if keyword in self.config['keywords']: self.config['keywords'].remove(keyword) self._save_config() logger.info(f'Удалено ключевое слово из ЧС: {keyword}') return True return False def _save_config(self): """Сохранить конфигурацию в файл""" try: self.config['updated_at'] = datetime.utcnow().strftime('%Y-%m-%d') with open('data/bad_domains.json', 'w', encoding='utf-8') as f: json.dump(self.config, f, ensure_ascii=False, indent=2) except Exception as e: logger.error(f'Не удалось сохранить bad_domains.json: {e}') def get_all_domains(self) -> List[str]: """Получить все домены из чёрного списка""" return self.config.get('domains', []) def get_all_patterns(self) -> List[str]: """Получить все паттерны из чёрного списка""" return self.config.get('patterns', []) def get_all_keywords(self) -> List[str]: """Получить все ключевые слова из чёрного списка""" return self.config.get('keywords', []) def get_stats(self) -> Dict: """Получить статистику чёрного списка""" return { 'domains_count': len(self.config.get('domains', [])), 'patterns_count': len(self.config.get('patterns', [])), 'keywords_count': len(self.config.get('keywords', [])), 'settings': self.config.get('settings', {}) }