Как я написал Telegram-бота для SEO-аудита и не дал ему стать инструментом для атак

Вступление (до ката)

Проверить мета-теги на одной странице — дело пяти минут. Найти битые ссылки на сайте из 500 страниц — уже задача на вечер. А если нужно ещё проанализировать тексты на переоптимизацию, сравнить SEO-метрики с конкурентом и получить красивый отчёт?

Я решил автоматизировать рутину SEO-специалиста (Search Engine Optimization — поисковая оптимизация) через Telegram-бота. Начиналось всё с простой проверки редиректов, а выросло в полноценный инструмент: аудит страниц по 40+ параметрам, краулер битых ссылок, анализ текстов на фильтр «Баден-Баден» Яндекса, конкурентный анализ и мониторинг.

В этой статье — технические решения, которые пригодятся при создании любого бота, работающего с пользовательскими URL: защита от SSRF (Server-Side Request Forgery — подделка серверных запросов), включая DNS rebinding, rate limiting через Redis, многопоточная обработка и генерация отчётов.

---

Содержание

1. Что умеет бот

2. Архитектура

3. Парсинг sitemap.xml

4. SEO-аудит страницы: 40+ проверок

5. Анализ текста на фильтр «Баден-Баден»

6. Краулер битых ссылок

7. Защита от SSRF-атак и DNS rebinding

8. Rate Limiting с Redis

9. Многопоточность и очередь проверок

10. Генерация отчётов

11. Уроки и выводы

---

Что умеет бот

Начиналось с проверки редиректов, а выросло в 8 модулей:

Всё это работает через Telegram — без веб-интерфейса, без регистрации, с inline-кнопками для подтверждения.

---

Архитектура

Стек

Python 3.12 + pyTelegramBotAPI (FSM — конечный автомат состояний)
PostgreSQL + SQLAlchemy (ORM — объектно-реляционное отображение)
Redis (rate limiting — ограничение частоты запросов)
requests + BeautifulSoup (HTTP + парсинг HTML)
Windows Service через NSSM (деплой)

Принцип модульности

Каждая функция — отдельный модуль-хендлер, который регистрируется в главном файле:

# Порядок регистрации важен — хендлеры проверяются последовательно
handlers = [
    start,          # /start, /help, меню
    seo,            # SEO-проверки
    broken_links,   # Битые ссылки
    competitor,     # Конкурентный анализ
    page_audit,     # SEO-аудит страницы
    baden_baden,    # Баден-Баден
    diff,           # Сравнение проверок
    monitoring,     # Мониторинг
    check,          # Массовая проверка URL (FSM-состояния)
    payments,       # Платежи
    reports,        # Отчёты и история
    admin,          # Админ-панель (последний — catch-all callback)
]

for handler in handlers:
    handler.register(bot, shared_state)

Каждый модуль получает экземпляр бота и общее состояние. Внутри — свои FSM-состояния, callback-хендлеры, фоновые потоки.

FSM-состояния

pyTelegramBotAPI поддерживает конечные автоматы через StateMemoryStorage. Каждый модуль объявляет свои состояния:

class BotStates(StatesGroup):
    # Проверка URL
    awaiting_sitemap_url = State()
    awaiting_url_list = State()
    awaiting_confirmation = State()
    awaiting_speed_selection = State()
    
    # SEO-аудит
    page_audit_awaiting_url = State()
    
    # Баден-Баден
    baden_awaiting_input = State()
    
    # Конкурентный анализ
    competitor_awaiting_url1 = State()
    competitor_awaiting_url2 = State()
    
    # ... и т.д.

Это позволяет боту понимать, на каком шаге находится пользователь, и корректно обрабатывать ввод.

---

Парсинг sitemap.xml

Нужно извлечь все URL из sitemap, включая вложенные (sitemap index). Ключевые моменты:

import xml.etree.ElementTree as ET

