selenium 爬取淘宝相关数据

import json

def save(data):
    # 保存数据的文件路径
    file_path = './taobao_data.json'

    # 将数据逐条写入文件，每条数据后添加换行符
    with open(file_path, 'a', encoding='utf-8') as file:
        # 将字典转换为 JSON 字符串
        json_line = json.dumps(data, ensure_ascii=False)
        # 写入文件并添加换行符
        file.write(json_line + '\n')

import csv

def save_product_to_csv(product):
    fieldnames = ['image', 'price', 'deal', 'title', 'shop', 'location']
    csv_file_path = "./taobao_data.csv"

    with open(csv_file_path, 'a', encoding='utf-8', newline='') as file:
        writer = csv.DictWriter(file, fieldnames=fieldnames)

        # 检查文件是否为空
        if file.tell() == 0:
            writer.writeheader()

        writer.writerow(product)

from pymongo import MongoClient

"""
# 连接到 MongoDB（如果尚未连接）
mongo -u root -p 123456 --authenticationDatabase admin

# 切换到 products_db 数据库
use products_db

# 查看 products_db 数据库中的所有集合
show collections

# 查看 products 集合中的前5数据
db.products.find().limit(5).pretty()

# 删除一个集合
db.products.drop()

# 退出 MongoDB Shell
exit
"""

def save_product_to_mongodb(product, db_name, collection_name):

    # TODO MongoDB 连接字符串
    connection_string = "mongodb://root:123456@192.168.10.66:27017/"
    # TODO 连接到 MongoDB
    client = MongoClient(connection_string)

    # TODO 选择数据库
    db = client[db_name]
    # TODO 选择集合
    collection = db[collection_name]

    # TODO 插入数据
    collection.insert_one(product)

from selenium.common.exceptions import TimeoutException, NoSuchElementException
from selenium import webdriver
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from selenium.webdriver.common.by import By
from pyquery import PyQuery as pq
import time
import random
import save_to_json
import save_to_mongo
import save_to_csv


# TODO 浏览器相关配置
options = webdriver.EdgeOptions()
driver = webdriver.Edge(options = options)

# TODO 关键词和最大爬取页数
KEYWORD = "ipad"
MAX_PAGE = 10

# TODO 模拟登录
login_url = f"https://login.taobao.com/member/login.jhtml?spm=a21n57.1.754894437.1.281d523cnqsuAo&f=top&redirectURL=https%3A%2F%2Fs.taobao.com%2Fsearch%3Fq%3D{KEYWORD}"
driver.get(url = login_url)

print("登录成功，等待主页面加载...")
wait = WebDriverWait(driver, 20)

# TODO 数据爬取
def get_data():
    try:
        # TODO 把页面进行从上到下滚动，确保元素加载完全，应对懒加载技术
        for i in range(0, 8):
            driver.execute_script(f'window.scrollTo(0,{i * 1000 - 500})')
            time.sleep(random.randint(1, 2))

        divs = wait.until(EC.visibility_of_all_elements_located((By.CSS_SELECTOR, ".doubleCard--gO3Bz6bu")))
        for div in divs:
            try:
                # TODO 获取元素的innerHTML以便使用Pyquery解析
                inner_html = div.get_attribute('innerHTML')
                doc = pq(inner_html)

                # TODO 获取图像链接
                try:
                    image_element = doc('.mainPic--Ds3X7I8z')
                    image_src = image_element.attr('src')
                except Exception as e:
                    pass

                # TODO 获取价格元素
                price_element1 = doc('.priceInt--yqqZMJ5a').text()
                price_element2 = doc('.priceFloat--XpixvyQ1').text()

                # TODO 获取销售数量，使用try-except防止找不到元素
                try:
                    deal_element = doc('.realSales--XZJiepmt').text()
                except NoSuchElementException:
                    deal_element = "未提供"

                # TODO 获取其他信息
                title_element = doc('.title--qJ7Xg_90').text()
                shop_element = doc('.shopNameText--DmtlsDKm').text()
                location_element = doc('.procity--wlcT2xH9').text()

                product = {
                    'image': image_src,
                    'price': price_element1 + price_element2,
                    'deal': deal_element,
                    'title': title_element,
                    'shop': shop_element,
                    'location': location_element
                }

                # TODO 控制台数据打印
                print(product)

                # TODO 调用函数保存数据到json
                save_to_json.save(product)
                # TODO 调用函数保存数据到csv
                save_to_csv.save_product_to_csv(product)
                # TODO MongoDB数据库和集合名称
                db_name = 'products_db'
                collection_name = 'products'
                # TODO 调用函数保存数据到MongoDB
                save_to_mongo.save_product_to_mongodb(product, db_name, collection_name)

            except NoSuchElementException as e:
                print(f"处理div时发生错误：{e}")
            except TimeoutException:
                print("超时：未能找到元素")

    except TimeoutException:
        print("超时：未找到任何匹配的元素。")

# TODO 翻页爬取
def index_page(page):
    print('正在爬取第 ', page, ' 页')
    if page > 0:
        input = wait.until(EC.presence_of_element_located((By.CSS_SELECTOR, '#search-content-leftWrap > div.leftContent--BdYLMbH8 > div.pgWrap--RTFKoWa6 > div > div > span.next-input.next-medium.next-pagination-jump-input > input')))
        submit = wait.until(EC.element_to_be_clickable((By.CSS_SELECTOR, '#search-content-leftWrap > div.leftContent--BdYLMbH8 > div.pgWrap--RTFKoWa6 > div > div > button.next-btn.next-medium.next-btn-normal.next-pagination-jump-go')))

        input.clear()
        input.send_keys(page)
        submit.click()

        # TODO 调用数据爬取函数
        get_data()

# TODO 主函数，调度翻页批量爬取
def main():
    for i in range(1, MAX_PAGE + 1):
        index_page(i)
        time.sleep(3)

main()