How to make a parser of Telegram channels without garbage: filtering spam, doubles, vacancies and advertising

pip install telethon asyncio aiofiles python-dotenv
# Для хранения в SQLite
pip install aiosqlite
# Для симхэша (дедупликация)
pip install simhash

# .env
API_ID=12345678
API_HASH=your_api_hash_here
SESSION_NAME=parser_session

import asyncio
import os
from telethon import TelegramClient
from dotenv import load_dotenv

load_dotenv()

client = TelegramClient(
    os.getenv('SESSION_NAME'),
    int(os.getenv('API_ID')),
    os.getenv('API_HASH')
)

CHANNELS = [
    'https://t.me/example_channel_1',
    'https://t.me/example_channel_2',
]

async def parse_channel(channel_url: str, limit: int = 500):
    messages = []
    async for message in client.iter_messages(channel_url, limit=limit):
        messages.append(message)
    return messages

async def main():
    await client.start()
    for channel in CHANNELS:
        msgs = await parse_channel(channel)
        print(f'{channel}: получено {len(msgs)} сообщений')

with client:
    client.loop.run_until_complete(main())

def is_structural_garbage(message) -> tuple[bool, str]:
    """
    Возвращает (True, причина) если сообщение — структурный мусор.
    Проверки от дешёвых к дорогим.
    """

    # Пустое сообщение (только медиа без подписи, или сервисное)
    if not message.text and not message.caption:
        return True, 'no_text'

    # Сервисное сообщение: вступление, смена фото, пин и т.д.
    if message.action is not None:
        return True, 'service_action'

    # Форвард — чужой контент
    if message.fwd_from is not None:
        return True, 'forward'

    text = message.text or message.caption or ''

    # Слишком короткое — скорее всего анонс или эмодзи-реакция
    if len(text.strip()) < 50:
        return True, 'too_short'

    # Спонсированное сообщение (реклама через Telegram Ads)
    # Помечается флагом sponsored в некоторых версиях API
    if getattr(message, 'sponsored', False):
        return True, 'sponsored'

    return False, ''

import

VACANCY PATTERNS = [
# Direct job markers
R'required?
R'Are you open?\S+Vacancy?
r'vacancy\s*:',
r'we are looking for \s+ (developer |Manager |Analytics)',
r'invite \s+(in\s+team |developer |specialist)',
Working conditions
r'formation\s+by\s+tc',
R'Official \s + Employment',
r'complete\s+(work\s+)?day',
R'remote? \s+work',
r'hybrid\s+format',
#Salary
r's salary\s+from',
r'grade\s*(junior|middle|senior)',
r'\d+\s*(thousand|k)\s*/?\s*me',
r'Income\s+ from\s+\d+',
#HR markers
r'summary\s+on',
r'hr@',
r'response\s+in\s+ls',
r'write\s+in\s+l',
r'stack\s*:',
r'requirements\s*:',
r'obligations\s*:',
r'conditions\s*:',
]

vacancy re = re.compile()
'|'.join (VACANCY PATTERNS),
re.IGNORECASE | re.UNICODE
)

def is vacancy(text: str) -> bool:
return bool( vacancy re.search(text))

FORBIDDEN_PATTERNS = {
    'gambling': [
        r'казино', r'ставки\s+на\s+спорт', r'онлайн.?казино',
        r'слоты', r'рулетка', r'1xbet', r'1win', r'melbet',
        r'бонус\s+за\s+депозит',
    ],
    'crypto_scam': [
        r'х\d+\s+за\s+\d+\s+дней',
        r'пассивный\s+доход\s+от',
        r'вложи\s+и\s+получи',
        r'майнинг.{0,20}доход',
        r'крипт[оа].{0,20}заработ',
        r'pump.{0,10}группа',
        r'инсайд.{0,10}сигнал',
    ],
    'loans': [
        r'займ\s+без\s+отказа',
        r'микрозайм',
        r'мфо',
        r'кредит\s+без\s+справок',
        r'деньги\s+в\s+долг\s+срочно',
    ],
    'adult': [
        r'18\+.{0,20}(фото|видео|контент)',
        r'только\s+для\s+взрослых',
    ],
}

