first

.gitignore

+.idea
+.vscode
+.venv
+*.pyc
\ No newline at end of file

.idea/.gitignore

+# 默认忽略的文件
+/shelf/
+/workspace.xml

.idea/inspectionProfiles/Project_Default.xml

+<component name="InspectionProjectProfileManager">
+  <profile version="1.0">
+    <option name="myName" value="Project Default" />
+    <inspection_tool class="PyPep8NamingInspection" enabled="true" level="WEAK WARNING" enabled_by_default="true">
+      <option name="ignoredErrors">
+        <list>
+          <option value="N802" />
+        </list>
+      </option>
+    </inspection_tool>
+  </profile>
+</component>
\ No newline at end of file

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="Black">
+    <option name="sdkName" value="Python 3.10 (test)" />
+  </component>
+  <component name="ProjectRootManager" version="2" project-jdk-name="Python 3.10 (amazon) (2)" project-jdk-type="Python SDK" />
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/test.iml" filepath="$PROJECT_DIR$/.idea/test.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/test.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$">
+      <excludeFolder url="file://$MODULE_DIR$/.venv" />
+    </content>
+    <orderEntry type="jdk" jdkName="Python 3.10 (amazon) (2)" jdkType="Python SDK" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+  <component name="PackageRequirementsSettings">
+    <option name="versionSpecifier" value="强相等 (==x.y.z)" />
+    <option name="removeUnused" value="true" />
+    <option name="modifyBaseFiles" value="true" />
+  </component>
+</module>
\ No newline at end of file

.idea/vcs.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="VcsDirectoryMappings">
+    <mapping directory="$PROJECT_DIR$" vcs="Git" />
+  </component>
+</project>
\ No newline at end of file

bootstrap.py

+# coding: utf-8
+
+import configparser
+import os
+
+from sqlalchemy import create_engine, text
+from sqlalchemy.orm import sessionmaker
+
+from models import Base
+
+# 创建 configparser 对象
+config = configparser.ConfigParser()
+db_session = None
+
+
+def init():
+    # 配置文件路径
+    config_file = 'config.ini'
+    # 检查配置文件是否存在
+    if not os.path.exists(config_file):
+        raise FileNotFoundError(f"配置文件 {config_file} 不存在.")
+
+    # 读取配置文件
+    config.read(config_file)
+
+    init_mysql()
+
+
+def init_mysql():
+    global db_session
+    try:
+        host = config.get('database', 'host')
+        port = config.get('database', 'port')
+        database = config.get('database', 'database')
+        username = config.get('database', 'username')
+        password = config.get('database', 'password')
+
+        engine = create_engine(url=f"mysql+pymysql://{username}:{password}@{host}:{port}/{database}",
+                               pool_recycle=3600,
+                               echo=True)
+
+        # 测试连接是否正常
+        with engine.connect() as connection:
+            connection.execute(text("SELECT 1"))
+
+        Base.metadata.create_all(engine)
+        db_session = sessionmaker(bind=engine)
+
+        print("数据库连接成功，Session 初始化完成。")
+    except Exception as e:
+        raise FileNotFoundError(f"数据库连接失败: {e}")

config.ini

+[database]
+host = localhost
+port = 3306
+database = spring
+username = root
+password = 123456
+
+[redis]
+host = localhost
+port = 6379
+password =
+db = 0
\ No newline at end of file

example/cc.py

