Python爬取当当网APP数据

• Z时代
• 2024-01-10
• 分类：综合

本文的文字及图片来源于网络,仅供学习、交流使用,不具有任何商业用途,版权归原作者所有,如有问题请及时联系我们以作处理

以下文章来源于AirPython ，作者星安果

目标

场景：有时候通过传统的方法去爬一些 Web 网页或者 APP，受限于对方的反爬方案，很难爬到想要的数据，这个时候可以考虑使用「Appium」结合「mitmproxy」的方式去爬取数据。

其中，Appium 负责驱动 App 端自动化运行，mitmproxy 负责截取请求数据并解析保存到数据库。

今天的目的是爬取「当当网」的所有数据，并保存到 MongoDB 数据库当中。

准备工作

首先，需要在 PC 上安装好 Charles 和 Appium Desktop，并配置好 mitmproxy 环境。

# 安装mitmproxy依赖包
pip3 install mitmproxy
# 安装pymongodb
pip3 install pymongo

另外，需要准备一台 Android 手机，另外 PC 端配置好 Android 开发环境。

爬取思路

1. 在配置好手动代理的情况下，打开 Charles 实时捕获客户端的发起的网络请求。

打开当当网搜索商品的页面，搜索关键字「Python」,可以在 Charles 查看到当前请求的 URL 地址包含：「word=Python」

编写 mitmproxy 的执行脚本文件，重写 response() 函数，通过对请求的 URL 进行过滤，对有用的数据进行整理并保存到 MongoDB 数据库当中。

class DangDangMongo(object):
    """
    初始化MongoDB数据库
    """
    def __init__(self):
        self.client = MongoClient("localhost")
        self.db = self.client["admin"]
        self.db.authenticate("root", "xag")
        self.dangdang_book_collection = self.db["dangdang_book"]
def response(flow):
    # 过滤请求的URL
    if "keyword=Python" in request.url:
        data = json.loads(response.text.encode("utf-8"))
        # 书籍
        products = data.get("products") or None
        product_datas = []
        for product in products:
            # 书ID
            product_id = product.get("id")
            # 书名
            product_name = product.get("name")
            # 书价格
            product_price = product.get("price")
            # 作者
            authorname = product.get("authorname")
            # 出版社
            publisher = product.get("publisher")
            product_datas.append({
                "product_id": product_id,
                "product_name": product_name,
                "product_price": product_price,
                "authorname": authorname,
                "publisher": publisher
            })
        DangDangMongo().dangdang_book_collection.insert_many(product_datas)
        print("成功插入数据成功")

先打开客户端的手动代理监听 8080 端口，然后执行「mitmdump」命令，然后滚动商品界面，发现数据到写入到数据库中了。

 mitmdump -s script_dangdang.py 

2. 下面我们要利用 Appium 帮我们实现 自动化。

首先打开 Appium Desktop，并启动服务。

打开 Android Studio，利用菜单栏的 Build-Analyze APK 分析当当网的安卓应用，打开 AndroidManifest.xml

可以发现应用包名和初始化 Activity 分别为：

com.dangdang.buy2、com.dangdang.buy2.StartupActivity

获取到包名和初始 Activity 后，就可以利用 WebDriver 去模拟打开当当网 APP。

self.caps = {
            "automationName": DRIVER,
            "platformName": PLATFORM,
            "deviceName": DEVICE_NAME,
            "appPackage": APP_PACKAGE,
            "appActivity": APP_ACTIVITY,
            "platformVersion": ANDROID_VERSION,
            "autoGrantPermissions": AUTO_GRANT_PERMISSIONS,
            "unicodeKeyboard": True,
            "resetKeyboard": True
        }
self.driver = webdriver.Remote(DRIVER_SERVER, self.caps)

接着使用 Android SDK 自带的工具 uiautomatorviewer 获取到元素信息，使用 Appium 中的 WebDriver 去操作 UI 元素。

第一次打开应用的时候，可能会出现红包雨对话框、新人专享红包对话框、切换城市对话框，这里需要通过元素 ID 获取到关闭按钮，执行点击操作来关闭这些对话框。

这里创建一个 新的线程 来单独处理这些对话框。

class ExtraJob(threading.Thread):
   def run(self):
        while self.__running.isSet():
            # 为True时立即返回, 为False时阻塞直到内部的标识位为True后返回
            self.__flag.wait()
            # 1.0 【红包雨】对话框
            red_packet_element = is_element_exist(self.driver, "com.dangdang.buy2:id/close")
            if red_packet_element:
                red_packet_element.click()
            # 1.1 【新人专享券】对话框
            new_welcome_page_sure_element = is_element_exist(self.driver, "com.dangdang.buy2:id/dialog_cancel_tv")
            if new_welcome_page_sure_element:
                new_welcome_page_sure_element.click()
            # 1.2 【切换位置】对话框
            change_city_cancle_element = is_element_exist(self.driver, "com.dangdang.buy2:id/left_bt")
            if change_city_cancle_element:
                change_city_cancle_element.click()
extra_job = ExtraJob(dangdang.driver)
extra_job.start()

接下来就是点击搜索按钮，然后输入内容，执行点击搜索对话框。

 # 1.搜索框
search_element_pro = self.wait.until(
            EC.presence_of_element_located((By.ID, "com.dangdang.buy2:id/index_search")))
search_element_pro.click()
search_input_element = self.wait.until(
            EC.presence_of_element_located((By.ID, "com.dangdang.buy2:id/search_text_layout")))
search_input_element.set_text(KEY_WORD)
# 2.搜索对话框，开始检索
search_btn_element = self.wait.until(
            EC.element_to_be_clickable((By.ID, "com.dangdang.buy2:id/search_btn_search")))
search_btn_element.click()
# 3.休眠3秒，保证第一页的内容加载完全
time.sleep(3)

待第一页的数据加载完全之后，可以一直向上滚动页面，直到数据全部被加载完全，数据会由 mitmproxy 自动保存到 MongoDB 数据库当中。

while True:
      str1 = self.driver.page_source
      self.driver.swipe(FLICK_START_X, FLICK_START_Y + FLICK_DISTANCE, FLICK_START_X, FLICK_START_X)
      time.sleep(1)
      str2 = self.driver.page_source
      if str1 == str2:
            print("停止滑动")
            # 停止线程
            extra_job.stop()
            break
      print("继续滑动"

结果

首先使用 mitmdump 开启请求监听的服务，然后执行爬取脚本。

App 会自动打开，执行一系列操作后，到达商品界面，然后自动滑动界面，通过 mitmproxy 自动把有用的数据保存到 MongoDB 数据库中。

以上是 Python爬取当当网APP数据 的全部内容， 来源链接： utcz.com/z/531096.html