SITEMAP_NS = {'sm': 'http://www.sitemaps.org/schemas/sitemap/0.9'}

def parse_sitemap(url: str, depth: int = 0, max_depth: int = 5) -> list:
    """Рекурсивный парсинг sitemap с защитой от бесконечной вложенности"""
    if depth > max_depth:
        raise ValueError("Слишком глубокая вложенность sitemap")
    
    response = requests.get(url, timeout=10)
    root = ET.fromstring(response.content)
    
    # Sitemap index → рекурсия
    sitemaps = root.findall('.//sm:sitemap/sm:loc', SITEMAP_NS)
    if sitemaps:
        urls = []
        for s in sitemaps:
            urls.extend(parse_sitemap(s.text, depth + 1, max_depth))
        return urls
    
    # Обычный sitemap → извлекаем URL
    return [u.text for u in root.findall('.//sm:url/sm:loc', SITEMAP_NS)]

Важно: без namespace findall('.//url/loc') ничего не найдёт — sitemap использует XML namespace.

---

SEO-аудит страницы: 40+ проверок

Это самая объёмная функция. Один URL → один HTTP-запрос → парсинг HTML через BeautifulSoup → 40+ проверок, сгруппированных по категориям.

Архитектура аудита

class PageAuditor:
    def audit(self, url: str) -> dict:
        """Полный SEO-аудит страницы"""
        response = requests.get(url, timeout=15)
        soup = BeautifulSoup(response.text, 'html.parser')
        
        result = {'url': url, 'checks': [], 'score': 0}
        
        # Фазы аудита — каждая добавляет проверки в result['checks']
        phases = [
            self._check_meta,       # Title, Description, Canonical, Robots
            self._check_headings,   # H1-H6, иерархия, дубли
            self._check_links,      # Внутренние/внешние, nofollow, якоря
            self._check_images,     # Alt, размеры, lazy loading
            self._check_content,    # Длина, уникальные слова, читаемость
            self._check_technical,  # Скорость, размер HTML, сжатие
            self._check_security,   # HTTPS, HSTS, X-Frame-Options
            self._check_robots,     # robots.txt, sitemap.xml
            self._check_baden_baden # Фильтр Яндекса
        ]
        
        for phase in phases:
            phase(result, soup, url)
        
        result['score'] = self._calculate_score(result['checks'])
        return result

Формат проверки

Каждая проверка — словарь с единым форматом:

{
    'name': 'Title',
    'category': 'meta',
    'score': 'pass',          # pass | warning | fail | info
    'value': 'Купить окна в Москве — ОкнаПро',
    'recommendation': None     # или текст рекомендации
}

Это позволяет единообразно отображать результаты в Telegram-сообщении и HTML-отчёте.

Примеры проверок

Мета-теги:

• Title: длина 30-60 символов, наличие, не дубль H1

• Description: длина 120-160 символов

• Canonical: наличие, совпадение с текущим URL

• Robots: нет ли noindex

Заголовки:

• Ровно один H1

• H1 не пустой, не слишком длинный

• Правильная иерархия (нет H3 без H2)

Изображения:

• Все <img> имеют alt

• Нет пустых alt=""

• Есть width/height (CLS — Cumulative Layout Shift, сдвиг макета)

• Используется lazy loading

Безопасность:

• HTTPS

• HSTS заголовок

• X-Content-Type-Options

• X-Frame-Options

Подсчёт оценки

def _calculate_score(self, checks: list) -> int:
    """Оценка 0-100 на основе пройденных проверок"""
    scorable = [c for c in checks if c['score'] != 'info']
    if not scorable:
        return 0
    
    weights = {'pass': 1.0, 'warning': 0.5, 'fail': 0.0}
    total = sum(weights.get(c['score'], 0) for c in scorable)
    return round(total / len(scorable) * 100)

---

Анализ текста на фильтр «Баден-Баден»

«Баден-Баден» — фильтр Яндекса, который наказывает за переоптимизированные тексты. Я реализовал эвристический анализ по 11 параметрам.