+# coding: utf-8
+
+import random
+import time
+
+from curl_cffi import requests
+from lxml import etree
+
+
+def proxies():
+    proxies = {
+        "http": "http://127.0.0.1:7897",
+        "https": "http://127.0.0.1:7897",
+    }
+    return proxies
+
+
+def get_headers():
+    headers = {
+        "accept": "text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8,application/signed-exchange;v=b3;q=0.7",
+        "accept-language": "zh-CN,zh;q=0.9",
+        "cache-control": "max-age=0",
+        "device-memory": "8",
+        "downlink": "9.35",
+        "dpr": "1",
+        "ect": "4g",
+        "priority": "u=0, i",
+        "rtt": "250",
+        "sec-ch-device-memory": "8",
+        "sec-ch-dpr": "1",
+        "sec-ch-ua": "\"Chromium\";v=\"128\", \"Not;A=Brand\";v=\"24\", \"Google Chrome\";v=\"128\"",
+        "sec-ch-ua-mobile": "?0",
+        "sec-ch-ua-platform": "\"Windows\"",
+        "sec-ch-ua-platform-version": "\"15.0.0\"",
+        "sec-ch-viewport-width": "1920",
+        "sec-fetch-dest": "document",
+        "sec-fetch-mode": "navigate",
+        "sec-fetch-site": "none",
+        "sec-fetch-user": "?1",
+        "upgrade-insecure-requests": "1",
+        "viewport-width": "1920"
+    }
+
+    return headers
+
+
+def gen_ac_buk():
+    return f"{random.randint(100, 999)}-{random.randint(1000000, 9999999)}-{random.randint(1000000, 9999999)}"
+
+
+def goodsSearch(keywords):
+    """
+    商品搜索
+    :param keywords:
+    :return:
+    """
+    headers = get_headers()
+    ac_buk = gen_ac_buk()
+    cookies = {
+        "lc-main": "zh_CN",
+        "ubid-acbuk": ac_buk,
+        "crid": "MEKLIWGGWBQG",
+        "sprefix": "沙发,aps,373",
+    }
+
+    url = "https://www.amazon.com/s"
+    params = {
+        "k": keywords,
+        "ref": "nb_sb_noss"
+    }
+
+    response = requests.get(url, headers=headers, cookies=cookies, params=params, impersonate="chrome110", proxies="")
+
+    return response
+
+
+def goodsInfo(url):
+    """
+    获取商品详情页
+    :param url:
+    :return:
+    """
+    headers = get_headers()
+    ac_buk = gen_ac_buk()
+    cookies = {
+        "lc-main": "zh_CN",
+        "ubid-acbuk": ac_buk,
+    }
+    response = requests.get(url, headers=headers, cookies=cookies, params="", impersonate="chrome110", proxies="")
+    return response
+
+
+def goodsListPage(keywords, page):
+    """
+    商品分页页面
+    :param keywords:
+    :param page:
+    :return:
+    """
+    headers = get_headers()
+    ac_buk = gen_ac_buk()
+    cookies = {
+        "lc-main": "zh_CN",
+        # "lc-main": "en_US",
+        "ubid-acbuk": ac_buk,
+    }
+
+    url = "https://www.amazon.com/s"
+    params = {
+        "k": keywords,  # 关键词参数
+        "qid": int(time.time()),  # 当前时间戳
+        "page": page,  # 当前页码
+        "crid": "MEKLIWGGWBQG",
+        "sprefix": "沙发,aps,373",
+    }
+
+    response = requests.get(url, headers=headers, cookies=cookies, params=params, impersonate="chrome110", proxies="")
+
+    return response
+
+
+data_total = 0
+
+
+def get_total_pages(response):
+    """
+    分页总数
+    :param response:
+    :return:
+    """
+    # 解析 HTML 内容
+    tree = etree.HTML(response.text)
+
+    # 找到分页总数的元素 (通过类名找到总页数，如 "s-pagination-item s-pagination-disabled")
+    total_pages = tree.xpath('//span[@class="s-pagination-item s-pagination-disabled"]/text()')
+
+    # 返回总页数，确保返回的是整数
+    return int(total_pages[-1].strip()) if total_pages else 1
+
+
+def process_page_content(response):
+    # 在这里解析每页的响应内容，提取需要的数据
+    tree = etree.HTML(response.text)
+
+    # 查找包含特定 class 的元素
+    # 注意：因为 class 属性可以包含多个值，所以使用 contains() 函数来匹配 class
+    elements = tree.xpath('//div[contains(@class, "a-section") and contains(@class, "a-spacing-base")]')
+
+    # 输出匹配的元素
+    for element in elements:
+        # 检查元素是否包含 "Sponsored" 商品是否为广告
+        is_sponsored = element.xpath('.//span[contains(text(), "Sponsored")]')
+        if is_sponsored:
+            global data_total
+            data_total = data_total + 1
+
+            # 提取 class='a-size-base-plus a-color-base a-text-normal' 的元素的文本值
+            title = element.xpath(
+                './/span[contains(@class, "a-size-base-plus") and contains(@class, "a-color-base") and contains(@class, "a-text-normal")]/text()')
+            if title:
+                print(f"商品名称: {title[0]}")
+
+            # 提取 class='s-image' 的 img 标签中的 src 属性 (图片链接)
+            img_src = element.xpath('.//img[contains(@class, "s-image")]/@src')
+            if img_src:
+                print(f"图片链接: {img_src[0]}")
+
+            # 提取 class='a-link-normal s-no-outline' 的链接
+            link = element.xpath(
+                './/a[contains(@class, "a-link-normal") and contains(@class, "s-no-outline")]/@href')
+            # 输出商品标题和图片链接和产品链接
+            if link:
+                url = "https://www.amazon.com" + link[0]
+                print(f"详情链接: {url}")
+                info_resp = goodsInfo(url)
+                new_tree = etree.HTML(info_resp.text)
+                # 提取 <a id="bylineInfo"> 标签的文本值
+                byline_info = new_tree.xpath('//a[@id="bylineInfo"]/text()')
+                if byline_info:
+                    print(f"店铺名: {byline_info[0]}")
+
+            print("-" * 50)
+
+
+def main():
+    keywords = "沙发"
+
+    # 获取第一页内容
+    first_page_response = goodsSearch(keywords)
+
+    # 获取总页数
+    total_pages = get_total_pages(first_page_response)
+    print(f"总页数: {total_pages}")
+
+    print(f"抓取第 1 页内容...")
+    # 获取第一页数据
+    process_page_content(first_page_response)
+
+    # 从第二页开始循环抓取，直到总页数
+    for page in range(2, total_pages + 1):
+        print(f"抓取第 {page} 页内容...")
+        response = goodsListPage(keywords, page)
+        # 处理每页的响应内容 (可以解析并提取需要的信息)
+        process_page_content(response)
+
+    print("共计:", data_total, "条")
+
+
+if __name__ == "__main__":
+    main()

