اسکریپینگ پایتون: چارچوب ساده و عملی

مقدمه

در این راهنما قدم‌به‌قدم می‌آموزید چگونه یک چارچوب یا کلاینت عمومی برای اسکریپینگ وب با Python بسازید. هدف این چارچوب: مدیریت retries، اتصال به پراکسی، ذخیره‌سازی نتایج، و ارسال درخواست‌های همزمان است تا اسکریپرهای شما آماده تولید شوند. تا پایان، نمونه‌های کد عملی، ساختار پروژه و نکات مهم درباره پایداری و امنیت را خواهید داشت.

معماری کلی: Crawler و Scraper

به‌طور معمول پروژه را به دو بخش تقسیم می‌کنیم تا مسئولیت‌ها واضح باشد:

• Crawler: اجرا یک جستجو یا فهرست‌بندی، استخراج نتایج اولیه (مثلاً عنوان و لینک) و ذخیرهٔ آن‌ها در فایل (CSV).
• Scraper: خواندن فایل خروجی از کرالر و بازدید از هر لینک برای استخراج داده‌های جزئی‌تر (مثل متادیتا، قیمت، توضیحات) و ذخیرهٔ نهایی (CSV یا JSON).

تقسیم مسئولیت کمک می‌کند که هر بخش را بهینه، تست‌پذیر و قابل مقیاس کنیم.

ارسال درخواست HTTP

برای گرفتن صفحات از وب معمولا از requests استفاده می‌کنیم. نکات مهم:

• همیشه timeout و headers مناسب تنظیم کنید.
• از یک Session برای نگهداری کوکی‌ها و اتصال مجدد استفاده کنید.
• برای GET صفحات ساده کد نمونه زیر پایهٔ خوبی است.

import requests

with requests.Session() as session:
    headers = {"User-Agent": "my-scraper/1.0"}
    resp = session.get("https://duckduckgo.com/?q=learn+rust", headers=headers, timeout=10)
    print(resp.status_code)
    html = resp.text
# ورودی: URL و headers. خروجی: متن HTML یا خطا در صورت timeout.

توضیح خط‌به‌خط: Session اتصال را نگه می‌دارد، timeout از بلوکه شدن جلوگیری می‌کند، و header مانند User-Agent به کاهش ریسک بلاک شدن کمک می‌کند.

پارسینگ با BeautifulSoup

برای استخراج عناصر HTML از BeautifulSoup استفاده کنید. ابتدا ساختار صفحه را بررسی کنید (inspect) و selector مناسب را پیدا کنید.

from bs4 import BeautifulSoup

soup = BeautifulSoup(html, "html.parser")
results = []
for h2 in soup.find_all("h2"):
    a = h2.find("a")
    if not a:
        continue
    title = h2.get_text(strip=True)
    href = a.get("href")
    results.append({"name": title, "url": href})
# ورودی: html، خروجی: لیستی از دیکشنری‌ها

نکته: همیشه برای مقادیر احتمالا خالی از شرط‌ها استفاده کنید تا خطا هنگام None پیش نیاید.

مدیریت صفحه‌بندی (Pagination)

بعضی صفحات با لینک‌های Next یا پارامترهای صفحه کار می‌کنند. الگو:

1. از صفحهٔ فعلی لینک href مربوط به صفحهٔ بعد را استخراج کنید.
2. تا زمانی که لینک وجود دارد یا شرط دلخواه برقرار است، درخواست بزنید و نتایج را جمع‌آوری کنید.

در سایت‌هایی که pagination با جاوااسکریپت بارگذاری می‌شود، باید رندر کردن را در نظر بگیرید (بخش پراکسی و render).

استفاده از پراکسی و رندر کردن (Geolocation & Anti-bot)

برای کنترل موقعیت جغرافیایی و کنار زدن برخی قوانین anti-bot، می‌توانیم درخواست‌ها را از طریق یک سرویس پراکسی با پارامترهای کشور و زمان انتظار برای رندر ارسال کنیم.

from urllib.parse import urlencode

API_KEY = "..."

def get_proxy_url(url, country="us", wait=None):
    payload = {"api_key": API_KEY, "url": url, "country": country}
    if wait:
        payload["wait"] = wait
    proxy_url = "https://proxy.example/v1/?" + urlencode(payload)
    return proxy_url

