淘宝数据爬虫方案

爬取淘宝数据是一个有挑战性的任务，因为淘宝有严格的反爬机制。下面我提供一个基本框架，使用 Selenium 模拟浏览器行为来获取商品信息，包括到手价。
taobao_scraper.py
import time
import random
import json
import pandas as pd
from selenium import webdriver
from selenium.webdriver.chrome.service import Service
from selenium.webdriver.common.by import By
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.common.exceptions import TimeoutException
from webdriver_manager.chrome import ChromeDriverManager
import logging

配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(name)

class TaobaoScraper:
def init(self, username=None, password=None):

    # 初始化浏览器配置
    options = webdriver.ChromeOptions()
    # 添加一些参数来避免被检测为爬虫
    options.add_argument('--disable-blink-features=AutomationControlled')
    options.add_argument('--no-sandbox')
    options.add_argument('--disable-dev-shm-usage')
    options.add_experimental_option('excludeSwitches', ['enable-automation'])
    options.add_experimental_option('useAutomationExtension', False)

    # 初始化浏览器
    self.driver = webdriver.Chrome(service=Service(ChromeDriverManager().install()), options=options)
    self.driver.execute_script("Object.defineProperty(navigator, 'webdriver', {get: () => undefined})")

    self.username = username
    self.password = password
    self.products_data = []

def login(self):
    """登录淘宝账号"""
    if not self.username or not self.password:
        logger.warning("未提供用户名和密码，将以游客身份访问")
        return False

    try:
        self.driver.get("https://login.taobao.com/")
        logger.info("请在30秒内完成扫码登录")
        time.sleep(30)  # 给用户时间扫码登录
        logger.info("登录完成")
        return True
    except Exception as e:
        logger.error(f"登录失败: {str(e)}")
        return False

def search_products(self, keyword, pages=3):
    """搜索商品并爬取多页数据"""
    try:
        self.driver.get(f"https://s.taobao.com/search?q={keyword}")
        logger.info(f"搜索关键词: {keyword}")

        # 等待搜索结果加载
        WebDriverWait(self.driver, 10).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, ".items .item"))
        )

        # 爬取多页数据
        for page in range(1, pages + 1):
            logger.info(f"正在爬取第 {page} 页")
            self._scrape_current_page()

            # 如果不是最后一页，点击下一页
            if page < pages:
                try:
                    next_button = WebDriverWait(self.driver, 10).until(
                        EC.element_to_be_clickable((By.CSS_SELECTOR, ".next"))
                    )
                    next_button.click()
                    # 等待下一页加载
                    time.sleep(random.uniform(2, 5))
                except Exception as e:
                    logger.error(f"点击下一页失败: {str(e)}")
                    break

        return True
    except Exception as e:
        logger.error(f"搜索商品失败: {str(e)}")
        return False

def _scrape_current_page(self):
    """爬取当前页面的商品数据"""
    try:
        # 等待商品列表加载
        WebDriverWait(self.driver, 10).until(
            EC.presence_of_element_located((By.CSS_SELECTOR, ".items .item"))
        )

        # 获取所有商品元素
        items = self.driver.find_elements(By.CSS_SELECTOR, ".items .item")

        for item in items:
            try:
                product_data = self._extract_product_data(item)
                if product_data:
                    self.products_data.append(product_data)
            except Exception as e:
                logger.error(f"提取商品数据失败: {str(e)}")

        logger.info(f"当前页面爬取完成，获取 {len(items)} 个商品")
    except Exception as e:
        logger.error(f"爬取当前页面失败: {str(e)}")

