Telegram Channel Parser: Export Messages to CSV

import asyncio
import csv
import logging
import os
from datetime import datetime, timezone
from html.parser import HTMLParser
from typing import Optional

from dotenv import load_dotenv
from telethon import TelegramClient
from telethon.errors import ChannelPrivateError, FloodWaitError, UsernameNotOccupiedError
from telethon.tl.types import Message
from tqdm.asyncio import tqdm

load_dotenv()
logging.basicConfig(filename="parser.log", level=logging.WARNING)
logger = logging.getLogger(__name__)

API_ID: int = int(os.environ["API_ID"])
API_HASH: str = os.environ["API_HASH"])
PHONE: str = os.environ["PHONE"]

class _HTMLStripper(HTMLParser):
    """Strips HTML tags from a string."""
    def __init__(self) -> None:
        super().__init__()
        self._parts: list[str] = []

    def handle_data(self, data: str) -> None:
        self._parts.append(data)

    def get_text(self) -> str:
        return "".join(self._parts)

def strip_html(text: str) -> str:
    """Remove HTML tags from message text."""
    s = _HTMLStripper()
    s.feed(text)
    return s.get_text()

def get_media_type(msg: Message) -> str:
    """Determine media type of a Telegram message."""
    if msg.photo:
        return "photo"
    if msg.video:
        return "video"
    if msg.document:
        return "document"
    return "none"

async def parse_channel(
    channel: str,
    limit: int,
    output_file: str,
    date_from: Optional[datetime] = None,
) -> None:
    """Parse Telegram channel and export messages to CSV."""
    async with TelegramClient("session", API_ID, API_HASH) as client:
        await client.start(phone=PHONE)
        try:
            entity = await client.get_entity(channel)
        except (ChannelPrivateError, UsernameNotOccupiedError) as e:
            logger.warning("Cannot access channel %s: %s", channel, e)
            return

        rows: list[dict] = []
        try:
            async for msg in tqdm(client.iter_messages(entity, limit=limit), total=limit):
                if not isinstance(msg, Message):
                    continue
                if date_from and msg.date.replace(tzinfo=timezone.utc) < date_from:
                    break
                reactions_count = 0
                if msg.reactions:
                    reactions_count = sum(r.count for r in msg.reactions.results)
                rows.append({
                    "id": msg.id,
                    "date": msg.date.isoformat(),
                    "text": strip_html(msg.text or ""),
                    "media_type": get_media_type(msg),
                    "views": msg.views or 0,
                    "reactions": reactions_count,
                    "link": f"https://t.me/{channel}/{msg.id}",
                })
        except FloodWaitError as e:
            logger.warning("FloodWait: sleeping %s seconds", e.seconds)
            await asyncio.sleep(e.seconds)

        with open(f"{output_file}.csv", "w", newline="", encoding="utf-8-sig") as f:
            writer = csv.DictWriter(f, fieldnames=rows[0].keys(), delimiter=";")
            writer.writeheader()
            writer.writerows(rows)
        print(f"✅ Saved {len(rows)} messages → {output_file}.csv")

if __name__ == "__main__":
    asyncio.run(parse_channel(
        channel="bbcrussian",
        limit=500,
        output_file="telegram_export",
        date_from=datetime(2024, 1, 1, tzinfo=timezone.utc),
    ))

Dependencies: pip install telethon python-dotenv tqdm

.env: API_ID=12345678 | API_HASH=abcdef... | PHONE=+79001234567