Параметры анализа

class BadenBadenChecker:
    def check_text(self, text: str = None, url: str = None) -> dict:
        """Анализ текста на признаки переоптимизации"""
        
        # Извлекаем текст из HTML, если передан URL
        if url:
            text = self._extract_text(url)
        
        words = self._tokenize(text)
        word_freq = Counter(words)
        
        checks = []
        risk_score = 0
        
        # 1. Плотность топ-1 слова (>3% — warning, >5% — fail)
        # 2. Суммарная плотность топ-3 слов (>8% — warning, >12% — fail)
        # 3. Водность текста (вводные слова, паразиты)
        # 4. SEO-клише ("широкий ассортимент", "команда профессионалов")
        # 5. Ключевое слово в заголовках H1-H6
        # 6. Выделение ключей жирным (<b>, <strong>)
        # 7. Переспам ключа в Title
        # 8. Переспам ключа в Description
        # 9. Скрытый текст (display:none, font-size:0)
        # 10. Разнообразие лексики — TTR (Type-Token Ratio, отношение уникальных слов к общему числу)
        # 11. Неестественные конструкции ("купить X недорого москва")
        
        return {
            'risk_score': risk_score,      # 0-100
            'risk_level': self._get_level(risk_score),
            'checks': checks,
            'stats': {...},
            'top_words': word_freq.most_common(10)
        }

Водность текста

Водность — процент «пустых» слов в тексте. Я собрал список из 100+ водных слов:

WATER_WORDS = {
    'который', 'также', 'однако', 'поэтому', 'кроме',
    'именно', 'конечно', 'действительно', 'безусловно',
    'несомненно', 'разумеется', 'естественно', ...
}

def _check_water(self, words: list) -> float:
    """Процент водных слов в тексте"""
    water_count = sum(1 for w in words if w in WATER_WORDS)
    return water_count / len(words) * 100

Порог: >40% — предупреждение, >60% — проблема.

SEO-клише

50+ шаблонных фраз, которые сигнализируют о переоптимизации:

SEO_CLICHES = [
    'широкий ассортимент', 'команда профессионалов',
    'индивидуальный подход', 'оптимальное соотношение',
    'высокое качество', 'доступные цены',
    'в кратчайшие сроки', 'богатый опыт',
    'динамично развивающаяся компания', ...
]

Два режима работы

Пользователь может отправить URL (бот скачает и проанализирует) или вставить текст прямо в чат (для проверки до публикации).

---

Краулер битых ссылок

Краулер обходит сайт в ширину (BFS — Breadth-First Search), извлекает все ссылки и проверяет каждую на доступность.

Архитектура

class LinkCrawler:
    def __init__(self, max_pages=100, max_links=500, delay=0.5):
        self.max_pages = max_pages
        self.max_links = max_links
        self.delay = delay
    
    def crawl(self, start_url: str) -> dict:
        """Обход сайта в ширину (BFS)"""
        domain = urlparse(start_url).netloc
        visited = set()
        queue = deque([start_url])
        all_links = {}  # url → {status, source_pages}
        
        while queue and len(visited) < self.max_pages:
            url = queue.popleft()
            if url in visited:
                continue
            visited.add(url)
            
            # Скачиваем страницу
            soup = self._fetch(url)
            if not soup:
                continue
            
            # Извлекаем все ссылки: <a>, <img>, <link>, <script>
            links = self._extract_links(soup, url)
            
            for link in links:
                # Запоминаем, на какой странице нашли ссылку
                if link not in all_links:
                    all_links[link] = {'sources': []}
                all_links[link]['sources'].append(url)
                
                # Внутренние ссылки → добавляем в очередь
                if urlparse(link).netloc == domain:
                    queue.append(link)
            
            time.sleep(self.delay)  # Вежливый краулинг
        
        # Проверяем все найденные ссылки
        return self._check_links(all_links)