_forbidden_res = {
    category: re.compile('|'.join(patterns), re.IGNORECASE | re.UNICODE)
    for category, patterns in FORBIDDEN_PATTERNS.items()
}

def get_forbidden_category(text: str) -> str | None:
    for category, pattern in _forbidden_res.items():
        if pattern.search(text):
            return category
    return None

from simhash import Simhash

def text_to_simhash(text: str) -> int:
    """Нормализуем текст и считаем симхэш."""
    # Убираем всё кроме слов — эмодзи, пунктуация, лишние пробелы
    normalized = re.sub(r'[^\w\s]', ' ', text.lower())
    normalized = re.sub(r'\s+', ' ', normalized).strip()
    # Разбиваем на токены (слова и биграммы для лучшей точности)
    words = normalized.split()
    tokens = words + [f'{a} {b}' for a, b in zip(words, words[1:])]
    return Simhash(tokens).value

def simhash_distance(h1: int, h2: int) -> int:
    """Расстояние Хэмминга между двумя симхэшами."""
    xor = h1 ^ h2
    return bin(xor).count('1')

class DuplicateDetector:
    def __init__(self, threshold: int = 5):
        # threshold: максимальное расстояние Хэмминга для «дубля»
        # 0 = точное совпадение, 5 = тексты ~90% похожи, 10 = ~70%
        self.threshold = threshold
        self.seen_hashes: list[tuple[int, int]] = []  # (message_id, simhash)

    def is_duplicate(self, message_id: int, text: str) -> bool:
        h = text_to_simhash(text)
        for seen_id, seen_hash in self.seen_hashes:
            if seen_id == message_id:
                continue
            if simhash_distance(h, seen_hash) <= self.threshold:
                return True
        self.seen_hashes.append((message_id, h))
        return False

AD PATTERNS = []
# Calls and UTM patterns
R'click \s+ on \s+link',
r'click here',
r'utm source',
r'utm medium',
r'utm campaign',
#Partner markers
r'advertising',
r'advertising\s+post',
r'on\s+rights\s+advertising',
r'partner's/s+material',
r'promo',
#Referral patterns
r'promocode\s+[A-Z0-9]{3,}',
r'discount\s+\d+%\s+ by\s+promo code',
# Characteristic designs
r'Subscribe \s+ to \s+channel',
r's+ to s+@\w+',
r' go \s+ to \s+@\w+',
]

ad re = re.compile('|'.join(AD PATTERNS), re.IGNORECASE | re.UNICODE)

def looks like ad(message) -> bool:
text = message.text or message.caption or '

# Direct markers
if ad re.search(text):
True, true return.

# buttons with UTM or referral links
if message.reply markup:
for row in (message.reply markup.rows or []):
for button in (row.buttons or []):
url = getattr(button, 'url', '' or '')
if 'utm ' in url or any()
p in url for p in ['ref=', 'promo=', 'aff=']
:
True, true return.

return

import asyncio
import logging
from dataclasses import dataclass
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger('tg_parser')

@dataclass
class ParsedMessage:
    message_id: int
    channel: str
    text: str
    date: datetime
    views: int | None
    simhash: int

