Telegram Lead Scraping: The Complete Technical Guide to Extracting Qualified Leads

Telegram Lead Scraping: The Complete Technical Guide

Introduction

Telegram lead scraping is fundamentally different from scraping LinkedIn or email. The platform''s open API, structured group membership, and accessible message history make it the most scrapable major platform for B2B lead generation.

This guide covers every technical aspect: group discovery, member extraction, comment mining, profile enrichment, intent detection, and building a complete data pipeline.

Part 1: Infrastructure Setup

API Access Configuration

Telegram provides two API layers:

Bot API - Limited but stable. Bots can only access groups they are added to. User API (MTProto) - Full access through personal accounts. This is the primary tool for lead scraping.

Setting Up 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''Logged in as: {me.first_name}'')
with client:
    client.loop.run_until_complete(initialize())

Multi-Account Setup

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

Part 2: Group Discovery

Automated Group Discovery

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
        }

Part 3: Member Extraction

Basic Member Scraping

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

Part 4: Comment-Based Lead Extraction

Intent Detection

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

Part 5: Profile Enrichment

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

Part 6: Intent Scoring

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)

Part 7: Data Pipeline Architecture

[Group Discovery] -> [Group Scoring] -> [Member Extraction] -> [Profile Enrichment]
        |                    |                   |                    |
   TGStat API          NLP Analysis       Telethon API        External APIs
        |                    |                   |                    |
  [Comment Mining] -> [Intent Detection] -> [Behavioral Analysis] -> [Intent Scoring]
        |                    |                   |                    |
   Message API       Keyword Match       Pattern Detection     Weighted Score
        |                    |                   |                    |
  [Data Storage] -> [Lead Qualification] -> [CRM Integration] -> [Outreach Ready]
        |                    |                   |                    |
   PostgreSQL       Score Threshold     Webhook/API         Campaign Ready

Database Schema

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);

Part 8: Scaling and Optimization

Parallel Processing

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

Caching

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

Part 9: Compliance and Best Practices

• Only scrape public groups
• Do not store phone numbers without explicit consent
• Provide opt-out mechanisms in all outreach
• Respect rate limits
• Document data processing activities

Conclusion

Telegram lead scraping provides the foundation for automated lead generation. The technical infrastructure described here enables continuous discovery, extraction, enrichment, and qualification of prospects. Start with basics, then progressively add sophistication: comment mining, intent detection, external enrichment, and predictive scoring.