broshutjava
Senior Member
Em chào các bác, em đang có BTL crawl data điện thoại di động (tên, giá, url, thông số kỹ thuật) từ Cellphones. Code đã chạy được rồi, nhưng cực kỳ tốn thời gian (em để qua đêm >8 tiếng mới xong 1079 sản phẩm), bác nào giúp em cải tiến hoặc dùng cách khác để crawl sao cho nhanh được không ạ (hình như Cellphones ko public api), chứ em loay hoay cả tuần rồi nản quá
em dev BE, mà ở lớp thầy lại giao cái này để làm nên khá khó khăn vì không có kiến thức nhiều.
Em có vài cải tiến như bỏ cuộn trang, dùng javascript để ấn Button liên tục nhưng code chạy vẫn rất rất lâu. Hỏi bọn gpt, claude thì toàn bug
Em chân thành cảm ơn nhiều ạ!
em dev BE, mà ở lớp thầy lại giao cái này để làm nên khá khó khăn vì không có kiến thức nhiều.Em có vài cải tiến như bỏ cuộn trang, dùng javascript để ấn Button liên tục nhưng code chạy vẫn rất rất lâu. Hỏi bọn gpt, claude thì toàn bug
Em chân thành cảm ơn nhiều ạ!
Python:
import os
import time
import pandas as pd
from selenium import webdriver
from selenium.webdriver.common.by import By
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.keys import Keys
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from fake_useragent import UserAgent
import csv
class CellphonesCrawler:
def __init__(self):
self.chrome_driver_path = r"C:\\Users\\Quang Anh\\Downloads\\chromedriver-win64\\chromedriver-win64\\chromedriver.exe"
os.environ["webdriver.chrome.driver"] = self.chrome_driver_path
self.driver = None
self.options = webdriver.ChromeOptions()
self.ua = UserAgent()
def init_driver(self):
service = Service(self.chrome_driver_path)
self.driver = webdriver.Chrome(service=service, options=self.options)
def close_driver(self):
if self.driver:
self.driver.quit()
def wait_and_scroll(self, wait_time=2):
"""Helper function to wait for page load and scroll gradually."""
time.sleep(wait_time)
total_height = int(self.driver.execute_script("return document.body.scrollHeight"))
for height in range(0, total_height, 300): # Scroll 300px at a time
self.driver.execute_script(f"window.scrollTo(0, {height});")
time.sleep(0.3)
self.driver.execute_script("window.scrollTo(0, document.body.scrollHeight);")
time.sleep(1)
def scrape_product(self, url):
"""Scrape the technical specifications of a product."""
print(f"Đang truy cập URL sản phẩm: {url}")
self.driver.get(url)
self.wait_and_scroll()
specs_dict = {}
try:
# Selectors for technical specifications
spec_container_selectors = [
"ul.technical-content",
"div.technical-info",
"div.product-specs"
]
for container_selector in spec_container_selectors:
try:
spec_container = WebDriverWait(self.driver, 15).until(
EC.presence_of_element_located((By.CSS_SELECTOR, container_selector))
)
spec_items = self.driver.find_elements(By.CSS_SELECTOR,
"li.technical-content-item.is-flex.is-align-items-center.is-justify-content-space-between.p-2, div.spec-item")
for item in spec_items:
try:
title = item.find_element(By.CSS_SELECTOR, "p, span.spec-title").text.strip()
value = item.find_element(By.CSS_SELECTOR, "div, span.spec-value").text.strip()
if title and value and value != "∞":
specs_dict[title] = value
except Exception as e:
print(f"Lỗi khi lấy thông tin từ thông số: {e}")
if specs_dict:
break
except Exception as e:
print(f"Lỗi khi xử lý container thông số: {e}")
except Exception as e:
print(f"Lỗi khi truy cập thông số kỹ thuật: {e}")
return specs_dict
def scrape_all_products(self, main_url):
"""Scrape the product list from the homepage and handle pagination."""
print("Bắt đầu lấy danh sách sản phẩm từ trang chủ...")
self.driver.get(main_url)
self.wait_and_scroll()
products = []
seen_urls = set() # Avoid duplicates
previous_product_count = 0
while True: # Continue until no more products can be loaded
print("Đang crawl sản phẩm hiện tại...")
product_elements = self.driver.find_elements(By.CSS_SELECTOR, "div.product-item")
current_product_count = len(seen_urls)
# Get products that haven't been collected yet
for product in product_elements:
try:
name_element = product.find_element(By.CSS_SELECTOR, "div.product__name > h3")
price_element = product.find_element(By.CSS_SELECTOR, "p.product__price--show")
url_element = product.find_element(By.CSS_SELECTOR, "a.product__link")
name = name_element.text.strip()
price = price_element.text.strip().replace('đ', '').replace('.', '')
url = url_element.get_attribute('href')
if url not in seen_urls: # Avoid duplicates
products.append({
'Tên sản phẩm': name,
'Giá': f"{price}đ",
'URL': url
})
seen_urls.add(url)
except Exception as e:
print(f"Lỗi khi lấy thông tin sản phẩm: {e}")
# Check if we got any new products
if len(seen_urls) == previous_product_count:
print("Không có sản phẩm mới, dừng lại.")
break
previous_product_count = len(seen_urls)
# Try to click 'Load More'
try:
load_more_button = WebDriverWait(self.driver, 10).until(
EC.presence_of_element_located((By.CSS_SELECTOR, "a.button.btn-show-more.button__show-more-product"))
)
if load_more_button.is_displayed():
self.driver.execute_script("arguments[0].scrollIntoView();", load_more_button)
self.driver.execute_script("arguments[0].click();", load_more_button)
print(f"Đã tải thêm sản phẩm. Tổng số sản phẩm hiện tại: {len(seen_urls)}")
self.wait_and_scroll()
else:
print("Không tìm thấy nút 'Xem thêm sản phẩm', dừng lại.")
break
except Exception as e:
print("Không thể bấm nút 'Xem thêm sản phẩm' nữa, dừng lại.")
break
print(f"Đã lấy được {len(products)} sản phẩm.")
return products
def crawl(self, main_url, output_file):
"""Main crawl function."""
self.init_driver()
try:
# Scrape products
products = self.scrape_all_products(main_url)
# Gather all specifications
all_specs = set()
for product in products:
product_specs = self.scrape_product(product['URL'])
product.update(product_specs)
all_specs.update(product_specs.keys())
# Sort specifications for consistency
all_specs = sorted(all_specs)
# Write to CSV
with open(output_file, mode='w', encoding='utf-8-sig', newline='') as file:
writer = csv.writer(file)
headers = ["Tên sản phẩm", "Giá", "URL"] + all_specs
writer.writerow(headers)
for product in products:
row = [
product.get("Tên sản phẩm", "N/A"),
product.get("Giá", "N/A"),
product.get("URL", "N/A"),
] + [product.get(spec, "N/A") for spec in all_specs]
writer.writerow(row)
print(f"Đã lưu dữ liệu vào {output_file}")
finally:
self.close_driver()
if __name__ == "__main__":
crawler = CellphonesCrawler()
crawler.crawl("https://cellphones.com.vn/mobile.html", "cellphones_data.csv")
