Twitter Manager: автоматический engagement через браузер

Запустить бота, который постит твиты — это полдела. Настоящий рост в Twitter начинается тогда, когда ты не просто вещаешь в пустоту, а активно участвуешь в чужих дискуссиях. Именно с этого понимания начался следующий большой этап в развитии моего twitter-manager.

Спойлер: всё получилось, 107 тестов проходят, бот теперь лайкает, отвечает на упоминания и активно участвует в треды через browser automation — без единого платного API-вызова.

Контекст проекта

Twitter Manager — это автономный Python-бот для управления моим Twitter-аккаунтом. Базовая версия умела:

• Собирать контент из RSS-лент
• Оценивать его через Claude (Anthropic)
• Переписывать и постить твиты
• Уведомлять меня в Telegram о каждом действии
• Соблюдать дневные квоты (17 твитов, 7 реплаев, 3 цитаты, 15 лайков)

Вся публикация шла через Playwright — браузерную автоматизацию. Ноль расходов на Twitter API, который в 2023-2024 годах резко подорожал и стал практически недоступным для indie-разработчиков.

Но была проблема: бот умел только говорить, не умел слушать и реагировать. Модули для engagement (engagement.py, mention_monitor.py) были написаны, но завязаны на платный Twitter API для чтения. В итоге они просто висели отключёнными.

Проблема: engagement без API

Twitter-алгоритм устроен так, что простые посты дают минимальный охват. Реальный рост даёт reply game — умные ответы на твиты крупных аккаунтов в твоей нише. Это создаёт видимость, привлекает подписчиков, строит репутацию.

Для этого нужно уметь:

1. Читать чужие таймлайны (чтобы находить интересные твиты)
2. Читать упоминания (чтобы отвечать когда тебя тегают)
3. Лайкать твиты (сигнал алгоритму)
4. Постить реплаи

Пункты 1-3 требуют чтения данных из Twitter, что через официальный API стоит денег. А через браузер — ничего не стоит.

Решение напрашивалось само: расширить BrowserClient, который уже умел постить, возможностями чтения.

Архитектура решения

Перед реализацией я прошёл через структурированный brainstorming-процесс (у меня для этого есть специальный скилл в Claude). Рассматривали два подхода:

Подход A: Расширить BrowserClient — добавить методы чтения в существующий browser_client.py, переключить engagement.py и mention_monitor.py с TwitterClient на BrowserClient.

Подход B: Отдельный read-клиент — создать BrowserReader как отдельный модуль.

Выбрали Подход A. Причина простая: не нужно плодить абстракции. YAGNI в действии.

Новые методы BrowserClient

async def get_mentions(self, since_id=None) -> list[dict]:
    """Парсим /notifications/mentions через браузер"""
    await self.page.goto('https://twitter.com/notifications/mentions')
    await self.page.wait_for_selector('[data-testid="tweet"]', timeout=10000)
    
    tweets = []
    elements = await self.page.query_selector_all('[data-testid="tweet"]')
    
    for el in elements:
        tweet_data = await self._extract_tweet_data(el)
        if since_id and tweet_data['tweet_id'] <= since_id:
            break
        tweets.append(tweet_data)
    
    return tweets
 
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"]', timeout=10000)
    
    tweets = []
    elements = await self.page.query_selector_all('[data-testid="tweet"]')
    
    for el in elements[:max_results]:
        tweet_data = await self._extract_tweet_data(el)
        if not tweet_data.get('is_retweet'):
            tweets.append(tweet_data)
    
    return tweets
 
async def like_tweet(self, tweet_id: str) -> bool:
    """Лайкаем твит через UI"""
    try:
        like_btn = await self.page.query_selector(
            f'[data-testid="like"][aria-label*="{tweet_id}"]'
        )
        if like_btn:
            await like_btn.click()
            return True
    except Exception as e:
        logger.error(f"Like failed for {tweet_id}: {e}")
    return False

Persistent Browser Context

Раньше браузер создавался и убивался на каждый вызов. Это медленно (3-5 секунд на запуск) и ненадёжно. Переделал на persistent context:

class BrowserClient:
    def __init__(self):
        self._playwright = None
        self._browser = None
        self._context = None
        self._page = None
    
    async def __aenter__(self):
        self._playwright = await async_playwright().start()
        self._browser = await self._playwright.chromium.launch(headless=True)
        # Загружаем сохранённую сессию
        self._context = await self._browser.new_context(
            storage_state='twitter_session.json'
        )
        self._page = await self._context.new_page()
        return self
    
    async def __aexit__(self, *args):
        await self._context.storage_state(path='twitter_session.json')
        await self._browser.close()
        await self._playwright.stop()

Это даёт два преимущества: сессия сохраняется между запусками (не нужно логиниться каждый раз), и браузер не перезапускается при каждой операции.

Retry Decorator

Сеть нестабильна. Playwright падает. Добавили универсальный retry:

def browser_retry(max_attempts: int = 3, backoff: float = 2.0):
    def decorator(func):
        @wraps(func)
        async def wrapper(*args, **kwargs):
            last_exception = None
            for attempt in range(max_attempts):
                try:
                    return await func(*args, **kwargs)
                except (TimeoutError, Error) as e:
                    last_exception = e
                    if attempt < max_attempts - 1:
                        wait_time = backoff ** attempt
                        logger.warning(
                            f"{func.__name__} attempt {attempt+1} failed, "
                            f"retrying in {wait_time}s: {e}"
                        )
                        await asyncio.sleep(wait_time)
            raise last_exception
        return wrapper
    return decorator

Exponential backoff: первый retry через 1 секунду, второй через 2, третий через 4. Если все три провалились — Telegram-алерт и пропуск задачи.

Reply Game: новый модуль

Помимо реактивных ответов на упоминания, добавили проактивный режим — reply_engine.py. Это основной двигатель роста.

Логика:

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

async def score_tweet_for_reply(tweet: dict, claude_client) -> int:
    prompt = f"""Rate this tweet for reply opportunity (0-10):
    
 Tweet: {tweet['text']}
 Author followers: {tweet.get('author_followers', 0)}
 
 Consider: engagement potential, topic relevance (AI/tech/startups),
 conversation opportunity, recency.
 
 Return only a number 0-10."""
    
    response = await claude_client.message(prompt)
    try:
        return int(response.strip())
    except ValueError:
        return 0

Важный момент: реплаи не должны быть спамом. Claude генерирует реплаи с чётким промптом — добавляй ценность, будь конкретным, не будь банальным. Никаких «Great point!» и «Totally agree!».

Health Monitoring

Новый модуль health.py — потому что у автономного бота должен быть способ сказать «я живой» и «вот что произошло».

class ServiceHealth:
    def record_job_success(self, job_name: str):
        self._state[job_name] = {
            'last_success': datetime.now(timezone.utc).isoformat(),
            'consecutive_failures': 0
        }
    
    def record_job_failure(self, job_name: str, error: str):
        state = self._state.get(job_name, {'consecutive_failures': 0})
        state['consecutive_failures'] = state.get('consecutive_failures', 0) + 1
        state['last_error'] = error
        self._state[job_name] = state
        
        if state['consecutive_failures'] >= 3:
            self._alert_telegram(f"Job {job_name} failed {state['consecutive_failures']} times")
    
    def check_stale(self, job_name: str, max_age_hours: float = 2.0) -> bool:
        state = self._state.get(job_name, {})
        last_success = state.get('last_success')
        if not last_success:
            return True
        
        last_dt = datetime.fromisoformat(last_success)
        age = datetime.now(timezone.utc) - last_dt
        return age.total_seconds() > max_age_hours * 3600

Если джоб не запускался 2+ часа — алерт в Telegram. Если 3 фейла подряд — алерт. Просто и эффективно.

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

Интересный момент в процессе разработки: план из 11 задач реализовывался через параллельных Claude-агентов в изолированных git worktree. Четыре агента работали одновременно над независимыми частями, потом изменения сливались в master.

Это позволило значительно ускорить реализацию — задачи, у которых не было зависимостей между собой, выполнялись параллельно. После слияния — проверка конфликтов и интеграционные тесты.