# سپس از proxy_url در requests.get استفاده کنید.

نکات امنیتی و عملی: هرگز API key را مستقیم در سورس عمومی نگه ندارید؛ از فایل config با دسترسی محدود یا متغیر محیطی استفاده کنید. همچنین پارامتر wait زمانی مفید است که سایت محتوا را با جاوااسکریپت بارگذاری می‌کند.

دیتاکلاس‌ها و پایپلاین ذخیره‌سازی

برای ایمن‌تر و قابل‌اعتمادتر شدن داده‌ها از dataclass استفاده کنید و یک کلاس pipeline برای صف‌بندی و ذخیره‌سازی بنویسید. مزایا: تایپ‌سیف بودن، راحتی تبدیل به CSV/JSON و حذف موارد تکراری.

from dataclasses import dataclass, asdict, fields
import csv, json, os

@dataclass
class SearchData:
    name: str = ""
    base_url: str = ""
    url: str = ""
    result_number: int = 0

class DataPipeline:
    def __init__(self, filename="", storage_queue_limit=50, output_format="csv"):
        self.storage_queue = []
        self.storage_queue_limit = storage_queue_limit
        self.filename = filename
        self.output_format = output_format.lower()

    def save_to_csv(self):
        data = [asdict(i) for i in self.storage_queue]
        if not data:
            return
        keys = list(data[0].keys())
        file_exists = os.path.isfile(self.filename) and os.path.getsize(self.filename) > 0
        with open(self.filename, "a", newline="", encoding="utf-8") as f:
            writer = csv.DictWriter(f, fieldnames=keys)
            if not file_exists:
                writer.writeheader()
            writer.writerows(data)
        self.storage_queue.clear()

    def add_data(self, item):
        self.storage_queue.append(item)
        if len(self.storage_queue) >= self.storage_queue_limit:
            self.save_to_csv()

    def save(self):
        if self.output_format == "csv":
            self.save_to_csv()
        elif self.output_format == "json":
            # معادل ذخیره JSON
            pass

شرح: add_data داده را به صف اضافه می‌کند و در صورتی که به حد نساب رسیده باشد، آن‌ها را ذخیره می‌کند. save_to_csv فایل را باز کرده، header را اضافه می‌کند و ردیف‌ها را می‌نویسد.

تابع کرالینگ نمونه با مدیریت Retries

الگوی عمومی برای تلاش مجدد (retry) و لاگ‌گیری:

import logging
from urllib.parse import urlparse
from bs4 import BeautifulSoup

logger = logging.getLogger(__name__)

def scrape_search_results(keyword, location, data_pipeline=None, retries=3):
    formatted = keyword.replace(" ", "+")
    url = f"https://duckduckgo.com/?q={formatted}&t=h_&ia=web"
    tries = 0
    success = False
    while tries <= retries and not success:
        try:
            resp = requests.get(get_proxy_url(url, country=location, wait=5))
            logger.info(f"Received [{resp.status_code}] from: {url}")
            if resp.status_code != 200:
                raise Exception(f"Failed request: {resp.status_code}")
            soup = BeautifulSoup(resp.text, "html.parser")
            headers = soup.find_all("h2")
            for idx, header in enumerate(headers):
                link = header.find("a")
                if not link:
                    continue
                href = link.get("href")
                parsed = urlparse(href)
                base = f"{parsed.scheme}://{parsed.netloc}"
                sd = SearchData(name=header.get_text(strip=True), base_url=base, url=href, result_number=idx)
                if data_pipeline:
                    data_pipeline.add_data(sd)
            success = True
        except Exception as e:
            logger.error(f"Error processing {url}: {e}")
            tries += 1
            logger.info(f"Retrying ({tries}/{retries})")
    if not success:
        raise Exception("Max retries exceeded")

توضیح: حلقهٔ while تا زمانی که tries کمتر از حد مجاز است تلاش می‌کند. در خطاها شمارنده افزایش می‌یابد و در صورت تکرار ناکام، استثنا پرتاب می‌شود.

اسکرپر: پردازش نتایج و استخراج متادیتا

تابع پردازش هر نتیجه که از CSV خوانده می‌شود:

