Twitter-бот с reply game: автопродвижение без API

Есть такая ловушка при автоматизации Twitter: сделаешь бота, который просто постит твиты — и думаешь, что дело сделано. Контент идёт, очередь работает, Telegram шлёт уведомления. Красота. Но подписчики не растут, охваты не растут, и вообще ничего не происходит. Потому что Twitter — это не блог. Это разговор. И если ты только говоришь, но не участвуешь в чужих разговорах, алгоритм тебя просто не замечает.

Именно с этого осознания начался следующий этап развития моего проекта twitter-manager — автономного бота для продвижения аккаунта.

Что было до: просто постер

Исходно twitter-manager умел следующее:

• Собирать контент из RSS-лент
• Оценивать его через Claude на релевантность
• Переписывать в твиттер-стиле
• Постить через Playwright (browser automation — никакого платного Twitter API)
• Слать уведомления в Telegram

Квоты были зашиты жёстко: до 17 твитов в день, 7 реплаев на упоминания, 3 цитаты, 15 лайков. Проблема в том, что это был реактивный бот — он только отвечал на упоминания и постил своё. Никакого проактивного участия в чужих тредах.

Когда я спросил себя «а что реально двигает аккаунт в Twitter?» — ответ оказался очевидным: reply game. Умные, содержательные ответы под твитами авторитетных аккаунтов в нише. Это и есть главный рычаг органического роста.

Проблема дороговизны Twitter API

Здесь сразу возникает техническая развилка. Twitter (теперь X) закрутил гайки на API до неприличия — базовый тариф для чтения тайmlайнов стоит сотни долларов в месяц. Для indie-проекта это нереально.

Решение, которое уже работало для постинга — Playwright browser automation — нужно было расширить на операции чтения. Парсить страницы профилей, уведомления, тайmlайны прямо через браузер. Zero API cost.

Архитектурно это означало расширить BrowserClient тремя новыми методами чтения и переключить все engagement-модули с TwitterClient (API) на BrowserClient (browser).

Дизайн нового движка

Перед реализацией я прошёл через полный цикл brainstorming → дизайн-документ → план реализации. Это важный момент: не бросаться кодить сразу, а сначала зафиксировать архитектурные решения.

Дизайн разбился на три секции:

Секция 1: Browser Read Methods

Добавляем в browser_client.py три новых метода:

async def get_mentions(self, since_id=None) -> list[dict]:
    """Парсим /notifications/mentions через browser"""
    await self.page.goto('https://twitter.com/notifications/mentions')
    await self.page.wait_for_selector('[data-testid="tweet"]')
    tweets = await self.page.query_selector_all('[data-testid="tweet"]')
    results = []
    for tweet in tweets:
        tweet_data = await self._parse_tweet_element(tweet)
        if since_id and tweet_data['tweet_id'] <= since_id:
            break
        results.append(tweet_data)
    return results
 
async def get_user_timeline(self, username: str, max_results: int = 5) -> list[dict]:
    """Парсим профиль пользователя"""
    await self.page.goto(f'https://twitter.com/{username}')
    await self.page.wait_for_selector('[data-testid="tweet"]')
    tweets = await self.page.query_selector_all('[data-testid="tweet"]')
    return [await self._parse_tweet_element(t) for t in tweets[:max_results]]
 
async def like_tweet(self, tweet_id: str) -> bool:
    """Кликаем лайк через UI"""
    like_btn = await self.page.query_selector(
        f'[data-testid="like"][href*="{tweet_id}"]'
    )
    if like_btn:
        await like_btn.click()
        return True
    return False

Секция 2: Reply Engine — проактивные реплаи

Это главная новинка — модуль reply_engine.py. В отличие от mention_monitor.py (реактивный, отвечает на входящие упоминания), ReplyEngine проактивно обходит мониторируемые аккаунты и ищет возможности для умных ответов.

Логика работы:

1. Обходим список мониторируемых аккаунтов (расширяем с 11 до 30-50)
2. Для каждого — get_user_timeline(), берём свежие твиты (< 2 часов)
3. Фильтруем: пропускаем ретвиты, рекламу, уже обработанные
4. Claude скорит каждый твит на «reply-worthiness» по шкале 0-10
5. Порог 7+ — генерируем умный реплай в стиле аккаунта
6. Постим через reply_to_tweet(), логируем в БД

