移动端数据采集：从抓包到自动化

mitmdump --version

# 要求 Python 3.10+
from mitmproxy import ctx

def request(flow):
    """请求钩子：每次收到请求时自动调用"""
    # 修改 User-Agent
    flow.request.headers['User-Agent'] = 'MitmProxy/1.0'
    # 打印请求信息
    ctx.log.info(f"[Request] {flow.request.method} {flow.request.url}")

# 要求 Python 3.10+
import json
from mitmproxy import ctx

TARGET_DOMAIN = 'api.example.com'

def response(flow):
    """响应钩子：提取目标接口的数据"""
    if TARGET_DOMAIN not in flow.request.url:
        return
    try:
        data = json.loads(flow.response.text)
        ctx.log.info(f"[Data] {json.dumps(data, ensure_ascii=False, indent=2)}")
    except json.JSONDecodeError:
        ctx.log.warning("[Warning] 响应不是有效的 JSON")

mitmdump -s script.py

# 要求 Python 3.10+
"""
mitmdump 爬取"得到"App 电子书信息
用法：mitmdump -s script.py
"""
import json
import pymongo
from mitmproxy import ctx

# 连接 MongoDB
client = pymongo.MongoClient('localhost', 27017)
db = client['igetget']
collection = db['books']
# 创建唯一索引，防止重复插入
collection.create_index([('title', pymongo.ASCENDING)], unique=True)

TARGET_URL = 'https://dedao.igetget.com/v3/discover/bookList'

def response(flow):
    if not flow.request.url.startswith(TARGET_URL):
        return
    try:
        data = json.loads(flow.response.text)
        books = data.get('c', {}).get('list', [])
        for book in books:
            book_data = {
                'title': book.get('operating_title', ''),
                'cover': book.get('cover', ''),
                'summary': book.get('other_share_summary', ''),
                'price': book.get('price', 0),
            }
            ctx.log.info(str(book_data))
            # upsert：存在则更新，不存在则插入（pymongo 4.0+ 用 update_one）
            collection.update_one(
                {'title': book_data['title']},
                {'$set': book_data},
                upsert=True
            )
    except (json.JSONDecodeError, KeyError) as e:
        ctx.log.error(f"解析失败: {e}")

# 检查 Appium 版本
appium --version

# 检查 ADB 连接
adb devices -l

# 要求 Python 3.10+
from appium import webdriver
from appium.options.android import UiAutomator2Options
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC

# Appium Server 地址
SERVER = 'http://localhost:4723'

# Desired Capabilities 配置
desired_caps = {
    'platformName': 'Android',
    'deviceName': 'MI_NOTE_Pro',            # adb devices 获取
    'appPackage': 'com.tencent.mm',         # 微信包名
    'appActivity': '.ui.LauncherUI',        # 入口 Activity
    'noReset': True,                        # 不重置 App 数据
    # 注：unicodeKeyboard/resetKeyboard 仅旧版 Appium 支持，
    # Appium 2.x + UiAutomator2 已不支持这两个能力，中文输入请直接用 set_text/send_keys
    'unicodeKeyboard': True,                # 支持中文输入（仅 Appium 1.x）
    'resetKeyboard': True,
}

# appium-python-client 3.0+ 移除了 desired_capabilities 参数，须通过 Options 传入
driver = webdriver.Remote(SERVER, options=UiAutomator2Options().load_capabilities(desired_caps))
wait = WebDriverWait(driver, 30)

# 点击登录按钮
login_btn = wait.until(
    EC.presence_of_element_located((AppiumBy.ID, 'com.tencent.mm:id/cjk'))
)
login_btn.click()

# 输入手机号
phone_input = wait.until(
    EC.presence_of_element_located((AppiumBy.ID, 'com.tencent.mm:id/h2'))
)
phone_input.set_text('18888888888')

# 要求 Python 3.10+
def crawl(self):
    """持续滑动并提取朋友圈动态"""
    while True:
        # 上滑加载更多
        self.driver.swipe(300, 1000, 300, 300)
        # 获取当前页面所有动态元素
        items = self.wait.until(
            EC.presence_of_all_elements_located(
                (AppiumBy.XPATH,
                 '//*[@resource-id="com.tencent.mm:id/cve"]'
                 '//android.widget.FrameLayout')
            )
        )
        for item in items:
            try:
                nickname = item.find_element(AppiumBy.ID, 'com.tencent.mm:id/aig').get_attribute('text')
                content = item.find_element(AppiumBy.ID, 'com.tencent.mm:id/cwm').get_attribute('text')
                date_str = item.find_element(AppiumBy.ID, 'com.tencent.mm:id/crh').get_attribute('text')
                # 日期转换（"5分钟前" -> "2026-06-05"）
                date = self.processor.date(date_str)
                # 去重存储（pymongo 4.0+ 用 update_one）
                self.collection.update_one(
                    {'nickname': nickname, 'content': content},
                    {'$set': {'nickname': nickname, 'content': content, 'date': date}},
                    upsert=True
                )
            except NoSuchElementException:
                pass  # 部分动态可能缺少字段，跳过