def process_result(row, location, retries=3, data_pipeline=None):
    url = row["url"]
    tries = 0
    success = False
    while tries <= retries and not success:
        resp = requests.get(get_proxy_url(url, location))
        try:
            if resp.status_code == 200:
                soup = BeautifulSoup(resp.text, "html.parser")
                head = soup.find("head")
                title = head.find("title").text if head and head.find("title") else "n/a"
                desc_el = head.select_one("meta[name='description']") if head else None
                description = desc_el.get("content") if desc_el else "n/a"
                meta = {"name": title, "url": url, "description": description}
                print(meta)
                if data_pipeline:
                    data_pipeline.add_data(meta)
                success = True
            else:
                raise Exception(f"Bad status: {resp.status_code}")
        except Exception as e:
            logger.error(f"Exception: {e}")
            tries += 1
    if not success:
        raise Exception("Max retries exceeded")

این تابع متادیتا را استخراج کرده و به پایپلاین اضافه می‌کند. توجه کنید که برای سایت‌های مختلف باید قوانین پارسینگ اختصاصی بنویسید.

افزودن همزمانی با ThreadPoolExecutor

برای سرعت‌بخشی پردازش نتایج، از concurrent.futures.ThreadPoolExecutor استفاده کنید:

import concurrent.futures

def process_results(csv_file, location, max_threads=5, retries=3, data_pipeline=None):
    with open(csv_file, newline="") as f:
        rows = list(csv.DictReader(f))
    with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
        executor.map(process_result,
                     rows,
                     [location] * len(rows),
                     [retries] * len(rows),
                     [data_pipeline] * len(rows))

نکته: ThreadPool مناسب I/O-bound tasks است. اگر نیاز به اجرای جاوااسکریپت (browser rendering) دارید، ممکن است از ابزارهایی مثل headless browser استفاده کنید که هزینهٔ سنگین‌تری دارند.

نکات پایداری، امنیت و بهترین شیوه‌ها

• توقیت‌ها: همیشه timeout و حداکثر retry منطقی تعیین کنید و از backoff نمایی بین تلاش‌ها استفاده کنید.
• Rate limiting: به سایت احترام بگذارید؛ با sleep بین درخواست‌ها یا تنظیم نرخ درخواست از بلوک شدن جلوگیری کنید.
• Headers و استراتژی‌های rotate پراکسی / user-agent برای کاهش شناسایی استفاده کنید.
• محیط تولید: لاگ‌گذاری ساختاربندی‌شده، مانیتورینگ و هشدار برای خطاها ضروری است.
• ایمنی: کلیدها و فایل‌های حساس را در متغیرهای محیطی نگه دارید، لاگ حساس را فیلتر کنید.

ملاحظات حقوقی و اخلاقی

قبل از اسکریپ کردن هر سایت:

• robots.txt و terms of service سایت را بررسی کنید.
• از گرفتن داده‌های خصوصی یا محتوای محافظت‌شده خودداری کنید.
• دادهٔ شخصی را منتشر یا به اشتراک نگذارید؛ قوانین حریم خصوصی را رعایت کنید.
• اگر ترافیک زیادی ایجاد می‌کنید، با صاحبان سایت هماهنگ کنید یا از API رسمی استفاده کنید.

جمع‌بندی

این مقاله چارچوبی عملی برای ساخت یک کلاینت اسکریپینگ با Python ارائه کرد: معماری crawler/scraper، ارسال امن درخواست‌ها، پارسینگ با BeautifulSoup، ذخیره‌سازی با dataclass و pipeline، استفاده از پراکسی برای رندر و موقعیت، و مدیریت همزمانی با ThreadPoolExecutor. با رعایت نکات پایداری، امنیت و اخلاقی می‌توانید اسکریپرهایی قابل‌اعتماد و آمادهٔ تولید بسازید.

توصیه‌های بعدی

• برای سایت‌های دینامیک بررسی کنید که آیا رندر سمت سرور یا headless browser نیاز است.
• برای حجم بزرگ از داده به صف‌بندی کاری (job queue) و مانیتورینگ فکر کنید.
• همیشه تست‌های واحد برای بخش‌های پارسینگ بنویسید تا تغییرات ساختار HTML باعث شکست کلی نشود.