example/dp.py

+import time
+
+import ddddocr
+from DrissionPage import Chromium
+from DrissionPage import ChromiumPage
+from DrissionPage.errors import ElementNotFoundError, PageDisconnectedError
+from lxml import etree
+
+data_count = 0
+
+
+def get_total_pages(page):
+    total_pages = page.ele('.s-pagination-item s-pagination-disabled').text
+    return int(total_pages) if total_pages else 1
+
+
+def search_goods(page, keywords):
+    # 定位输入框并输入关键字
+    page.ele('#twotabsearchtextbox').input(keywords)
+    page.ele('#nav-search-submit-button').click()
+
+    page.wait.title_change(keywords)
+    # 获取总分页
+    total_pages = get_total_pages(page)
+    print(f"总页数: {total_pages}")
+
+    # 获取商品信息
+    goods_list(page)
+
+    # 从第二页开始循环抓取，直到总页数
+    for pg in range(2, total_pages + 1):
+        pagination = page.ele('.a-section a-text-center s-pagination-container')
+        pagination.ele('.s-pagination-item s-pagination-next s-pagination-button s-pagination-separator').click()
+        print(f"抓取第 {pg} 页内容...")
+        time.sleep(10)
+        goods_list(page)
+
+
+def goods_list(page):
+    # 解析 HTML 内容
+    tree = etree.HTML(page.html)
+    elements = tree.xpath('//div[contains(@class, "a-section") and contains(@class, "a-spacing-base")]')
+
+    table = Chromium()
+
+    for element in elements:
+        try:
+            # 检查元素是否包含 "Sponsored" 商品是否为广告
+            is_sponsored = element.xpath('.//span[contains(text(), "Sponsored")]')
+            if is_sponsored:
+                global data_count
+                data_count = data_count + 1
+
+                # 提取 class='a-size-base-plus a-color-base a-text-normal' 的元素的文本值
+                title = element.xpath(
+                    './/span[contains(@class, "a-size-base-plus") and contains(@class, "a-color-base") and contains(@class, "a-text-normal")]/text()')
+                if title:
+                    print(f"商品名称: {title[0]}")
+
+                # 提取 class='s-image' 的 img 标签中的 src 属性 (图片链接)
+                img_src = element.xpath('.//img[contains(@class, "s-image")]/@src')
+                if img_src:
+                    print(f"图片链接: {img_src[0]}")
+
+                # 提取 class='a-link-normal s-no-outline' 的链接
+                link = element.xpath(
+                    './/a[contains(@class, "a-link-normal") and contains(@class, "s-no-outline")]/@href')
+
+                if link:
+                    url = "https://www.amazon.com" + link[0]
+                    tab = table.new_tab(url=url)
+                    new_tree = etree.HTML(tab.html)
+                    # 提取 <a id="bylineInfo"> 标签的文本值
+                    store_name_text = new_tree.xpath('//a[@id="bylineInfo"]/text()')
+                    if store_name_text:
+                        store_name = store_name_text[0].replace("Visit the ", "")
+                        print(f"店铺名: {store_name}")
+
+                    tab.close()
+
+                    print(f"详情链接: {url}")
+
+                print("-" * 50)
+
+        except ElementNotFoundError:
+            pass
+
+
+def check_captcha(page):
+    """
+    检测图形码
+    :param page:
+    :return:
+    """
+    container = page.ele('.a-row a-text-center')
+    return True if container else False
+
+
+def open_url(page, url):
+    # 访问网页
+    page.get(url)
+
+    # 延迟2秒等待图形码加载
+    page.wait(2)
+
+    while True:
+        time.sleep(1)
+
+        if not check_captcha(page):
+            break
+
+        print("存在图形码开始处理...")
+
+        # 下载图形码
+        img = page('tag:img')
+        img.get_screenshot(name='captcha_image.jpg')
+
+        # 读取图像并进行识别
+        ocr = ddddocr.DdddOcr(show_ad=False)
+        image = open("captcha_image.jpg", "rb").read()
+        result = ocr.classification(image).upper()
+        if result:
+            # 输入验证码提交
+            page.ele('#captchacharacters').input(result)
+            time.sleep(2)
+            page.ele('.a-button-text').click()
+
+
+def main(keywords):
+    page = ChromiumPage()
+    page.set.load_mode.normal()  # 设置为normal模式
+
+    # 访问网页
+    open_url(page, 'https://www.amazon.com')
+
+    search_goods(page, keywords)
+
+    print(f"数据总量 {data_count} 条")
+
+
+if __name__ == '__main__':
+    try:
+        print("请输入你要搜索的商品")
+        search_input = input()
+        print("你输入的是：", search_input)
+        # search_input = "皮鞋"
+        main(search_input)
+    except KeyboardInterrupt:
+        pass
+    except PageDisconnectedError as e:
+        print("与页面的连接已断开")

