构建1688店铺商品数据集：Python爬虫数据采集与格式化实践

import requests
from parsel import Selector
import pandas as pd
import time
import random

class Ali1688Spider:
    def __init__(self):
        self.session = requests.Session()
        # 设置通用的请求头，模拟浏览器
        self.headers = {
            'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/91.0.4472.124 Safari/537.36',
            'Accept-Language': 'zh-CN,zh;q=0.9,en;q=0.8',
        }
        self.session.headers.update(self.headers)
        
        # 代理配置信息
        self.proxyHost = "www.16yun.cn"
        self.proxyPort = "5445"
        self.proxyUser = "16QMSOML"
        self.proxyPass = "280651"
        
        # 构建代理认证信息
        self.proxyMeta = f"http://{self.proxyUser}:{self.proxyPass}@{self.proxyHost}:{self.proxyPort}"
        self.proxies = {
            "http": self.proxyMeta,
            "https": self.proxyMeta,
        }

    def get_page(self, url, max_retries=3, use_proxy=True):
        """获取页面HTML，包含简单的重试机制和代理支持"""
        for i in range(max_retries):
            try:
                # 根据参数决定是否使用代理
                if use_proxy:
                    resp = self.session.get(url, timeout=10, proxies=self.proxies)
                else:
                    resp = self.session.get(url, timeout=10)
                    
                resp.raise_for_status() # 如果状态码不是200，抛出异常
                
                # 检查页面内容是否包含反爬提示（根据实际情况调整）
                if "访问受限" in resp.text or "验证码" in resp.text:
                    print(f"第{i+1}次请求可能被反爬，正在重试...")
                    time.sleep(2)
                    continue
                    
                # 检查代理是否正常工作
                if use_proxy and resp.status_code == 200:
                    print(f"第{i+1}次请求成功（使用代理）")
                    
                return resp.text
                
            except requests.exceptions.ProxyError as e:
                print(f"代理连接失败: {e}, 第{i+1}次重试...")
                # 如果代理失败，可以尝试不使用代理
                if i == max_retries - 1:  # 最后一次重试
                    print("尝试不使用代理...")
                    try:
                        resp = self.session.get(url, timeout=10)
                        resp.raise_for_status()
                        return resp.text
                    except:
                        pass
                time.sleep(2)
                
            except requests.exceptions.ConnectTimeout as e:
                print(f"连接超时: {e}, 第{i+1}次重试...")
                time.sleep(2)
                
            except requests.exceptions.RequestException as e:
                print(f"请求失败: {e}, 第{i+1}次重试...")
                time.sleep(2)
                
        return None

    def test_proxy_connection(self):
        """测试代理连接是否正常"""
        test_url = "http://httpbin.org/ip"
        try:
            resp = self.session.get(test_url, timeout=10, proxies=self.proxies)
            if resp.status_code == 200:
                print("代理连接测试成功")
                print(f"当前代理IP: {resp.json()['origin']}")
                return True
            else:
                print("代理连接测试失败")
                return False
        except Exception as e:
            print(f"代理测试异常: {e}")
            return False

# 初始化爬虫
spider = Ali1688Spider()

# 测试代理连接
print("正在测试代理连接...")
spider.test_proxy_connection()

# 示例使用
if __name__ == "__main__":
    # 测试爬取一个页面
    test_url = "https://shop.abc.1688.com/page/offerlist_1.htm"
    html_content = spider.get_page(test_url, use_proxy=True)
    
    if html_content:
        print("页面获取成功！")
        # 这里可以添加后续的解析逻辑
    else:
        print("页面获取失败！")

def parse_product_links(self, html_content):
        """从列表页HTML中解析出所有商品的详情页链接"""
        if not html_content:
            return []
        selector = Selector(text=html_content)
        product_links = []
        
        # 使用XPath定位商品链接元素
        # !!! 注意：此XPath为示例，需要根据目标店铺的实际HTML结构进行调整 !!!
        link_elements = selector.xpath('//div[@class="offer-list-row"]//a[contains(@class, "offer-title")]/@href').getall()
        
        for link in link_elements:
            # 确保链接是完整的URL
            if link.startswith('//'):
                full_link = 'https:' + link
            elif link.startswith('/'):
                # 假设我们知道店铺主域名，这里需要替换成实际的
                full_link = 'https://shop.abc.1688.com' + link
            else:
                full_link = link
            product_links.append(full_link)
        
        return list(set(product_links)) # 去重

# 示例：获取第一页的商品链接
list_page_url = "https://shop.abc.1688.com/page/offerlist_1.htm"
html = spider.get_page(list_page_url)
product_links = spider.parse_product_links(html)
print(f"从第一页获取到 {len(product_links)} 个商品链接")

