Скрэппинг лидов в Telegram: полное техническое руководство по извлечению квалифицированных лидов

Скрэппинг лидов в Telegram: полное техническое руководство

Введение

Скрэппинг лидов в Telegram принципиально отличается от сканирования LinkedIn или email. Открытый API платформы, структурированное членство в группах и доступная история сообщений делают её самой сканируемой крупной платформой для B2B генерации лидов.

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

Часть 1: Настройка инфраструктуры

Конфигурация доступа к API

Telegram предоставляет два уровня API:

Bot API - Ограниченный, но стабильный. Боты могут обращаться только к группам, в которые они добавлены. User API (MTProto) - Полный доступ через личные аккаунты. Это основной инструмент для скрэппинга лидов.

Настройка Telethon

pip install telethon

from telethon import TelegramClient
api_id = 12345678
api_hash = ''your_api_hash_here''
phone = ''+1234567890''
client = TelegramClient(f''sessions/{phone}'', api_id, api_hash)
async def initialize():
    await client.start(phone)
    me = await client.get_me()
    print(f''Залогинен как: {me.first_name}'')
with client:
    client.loop.run_until_complete(initialize())

Настройка нескольких аккаунтов

class AccountPool:
    def __init__(self, config_path=''accounts.json''):
        with open(config_path) as f:
            self.accounts = json.load(f)
        self.clients = {}
        self.usage = {}
    
    def get_best_account(self):
        min_usage = float(''inf'')
        best_index = 0
        for i, account in enumerate(self.accounts):
            usage = self.usage.get(i, 0)
            if usage < min_usage and usage < account.get(''daily_limit'', 200):
                min_usage = usage
                best_index = i
        return best_index

Часть 2: Обнаружение групп

Автоматическое обнаружение групп

class GroupDiscovery:
    def __init__(self, client):
        self.client = client
    
    async def evaluate_group(self, group_username):
        entity = await self.client.get_entity(group_username)
        full_chat = await self.client(GetFullChatRequest(entity))
        participants_count = full_chat.full_chat.participants_count or 0
        
        messages = []
        async for msg in self.client.iter_messages(entity, limit=100):
            messages.append(msg)
        
        if not messages:
            return None
        
        oldest_msg = messages[-1].date
        newest_msg = messages[0].date
        time_span_days = max((newest_msg - oldest_msg).days, 1)
        messages_per_day = len(messages) / time_span_days
        
        score = 0
        if 5000 <= participants_count <= 50000:
            score += 30
        elif 1000 <= participants_count <= 5000:
            score += 20
        
        score += min(messages_per_day / 10, 30)
        
        return {
            ''username'': group_username,
            ''participants_count'': participants_count,
            ''messages_per_day'': messages_per_day,
            ''overall_score'': score
        }

Часть 3: Извлечение участников

Базовое сканирование участников