class ReplyEngine:
    def __init__(self, browser_client, claude_client, db):
        self.browser = browser_client
        self.claude = claude_client
        self.db = db
        self.daily_limit = 20  # проактивных реплаев в день
    
    async def run(self) -> int:
        """Основной цикл. Возвращает количество posted replies."""
        posted = 0
        today_count = self.db.get_reply_count_today()
        
        if today_count >= self.daily_limit:
            return 0
        
        for username in self._get_monitored_accounts():
            if posted >= (self.daily_limit - today_count):
                break
            
            tweets = await self.browser.get_user_timeline(username, max_results=10)
            fresh_tweets = self._filter_fresh(tweets, max_age_hours=2)
            
            for tweet in fresh_tweets:
                if self.db.already_replied(tweet['tweet_id']):
                    continue
                
                score = await self._score_reply_worthiness(tweet)
                if score >= 7:
                    reply_text = await self._generate_reply(tweet)
                    success = await self.browser.reply_to_tweet(
                        tweet['tweet_id'], reply_text
                    )
                    if success:
                        self.db.mark_replied(tweet['tweet_id'])
                        posted += 1
        
        return posted

Секция 3: Надёжность и ops

Браузерная автоматизация — штука хрупкая. Страницы падают, таймауты случаются, сеть ведёт себя непредсказуемо. Поэтому третья секция дизайна — это retry-логика и health monitoring.

Retry декоратор:

def browser_retry(max_attempts: int = 3, backoff: float = 2.0):
    def decorator(func):
        @functools.wraps(func)
        async def wrapper(*args, **kwargs):
            last_error = None
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except (TimeoutError, PlaywrightError) as e:
                    last_error = e
                    if attempt < max_attempts - 1:
                        await asyncio.sleep(backoff ** attempt)
            raise last_error
        return wrapper
    return decorator

Новый модуль health.py — heartbeat, детекция зависших джобов, трекинг ошибок:

class ServiceHealth:
    def record_job_success(self, job_name: str):
        self.db.execute(
            "UPDATE job_health SET last_success=?, consecutive_failures=0 WHERE job=?",
            [datetime.now(timezone.utc).isoformat(), job_name]
        )
    
    def check_stale(self, job_name: str, max_age_minutes: int) -> bool:
        row = self.db.fetchone(
            "SELECT last_success FROM job_health WHERE job=?", [job_name]
        )
        if not row:
            return True
        last = datetime.fromisoformat(row['last_success'])
        age = datetime.now(timezone.utc) - last
        return age.total_seconds() > max_age_minutes * 60

Параллельная реализация через агентов

Одна из интересных деталей этого проекта — способ реализации. Вместо линейного «написал задачу → выполнил → следующая» я использовал параллельных агентов в изолированных git worktree.

План из 11 задач разбился на 4 группы, которые можно выполнять параллельно без конфликтов:

• Группа A: persistent browser context + retry decorator
• Группа B: browser read methods (timeline, mentions, like)
• Группа C: адаптация engagement.py и mention_monitor.py
• Группа D: модуль health.py

Каждая группа — отдельный worktree, отдельная ветка, независимый агент. После завершения всех четырёх — merge в master, затем последовательные задачи 9-11 (structured logging, интеграция в main.py, финальные тесты).

Результат: 107 тестов, 0 failures, 0 warnings.

Проблемы, которые пришлось решить

Datetime timezone hell

Классическая Python-ловушка: datetime.utcnow() возвращает naive datetime (без timezone info), а datetime.now(timezone.utc) — aware. Когда mixing происходит в сравнениях — падаешь с TypeError.

В health.py это проявилось при проверке stale джобов:

# Было (падало):
last = datetime.fromisoformat(row['last_success'])  # может быть naive
now = datetime.now(timezone.utc)  # aware
age = now - last  # TypeError!
 
# Стало:
last = datetime.fromisoformat(row['last_success'])  # aware, т.к. хранится с +00:00
now = datetime.now(timezone.utc)  # aware
age = now - last  # работает