def parse_product_detail(self, html_content, product_url):
        """解析单个商品详情页，提取商品信息"""
        if not html_content:
            return None
        
        selector = Selector(text=html_content)
        product_info = {}

        # 1. 商品标题
        # !!! 以下所有XPath路径均为示例，必须根据实际页面结构调整 !!!
        title = selector.xpath('//h1[@class="d-title"]/text()').get()
        product_info['title'] = title.strip() if title else None

        # 2. 商品价格 - 1688价格通常复杂，可能有区间，需要拼接
        price_elements = selector.xpath('//span[contains(@class, "price-num")]/text()').getall()
        product_info['price_range'] = '-'.join([p.strip() for p in price_elements if p.strip()]) if price_elements else None

        # 3. 月销量
        sales = selector.xpath('//span[contains(text(), "月销量")]/following-sibling::span/text()').get()
        product_info['monthly_sales'] = sales.strip() if sales else '0'

        # 4. 库存
        stock = selector.xpath('//span[contains(text(), "库存")]/following-sibling::span/text()').get()
        product_info['stock'] = stock.strip() if stock else None

        # 5. 公司名称
        company = selector.xpath('//a[contains(@class, "company-name")]/text()').get()
        product_info['company'] = company.strip() if company else None

        # 6. 商品图片链接
        image_urls = selector.xpath('//div[contains(@class, "image-view")]//img/@src').getall()
        # 处理图片链接，确保是HTTP/HTTPS格式
        processed_image_urls = []
        for img_url in image_urls:
            if img_url.startswith('//'):
                processed_image_urls.append('https:' + img_url)
            else:
                processed_image_urls.append(img_url)
        product_info['image_urls'] = ' | '.join(processed_image_urls) # 用竖线分隔多个图片URL

        # 7. 商品URL
        product_info['product_url'] = product_url

        # 8. 采集时间戳
        product_info['crawl_time'] = pd.Timestamp.now().strftime('%Y-%m-%d %H:%M:%S')

        return product_info

# 示例：解析第一个商品
if product_links:
    first_product_url = product_links[0]
    print(f"正在采集: {first_product_url}")
    detail_html = spider.get_page(first_product_url)
    product_data = spider.parse_product_detail(detail_html, first_product_url)
    print(product_data)
    # 礼貌性延迟，避免请求过快
    time.sleep(random.uniform(1, 3))

def crawl_entire_shop(self, shop_list_url_pattern, start_page=1, max_pages=10):
        """爬取整个店铺的多页商品"""
        all_products_data = []
        current_page = start_page

        while current_page <= max_pages:
            # 构造列表页URL
            list_page_url = shop_list_url_pattern.format(page=current_page)
            print(f"正在爬取第 {current_page} 页: {list_page_url}")

            html_content = self.get_page(list_page_url)
            if not html_content:
                print(f"第 {current_page} 页获取失败，终止爬取。")
                break

            product_links = self.parse_product_links(html_content)
            if not product_links:
                print(f"第 {current_page} 页未找到商品链接，可能已到末页。")
                break

            # 遍历当前页的所有商品链接
            for link in product_links:
                print(f"  正在处理商品: {link}")
                detail_html = self.get_page(link)
                product_info = self.parse_product_detail(detail_html, link)
                if product_info:
                    all_products_data.append(product_info)
                # 重要：在每个商品请求间设置随机延迟，友好爬取
                time.sleep(random.uniform(1, 2))

            current_page += 1
            # 在每页请求后设置一个稍长的延迟
            time.sleep(random.uniform(2, 4))

        return all_products_data

    def save_to_csv(self, data, filename='1688_shop_products.csv'):
        """将数据保存到CSV文件"""
        if not data:
            print("没有数据可保存。")
            return
        df = pd.DataFrame(data)
        df.to_csv(filename, index=False, encoding='utf_8_sig') # 使用utf_8_sig编码支持Excel直接打开中文
        print(f"数据已成功保存到 {filename}, 共计 {len(data)} 条商品记录。")

# 主执行流程
if __name__ == '__main__':
    spider = Ali1688Spider()
    # 假设我们已经分析出了店铺列表页的URL模式，其中 {page} 是页码占位符
    # 请将此模式替换为真实的目标店铺URL模式
    shop_url_pattern = "https://shop.abc.1688.com/page/offerlist_{page}.htm"
    
    # 开始爬取，例如最多爬5页
    all_products = spider.crawl_entire_shop(shop_url_pattern, start_page=1, max_pages=5)
    
    # 保存数据
    spider.save_to_csv(all_products)