async def scrape_group_members(group_username, limit=1000):
    client = TelegramClient(''session'', api_id, api_hash)
    await client.start()
    
    members = []
    entity = await client.get_entity(group_username)
    
    offset = 0
    batch_size = 100
    
    while offset < limit:
        result = await client(GetParticipantsRequest(
            channel=entity,
            filter=ChannelParticipantsSearch(''''),
            offset=offset,
            limit=batch_size,
            hash=0
        ))
        
        if not result.users:
            break
        
        for user in result.users:
            if not user.bot and not user.deleted:
                members.append({
                    ''user_id'': user.id,
                    ''username'': user.username,
                    ''first_name'': user.first_name,
                    ''last_name'': user.last_name,
                    ''phone'': user.phone,
                    ''is_premium'': getattr(user, ''premium'', False),
                    ''group'': group_username
                })
        
        offset += len(result.users)
        await asyncio.sleep(2)
    
    await client.disconnect()
    return members

Часть 4: Извлечение лидов по комментариям

Детекция намерений

class CommentExtractor:
    def __init__(self, client):
        self.client = client
        self.intent_keywords = {
            ''high'': [''recommend'', ''suggest'', ''looking for'', ''need help'',
                     ''рекомендуете'', ''посоветуйте'', ''ищу'', ''нужна помощь''],
            ''medium'': [''how to'', ''what is'', ''best way'',
                       ''как сделать'', ''что такое'', ''лучший способ''],
            ''low'': [''interesting'', ''tell me more'',
                    ''интересно'', ''расскажи больше'']
        }
    
    async def extract_comments_with_intent(self, group_username, hours_back=24):
        entity = await self.client.get_entity(group_username)
        cutoff_time = datetime.now() - timedelta(hours=hours_back)
        
        leads = []
        async for message in self.client.iter_messages(entity, offset_date=cutoff_time):
            if not message.text or message.reply_to is None:
                continue
            
            text = message.text.lower()
            intent_level = self.detect_intent(text)
            if not intent_level:
                continue
            
            sender = await message.get_sender()
            if sender.bot or sender.deleted:
                continue
            
            leads.append({
                ''user_id'': sender.id,
                ''username'': sender.username,
                ''first_name'': sender.first_name,
                ''bio'': sender.about or '''',
                ''comment'': message.text[:500],
                ''intent_level'': intent_level,
                ''timestamp'': message.date.isoformat(),
                ''group'': group_username
            })
        
        return leads
    
    def detect_intent(self, text):
        for level, keywords in self.intent_keywords.items():
            for keyword in keywords:
                if keyword in text:
                    return level
        return None

Часть 5: Обогащение профилей

async def enrich_profile(user, client):
    profile = {
        ''telegram_id'': user.id,
        ''username'': user.username,
        ''first_name'': user.first_name,
        ''last_name'': user.last_name,
        ''phone'': user.phone,
        ''bio'': user.about or '''',
        ''is_premium'': getattr(user, ''premium'', False),
        ''has_photo'': bool(user.photo),
    }
    
    bio_text = profile[''bio''].lower()
    
    company_patterns = [r''at (\w+)'', r''@ (\w+)'', r''working at (\w+)'']
    for pattern in company_patterns:
        match = re.search(pattern, bio_text)
        if match:
            profile[''company''] = match.group(1)
            break
    
    role_keywords = {
        ''cto'': [''cto'', ''technical director'', ''технический директор''],
        ''ceo'': [''ceo'', ''founder'', ''основатель''],
        ''cmo'': [''cmo'', ''marketing director'', ''директор по маркетингу''],
        ''developer'': [''developer'', ''engineer'', ''разработчик'']
    }
    for role, keywords in role_keywords.items():
        if any(kw in bio_text for kw in keywords):
            profile[''role''] = role
            break
    
    return profile

Часть 6: Оценка намерений

class IntentScorer:
    def __init__(self):
        self.weights = {
            ''comment_intent'': 0.25,
            ''profile_match'': 0.20,
            ''activity_level'': 0.15,
            ''recency'': 0.15,
            ''question_asked'': 0.15,
            ''company_match'': 0.10
        }
    
    def calculate_intent_score(self, lead_data):
        scores = {}
        
        if lead_data.get(''intent_level'') == ''high'':
            scores[''comment_intent''] = 90
        elif lead_data.get(''intent_level'') == ''medium'':
            scores[''comment_intent''] = 60
        else:
            scores[''comment_intent''] = 10
        
        scores[''profile_match''] = self.calculate_profile_match(lead_data)
        scores[''activity_level''] = min(lead_data.get(''message_count'', 0) * 10, 100)
        scores[''question_asked''] = 80 if lead_data.get(''question'') else 20
        scores[''company_match''] = 70 if lead_data.get(''company_match'') else 30
        
        return sum(scores[k] * self.weights[k] for k in self.weights)

Часть 7: Архитектура пайплайна данных

[Обнаружение групп] -> [Оценка групп] -> [Извлечение участников] -> [Обогащение профилей]
        |                    |                   |                    |
   API TGStat          NLP-анализ          API Telethon        Внешние API
        |                    |                   |                    |
  [Майнинг комментариев] -> [Детекция намерений] -> [Поведенческий анализ] -> [Оценка намерений]
        |                    |                   |                    |
   API сообщений      Ключевое слово     Детекция паттернов     Взвешенная оценка
        |                    |                   |                    |
  [Хранение данных] -> [Квалификация лидов] -> [CRM интеграция] -> [Готовность к outreach]
        |                    |                   |                    |
   PostgreSQL         Порог оценки      Webhook/API         Готовность кампании

Схема базы данных

CREATE TABLE leads (
    id SERIAL PRIMARY KEY,
    telegram_id BIGINT UNIQUE,
    username VARCHAR(255),
    first_name VARCHAR(255),
    last_name VARCHAR(255),
    phone VARCHAR(50),
    bio TEXT,
    is_premium BOOLEAN DEFAULT FALSE,
    has_photo BOOLEAN DEFAULT FALSE,
    scraped_at TIMESTAMP DEFAULT NOW(),
    intent_score FLOAT DEFAULT 0,
    status VARCHAR(50) DEFAULT ''new'',
    metadata JSONB DEFAULT ''{}''
);
CREATE TABLE lead_sources (
    id SERIAL PRIMARY KEY,
    lead_id INTEGER REFERENCES leads(id),
    source_type VARCHAR(50),
    source_group VARCHAR(255),
    source_context TEXT,
    discovered_at TIMESTAMP DEFAULT NOW()
);
CREATE INDEX idx_leads_score ON leads(intent_score DESC);
CREATE INDEX idx_leads_status ON leads(status);

Часть 8: Масштабирование и оптимизация

Параллельная обработка

async def parallel_scrape(groups, max_concurrent=5):
    semaphore = asyncio.Semaphore(max_concurrent)
    
    async def scrape_with_semaphore(group):
        async with semaphore:
            return await scrape_group(group)
    
    tasks = [scrape_with_semaphore(g) for g in groups]
    results = await asyncio.gather(*tasks, return_exceptions=True)
    
    successful = [r for r in results if not isinstance(r, Exception)]
    return successful

Кеширование

class CacheManager:
    def __init__(self, redis_url=''redis://localhost:6379''):
        self.redis = redis.from_url(redis_url)
        self.default_ttl = 3600 * 24
    
    def cached(self, ttl=None):
        def decorator(func):
            @wraps(func)
            async def wrapper(args, *kwargs):
                cache_key = f''{func.__name__}:{hash(str(args) + str(kwargs))}''
                cached = self.redis.get(cache_key)
                if cached:
                    return json.loads(cached)
                result = await func(args, *kwargs)
                self.redis.setex(cache_key, ttl or self.default_ttl, json.dumps(result, default=str))
                return result
            return wrapper
        return decorator

Часть 9: Соответствие и лучшие практики

• Сканируйте только публичные группы
• Не храните номера телефонов без явного согласия
• Предоставляйте механизм отказа во всех outreach
• Уважайте лимиты速率
• Документируйте действия по обработке данных

Заключение

Скрэппинг лидов в Telegram обеспечивает фундамент для автоматической генерации лидов. Описанная здесь техническая инфраструктура обеспечивает непрерывное обнаружение, извлечение, обогащение и квалификацию prospects. Начните с основ, затем постепенно добавляйте сложность: майнинг комментариев, детекция намерений, внешнее обогащение и предиктивная оценка.