class TelegramParser:
    def __init__(self, channels: list[str], limit_per_channel: int = 1000):
        self.channels = channels
        self.limit = limit_per_channel
        self.dedup = DuplicateDetector(threshold=5)
        self.stats = {
            'total': 0,
            'passed': 0,
            'filtered': {},
        }

    def _record_filtered(self, reason: str):
        self.stats['filtered'][reason] = self.stats['filtered'].get(reason, 0) + 1

    def filter_message(self, message, channel: str) -> ParsedMessage | None:
        self.stats['total'] += 1
        text = message.text or message.caption or ''

        # Слой 1: структурный мусор
        is_garbage, reason = is_structural_garbage(message)
        if is_garbage:
            self._record_filtered(reason)
            return None

        # Слой 2: реклама
        if looks_like_ad(message):
            self._record_filtered('ad')
            return None

        # Слой 3: вакансии
        if is_vacancy(text):
            self._record_filtered('vacancy')
            return None

        # Слой 4: запрещённые тематики
        forbidden = get_forbidden_category(text)
        if forbidden:
            self._record_filtered(f'forbidden_{forbidden}')
            return None

        # Слой 5: дубли
        if self.dedup.is_duplicate(message.id, text):
            self._record_filtered('duplicate')
            return None

        self.stats['passed'] += 1
        return ParsedMessage(
            message_id=message.id,
            channel=channel,
            text=text,
            date=message.date,
            views=getattr(message, 'views', None),
            simhash=text_to_simhash(text),
        )

    async def parse_channel(self, channel: str) -> list[ParsedMessage]:
        results = []
        logger.info(f'Парсим {channel}...')
        try:
            async for message in client.iter_messages(channel, limit=self.limit):
                parsed = self.filter_message(message, channel)
                if parsed:
                    results.append(parsed)
                # Небольшая пауза между сообщениями — флоуд-контроль
                await asyncio.sleep(0.05)
        except Exception as e:
            logger.error(f'Ошибка парсинга {channel}: {e}')
        return results

    async def run(self) -> list[ParsedMessage]:
        all_messages = []
        for channel in self.channels:
            msgs = await self.parse_channel(channel)
            all_messages.extend(msgs)
            # Пауза между каналами
            await asyncio.sleep(2)

        logger.info(f'Итого: {self.stats["total"]} сообщений, прошло фильтры: {self.stats["passed"]}')
        logger.info(f'Отфильтровано: {self.stats["filtered"]}')
        return all_messages

from telethon.errors import FloodWaitError

async def safe_iter_messages(channel, limit=1000, pause=0.1):
    """iter_messages с автоматическим retry при FloodWait."""
    retries = 0
    async for message in client.iter_messages(channel, limit=limit):
        try:
            yield message
            await asyncio.sleep(pause)
        except FloodWaitError as e:
            wait_seconds = e.seconds + 5
            logger.warning(f'FloodWait: ждём {wait_seconds} секунд...')
            await asyncio.sleep(wait_seconds)
            retries += 1
            if retries > 5:
                logger.error('Слишком много FloodWait, прерываем канал')
                break

async with client.takeout(messages=True) as takeout:
    async for message in takeout.iter_messages(channel, wait_time=0.5):
        # обрабатываем
        pass

import aiosqlite

CREATE_TABLE = """
CREATE TABLE IF NOT EXISTS messages (
    id          INTEGER PRIMARY KEY,
    message_id  INTEGER NOT NULL,
    channel     TEXT NOT NULL,
    text        TEXT NOT NULL,
    date        TEXT NOT NULL,
    views       INTEGER,
    simhash     INTEGER NOT NULL,
    created_at  TEXT DEFAULT (datetime('now')),
    UNIQUE(channel, message_id)
);
CREATE INDEX IF NOT EXISTS idx_simhash ON messages(simhash);
CREATE INDEX IF NOT EXISTS idx_channel_date ON messages(channel, date);
"""

async def save_messages(messages: list[ParsedMessage], db_path: str = 'parser.db'):
    async with aiosqlite.connect(db_path) as db:
        await db.executescript(CREATE_TABLE)
        await db.executemany(
            """
            INSERT OR IGNORE INTO messages
                (message_id, channel, text, date, views, simhash)
            VALUES (?, ?, ?, ?, ?, ?)
            """,
            [
                (m.message_id, m.channel, m.text,
                 m.date.isoformat(), m.views, m.simhash)
                for m in messages
            ]
        )
        await db.commit()