# 要求 Python 3.10+
from appium import webdriver
from appium.options.android import UiAutomator2Options
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
from time import sleep

class JDAction:
    def __init__(self):
        desired_caps = {
            'platformName': 'Android',
            'deviceName': 'MI_NOTE_Pro',
            'appPackage': 'com.jingdong.app.mall',
            'appActivity': 'main.MainActivity',
            'noReset': True,
        }
        # appium-python-client 3.0+ 须通过 Options 传入能力字典
        self.driver = webdriver.Remote(
            'http://localhost:4723',
            options=UiAutomator2Options().load_capabilities(desired_caps),
        )
        self.wait = WebDriverWait(self.driver, 60)

    def navigate_to_comments(self, keyword='iPhone'):
        """导航到商品评论页面"""
        # 点击搜索框
        search = self.wait.until(
            EC.presence_of_element_located((AppiumBy.ID, 'com.jingdong.app.mall:id/mp'))
        )
        search.click()
        # 输入关键词
        box = self.wait.until(
            EC.presence_of_element_located((AppiumBy.ID, 'com.jd.lib.search:id/search_box_layout'))
        )
        box.set_text(keyword)
        # 点击搜索
        self.wait.until(
            EC.presence_of_element_located((AppiumBy.ID, 'com.jd.lib.search:id/search_btn'))
        ).click()
        # 进入商品详情
        self.wait.until(
            EC.presence_of_element_located((AppiumBy.ID, 'com.jd.lib.search:id/product_list_item'))
        ).click()
        # 点击评论标签
        self.wait.until(
            EC.presence_of_element_located((AppiumBy.ID, 'com.jd.lib.productdetail:id/pd_tab3'))
        ).click()

    def scroll_forever(self):
        """持续滑动加载评论"""
        while True:
            self.driver.swipe(300, 1000, 300, 300)
            sleep(2)  # 控制滑动频率

    def main(self):
        self.navigate_to_comments()
        self.scroll_forever()

if __name__ == '__main__':
    JDAction().main()

# 要求 Python 3.10+
"""
Appium + mitmdump 联合爬取京东 - mitmdump 脚本
用法：mitmdump -s jd_script.py
"""
import json
import re
from urllib.parse import unquote
from mitmproxy import ctx
import pymongo

client = pymongo.MongoClient('localhost', 27017)
db = client['jd']
products_col = db['products']
comments_col = db['comments']

def response(flow):
    # ---- 商品详情 ----
    if 'cdnware.m.jd.com' in flow.request.url:
        try:
            data = json.loads(flow.response.text)
            info = data.get('wareInfo', {}).get('basicInfo', {})
            if info:
                product = {
                    'id': info.get('wareId'),
                    'name': info.get('name'),
                    'images': info.get('wareImage'),
                }
                products_col.update_one({'id': product['id']}, {'$set': product}, upsert=True)
                ctx.log.info(f"[商品] {product['id']} - {product['name']}")
        except Exception as e:
            ctx.log.error(f"[商品] 解析失败: {e}")

    # ---- 评论数据 ----
    if 'api.m.jd.com/client.action' in flow.request.url:
        try:
            # 从请求参数中提取商品 ID
            body = unquote(flow.request.text)
            match = re.search(r'sku".*?"(\d+)"', body)
            product_id = match.group(1) if match else None

            data = json.loads(flow.response.text)
            for comment in data.get('commentInfoList', []):
                info = comment.get('commentInfo', {})
                if info.get('commentData'):
                    comment_data = {
                        'product_id': product_id,
                        'nickname': info.get('userNickName', ''),
                        'content': info.get('commentData', ''),
                        'date': info.get('commentDate', ''),
                        'pictures': info.get('pictureInfoList', []),
                    }
                    comments_col.update_one(
                        {'product_id': product_id, 'nickname': comment_data['nickname'],
                         'content': comment_data['content']},
                        {'$set': comment_data},
                        upsert=True
                    )
        except Exception as e:
            ctx.log.error(f"[评论] 解析失败: {e}")