Ключевые решения

1. Ограничения — без них краулер может обходить сайт бесконечно. Лимиты: 100 страниц, 500 ссылок, задержка 0.5 сек.

2. Source tracking — для каждой битой ссылки запоминаем, на каких страницах она найдена. Это критически важно для исправления.

3. Типы ссылок — проверяем не только <a href>, но и <img src>, <link href>, <script src>. Битая картинка или CSS — тоже проблема.

---

Защита от SSRF-атак и DNS rebinding

Это самая важная часть для любого сервиса, который делает HTTP-запросы по пользовательским URL.

Что такое SSRF?

SSRF — атака, при которой злоумышленник заставляет сервер делать запросы к внутренним ресурсам:

# Злоумышленник отправляет:
check_url('https://localhost:5432')                     # БД
check_url('https://169.254.169.254/latest/meta-data')   # AWS metadata
check_url('https://192.168.1.1/admin')                  # Роутер

Многоуровневая защита

import socket
import ipaddress
from urllib.parse import urlparse

BLOCKED_HOSTS = ['localhost', '127.0.0.1', '0.0.0.0', '::1']

def is_safe_url(url: str) -> tuple[bool, str | None]:
    parsed = urlparse(url.lower())
    
    # 1. Только HTTP/HTTPS
    if parsed.scheme not in ['http', 'https']:
        return False, "Разрешены только HTTP/HTTPS"
    
    # 2. Блокируем известные опасные хосты
    if parsed.hostname in BLOCKED_HOSTS:
        return False, f"Доступ к {parsed.hostname} запрещён"
    
    # 3. Проверяем IP-адреса через ipaddress
    try:
        ip = ipaddress.ip_address(parsed.hostname)
        if ip.is_private or ip.is_loopback or ip.is_link_local:
            return False, "Доступ к внутренним IP запрещён"
    except ValueError:
        pass  # Это доменное имя — проверяем через DNS
    
    # 4. DNS rebinding защита
    try:
        resolved = socket.getaddrinfo(parsed.hostname, None)
        for _, _, _, _, sockaddr in resolved:
            ip = ipaddress.ip_address(sockaddr[0])
            if ip.is_private or ip.is_loopback or ip.is_reserved:
                return False, f"Домен резолвится во внутренний IP"
    except socket.gaierror:
        return False, "Не удалось разрешить домен"
    
    # 5. Ограничение длины
    if len(url) > 2048:
        return False, "URL слишком длинный"
    
    return True, None

Почему DNS rebinding — это важно

Без шага 4 злоумышленник может:

1. Зарегистрировать домен evil.example.com

2. Настроить DNS: evil.example.com → 8.8.8.8 (публичный IP)

3. Бот проверяет IP — всё ок, пропускает

4. Злоумышленник меняет DNS: evil.example.com → 127.0.0.1

5. Бот делает HTTP-запрос — и попадает на localhost

Решение: резолвим DNS сами через socket.getaddrinfo() и проверяем все IP-адреса, в которые резолвится домен, до того как делать HTTP-запрос.

Что ещё блокируем

# Октальная запись: 0177.0.0.1 = 127.0.0.1
# Десятичная: 2130706433 = 127.0.0.1
# IPv6: ::ffff:127.0.0.1
# Всё это ловит ipaddress.ip_address()

>>> ipaddress.ip_address('127.0.0.2').is_loopback
True
>>> ipaddress.ip_address('10.0.0.1').is_private
True
>>> ipaddress.ip_address('169.254.1.1').is_link_local
True

Модуль ipaddress из стандартной библиотеки покрывает все эти случаи.

---

Rate Limiting с Redis

Зачем

Без ограничений бот может быть использован для DDoS-атак (Distributed Denial of Service — распределённый отказ в обслуживании) — например, проверка тысяч URL конкурента, сканирование сетей или просто перегрузка сервера.

