Divar to Telegram Notifier Bot

divar_notifier.py

"""
Divar to Telegram Notifier Bot

This script monitors a filtered Divar page and sends newly posted items to a specified Telegram channel every 60 seconds.

Setup Instructions:
1. Create a Telegram bot using BotFather.
2. Add the bot to your Telegram group or channel.
3. Promote the bot as an admin in your channel so it can post messages.
4. Use @username_to_id_bot to get the numeric ID of your channel (e.g., -1001234567890).
5. Install the required library: `pip install requests`
6. Run the script with the following arguments:
    --bot-token     Your Telegram bot token from BotFather
    --channel-id    The numeric channel ID retrieved via @username_to_id_bot
    --target-url    The URL of the filtered Divar page you want to scrape

Example:
    python divar_notifier.py \
        --bot-token 123456789:ABCDEF_your_bot_token_here \
        --channel-id -1001234567890 \
        --target-url "https://divar.ir/s/mashhad/mobile-phones?price=-40000000&q=s%2024%20ultra&sort=sort_date"

The bot will check the page every 60 seconds and forward any new listings to your Telegram channel.
"""


import json
import time
import argparse

import requests


def parse_arguments() -> tuple[str, str, str]:
    """
    Parse command line arguments for bot token, channel ID, and target URL.
    
    Returns:
        tuple: (bot_token, channel_id, target_url)
    """
    parser = argparse.ArgumentParser(description='Divar scraper and Telegram notifier')
    parser.add_argument('--bot-token', required=True, help='Telegram bot token')
    parser.add_argument('--channel-id', required=True, help='Telegram channel ID')
    parser.add_argument('--target-url', required=True, help='Divar target URL to scrape')
    
    args = parser.parse_args()
    return args.bot_token, args.channel_id, args.target_url


def send_telegram_message(bot_token: str, channel_id: str, message: str) -> None:
    """
    Send a message to a Telegram channel using the bot API.
    
    Args:
        bot_token: Telegram bot token
        channel_id: Target channel ID
        message: Message content to send
    """
    try:
        api_url = f"https://api.telegram.org/bot{bot_token}/sendMessage"
        payload = {
            "chat_id": channel_id,
            "text": message,
            "parse_mode": "HTML",
        }
        response = requests.post(url=api_url, data=payload, timeout=10)
        response.raise_for_status()
    except Exception as error:
        print(f"Failed to send Telegram message: {error}")


def fetch_divar_posts(target_url: str, seen_tokens: set[str]) -> list[str]:
    """
    Fetch new posts from Divar website and filter out already seen ones.
    
    Args:
        target_url: URL to scrape for posts
        seen_tokens: Set of already processed post tokens
        
    Returns:
        List of new post messages (each containing name and URL)
    """
    headers = {
        'accept': 'application/json, text/plain, */*',
        'accept-language': 'en-GB,en;q=0.9,fa-IR;q=0.8,fa;q=0.7,en-US;q=0.6',
        'user-agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/136.0.0.0 Safari/537.36',
    }
    
    try:
        response = requests.get(target_url, headers=headers)
        _, json_data = response.text.split('<script type="application/ld+json">')
        json_data, _ = json_data.split('</script>')
        posts_data = json.loads(json_data)
        
        new_posts = []
        for post in posts_data:
            token = post['url'].split('/')[-1]
            
            if token not in seen_tokens:
                seen_tokens.add(token)
                post_url = f'https://divar.ir/v/{token}'
                new_posts.append(f"{post['name']}\n{post_url}")
                
        return new_posts
        
    except Exception as error:
        print(f"Error fetching Divar posts: {error}")
        return []


if __name__ == '__main__':
    """Main execution loop for the scraper and notifier."""
    bot_token, channel_id, target_url = parse_arguments()
    seen_tokens: set[str] = set()

    # Ignore first run
    fetch_divar_posts(target_url, seen_tokens)
    
    while True:
        try:
            new_posts = fetch_divar_posts(target_url, seen_tokens)
            
            for post in new_posts:
                send_telegram_message(bot_token, channel_id, post)
                time.sleep(0.1)  # Rate limiting
                
        except Exception as error:
            error_message = f"Scraper error: {error}"
            print(error_message)
            send_telegram_message(bot_token, channel_id, error_message)
            
        time.sleep(60)