Torob Product Tracker

torob.py

"""
Scrape Torob API using curl_cffi to monitor product prices with Telegram and Bale alerts.

Installation:
    pip install curl_cffi

Configuration:
    Create a 'config.json' file in the same directory:
    {
        "bale_bot_token": "YOUR_BALE_TOKEN",
        "bale_chat_id": "YOUR_BALE_CHAT_ID",
        "bale_proxy": null,
        "telegram_bot_token": "YOUR_TELEGRAM_TOKEN",
        "telegram_chat_id": "YOUR_TELEGRAM_CHAT_ID",
        "telegram_proxy": "socks5h://127.0.0.1:1080"
    }

Usage:
    Run continuously:
        python main.py

    Run once:
        RUN_ONCE=1 python main.py
"""

import json
import logging
import os
import time
from pathlib import Path
from typing import Any, Optional

from curl_cffi import requests

TOROB_API_URL = 'https://api.torob.com/v4/base-product/search/?.......'
STATE_FILE = Path(__file__).with_name('torob_state.json')
CONFIG_FILE = Path(__file__).with_name('config.json')

CONFIG = json.loads(CONFIG_FILE.read_text(encoding='utf-8')) if CONFIG_FILE.exists() else {}

BALE_BOT_TOKEN = CONFIG.get('bale_bot_token')
BALE_CHAT_ID = CONFIG.get('bale_chat_id')
BALE_PROXY = CONFIG.get('bale_proxy')

TELEGRAM_BOT_TOKEN = CONFIG.get('telegram_bot_token')
TELEGRAM_CHAT_ID = CONFIG.get('telegram_chat_id')
TELEGRAM_PROXY = CONFIG.get('telegram_proxy')

POLL_INTERVAL_SECONDS = 60
MAX_RETRIES = 3
RETRY_DELAY_SECONDS = 5
MAX_PRICE_THRESHOLD = 50_000

# Full browser header stack required by Torob WAF
HEADERS = {
    'User-Agent': 'Mozilla/5.0 (X11; Linux x86_64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36',
    'Accept': 'application/json, text/plain, */*',
    'Accept-Language': 'fa-IR,fa;q=0.9,en-US;q=0.8,en;q=0.7',
    'Accept-Encoding': 'gzip, deflate, br',
    'Referer': 'https://torob.com/',
    'Origin': 'https://torob.com',
    'sec-ch-ua': '"Not/A)Brand";v="8", "Chromium";v="126", "Google Chrome";v="126"',
    'sec-ch-ua-mobile': '?0',
    'sec-ch-ua-platform': '"Linux"',
    'sec-fetch-dest': 'empty',
    'sec-fetch-mode': 'cors',
    'sec-fetch-site': 'same-site',
}

RUN_ONCE = os.getenv('RUN_ONCE', '').lower() in {'1', 'true', 'yes'}
STARTUP_MESSAGE = 'سلام! ربات رصد قیمت ترب بیدار شد 👀🛍️'
FETCH_ERROR_MESSAGE = 'متاسفانه اتصال به API ترب برقرار نشد 😔'

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')


def download_image(url: str) -> Optional[bytes]:
    """Download image bytes using curl_cffi browser impersonation."""
    if not url:
        return None

    if url.startswith('//'):
        url = f'https:{url}'
    elif not url.startswith('http'):
        url = f'https://torob.com{url}'

    try:
        response = requests.get(url, headers=HEADERS, impersonate='chrome', timeout=15)
        if response.status_code == 200:
            return response.content
    except Exception as e:
        logging.warning(f'Failed to download image ({url}): {e}')

    return None


def send_api_message(
    base_url: str,
    token: str,
    chat_id: Any,
    text: str,
    photo_bytes: Optional[bytes] = None,
    proxy: Optional[str] = None,
    parse_mode: Optional[str] = None,
) -> bool:
    """Send message or photo to Bot API endpoints (Telegram or Bale)."""
    if not token or not chat_id:
        return False

    proxies = {'http': proxy, 'https': proxy} if proxy else None
    current_photo_bytes = photo_bytes

    for attempt in range(MAX_RETRIES):
        try:
            if current_photo_bytes:
                url = f'{base_url}/bot{token}/sendPhoto'
                data = {'chat_id': chat_id, 'caption': text}
                if parse_mode:
                    data['parse_mode'] = parse_mode

                files = {'photo': ('image.jpg', current_photo_bytes, 'image/jpeg')}
                response = requests.post(url, data=data, files=files, proxies=proxies, timeout=25)
            else:
                url = f'{base_url}/bot{token}/sendMessage'
                data = {'chat_id': chat_id, 'text': text}
                if parse_mode:
                    data['parse_mode'] = parse_mode

                response = requests.post(url, json=data, proxies=proxies, timeout=20)

            if response.status_code == 200:
                return True

            logging.error(f'API error ({url}): {response.status_code} - {response.text}')
            if current_photo_bytes:
                current_photo_bytes = None

        except Exception as e:
            logging.warning(f'Send attempt {attempt + 1} failed: {e}')

        if attempt < MAX_RETRIES - 1:
            time.sleep(RETRY_DELAY_SECONDS)

    return False


