import argparse import json import random import re import time from concurrent.futures import ThreadPoolExecutor, as_completed from datetime import datetime from pathlib import Path from urllib.parse import quote import pandas as pd from selenium import webdriver from selenium.common.exceptions import ( StaleElementReferenceException, TimeoutException, WebDriverException, ) from selenium.webdriver.common.by import By from selenium.webdriver.chrome.service import Service from selenium.webdriver.common.action_chains import ActionChains from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.support.ui import WebDriverWait BASE_DIR = Path(__file__).resolve().parent CONFIG_PATH = BASE_DIR / "config.json" DOMAIN_PATTERN = re.compile(r"^(?:[a-z0-9](?:[a-z0-9-]{0,61}[a-z0-9])?\.)+[a-z]{2,24}$", re.IGNORECASE) WEBSITE_LINK_SELECTOR = "a[href*='api.torob.com/v4/internet-shop/profile-website/redirect/']" ONLINE_TAB_LABELS = ("خرید اینترنتی", "خرید آنلاین", "آنلاین") OFFLINE_TAB_LABELS = ("خرید حضوری", "حضوری") CAPTCHA_SELECTORS = ( "iframe[src*='recaptcha']", "iframe[src*='hcaptcha']", "iframe[src*='turnstile']", "[class*='captcha']", "[id*='captcha']", "#cf-challenge-running", "#challenge-running", "input[name='cf-turnstile-response']", ) CAPTCHA_TEXT_MARKERS = ( "verify you are human", "checking your browser", "security check", "captcha", "recaptcha", "hcaptcha", "تأیید کنید انسان هستید", "تایید کنید انسان هستید", "کپچا", ) def load_config(config_path=CONFIG_PATH): print("[LOG] Loading config from:", config_path, flush=True) defaults = { "target_url": "https://torob.com/browse/293/%D9%84%D9%88%D8%A7%D8%B2%D9%85-%DB%8C%D8%AF%DA%A9%DB%8C-%D8%AE%D9%88%D8%AF%D8%B1%D9%88-car-parts-airplane-motocycle/?shop_type=offline", "output_file": "Intro_autosave_autosave.xlsx", "max_missing_website_shops": 50, "unlimited_results": False, "only_without_website": True, "page_load_timeout_seconds": 40, "show_all_wait_seconds": 8, "shop_retry_count": 2, "headless": False, "manual_captcha_pause": True, "shop_worker_count": 1, } if not config_path.exists(): config_path.write_text(json.dumps(defaults, ensure_ascii=False, indent=2), encoding="utf-8") print(f"[LOG] Config file created: {config_path.name}", flush=True) return defaults with config_path.open(encoding="utf-8") as config_file: saved_config = json.load(config_file) if isinstance(saved_config, list): if saved_config and isinstance(saved_config[0], str): saved_config = {"target_url": saved_config[0]} else: saved_config = {} defaults.update(saved_config) defaults["only_without_website"] = bool(defaults["only_without_website"]) unlimited_value = defaults.get("unlimited_results", False) if isinstance(unlimited_value, str): defaults["unlimited_results"] = unlimited_value.strip().lower() in { "1", "true", "yes", "on", "unlimited", "بدون محدودیت", "بدون لیمیت" } else: defaults["unlimited_results"] = bool(unlimited_value) print("[LOG] Config loaded successfully", flush=True) return defaults def setup_driver(headless): print("[LOG] Setting up Chrome driver...", flush=True) print(f"[LOG] Headless mode: {headless}", flush=True) options = webdriver.ChromeOptions() chrome_path = str(BASE_DIR / "chrome-linux64" / "chrome") print(f"[LOG] Chrome binary path: {chrome_path}", flush=True) if not Path(chrome_path).exists(): raise FileNotFoundError(f"Chrome binary not found at: {chrome_path}") options.binary_location = chrome_path options.add_argument("--start-maximized") options.add_argument("--disable-gpu") options.add_argument("--disable-dev-shm-usage") options.page_load_strategy = "eager" options.add_argument("--log-level=3") options.add_argument("--disable-blink-features=AutomationControlled") options.add_experimental_option("excludeSwitches", ["enable-automation"]) options.add_experimental_option("useAutomationExtension", False) if headless: options.add_argument("--headless=new") options.add_argument("--window-size=1920,1080") driver_path = BASE_DIR / "chromedriver" print(f"[LOG] ChromeDriver path: {driver_path}", flush=True) if not driver_path.exists(): raise FileNotFoundError(f"ChromeDriver not found at: {driver_path}") print("[LOG] Starting ChromeDriver...", flush=True) driver = webdriver.Chrome(service=Service(str(driver_path)), options=options) driver.execute_cdp_cmd( "Page.addScriptToEvaluateOnNewDocument", {"source": "Object.defineProperty(navigator, 'webdriver', {get: () => undefined})"}, ) print("[LOG] ChromeDriver started successfully", flush=True) return driver # ... (بقیه توابع بدون تغییر) ... def scrape_torob_shops(config_path=CONFIG_PATH): print("[LOG] scrape_torob_shops() called", flush=True) try: print("[LOG] Loading config...", flush=True) config = load_config(config_path) print("[LOG] Config loaded:", config, flush=True) target_url = config["target_url"] output_file = str(config["output_file"]).strip() if not output_file.lower().endswith(".xlsx"): output_file = f"{output_file}.xlsx" print(f"هشدار: پسوند .xlsx به نام فایل خروجی اضافه شد: {output_file}", flush=True) output_path = BASE_DIR / output_file unlimited_results = bool(config.get("unlimited_results", False)) max_records = int(config["max_missing_website_shops"]) only_without_website = config["only_without_website"] worker_count = max(1, int(config["shop_worker_count"])) print("[LOG] Setting up driver...", flush=True) driver = setup_driver(bool(config["headless"])) driver.set_page_load_timeout(int(config["page_load_timeout_seconds"])) print("[LOG] Driver setup completed", flush=True) print("[LOG] Reading existing records...", flush=True) existing_records, known_shop_urls, output_path = read_existing_records(output_path) print(f"[LOG] Found {len(existing_records)} existing records", flush=True) new_records = [] processed_products = set() processed_shops = set(known_shop_urls) stalled_rounds = 0 MAX_STALLED = 10 if worker_count > 1: print("هشدار: هر Thread مرورگر جداگانه دارد؛ اگر کپچا نمایش داده شود، shop_worker_count را روی 1 بگذارید.") if only_without_website: print("حالت خروجی: فقط فروشگاه‌های بدون وب‌سایت") else: print("حالت خروجی: همه فروشگاه‌ها (با سایت و بدون سایت؛ تب‌های آنلاین و حضوری)") if unlimited_results: print("سقف ثبت جدید در این اجرا: بدون محدودیت", flush=True) else: print(f"سقف ثبت جدید در این اجرا: {max_records}", flush=True) print("[LOG] Starting main scraping loop...", flush=True) if not safe_get(driver, "https://torob.com", "صفحه اصلی ترب", config): print("[ERROR] Failed to load Torob homepage", flush=True) return if not safe_get(driver, target_url, "صفحه دسته‌بندی ترب", config): print("[ERROR] Failed to load category page", flush=True) return wait_between(3, 4) if only_without_website: enable_offline_filter(driver) while (unlimited_results or len(new_records) < max_records) and stalled_rounds < MAX_STALLED: product_urls = [url for url in extract_product_urls(driver) if url not in processed_products] if not product_urls: stalled_rounds += 1 print(f"-> محصول جدیدی یافت نشد، تلاش {stalled_rounds}/{MAX_STALLED} برای اسکرول...", flush=True) driver.execute_script("window.scrollBy(0, Math.max(900, window.innerHeight));") wait_between(2, 3) continue stalled_rounds = 0 for product_url in product_urls: if not unlimited_results and len(new_records) >= max_records: break processed_products.add(product_url) print(f"بررسی کالا: {product_url}") if not safe_get(driver, product_url, "صفحه کالا", config): continue wait_between(2, 3) seller_phones = collect_product_sellers( driver, include_online=not only_without_website, show_all_wait_seconds=int(config.get("show_all_wait_seconds", 8)), ) if seller_phones: print(f"-> مجموع فروشگاه‌های یکتای این کالا: {len(seller_phones)}", flush=True) else: print("-> فروشگاهی در تب‌های قابل‌دسترسی این کالا پیدا نشد.", flush=True) shop_urls = [url for url in seller_phones if url not in processed_shops] processed_shops.update(shop_urls) for shop_url, shop_name, website, shop_phone, error in inspect_shops(shop_urls, config): if not unlimited_results and len(new_records) >= max_records: break if error: print(f"خطا در بررسی {shop_url}: {error}", flush=True) elif website and only_without_website: print(f"رد شد؛ لینک رسمی وب‌سایت دارد: {shop_name} | {website} | {shop_url}", flush=True) else: new_records.append( { "نام فروشگاه": shop_name, "شماره تماس": shop_phone if shop_phone != "ندارد" else seller_phones.get(shop_url, "ندارد"), "دامنه سایت": website or "ندارد", "لینک ترب فروشگاه": shop_url, } ) print( f"ثبت جدید ({len(new_records)}/{'∞' if unlimited_results else max_records}): {shop_name} | " f"{shop_phone if shop_phone != 'ندارد' else seller_phones.get(shop_url, 'ندارد')} | {shop_url}", flush=True, ) output_path = save_records(output_path, existing_records, new_records) if not safe_get(driver, target_url, "بازگشت به دسته‌بندی ترب", config): stalled_rounds += 1 continue wait_between(2, 3) if only_without_website: enable_offline_filter(driver) output_path = save_records(output_path, existing_records, new_records) total_records = len({str(record.get("لینک ترب فروشگاه", "")).strip() for record in existing_records + new_records}) print( f"پایان عملیات. خروجی: {output_path} | ثبت جدید: {len(new_records)} | مجموع فایل: {total_records}", flush=True, ) except Exception as e: print(f"[ERROR] Exception in scrape_torob_shops: {e}", flush=True) import traceback traceback.print_exc() try: output_path = save_records(output_path, existing_records, new_records) except Exception as save_error: print(f"خطا در ذخیره خروجی: {save_error}", flush=True) finally: if 'driver' in locals(): close_driver(driver) if __name__ == "__main__": print("[LOG] Script started", flush=True) parser = argparse.ArgumentParser(description="اسکرپر فروشگاه‌های حضوری ترب") parser.add_argument("--config", type=Path, default=CONFIG_PATH, help="مسیر فایل تنظیمات JSON") arguments = parser.parse_args() print(f"[LOG] Config path: {arguments.config}", flush=True) scrape_torob_shops(arguments.config) print("[LOG] Script finished", flush=True)