多线程爬虫优化：快速爬取并写入CSV

import requests
from bs4 import BeautifulSoup
import pandas as pd
import threading
from queue import Queue
import time

# 代理信息
proxyHost = "www.16yun.cn"
proxyPort = "5445"
proxyUser = "16QMSOML"
proxyPass = "280651"

# 构造代理
proxies = {
    "http": f"http://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}",
    "https": f"https://{proxyUser}:{proxyPass}@{proxyHost}:{proxyPort}",
}

# 全局变量
url_queue = Queue()  # 存储待爬取的URL
data_queue = Queue()  # 存储爬取到的数据
headers = {
    "User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/58.0.3029.110 Safari/537.3"
}

# 爬取单个页面
def crawl_page(url):
    try:
        response = requests.get(url, headers=headers, proxies=proxies, timeout=10)
        response.raise_for_status()  # 检查请求是否成功
        soup = BeautifulSoup(response.text, "html.parser")
        
        products = soup.find_all("div", class_="product-item")
        for product in products:
            name = product.find("h2", class_="product-name").text.strip()
            price = product.find("span", class_="product-price").text.strip()
            sales = product.find("span", class_="product-sales").text.strip()
            reviews = product.find("span", class_="product-reviews").text.strip()
            
            data_queue.put({
                "商品名称": name,
                "价格": price,
                "销量": sales,
                "评价": reviews
            })
    except requests.exceptions.RequestException as e:
        print(f"请求失败：{url}, 错误：{e}")
        print("请检查网页链接的合法性，或者稍后重试。")

# 爬虫线程
def worker():
    while not url_queue.empty():
        url = url_queue.get()
        crawl_page(url)
        url_queue.task_done()

# 主函数
def main():
    # 假设目标网站有多个页面
    base_url = "https://example.com/products?page="
    for page in range(1, 11):  # 爬取前10页
        url_queue.put(base_url + str(page))

    # 创建多个线程
    threads = []
    for _ in range(5):  # 启动5个线程
        t = threading.Thread(target=worker)
        t.start()
        threads.append(t)

    # 等待所有线程完成
    for t in threads:
        t.join()

    # 将数据存储为CSV文件
    data_list = []
    while not data_queue.empty():
        data_list.append(data_queue.get())

    df = pd.DataFrame(data_list)
    df.to_csv("products.csv", index=False, encoding="utf-8-sig")
    print("数据已成功保存到CSV文件中。")

if __name__ == "__main__":
    start_time = time.time()
    main()
    end_time = time.time()
    print(f"总耗时：{end_time - start_time:.2f}秒")

四import pandas as pd

# 加载CSV文件
df = pd.read_csv("products.csv")

# 查看数据的前几行
print(df.head())

# 数据清洗
df.drop_duplicates(inplace=True)  # 删除重复数据
df.dropna(inplace=True)  # 删除缺失值

# 数据分析
print("平均价格：", df["价格"].mean())
print("最高销量商品：", df.loc[df["销量"].idxmax()]["商品名称"])
print("评价分布：", df["评价"].value_counts())

# 数据可视化
import matplotlib.pyplot as plt

df["价格"].plot(kind="hist", bins=20, title="价格分布")
plt.xlabel("价格")
plt.ylabel("商品数量")
plt.show()