ba4676b0af
✅ Регистрация и верификация жильцов ✅ Рейтинг активности (уровни) ✅ Детект шпионов (Score 0-100) ✅ Анти-мат система (3 предупреждения → бан) ✅ Админ-панель с рассылкой ✅ Учёт квартир (несколько жильцов) ✅ Телефоны экстренных служб и мастеров ✅ Еженедельный экспорт JSON ✅ Прокси (socks5) для обхода РКН Бекап: backups/versions/v1.0_mvp_2026-02-26/ Co-authored-by: Qwen-Coder <qwen-coder@alibabacloud.com>
221 lines
8.5 KiB
Python
221 lines
8.5 KiB
Python
"""
|
||
Анти-спам сервис — проверка ссылок и доменов
|
||
"""
|
||
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', {})
|
||
}
|