example/se.py

+import random
+
+from selenium import webdriver
+from selenium.webdriver.common.by import By
+
+
+def gen_ac_buk():
+    return f"{random.randint(100, 999)}-{random.randint(1000000, 9999999)}-{random.randint(1000000, 9999999)}"
+
+
+driver = webdriver.Chrome()
+cookie = {
+    "lc-main": "zh_CN",
+    "ubid-acbuk": gen_ac_buk(),
+}
+
+driver.add_cookie(cookie)
+
+driver.get("https://amazon.com/")
+
+driver.implicitly_wait(0.5)
+
+search_box = driver.find_element(by=By.ID, value="twotabsearchtextbox")
+driver.find_element(by=By.ID, value="nav-search-submit-button").click()
+
+# driver.quit()

main.py

+import re
+import time
+
+import ddddocr
+from DrissionPage import Chromium
+from DrissionPage import ChromiumPage
+from DrissionPage.errors import ElementNotFoundError, PageDisconnectedError
+from lxml import etree
+
+import tasks
+from bootstrap import init
+
+data_count = 0
+
+init()
+
+
+def get_total_pages(page):
+    total_pages = page.ele('.s-pagination-item s-pagination-disabled').text
+    return int(total_pages) if total_pages else 1
+
+
+def search_goods(page, keywords):
+    # 定位输入框并输入关键字
+    page.ele('#twotabsearchtextbox').input(keywords)
+    page.ele('#nav-search-submit-button').click()
+
+    page.wait.title_change(keywords)
+    # 获取总分页
+    total_pages = get_total_pages(page)
+    print(f"总页数: {total_pages}")
+
+    # 获取商品信息
+    goods_list(page)
+
+    # 从第二页开始循环抓取，直到总页数
+    for pg in range(2, total_pages + 1):
+        pagination = page.ele('.a-section a-text-center s-pagination-container')
+        pagination.ele('.s-pagination-item s-pagination-next s-pagination-button s-pagination-separator').click()
+        print(f"抓取第 {pg} 页内容...")
+        time.sleep(10)
+        goods_list(page)
+
+
+def goods_list(page):
+    # 解析 HTML 内容
+    tree = etree.HTML(page.html)
+    elements = tree.xpath('//div[contains(@class, "a-section") and contains(@class, "a-spacing-base")]')
+
+    table = Chromium()
+
+    for element in elements:
+        try:
+            # 检查元素是否包含 "Sponsored" 商品是否为广告
+            is_sponsored = element.xpath('.//span[contains(text(), "Sponsored")]')
+            if is_sponsored:
+                global data_count
+                data_count = data_count + 1
+
+                data = {}
+
+                # 提取 class='a-size-base-plus a-color-base a-text-normal' 的元素的文本值
+                goods_name = element.xpath(
+                    './/span[contains(@class, "a-size-base-plus") and contains(@class, "a-color-base") and contains(@class, "a-text-normal")]/text()')
+                if goods_name:
+                    # print(f"商品名称: {title[0]}")
+                    data['goods_name'] = goods_name[0]
+
+                # 提取 class='s-image' 的 img 标签中的 src 属性 (图片链接)
+                img_src = element.xpath('.//img[contains(@class, "s-image")]/@src')
+                if img_src:
+                    # print(f"图片链接: {img_src[0]}")
+                    data['img_src'] = img_src[0]
+
+                # 提取 class='a-link-normal s-no-outline' 的链接
+                link = element.xpath(
+                    './/a[contains(@class, "a-link-normal") and contains(@class, "s-no-outline")]/@href')
+
+                if link:
+                    url = "https://www.amazon.com" + link[0]
+                    data['url'] = url
+
+                    tab = table.new_tab(url=url)
+                    new_tree = etree.HTML(tab.html)
+                    # 提取 <a id="bylineInfo"> 标签的文本值
+                    store_name_text = new_tree.xpath('//a[@id="bylineInfo"]/text()')
+                    if store_name_text:
+                        store_name = re.sub(r'^(Visit the |访问 )', '', store_name_text[0])
+                        data['store_name'] = store_name
+
+                    tab.close()
+
+                    # print(f"详情链接: {url}")
+                # print(data)
+                tasks.process_data(data)
+                # print("-" * 50)
+
+        except ElementNotFoundError:
+            pass
+
+
+def check_captcha(page):
+    """
+    检测图形码
+    :param page:
+    :return:
+    """
+    container = page.ele('.a-row a-text-center')
+    return True if container else False
+
+
+def open_url(page, url):
+    # 访问网页
+    page.get(url)
+
+    # 延迟2秒等待图形码加载
+    page.wait(2)
+
+    while True:
+        time.sleep(1)
+
+        if not check_captcha(page):
+            break
+
+        print("存在图形码开始处理...")
+
+        # 下载图形码
+        img = page('tag:img')
+        img.get_screenshot(name='captcha_image.jpg')
+
+        # 读取图像并进行识别
+        ocr = ddddocr.DdddOcr(show_ad=False)
+        image = open("captcha_image.jpg", "rb").read()
+        result = ocr.classification(image).upper()
+        if result:
+            # 输入验证码提交
+            page.ele('#captchacharacters').input(result)
+            time.sleep(2)
+            page.ele('.a-button-text').click()
+
+
+def main(keywords):
+    page = ChromiumPage()
+    page.set.load_mode.normal()  # 设置为normal模式
+
+    # 访问网页
+    open_url(page, 'https://www.amazon.com')
+
+    search_goods(page, keywords)
+
+    print(f"数据总量 {data_count} 条")
+
+
+if __name__ == '__main__':
+    try:
+        print("请输入你要搜索的商品")
+        search_input = input()
+        print("你输入的是：", search_input)
+        # search_input = "皮鞋"
+        main(search_input)
+    except KeyboardInterrupt:
+        pass
+    except PageDisconnectedError as e:
+        print("与页面的连接已断开")