-- Найти все сообщения с похожим симхэшем (расстояние ≤ 5 бит)
-- В PostgreSQL нет битового расстояния из коробки, используем функцию
CREATE OR REPLACE FUNCTION hamming_distance(a BIGINT, b BIGINT)
RETURNS INTEGER AS $$
BEGIN
    RETURN bit_count((a # b)::bit(64));
END;
$$ LANGUAGE plpgsql IMMUTABLE;

-- Найти потенциальные дубли для конкретного симхэша
SELECT channel, text, date
FROM messages
WHERE hamming_distance(simhash, $1) <= 5
  AND message_id != $2
LIMIT 20;

from telethon import events

WATCH_CHANNELS = ['https://t.me/channel1', 'https://t.me/channel2']

parser = TelegramParser(channels=WATCH_CHANNELS)

@client.on(events.NewMessage(chats=WATCH_CHANNELS))
async def handle_new_message(event):
    channel = event.chat.username or str(event.chat_id)
    parsed = parser.filter_message(event.message, channel)
    if parsed:
        await save_messages([parsed])
        logger.info(f'Новое сообщение из {channel}: {parsed.text[:80]}...')

async def main():
    await client.start()
    logger.info('Слушаем новые сообщения...')
    await client.run_until_disconnected()

with client:
    client.loop.run_until_complete(main())

async def periodic_catchup():
    while True:
        await client.catch_up()
        await asyncio.sleep(30)

I have a list of jobs from Telegram channels.
Analyze 50 examples and regex patterns.
which cover all cases, but do not give false positives
Ordinary technical content. The result is a Python string list for re.compile.

# Для сложных случаев — классификация через API вместо словарей
import anthropic

claude = anthropic.Anthropic()

def classify_message(text: str) -> dict:
    response = claude.messages.create(
        model="claude-sonnet-4-20250514",
        max_tokens=100,
        messages=[{
            "role": "user",
            "content": f"""Классифицируй это сообщение из Telegram-канала.
Ответь ТОЛЬКО JSON: {{"is_ad": bool, "is_vacancy": bool, "category": str, "confidence": float}}
Категории: tech_news, tutorial, opinion, tool_review, other
Сообщение: {text[:500]}"""
        }]
    )
    import json
    return json.loads(response.content[0].text)

async def get_last_message_id(db_path: str, channel: str) -> int:
    async with aiosqlite.connect(db_path) as db:
        cursor = await db.execute(
            'SELECT MAX(message_id) FROM messages WHERE channel = ?',
            (channel,)
        )
        row = await cursor.fetchone()
        return row[0] or 0

async def parse_channel_incremental(channel: str, db_path: str):
    min_id = await get_last_message_id(db_path, channel)
    results = []
    async for message in client.iter_messages(channel, min_id=min_id):
        parsed = parser.filter_message(message, channel)
        if parsed:
            results.append(parsed)
        await asyncio.sleep(0.05)
    if results:
        await save_messages(results, db_path)
    logger.info(f'{channel}: +{len(results)} новых сообщений (с ID > {min_id})')

Adjustment:
Separate Telegram account (not the main one)
api id and api hash in .env, not in code
● Session file outside the repository (in .gitignore)

Filters:
Structural: deviation of empty, service, too short
Forwards: message.fwd from is not None
Advertising: Patterns + Button Check on UTM
Jobs: regex for HR keywords
Prohibited topics: gambling, crypto scam, adult, loans
Deduplication: SimHash with a 5-bit threshold

Floud control:
Asyncio.sleep between messages (0.05–0.5 seconds)
Asyncio.sleep between channels (2-5 seconds)
FloodWaitError processing with e.seconds waiting

Storage:
UNIQUE (channel, message id) - no re-saving
Simhash is stored in DB – deduplication between sessions
● Incremental parsing through min id

Monitoring:
● Logistics of the reasons for the rejection of each filter
● Final statistics: total / passed / by category n