def notify_all(text: str, photo_url: Optional[str] = None) -> None:
    """Broadcast notification to configured platforms."""
    photo_bytes = download_image(photo_url) if photo_url else None

    if BALE_BOT_TOKEN and BALE_CHAT_ID:
        send_api_message(
            base_url='https://tapi.bale.ai',
            token=BALE_BOT_TOKEN,
            chat_id=BALE_CHAT_ID,
            text=text,
            photo_bytes=photo_bytes,
            proxy=BALE_PROXY,
        )

    if TELEGRAM_BOT_TOKEN and TELEGRAM_CHAT_ID:
        send_api_message(
            base_url='https://api.telegram.org',
            token=TELEGRAM_BOT_TOKEN,
            chat_id=TELEGRAM_CHAT_ID,
            text=text,
            photo_bytes=photo_bytes,
            proxy=TELEGRAM_PROXY,
            parse_mode='Markdown',
        )


def fetch_products() -> list[dict[str, Any]]:
    """Fetch product items from Torob API using browser TLS impersonation."""
    for attempt in range(MAX_RETRIES):
        try:
            response = requests.get(
                TOROB_API_URL,
                headers=HEADERS,
                impersonate='chrome',
                timeout=20,
            )
            response.raise_for_status()
            data = response.json()
            return data.get('results', [])
        except Exception as e:
            logging.error(f'Fetch attempt {attempt + 1} failed: {e}')
            if attempt < MAX_RETRIES - 1:
                time.sleep(RETRY_DELAY_SECONDS)

    raise RuntimeError('Failed to fetch Torob products after multiple attempts')


def load_state() -> Optional[dict[str, dict[str, Any]]]:
    """Load saved product state from file."""
    if not STATE_FILE.exists():
        return None

    try:
        return json.loads(STATE_FILE.read_text(encoding='utf-8'))
    except (json.JSONDecodeError, OSError) as e:
        logging.warning(f'Could not load state: {e}')
        return None


def save_state(state: dict[str, dict[str, Any]]) -> None:
    """Save product state map to file."""
    try:
        STATE_FILE.write_text(json.dumps(state, ensure_ascii=False, indent=2), encoding='utf-8')
    except OSError as e:
        logging.error(f'Could not save state: {e}')


def format_message(event_type: str, item: dict[str, Any], old_price: Optional[int] = None) -> str:
    """Build message text for product events."""
    name = item.get('name1', 'محصول')
    price_text = item.get('price_text', '')
    shop_text = item.get('shop_text', '')
    web_url = item.get('web_client_absolute_url', '')
    link = f'https://torob.com{web_url}' if web_url else ''

    if event_type == 'new':
        header = '📦 **محصول جدید اضافه شد**'
    else:
        old_price_formatted = f'{old_price:,}' if old_price else ''
        header = f'📉 **کاهش قیمت!** (قیمت قبلی: {old_price_formatted} تومان)'

    lines = [
        header,
        '',
        f'📌 **نام:** {name}',
        f'💰 **قیمت:** {price_text}',
    ]
    if shop_text:
        lines.append(f'🏪 **فروشگاه:** {shop_text}')
    if link:
        lines.append(f'🔗 **لینک:** {link}')

    return '\n'.join(lines)


def main() -> None:
    """Main execution loop."""
    notify_all(STARTUP_MESSAGE)

    while True:
        try:
            raw_items = fetch_products()
            previous_state = load_state()
            current_state: dict[str, dict[str, Any]] = {}

            for item in raw_items:
                key = item.get('random_key')
                if key:
                    current_state[key] = {
                        'name': item.get('name1', ''),
                        'price': item.get('price', 0),
                        'price_text': item.get('price_text', ''),
                    }

            if previous_state is None:
                save_state(current_state)
                logging.info('Initial state recorded.')
            else:
                for item in raw_items:
                    key = item.get('random_key')
                    if not key:
                        continue

                    price = item.get('price', 0)
                    img_url = item.get('image_url')

                    if key not in previous_state:
                        if MAX_PRICE_THRESHOLD is None or price <= MAX_PRICE_THRESHOLD:
                            msg = format_message('new', item)
                            logging.info(f'New product found: {item.get("name1")}')
                            notify_all(msg, photo_url=img_url)
                    else:
                        old_price = previous_state[key].get('price', price)
                        if price < old_price:
                            if MAX_PRICE_THRESHOLD is None or price <= MAX_PRICE_THRESHOLD:
                                msg = format_message('price_drop', item, old_price=old_price)
                                logging.info(f'Price drop found for: {item.get("name1")}')
                                notify_all(msg, photo_url=img_url)

                save_state(current_state)

        except RuntimeError as e:
            logging.error(str(e))
            notify_all(FETCH_ERROR_MESSAGE)
        except Exception as e:
            logging.error(f'Unexpected error during execution cycle: {e}')

        if RUN_ONCE:
            break

        logging.info(f'Sleeping for {POLL_INTERVAL_SECONDS} seconds...')
        time.sleep(POLL_INTERVAL_SECONDS)


if __name__ == '__main__':
    try:
        main()
    except KeyboardInterrupt:
        logging.info('Script stopped by user.')
    except Exception as error:
        logging.critical(f'Fatal error: {error}')

Interesting😎👌

awesome 😍

Very Helpful, Thanks Mate🙌