models.py

+# coding: utf-8
+from sqlalchemy import Column, Integer, String, Text
+from sqlalchemy.orm import declarative_base
+
+# 创建基础类
+Base = declarative_base()
+
+
+class BaseModel(Base):
+    __abstract__ = True
+    __table_args__ = {
+        'mysql_engine': 'InnoDB',  # 指定全局的 InnoDB 引擎
+        'mysql_charset': 'utf8mb4'  # 设置字符集
+    }
+
+
+class Store(BaseModel):
+    __tablename__ = 'store'
+    id = Column(Integer, primary_key=True)
+    name = Column(String(100), comment='店铺名称')
+    code = Column(String(50), comment='店铺编码')
+    __table_args__ = ({'comment': '店铺表'})
+
+
+class Category(BaseModel):
+    __tablename__ = 'category'
+    id = Column(Integer, primary_key=True)
+    name = Column(String(50), comment='分类名称')
+    __table_args__ = ({'comment': '商品分类表'})
+
+
+class Goods(BaseModel):
+    __tablename__ = 'goods'
+    id = Column(Integer, primary_key=True)
+    store_id = Column(Integer, comment='门店id')
+    name = Column(String(255), comment='商品名称')
+    image = Column(String(255), comment='商品图片')
+    url = Column(Text, comment='商品地址')
+    __table_args__ = ({'comment': '商品表'})
+
+
+class Comment(BaseModel):
+    __tablename__ = 'comments'
+    id = Column(Integer, primary_key=True)
+    content = Column(String(50), comment='内容')
+    goods_id = Column(Integer, comment='商品id')
+    store_id = Column(Integer, comment='门店id')
+    __table_args__ = ({'comment': '商品信息表'})