Результаты

После реализации всех 11 задач:

107 tests passed, 0 warnings

Что изменилось в архитектуре:

| Компонент | До | После | |-----------|-----|-------| | browser_client.py | Только запись | + get_timeline, get_mentions, like_tweet, persistent context, retry | | engagement.py | Отключён (нужен платный API) | Работает через браузер | | mention_monitor.py | Отключён | Реактивные ответы на упоминания | | reply_engine.py | Не существовал | Проактивный reply game | | health.py | Не существовал | Heartbeat, stale detection, failure tracking |

Дневная активность бота теперь включает:

• До 17 оригинальных твитов из RSS
• До 15 лайков в день
• До 7 ответов на упоминания
• До 3 цитат от monitored accounts
• Плюс проактивные реплаи в рамках новых квот

Подводные камни которые встретились

datetime.utcnow() deprecation. Python 3.12 активно ворнит о utcnow(). Заменил везде на datetime.now(timezone.utc). Казалось бы мелочь, но при сравнении aware и naive datetime получаешь TypeError в рантайме. Причём тесты могут проходить, а в проде падать — потому что тест хранит значение из datetime.now(timezone.utc) (aware), а legacy код читает и сравнивает с datetime.utcnow() (naive).

RotatingFileHandler в pytest. Тест на structured logging с RotatingFileHandler падал, потому что logging.basicConfig() не добавляет хендлеры если логирование уже настроено — а pytest настраивает его раньше. Решение: явно добавлять хендлер через logger.addHandler() вместо basicConfig.

Duck typing вместо жёстких импортов. engagement.py изначально импортировал TwitterClient напрямую. Переделал на duck typing — модуль принимает любой объект с нужными методами. Это позволило передавать BrowserClient без изменения сигнатур функций и сохранить обратную совместимость.

Зависшая сессия Claude. Один из сеансов разработки завис на AskUserQuestion — AI-агент ждал ответа в UI, но промпт не отрисовался. Это известный баг в Cursor/Claude-интеграции. Всё что было сделано — сохранено в git, ничего не потеряно. Вывод: коммить дизайн-документы и планы сразу, до начала реализации.

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

Главный урок этого проекта: browser automation — это полноценная альтернатива платным API для задач автоматизации собственных действий. Playwright даёт доступ ко всему что видит пользователь, без ограничений rate limits и без ежемесячной платы. Цена — хрупкость к изменениям верстки сайта. Но для Twitter, где UI меняется редко, это приемлемый компромисс.

Второй урок: структура «brainstorm → дизайн-документ → план → реализация» экономит время даже когда работаешь в одиночку. Когда сессия зависла, я мог продолжить с другой сессии не теряя контекст — именно потому что дизайн и план были закоммичены. Документация — это не бюрократия, это защита от потери работы.

Третий урок: TDD при работе с browser automation особенно важен. Playwright-методы можно мокировать, и тесты дают уверенность что изменения в одном месте не сломали другое. 107 тестов — это не паранойя, это возможность рефакторить без страха.

Четвёртый урок: параллельные агенты в git worktree — мощный инструмент для ускорения разработки когда задачи независимы. Четыре агента выполнили работу которая последовательно заняла бы в 3-4 раза больше времени. Главное — правильно разбить задачи чтобы минимизировать конфликты при слиянии.

Пятый, и самый практичный: мониторинг нужен с первого дня. health.py — это не nice-to-have, это базовая инфраструктура автономного сервиса. Без него ты не знаешь что бот упал пока не посмотришь вручную. С ним — Telegram-алерт через 2 часа после первого тихого падения.

Следующий шаг — расширить список monitored accounts до 50, поднастроить промпты для reply game, и посмотреть на реальные метрики роста аккаунта через месяц работы.

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

• Playwright Python документация — всё про browser automation
• Anthropic Claude API — документация по API которое используется для scoring и генерации контента
• Python asyncio — основа для async/await паттернов в проекте
• pytest документация — фреймворк для 107 тестов
• Python logging HOWTO — structured logging и RotatingFileHandler