def _extract_product_data(self, item):
    """从商品元素中提取数据"""
    try:
        # 提取商品基本信息
        title = item.find_element(By.CSS_SELECTOR, ".title").text.strip()
        price = item.find_element(By.CSS_SELECTOR, ".price strong").text.strip()

        # 点击商品查看详情页以获取到手价
        product_url = item.find_element(By.CSS_SELECTOR, ".title a").get_attribute("href")
        original_window = self.driver.current_window_handle

        # 打开新标签页访问商品详情
        self.driver.execute_script(f"window.open('{product_url}');")
        self.driver.switch_to.window(self.driver.window_handles[1])

        # 等待详情页加载
        time.sleep(random.uniform(2, 4))

        # 获取到手价（这部分可能需要根据实际页面结构调整）
        try:
            final_price = self.driver.find_element(By.CSS_SELECTOR, ".tb-rmb-num").text.strip()
        except:
            # 如果找不到到手价，尝试从其他可能的元素获取
            try:
                final_price = self.driver.find_element(By.CSS_SELECTOR, ".tb-promo-price strong").text.strip()
            except:
                final_price = price  # 如果都找不到，使用搜索结果页的价格

        # 获取其他信息
        shop_name = self.driver.find_element(By.CSS_SELECTOR, ".tb-shop-name").text.strip()
        sales = self.driver.find_element(By.CSS_SELECTOR, ".tb-sell-counter").text.strip()

        # 关闭详情页，返回搜索结果页
        self.driver.close()
        self.driver.switch_to.window(original_window)

        # 构建商品数据
        product_data = {
            "title": title,
            "price": price,
            "final_price": final_price,
            "shop_name": shop_name,
            "sales": sales,
            "url": product_url
        }

        return product_data
    except Exception as e:
        logger.error(f"提取单个商品详情失败: {str(e)}")
        # 确保回到原始窗口
        if len(self.driver.window_handles) > 1:
            self.driver.close()
            self.driver.switch_to.window(self.driver.window_handles[0])
        return None

def save_to_csv(self, filename="taobao_products.csv"):
    """将爬取的数据保存为CSV文件"""
    if not self.products_data:
        logger.warning("没有数据可保存")
        return False

    try:
        df = pd.DataFrame(self.products_data)
        df.to_csv(filename, index=False, encoding='utf-8-sig')
        logger.info(f"数据已保存到 {filename}")
        return True
    except Exception as e:
        logger.error(f"保存数据失败: {str(e)}")
        return False

def save_to_json(self, filename="taobao_products.json"):
    """将爬取的数据保存为JSON文件"""
    if not self.products_data:
        logger.warning("没有数据可保存")
        return False

    try:
        with open(filename, 'w', encoding='utf-8') as f:
            json.dump(self.products_data, f, ensure_ascii=False, indent=4)
        logger.info(f"数据已保存到 {filename}")
        return True
    except Exception as e:
        logger.error(f"保存数据失败: {str(e)}")
        return False

def close(self):
    """关闭浏览器"""
    if self.driver:
        self.driver.quit()
        logger.info("浏览器已关闭")

使用示例
if name == "main":
scraper = TaobaoScraper()

try:
    # 登录淘宝（可选）
    scraper.login()

    # 搜索商品并爬取数据
    keyword = input("请输入要搜索的商品关键词: ")
    pages = int(input("请输入要爬取的页数: "))
    scraper.search_products(keyword, pages)

    # 保存数据
    scraper.save_to_csv()
    scraper.save_to_json()

finally:
    # 确保浏览器关闭
    scraper.close()

代码说明
这个爬虫程序主要通过以下方式工作：
浏览器模拟：使用 Selenium 模拟真实用户的浏览器行为，降低被检测的风险
数据获取：
从搜索结果页获取商品基本信息
访问商品详情页获取更详细的信息，包括到手价
数据存储：将爬取的数据保存为 CSV 和 JSON 格式，方便后续分析
使用注意事项
1.反爬机制：淘宝有严格的反爬机制，程序可能会遇到验证码或 IP 限制
2.登录需求：某些商品信息可能需要登录才能查看，程序提供了登录选项
3.页面结构变化：淘宝的页面结构可能会变化，需要根据实际情况调整 CSS 选择器
4.使用频率：建议适当控制爬取频率，避免 IP 被封
5.合法使用：请确保你的爬取行为符合相关法律法规和淘宝的用户协议
可能的改进
1.添加代理 IP 池，避免单个 IP 被封
2.增加验证码识别功能
3.优化数据提取逻辑，提高准确率
4.添加多线程或异步处理，提高爬取效率
5.增加更多的商品信息字段，如评价、规格等
6.请注意，由于淘宝的反爬机制，实际使用时可能需要根据最新的网页结构进行调整。