مقدمه
در این مقاله عملی یاد میگیریم چگونه با استفاده از Selenium و اصول مهندسی، پروفایلهای عمومی LinkedIn را جمعآوری (scrape) کنیم. هدف این راهنما ارائهٔ یک معماری قابلاعتماد شامل کراولر برای پیدا کردن پروفایلها، اسکریپر برای استخراج جزئیات از هر پروفایل، مدیریت ذخیرهسازی و همزمانی است. پس از خواندن این مطلب باید بتوانید یک پروژه پایتون با مکانیزم retry، استفاده از proxy برای عبور از آنتیباتها، و ذخیره ایمن نتایج در CSV راهاندازی کنید.
معماری کلی و جریان کاری
بهطور خلاصه ما دو مؤلفهٔ اصلی خواهیم داشت:
- کراولر (Profile Crawler): جستجو بر اساس کلمات کلیدی، پارس کردن کارتهای نتایج و ذخیرهٔ آدرسهای پروفایل در CSV.
- اسکریپر (Profile Scraper): خواندن CSV، باز کردن هر پروفایل، استخراج JSON داخل تگ script[type='application/ld+json'] و نوشتن دادهٔ نهایی.
عناصری که باید در پیادهسازی رعایت شوند:
- مدیریت خطا و retry.
- همزمانی با ThreadPoolExecutor برای افزایش throughput.
- استفاده از پروکسی و geolocation برای کاهش ریسک بلاک شدن.
- پایپلاین ذخیرهسازی (buffering و نوشتن دستهای در CSV) و حذف تکراریها.
راهاندازی پروژه
نکات سریع راهاندازی:
- ایجاد virtual environment و نصب selenium.
- دقت کنید Chromedriver یا Chrome for Testing نسخهٔ مناسب داشته باشید.
- یک فایل config.json برای نگهداری API_KEY پروکسی بسازید.
تابع تولید URL پروکسی (برای عبور از آنتیبات)
ایده: یک URL به API پروکسی میفرستیم که آن را برایمان ریدایرکت یا تونل میکند و با پارامتر country میتوانیم geolocation تعیین کنیم.
from urllib.parse import urlencode
API_KEY = "" # مقدار را از config.json بخوانید
def get_scrapeops_url(url, location="us"):
payload = {
"api_key": API_KEY,
"url": url,
"country": location,
}
proxy_url = "https://proxy.scrapeops.io/v1/?" + urlencode(payload)
return proxy_url
توضیح: ورودی url و location است. خروجی رشتهٔ URL است که به جای دسترسی مستقیم به سایت، از طریق پروکسی صدا زده میشود. این کار باعث میشود درخواستها از آیپیها و مکانهای متفاوت خارج شوند و احتمال بلاک شدن کمتر گردد.
ساختار داده و پایپلاین ذخیرهسازی
برای نگهداری نتایج از دو dataclass استفاده میکنیم: یکی برای نتایج جستجو (SearchData) و دیگری برای دادهٔ پروفایل کامل (ProfileData). پایپلاین وظیفهٔ بافر کردن، حذف تکراریها و نوشتن دستهای در CSV را دارد.
from dataclasses import dataclass, fields, asdict
import csv, os
@dataclass
class ProfileData:
name: str = ""
company: str = ""
company_profile: str = ""
job_title: str = ""
followers: int = 0
class DataPipeline:
def __init__(self, csv_filename="", storage_queue_limit=50):
self.names_seen = []
self.storage_queue = []
self.storage_queue_limit = storage_queue_limit
self.csv_filename = csv_filename
self.csv_file_open = False
def save_to_csv(self):
self.csv_file_open = True
data_to_save = []
data_to_save.extend(self.storage_queue)
self.storage_queue.clear()
if not data_to_save:
return
keys = [field.name for field in fields(data_to_save[0])]
file_exists = os.path.isfile(self.csv_filename) and os.path.getsize(self.csv_filename) > 0
with open(self.csv_filename, mode="a", newline="", encoding="utf-8") as output_file:
writer = csv.DictWriter(output_file, fieldnames=keys)
if not file_exists:
writer.writeheader()
for item in data_to_save:
writer.writerow(asdict(item))
self.csv_file_open = False
def is_duplicate(self, input_data):
if input_data.name in self.names_seen:
return True
self.names_seen.append(input_data.name)
return False
def add_data(self, scraped_data):
if not self.is_duplicate(scraped_data):
self.storage_queue.append(scraped_data)
if len(self.storage_queue) >= self.storage_queue_limit and not self.csv_file_open:
self.save_to_csv()
def close_pipeline(self):
if len(self.storage_queue) > 0:
self.save_to_csv()
توضیح: add_data ابتدا تکراریها را چک میکند، سپس داده را به بافر اضافه میکند و در صورت پر شدن بافر آن را به CSV مینویسد. save_to_csv header را زمانی که فایل وجود ندارد مینویسد.
ایجاد کراولر برای نتایج جستجو
وظیفهٔ کراولر این است که صفحهٔ نتایج جستجو را باز کند، کارتهای پروفایل را پیدا کند و اطلاعات اولیه را استخراج کند (مانند لینک پروفایل و نام نمایشی).
from selenium import webdriver
from selenium.webdriver.common.by import By
options = webdriver.ChromeOptions()
options.add_argument("--headless")
# دقت کنید: خاموش کردن جاوااسکریپت میتواند باعث تغییر نمایش صفحه و از دست رفتن اطلاعات شود
def crawl_profiles(name, location, data_pipeline=None, retries=3):
# نمونهٔ ورودی: name = "bill gates"
first_name = name.split()[0]
last_name = name.split()[-1]
url = f"https://www.linkedin.com/pub/dir?firstName={first_name}&lastName={last_name}&trk=people-guest_people-search-bar_search-submit"
tries = 0
success = False
while tries <= retries and not success:
driver = webdriver.Chrome(options=options)
try:
scrapeops_proxy_url = get_scrapeops_url(url, location=location)
driver.get(scrapeops_proxy_url)
profile_cards = driver.find_elements(By.CSS_SELECTOR, "div.base-search-card__info")
for card in profile_cards:
parent = card.find_element(By.XPATH, "..")
href = parent.get_attribute("href").split("?")[0]
display_name = card.find_element(By.CSS_SELECTOR, "h3.base-search-card__title").text
location_text = card.find_element(By.CSS_SELECTOR, "p.people-search-card__location").text
companies = "n/a"
has_companies = card.find_elements(By.CSS_SELECTOR, "span.entity-list-meta__entities-list")
if has_companies:
companies = has_companies[0].text
# ساخت نمونهٔ داده و اضافه به پایپلاین
from dataclasses import dataclass
@dataclass
class SearchData:
name: str
display_name: str
url: str
location: str
companies: str
search_data = SearchData(name=name, display_name=display_name, url=href, location=location_text, companies=companies)
if data_pipeline:
data_pipeline.add_data(search_data)
success = True
except Exception as e:
tries += 1
finally:
driver.quit()
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
توضیح خطبهخط: ابتدا URL جستجو میسازیم، سپس با پروکسی صفحه را باز میکنیم، با find_elements کارتها را میگیریم و از هر کارت لینک پروفایل، نام و موقعیت را استخراج میکنیم. دادهها داخل یک SearchData بسته میشوند و به پایپلاین فرستاده میشوند.
ساخت اسکریپر پروفایل (استخراج JSON داخل head)
در صفحهٔ مشخصِ یک پروفایل، بسیاری از اطلاعات ساختاریافته داخل یک اسکریپت JSON در head قرار دارد. بهجای پارس HTML سطحی، آن JSON قابلاعتمادتر و کاملتر است.
def scrape_profile(row, location, retries=3):
# ورودی row یک dict با فیلد 'url' و 'name' است
url = row['url']
tries = 0
success = False
while tries <= retries and not success:
driver = webdriver.Chrome(options=options)
try:
driver.get(get_scrapeops_url(url))
head = driver.find_element(By.CSS_SELECTOR, "head")
script = head.find_element(By.CSS_SELECTOR, "script[type='application/ld+json']")
json_data_graph = json.loads(script.get_attribute("innerHTML"))['@graph']
json_data = {}
for element in json_data_graph:
if element.get('@type') == 'Person':
json_data = element
break
# پیشفرضها
company = "n/a"
company_profile = "n/a"
job_title = "n/a"
followers = 0
if 'jobTitle' in json_data and isinstance(json_data['jobTitle'], list) and len(json_data['jobTitle']) > 0:
job_title = json_data['jobTitle'][0]
if 'worksFor' in json_data and len(json_data['worksFor']) > 0:
company = json_data['worksFor'][0].get('name', 'n/a')
if 'url' in json_data['worksFor'][0]:
company_profile = json_data['worksFor'][0]['url']
if 'interactionStatistic' in json_data:
stats = json_data['interactionStatistic']
if stats.get('name') == 'Follows' and stats.get('@type') == 'InteractionCounter':
followers = stats.get('userInteractionCount', 0)
profile_data = ProfileData(name=row['name'], company=company, company_profile=company_profile, job_title=job_title, followers=followers)
person_pipeline = DataPipeline(f"{row['name']}.csv")
person_pipeline.add_data(profile_data)
person_pipeline.close_pipeline()
success = True
except Exception as e:
tries += 1
finally:
driver.quit()
if not success:
raise Exception(f"Max Retries exceeded: {retries}")
توضیح: ورودی این تابع یک سطر خواندهشده از CSV است. ابتدا JSON داخل تگ اسکریپت را میخوانیم، دنبال گرهای با @type == 'Person' میگردیم و فیلدهای موردنیاز را با مقادیر پیشفرض استخراج میکنیم. سپس یک نمونهٔ ProfileData میسازیم و به پایپلاین محلی اضافه میکنیم.
همزمانی (Concurrency) و بهینهسازی
برای افزایش سرعت از ThreadPoolExecutor استفاده کنید. الگو این است که آرایهای از ورودیها و آرایههایی با پارامترهای ثابت بسازید و با executor.map توابع را روی تردها اجرا کنید.
import concurrent.futures
def start_crawl(profile_list, location, data_pipeline=None, max_threads=5, retries=3):
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(
crawl_profiles,
profile_list,
[location] * len(profile_list),
[data_pipeline] * len(profile_list),
[retries] * len(profile_list),
)
def process_results(csv_file, location, max_threads=5, retries=3):
with open(csv_file, newline="") as file:
reader = list(csv.DictReader(file))
with concurrent.futures.ThreadPoolExecutor(max_workers=max_threads) as executor:
executor.map(scrape_profile, reader, [location] * len(reader), [retries] * len(reader))
نکتهٔ مهم: هر ترد یک نمونهٔ مرورگر ایجاد میکند؛ پس میزان منابع (CPU/RAM) را هنگام انتخاب max_threads در نظر بگیرید.
نکات امنیتی، اخلاقی و محدودیتها
- پیش از اسکریپ کردن، robots.txt و قوانین سرویس را بررسی کنید؛ حتی اگر دادهها عمومی باشند، تخطی از قوانین پلتفرم ممکن است پیامدهای حقوقی یا مسدودی حساب داشته باشد.
- محدودیت نرخ (rate limiting) را رعایت کنید؛ استفاده از retry و backoff الگوریتمی (مثلاً exponential backoff) به پایداری کمک میکند.
- برای امنیت دادههای ذخیرهشده، فایلهای خروجی را با دسترسی محدود ذخیره کنید و کلیدها را در فایلهای config نگهداری کنید، نه داخل کد.
- خاموش کردن جاوااسکریپت در Selenium ممکن است برخی عناصر را از دست بدهد؛ اگر صفحه وابسته به JS است، headless بدون غیرفعالسازی JS را در نظر بگیرید.
جمعبندی
در این راهنما ساختار، کد نمونه و بهترین روشهای مبتدی تا نیمهپیشرفته برای اسکریپینگ پروفایلهای LinkedIn با Selenium را دیدید: از طراحی کراولر و پایپلاین ذخیرهسازی گرفته تا استخراج JSON در head، استفاده از پروکسی برای geolocation و همزمانی با ThreadPoolExecutor. این الگوها را میتوانید برای سایتهای دیگر با ساختار مشابه نیز تطبیق دهید، اما همواره نکات حقوقی و امنیتی را در نظر بگیرید.