Реализация: счётчик с TTL (Time To Live — время жизни ключа)

class RedisRateLimiter:
    def __init__(self, redis_client):
        self.redis = redis_client
        self.limits = {
            'commands':  {'limit': 10,  'window': 60},
            'callbacks': {'limit': 20,  'window': 60},
            'messages':  {'limit': 30,  'window': 60},
        }
    
    def check(self, user_id: int, category: str) -> tuple[bool, str]:
        cfg = self.limits.get(category)
        if not cfg:
            return True, ""
        
        key = f"rl:{user_id}:{category}"
        current = self.redis.get(key)
        
        if current is None:
            self.redis.setex(key, cfg['window'], 1)
            return True, ""
        
        if int(current) >= cfg['limit']:
            ttl = self.redis.ttl(key)
            return False, f"Лимит превышен. Повторите через {ttl} сек."
        
        self.redis.incr(key)
        return True, ""

Принцип: SETEX key ttl 1 при первом запросе, INCR key при последующих. Ключ автоматически удаляется по TTL — никакой ручной очистки.

Где применяется

Rate limit стоит на каждой точке входа:

• Команды (/start, /help, /balance)

• Кнопки меню (проверка сайта, SEO, битые ссылки)

• Inline-кнопки (подтверждение, отмена, пагинация)

• Загрузка файлов

• Callback-запросы

---

Многопоточность и очередь проверок

Тяжёлые операции (краулинг, SEO-аудит, массовая проверка) выполняются в фоновых потоках:

import threading

def start_check(bot, user_id, chat_id, urls, speed_level):
    """Запуск проверки в фоновом потоке"""
    progress_msg = bot.send_message(chat_id, "⏳ Проверка запущена...")
    
    thread = threading.Thread(
        target=_perform_check,
        args=(bot, user_id, chat_id, urls, progress_msg.id),
        daemon=True
    )
    thread.start()

def _perform_check(bot, user_id, chat_id, urls, progress_msg_id):
    """Выполняется в отдельном потоке"""
    for i, url in enumerate(urls):
        result = check_url(url)
        
        # Обновляем прогресс каждые 10 URL
        if i % 10 == 0:
            bot.edit_message_text(
                f"⏳ Проверено: {i}/{len(urls)}",
                chat_id, progress_msg_id
            )
    
    # Генерируем отчёт и отправляем
    report = generate_report(results)
    bot.send_document(chat_id, report)

Очередь проверок

Чтобы сервер не захлебнулся от параллельных проверок, используется очередь с ограничением одновременных задач:

class CheckQueue:
    def __init__(self, max_concurrent=3):
        self.max_concurrent = max_concurrent
        self.active = {}
    
    def can_start_immediately(self) -> bool:
        return len(self.active) < self.max_concurrent
    
    def add_to_queue(self, user_id, data) -> int:
        """Возвращает позицию в очереди"""
        ...

Пользователь видит свою позицию в очереди и примерное время ожидания.

---

Генерация отчётов

HTML-отчёт с тёмной темой

SEO-аудит генерирует HTML-отчёт с оценкой, категориями проверок и рекомендациями:

def generate_audit_report(result: dict, user_id: int) -> str:
    """Генерация HTML-отчёта SEO-аудита"""
    score = result['score']
    checks = result['checks']
    
    # Группируем проверки по категориям
    categories = {}
    for check in checks:
        cat = check['category']
        categories.setdefault(cat, []).append(check)
    
    # Для каждой категории считаем процент пройденных
    for cat, cat_checks in categories.items():
        scorable = [c for c in cat_checks if c['score'] != 'info']
        passed = sum(1 for c in scorable if c['score'] == 'pass')
        cat_pct = round(passed / len(scorable) * 100) if scorable else 100
    
    # Генерируем HTML с CSS (тёмная тема, адаптивная вёрстка)
    html = f"""<!DOCTYPE html>
    <html lang="ru">
    <head>
        <meta charset="UTF-8">
        <style>
            body {{ background: #0f172a; color: #e2e8f0; }}
            .score-circle {{ border: 4px solid {score_color}; }}
            .check-pass {{ color: #22c55e; }}
            .check-fail {{ color: #ef4444; }}
            .rec {{ border-left: 3px solid #3b82f6; }}
        </style>
    </head>
    ...
    """

