Telegram Channel Parser: How to Scrape Data from Any Channel

 1pip install telethon pandas python-dotenv

 1API_ID=your_api_id_here
 2API_HASH=your_api_hash_here
 3PHONE=+1234567890

 1import os
 2from dotenv import load_dotenv
 3from telethon import TelegramClient
 4
 5load_dotenv()
 6
 7api_id = int(os.getenv('API_ID'))
 8api_hash = os.getenv('API_HASH')
 9phone = os.getenv('PHONE')
10
11client = TelegramClient('session', api_id, api_hash)
12
13async def main():
14    await client.start(phone=phone)
15    print("Connected to Telegram!")
16    
17    # Test: get your own info
18    me = await client.get_me()
19    print(f"Logged in as: {me.first_name} (@{me.username})")
20
21with client:
22    client.loop.run_until_complete(main())

 1import os
 2import csv
 3import asyncio
 4from datetime import datetime
 5from dotenv import load_dotenv
 6from telethon import TelegramClient
 7from telethon.tl.functions.channels import GetParticipantsRequest
 8from telethon.tl.types import ChannelParticipantsSearch
 9
10load_dotenv()
11
12api_id = int(os.getenv('API_ID'))
13api_hash = os.getenv('API_HASH')
14
15client = TelegramClient('session', api_id, api_hash)
16
17async def parse_members(channel_username: str, output_file: str = 'members.csv'):
18    """Parse all members from a public Telegram channel."""
19    
20    try:
21        channel = await client.get_entity(channel_username)
22    except ValueError:
23        print(f"Channel {channel_username} not found. Make sure it is public.")
24        return
25    
26    print(f"Parsing members from: {channel.title}")
27    
28    all_participants = []
29    offset = 0
30    limit = 100  # Telegram allows max 100 per request
31    
32    while True:
33        try:
34            participants = await client(GetParticipantsRequest(
35                channel=channel,
36                filter=ChannelParticipantsSearch(''),
37                offset=offset,
38                limit=limit,
39                hash=0
40            ))
41            
42            if not participants.users:
43                break
44            
45            for user in participants.users:
46                all_participants.append({
47                    'user_id': user.id,
48                    'username': user.username or '',
49                    'first_name': user.first_name or '',
50                    'last_name': user.last_name or '',
51                    'phone': user.phone or '',
52                    'is_bot': user.bot,
53                    'is_verified': user.verified,
54                    'is_restricted': user.restricted,
55                })
56            
57            offset += len(participants.users)
58            print(f"  Fetched {offset} members so far...")
59            
60            # Rate limiting: wait 1 second between requests
61            await asyncio.sleep(1)
62            
63        except Exception as e:
64            print(f"Error at offset {offset}: {e}")
65            await asyncio.sleep(5)
66            continue
67    
68    # Export to CSV
69    with open(output_file, 'w', newline='', encoding='utf-8') as f:
70        writer = csv.DictWriter(f, fieldnames=all_participants[0].keys())
71        writer.writeheader()
72        writer.writerows(all_participants)
73    
74    print(f"\nDone! Saved {len(all_participants)} members to {output_file}")
75    return all_participants
76
77async def main():
78    # Replace with your target channel
79    await parse_members('@durov', 'durov_members.csv')
80
81with client:
82    client.loop.run_until_complete(main())

 1Parsing members from: durov
 2  Fetched 100 members so far...
 3  Fetched 200 members so far...
 4  ...
 5Done! Saved 1,847 members to durov_members.csv

 1import os
 2import csv
 3import asyncio
 4from datetime import datetime
 5from dotenv import load_dotenv
 6from telethon import TelegramClient
 7
 8load_dotenv()
 9