Решение: всегда хранить в ISO формате с timezone (datetime.now(timezone.utc).isoformat() даёт 2026-03-02T14:30:00+00:00), тогда fromisoformat вернёт aware datetime.

RotatingFileHandler в тестах

Тест для structured logging падал потому что logging.basicConfig() не добавляет handlers если логирование уже настроено — а pytest настраивает его раньше. Пришлось переписать тест:

# Было:
def test_log_file_handler():
    logging.basicConfig(handlers=[RotatingFileHandler('test.log')])
    assert len(logging.root.handlers) > 0  # падало!
 
# Стало:
def test_log_file_handler(tmp_path):
    logger = logging.getLogger('test_rotating')
    handler = RotatingFileHandler(tmp_path / 'test.log')
    logger.addHandler(handler)
    assert any(isinstance(h, RotatingFileHandler) for h in logger.handlers)

Зависшая сессия на AskUserQuestion

Отдельный эпизод, который стоит упомянуть как операционный урок: одна из сессий планирования зависла на этапе AskUserQuestion — UI не отрендерил промпт, сессия просто висела 20 минут без активности.

Важный момент: ничего не потерялось. Оба документа (дизайн + план) были уже закоммичены в git до зависания. Просто закрыл ту сессию, открыл новую и продолжил с того места, где остановился. Git как source of truth — именно для таких случаев.

Итоговая архитектура

twitter-manager/
├── browser_client.py    # +persistent context, retry, get_timeline, like, get_mentions
├── engagement.py        # переключён на BrowserClient (без TwitterClient)
├── mention_monitor.py   # browser .author support, упрощённый _is_bot()
├── reply_engine.py      # НОВЫЙ: проактивный reply game
├── health.py            # НОВЫЙ: heartbeat, stale detection, failure tracking
├── main.py              # +новые jobs: reply_game, health_check
└── docs/plans/
    ├── 2026-03-02-engagement-and-reliability-design.md
    └── 2026-03-02-engagement-and-reliability-plan.md

Квоты после обновления:

| Активность | Было | Стало | |------------|------|-------| | Свои твиты | 17/день | 17/день | | Реплаи на упоминания | 7/день | 7/день | | Цитаты | 3/день | 3/день | | Лайки | 15/день | 15/день | | Проактивные реплаи | 0/день | 20/день |

Что дало добавление reply game

Главный инсайт этого проекта: Twitter-рост = участие в чужих разговорах, а не просто публикация своего контента. Алгоритм поощряет аккаунты, которые генерируют engagement, а не просто broadcast.

Проактивные реплаи — это 20 дополнительных точек касания с аудиторией в день. При правильном скоринге (Claude оценивает 0-10, порог 7+) это не спам, а реально полезные комментарии под релевантными твитами авторитетов в нише.

Для ресурсов по технологиям, использованным в проекте: Playwright Python документация, Anthropic Claude API, Python asyncio, Python logging HOWTO, pytest документация.

Выводы и уроки

Первый урок: browser automation как замена платному API — не хак и не временное решение. Для задач, где скорость не критична (не real-time), это полноценная архитектура. Playwright держит сессию, умеет ждать элементы, справляется с динамическим контентом. Минус — хрупкость к изменениям вёрстки Twitter. Плюс — zero cost на чтение данных.

Второй урок: дизайн-документ перед кодом экономит время. Звучит банально, но именно структурированный дизайн позволил распараллелить реализацию на 4 независимых агента. Без явного разделения зон ответственности параллельная работа превращается в merge-hell.

Третий урок: TDD в browser automation — это не про юнит-тесты UI (они хрупкие), а про тесты логики вокруг browser-вызовов. Моки BrowserClient дают стабильный тест-сьют: 107 тестов, которые запускаются за секунды без реального браузера.

Четвёртый урок: health monitoring нужен с самого начала, а не когда что-то сломалось. Модуль health.py с heartbeat и stale detection — это не overengineering, это observability минимального уровня для любого бота, который работает в фоне 24/7. Когда джоб тихо перестаёт работать, а ты узнаёшь об этом через неделю по метрикам — это провал операционного мониторинга.