Защита от XSS в отчётах

URL от пользователей могут содержать вредоносный код:

https://example.com/<script>alert('XSS')</script>

Решение: все пользовательские данные проходят через html.escape():

import html

def _esc(text: str) -> str:
    return html.escape(str(text), quote=True) if text else ''

# В шаблоне:
f"<td>{_esc(check['value'])}</td>"

Проверка владельца отчёта

Отчёты доступны по check_id через inline-кнопки. Без проверки владельца любой пользователь мог бы скачать чужой отчёт, подставив ID:

@bot.callback_query_handler(func=lambda call: call.data.startswith('download_'))
def download_report(call):
    check_id = int(call.data.split('_')[-1])
    
    # Проверяем, что отчёт принадлежит этому пользователю
    check = db.get_site_check(check_id)
    if not check or check.user_id != call.from_user.id:
        bot.answer_callback_query(call.id, "Доступ запрещён")
        return
    
    # Отправляем отчёт
    ...

Это классическая IDOR-уязвимость (Insecure Direct Object Reference — небезопасная прямая ссылка на объект), которую легко пропустить.

---

Уроки и выводы

Что я узнал за время разработки

1. SSRF — реальная угроза, DNS rebinding — ещё реальнее

Через неделю после запуска появились попытки проверить localhost и 127.0.0.1. Базовая защита (блокировка по строке) недостаточна — нужен DNS-резолвинг с проверкой IP.

2. Rate limiting нужен с первого дня

Без лимитов один пользователь отправил 10 000 URL за час. Redis + счётчик с TTL — простое и надёжное решение.

3. Inline-кнопки лучше текстового ввода

Первая версия просила «напишите да/нет» для подтверждения. Пользователи ошибались, бот зависал. Inline-кнопки решили проблему полностью.

4. Проверка владельца — везде

Любой callback с check_id или report_id должен проверять, что ресурс принадлежит текущему пользователю. Это легко забыть и сложно заметить при тестировании.

5. Retry повышает точность в 10 раз

До retry: 5% URL помечались как недоступные из-за временных ошибок. После (exponential backoff, 3 попытки): 0.5%.

6. Фоновые потоки + прогресс = хороший UX

Тяжёлые операции в threading.Thread(daemon=True) + обновление прогресса через edit_message_text каждые N шагов. Пользователь видит, что бот работает, а не завис.

Что можно улучшить

• Lighthouse интеграция — проверка Core Web Vitals

• Webhook вместо polling — для масштабирования

• Celery вместо threading — для надёжной очереди задач

• Кэширование DNS — чтобы не резолвить один домен дважды

---

Полезные ссылки

• OWASP — SSRF Prevention

• RFC 7231 — HTTP Redirects

• Redis — Rate Limiting Pattern

• pyTelegramBotAPI — FSM

• Python ipaddress

---

Заключение

Telegram-бот для SEO-аудита — это не только удобный инструмент, но и хороший полигон для практики:

• Безопасность: SSRF, DNS rebinding, IDOR, XSS (Cross-Site Scripting — межсайтовый скриптинг) — всё это актуально для любого сервиса, принимающего URL от пользователей

• Архитектура: модульность, FSM, фоновые потоки, очереди

• Инфраструктура: Redis, PostgreSQL, Windows Service

Все описанные техники применимы не только к ботам, но и к любым веб-сервисам, которые работают с пользовательскими URL.

Если есть вопросы — пишите в комментариях.

---

P.S. Код обобщён для статьи. Реальная реализация содержит дополнительные проверки и обработку краевых случаев.