requirements.txt

+curl_cffi==0.7.1
+lxml==5.3.0
+DrissionPage==4.1.0.2
+ddddocr==1.5.5
+SQLAlchemy==2.0.35
+pymysql==1.1.1
+huey==2.5.1
+redis==5.0.8
+selenium==4.25.0
\ No newline at end of file

tasks.py

+# coding: utf-8
+from huey import RedisHuey
+
+import bootstrap
+from models import Store, Goods
+
+bootstrap.init()
+
+config = bootstrap.config
+host = config.get('redis', 'host')
+port = config.get('redis', 'port')
+db = config.get('redis', 'db')
+password = config.get('redis', 'password')
+# 创建 Huey 应用，指定 Redis 配置
+huey = RedisHuey(host=host, port=port, db=db, password=password)
+
+
+@huey.task()
+def process_data(data: dict):
+    orm = bootstrap.db_session()
+    try:
+        store_name = data['store_name']
+        # 查询店铺是否存在
+        store_res = orm.query(Store).filter_by(name=store_name).first()
+        if not store_res:
+            store_data = Store(name=store_name)
+            orm.add(store_data)
+            orm.commit()
+            store_id = store_data.id
+        else:
+            store_id = store_res.id
+
+        goods_name = data['goods_name']
+        goods_res = orm.query(Goods).filter_by(name=goods_name).first()
+        if not goods_res:
+            goods_data = Goods(store_id=store_id, name=goods_name, image=data['img_src'], url=data['url'])
+            orm.add(goods_data)
+            orm.commit()
+
+    except Exception as e:
+        print(f"Error: {e}")
+    finally:
+        orm.close()