10api_id = int(os.getenv('API_ID'))
11api_hash = os.getenv('API_HASH')
12
13client = TelegramClient('session', api_id, api_hash)
14
15async def parse_messages(channel_username: str, output_file: str = 'messages.csv', limit: int = None):
16    """Parse all messages from a public Telegram channel."""
17    
18    try:
19        channel = await client.get_entity(channel_username)
20    except ValueError:
21        print(f"Channel {channel_username} not found.")
22        return
23    
24    print(f"Parsing messages from: {channel.title}")
25    
26    messages_data = []
27    count = 0
28    
29    async for message in client.iter_messages(channel, limit=limit):
30        count += 1
31        
32        # Extract media type
33        media_type = 'none'
34        if message.photo:
35            media_type = 'photo'
36        elif message.video:
37            media_type = 'video'
38        elif message.document:
39            media_type = 'document'
40        elif message.voice:
41            media_type = 'voice'
42        
43        # Extract URLs from message text
44        urls = []
45        if message.entities:
46            for entity in message.entities:
47                if hasattr(entity, 'url'):
48                    urls.append(entity.url)
49        
50        messages_data.append({
51            'message_id': message.id,
52            'date': message.date.strftime('%Y-%m-%d %H:%M:%S'),
53            'text': message.text or '',
54            'views': message.views or 0,
55            'forwards': message.forwards or 0,
56            'replies': message.replies.replies if message.replies else 0,
57            'media_type': media_type,
58            'urls': '|'.join(urls),
59            'is_pinned': message.pinned or False,
60        })
61        
62        if count % 100 == 0:
63            print(f"  Parsed {count} messages...")
64            await asyncio.sleep(0.5)  # Rate limiting
65    
66    # Export to CSV
67    if messages_data:
68        with open(output_file, 'w', newline='', encoding='utf-8') as f:
69            writer = csv.DictWriter(f, fieldnames=messages_data[0].keys())
70            writer.writeheader()
71            writer.writerows(messages_data)
72    
73    print(f"\nDone! Saved {len(messages_data)} messages to {output_file}")
74    return messages_data
75
76async def main():
77    # Parse last 1000 messages from a channel
78    await parse_messages('@techcrunch', 'techcrunch_messages.csv', limit=1000)
79
80with client:
81    client.loop.run_until_complete(main())

 1import json
 2
 3def export_to_json(data, filename):
 4    with open(filename, 'w', encoding='utf-8') as f:
 5        json.dump(data, f, ensure_ascii=False, indent=2)
 6    print(f"Exported {len(data)} records to {filename}")
 7
 8# Usage
 9export_to_json(messages_data, 'messages.json')

 1import gspread
 2from oauth2client.service_account import ServiceAccountCredentials
 3
 4def export_to_sheets(data, spreadsheet_name, worksheet_name='Sheet1'):
 5    scope = ['https://spreadsheets.google.com/feeds',
 6             'https://www.googleapis.com/auth/drive']
 7    
 8    creds = ServiceAccountCredentials.from_json_keyfile_name(
 9        'credentials.json', scope
10    )
11    client = gspread.authorize(creds)
12    
13    sheet = client.open(spreadsheet_name).worksheet(worksheet_name)
14    
15    # Write headers
16    headers = list(data[0].keys())
17    sheet.append_row(headers)
18    
19    # Write data
20    for row in data:
21        sheet.append_row(list(row.values()))
22    
23    print(f"Exported {len(data)} rows to Google Sheets")

 1import pandas as pd
 2
 3df = pd.read_csv('messages.csv')
 4
 5# Top 10 most viewed messages
 6top_messages = df.nlargest(10, 'views')[['date', 'text', 'views']]
 7print(top_messages)
 8
 9# Messages per day
10daily = df.groupby('date').size()
11print(daily)
12
13# Average views by media type
14avg_views = df.groupby('media_type')['views'].mean()
15print(avg_views)

 1import asyncio
 2import random
 3
 4# Conservative rate limiting
 5SAFE_DELAY = 1.0        # 1 second between requests
 6BATCH_DELAY = 5.0       # 5 seconds between batches
 7BATCH_SIZE = 100         # 100 items per batch
 8MAX_RETRIES = 3          # Retry on error
 9
10async def safe_request(func, *args, **kwargs):
11    """Execute a request with automatic retry and rate limiting."""
12    for attempt in range(MAX_RETRIES):
13        try:
14            result = await func(*args, **kwargs)
15            # Random delay to look more human
16            await asyncio.sleep(SAFE_DELAY + random.uniform(0, 0.5))
17            return result
18        except Exception as e:
19            if 'FLOOD_WAIT' in str(e):
20                # Telegram says to wait X seconds
21                wait_time = int(str(e).split('FLOOD_WAIT_')[1].split()[0])
22                print(f"Rate limited! Waiting {wait_time} seconds...")
23                await asyncio.sleep(wait_time + 1)
24            elif attempt < MAX_RETRIES - 1:
25                print(f"Error: {e}. Retrying in 5 seconds...")
26                await asyncio.sleep(5)
27            else:
28                raise

 1import logging
 2
 3# Enable Telethon debug logging
 4logging.basicConfig(level=logging.INFO)
 5logger = logging.getLogger('telethon')
 6logger.setLevel(logging.DEBUG)
 7
 8# This will show all API requests and responses

 1# If you get auth errors, delete the session and re-authenticate
 2import os
 3
 4session_file = 'session.session'
 5if os.path.exists(session_file):
 6    os.remove(session_file)
 7    print("Session deleted. Re-run the script to re-authenticate.")