project/xhs/spiders/xhs/product_scraper.py

# coding=utf-8
"""
小红书商品详情页采集
基于 uiautomator2 真机自动化方案

使用方法:
  1. 填好下方变量区
  2. 手机打开小红书, 搜索目标关键词, 进入商品列表页
  3. python product_scraper.py
"""

import uiautomator2 as u2
import time
import re
import random
import subprocess
import datetime


# ============================================================
# ====== 变量区 ======
# ============================================================

DEVICE_ID = "XOYPOZDADU79VGVG"
XHS_PACKAGE = "com.xingin.xhs"
PRODUCT_NAME = "小儿感冒颗粒"
BRAND = "999"
SPEC_LIST = '24'

# ============================================================
# ====== 变量区结束 ======
# ============================================================


class XHS:
    """小红书商品详情页采集器"""

    # ---- 类常量 ----
    SEP = "=" * 60
    SLEEP_CLICK = 1.5       # 点击后等待页面响应
    SLEEP_APP_START = 3     # App 启动等待
    SLEEP_CLIPBOARD = 0.8   # 复制链接等待

    # 弹窗检测：命中 >= POPUP_THRESHOLD 个 xpath 即认定为弹窗
    POPUP_XPATHS = [
        '//*[@text="立即领取"]',
        '//*[@text="送你 66周年庆礼券 啦 "]',
        '//*[contains(@text, "可用")]',   # 匹配 "满150可用" 等变体
    ]
    POPUP_THRESHOLD = 2

    # 列表页判定锚点（命中 >= LIST_PAGE_HIT_THRESHOLD 个即认为在列表页）
    LIST_PAGE_ANCHORS = [
        '//*[@text="搜索"]',
        '//*[@text="全部"]',
        '//*[@text="用户"]',
        '//*[@text="商品"]',
        '//*[@content-desc="全部删除"]',
    ]
    LIST_PAGE_HIT_THRESHOLD = 2

    def __init__(self, device_id, product_name="", brand="", spec_list=None):
        self.package_name = XHS_PACKAGE
        self.device_id = device_id
        self.d = None
        self.product_name = str(product_name or "").strip()
        self.brand = str(brand or "").strip()
        self.spec_list = self._normalize_rule_list(spec_list)

    # ============================================================
    # 工具方法
    # ============================================================

    @staticmethod
    def _normalize_rule_list(value):
        if value is None:
            return []
        if isinstance(value, (list, tuple, set)):
            raw_values = value
        else:
            raw_values = [value]
        return [str(v).strip() for v in raw_values if str(v).strip()]

    @staticmethod
    def _normalize_match_text(value):
        return re.sub(r'\s+', '', str(value or '')).lower()

    @staticmethod
    def get_sleep_time():
        return random.uniform(0.5, 1.0)

    @staticmethod
    def get_current_date():
        return datetime.datetime.now().strftime('%Y/%m/%d')

    @staticmethod
    def _strip_content_desc_prefix(text, prefix):
        """去掉 content-desc 中的前缀标签（兼容中英文逗号）"""
        for sep in ("，", ","):
            text = text.replace(f"{prefix}{sep}", "")
        return text.strip()

    def _safe_get_attr(self, xpath, attr="text"):
        """安全获取元素属性（text 或 content_desc），避免元素缺失导致崩溃"""
        try:
            el = self.d.xpath(xpath)
            if not el.exists:
                return ""
            if attr == "content_desc":
                return (el.info.get('contentDescription') or "").strip()
            return (el.text or "").strip()
        except Exception:
            return ""

    def _scroll_until_found(self, xpath, max_swipes=5, direction="up", scale=0.3):
        """翻页直到找到目标元素，返回元素对象；找不到返回 None"""
        el = self.d.xpath(xpath)
        if el.exists:
            return el
        for _ in range(max_swipes):
            self.d.swipe_ext(direction, scale=scale)
            time.sleep(self.get_sleep_time())
            el = self.d.xpath(xpath)
            if el.exists:
                return el
        return None

    def _count_popup_elements(self):
        """统计当前页面命中弹窗特征 xpath 的数量"""
        count = 0
        for xpath in self.POPUP_XPATHS:
            if self.d.xpath(xpath).exists:
                count += 1
        return count

    def _dismiss_popup_if_exists(self):
        """检测弹窗，命中 >= 阈值则按 back 关闭，返回是否关闭"""
        count = self._count_popup_elements()
        if count >= self.POPUP_THRESHOLD:
            print(f"  [弹窗检测] 命中 {count} 个特征，关闭弹窗")
            self.d.press("back")
            time.sleep(self.get_sleep_time())
            return True
        return False

    def _find_with_popup_retry(self, find_func, step_name="", max_retries=2):
        """查找元素，失败则关闭弹窗后重试；返回找到的元素或 None"""
        for attempt in range(max_retries):
            result = find_func()
            if result:
                return result
            if attempt < max_retries - 1:
                print(f"  [{step_name}] 未找到，关闭弹窗后重试 ({attempt + 1}/{max_retries - 1})")
                self._dismiss_popup_if_exists()
                time.sleep(random.uniform(1.5, 2))
        return None

    @staticmethod
    def _match_any_keyword(text, keywords):
        """判断 text 是否命中 keywords 中的任一关键词；keywords 为空则放行"""
        keyword_list = XHS._normalize_rule_list(keywords)
        if not keyword_list:
            return True
        normalized = XHS._normalize_match_text(text)
        return any(XHS._normalize_match_text(k) in normalized for k in keyword_list)

    def is_on_list_page(self):
        """判断当前是否在商品列表页"""
        hits = sum(1 for xp in self.LIST_PAGE_ANCHORS if self.d.xpath(xp).exists)
        return hits >= self.LIST_PAGE_HIT_THRESHOLD

    def is_title_useful(self, title):
        """标题需同时包含产品名、品牌、规格（各自命中一个即可），打印不匹配原因"""
        if self.product_name and not self._match_any_keyword(title, self.product_name):
            print(f"    不匹配: 产品名「{self.product_name}」")
            return False
        if self.brand and not self._match_any_keyword(title, self.brand):
            print(f"    不匹配: 品牌「{self.brand}」")
            return False
        if self.spec_list and not self._match_any_keyword(title, self.spec_list):
            print(f"    不匹配: 规格「{self.spec_list}」")
            return False
        return True

    def get_product_cards(self):
        """获取当前屏幕可见的商品卡片（按子元素高度定位正确的 RecyclerView）"""
        cards = []
        for idx in (1, 2, 3):
            candidates = self.d.xpath(
                f'(//androidx.recyclerview.widget.RecyclerView)[{idx}]/android.widget.FrameLayout'
            ).all()
            for c in candidates:
                try:
                    bounds = c.info.get('bounds', {})
                    h = bounds.get('bottom', 0) - bounds.get('top', 0)
                    if h > 400:  # 商品卡片高 600+，tab 栏只有 ~140
                        cards = candidates
                        break
                except Exception:
                    continue
            if cards:
                break

        if not cards:
            return []
        visible = []
        for card in cards:
            try:
                bounds = card.info.get('bounds', {})
                top = bounds.get('top', 0)
                bottom = bounds.get('bottom', 0)
                h = bottom - top
                if h > 400 and top >= 554 and bottom <= 2600:
                    visible.append(card)
            except Exception:
                continue
        return visible

    def back_to_list_page(self, max_attempts=5):
        """按 back 直到回到列表页"""
        for _ in range(max_attempts):
            if self.is_on_list_page():
                return True
            self.d.press("back")
            time.sleep(self.get_sleep_time())
        return self.is_on_list_page()

    def _find_shop_avatar(self):
        """在店铺页按位置范围找店铺头像 ImageView"""
        imgs = self.d.xpath('//android.widget.ImageView').all()
        for img in imgs:
            try:
                bounds = img.info.get('bounds', {})
                left = bounds.get('left', 0)
                top = bounds.get('top', 0)
                right = bounds.get('right', 0)
                bottom = bounds.get('bottom', 0)
                w = right - left
                h = bottom - top
                if 30 <= left <= 120 and 300 <= top <= 370 and abs(w - h) <= 30:
                    return img
            except Exception:
                continue
        return None

    # ============================================================
    # 设备连接 & App 控制
    # ============================================================

    def connect_device(self):
        try:
            self.d = u2.connect_usb(self.device_id)
            self._restart_uiautomator_services()
            print(f'[连接成功] 设备: {self.device_id}')
            return True
        except Exception as e:
            print(f'[连接失败] {self.device_id}: {e}')
            return False

    def _restart_uiautomator_services(self):
        stop_cmd = f'adb -s {self.device_id} shell /data/local/tmp/atx-agent server -d --stop'
        start_cmd = f'adb -s {self.device_id} shell /data/local/tmp/atx-agent server -d'
        subprocess.run(stop_cmd, capture_output=True, text=True, shell=True)
        time.sleep(self.get_sleep_time())
        subprocess.run(start_cmd, capture_output=True, text=True, shell=True)
        time.sleep(self.get_sleep_time())

    def start_app(self):
        self.d.app_start(self.package_name)
        time.sleep(self.SLEEP_APP_START)

    # ============================================================
    # 1. 提取标题
    # ============================================================

    def get_title(self):
        print("\n[1/7] 提取标题...")
        # 优先从 content-desc 取（直接进入的详情页）
        title = self._safe_get_attr(
            '//*[contains(@content-desc, "商品名称")]', "content_desc"
        )
        if title:
            title = self._strip_content_desc_prefix(title, "商品名称")
            print(f"  标题={title}")
            return title
        # fallback: 取页面上最长的 TextView（搜索结果点进来的详情页）
        try:
            text_els = self.d.xpath('//android.widget.TextView').all()
            longest = ""
            for el in text_els:
                t = (el.text or "").strip()
                if len(t) > len(longest):
                    longest = t
            if longest and len(longest) > 5:
                print(f"  标题(fallback)={longest}")
                return longest
        except Exception:
            pass
        print("  [失败] 未找到标题")
        return ""

    # ============================================================
    # 2. 提取价格
    # ============================================================

    def get_price(self):
        print("\n[2/7] 提取价格...")

        buy_btn = self.d.xpath('//*[@text="立即购买"]')
        if not buy_btn.exists:
            buy_btn = self.d.xpath('//*[@text="领券购买"]')
        if not buy_btn.exists:
            print("  [失败] 未找到「购买选项」")
            return None

        buy_btn.click()
        print("  点击「购买」")
        time.sleep(self.SLEEP_CLICK)

        # 优先到手价，兜底 ¥
        price_str = ""
        price_xpath = '//*[contains(@content-desc, "到手价")]'
        if self.d.xpath(price_xpath).exists:
            price_str = self._safe_get_attr(price_xpath, "content_desc")
            print(f"  [到手价] {price_str}")
        else:
            fallback_xpath = '//*[contains(@content-desc, "¥")]'
            if self.d.xpath(fallback_xpath).exists:
                price_str = self._safe_get_attr(fallback_xpath, "content_desc")
                print(f"  [兜底¥] {price_str}")

        self.d.press("back")
        time.sleep(self.get_sleep_time())

        if not price_str:
            return None

        match = re.search(r'¥([\d\.]+)', price_str)
        if match:
            price = float(match.group(1))
            print(f"  价格={price} 元")
            return price

        print(f"  [失败] 无法解析价格: {price_str}")
        return None

    # ============================================================
    # 3. 提取规格
    # ============================================================

    def get_spec(self):
        print("\n[3/7] 提取规格...")
        spec = self._safe_get_attr(
            '//*[contains(@content-desc, "已选规格")]', "content_desc"
        )
        if spec:
            spec = self._strip_content_desc_prefix(spec, "已选规格")
            print(f"  规格={spec}")
        else:
            print("  [失败] 未找到规格")
        return spec

    # ============================================================
    # 4. 提取店铺名
    # ============================================================

    def get_shop_name(self):
        print("\n[4/7] 提取店铺名...")

        self._scroll_until_found('//*[@text="进店"]', max_swipes=5)

        shop_name = self._safe_get_attr(
            '//*[contains(@content-desc, "旗舰店")]', "content_desc"
        )
        if not shop_name:
            shop_name = self._safe_get_attr('//*[contains(@text, "旗舰店")]')
        if shop_name:
            print(f"  店铺名={shop_name}")
        else:
            print("  [失败] 未找到店铺名")
        return shop_name

    # ============================================================
    # 5. 提取公司名
    # ============================================================

    def get_company_name(self):
        print("\n[5/7] 提取公司名...")

        enter_btn = self._scroll_until_found('//*[@text="进店"]', max_swipes=5)
        if not enter_btn:
            print("  [失败] 未找到「进店」")
            return ""

        enter_btn.click()
        time.sleep(random.uniform(1.5, 2))
        print("  点击「进店」")
        self._dismiss_popup_if_exists()

        # 优先点头像进资质页，没有头像再找"X篇笔记"
        avatar = self._find_shop_avatar()
        if avatar:
            avatar.click()
            time.sleep(random.uniform(1.5, 2))
            print("  点击「店铺头像」")
        else:
            notes_btn = self.d.xpath('//*[contains(@text, "篇笔记")]')
            if not notes_btn.exists:
                print("  [失败] 未找到店铺头像和「X篇笔记」")
                self.d.press("back")
                time.sleep(self.get_sleep_time())
                return ""
            notes_btn.click()
            time.sleep(random.uniform(1.5, 2))
            print("  点击「X篇笔记」资质入口")
        self._dismiss_popup_if_exists()
        self.get_sleep_time()
        company = self._safe_get_attr('//*[contains(@text, "公司")]')
        if company:
            print(f"  公司名={company}")

        self.d.press("back")
        time.sleep(self.get_sleep_time())
        self._dismiss_popup_if_exists()
        self.d.press("back")
        time.sleep(self.get_sleep_time())
        self._dismiss_popup_if_exists()
        return company

    # ============================================================
    # 6. 提取商品链接
    # ============================================================

    def get_product_link(self):
        print("\n[6/7] 提取商品链接...")

        try:
            self.d.set_clipboard("")
        except Exception:
            pass

        share_btn = self.d.xpath('//*[@content-desc="分享商品"]')
        if not share_btn.exists:
            print("  [失败] 未找到「分享商品」")
            return ""

        share_btn.click()
        time.sleep(1)
        print("  点击「分享商品」")

        copy_btn = self.d.xpath('//*[@text="复制链接"]')
        if copy_btn.exists:
            copy_btn.click()
            time.sleep(self.SLEEP_CLIPBOARD)

        link = (self.d.clipboard or "").strip()
        if link:
            match = re.search(r'https?://xhslink\.com/\S+', link)
            if match:
                link = match.group(0)
            print(f"  商品链接={link}")
        else:
            print("  [失败] 剪贴板为空")
        return link

    # ============================================================
    # 7. 提取批准文号
    # ============================================================

    def get_approval_number(self):
        print("\n[7/7] 提取批准文号...")

        # 步骤1: 小幅度滑动到"批准文号"标签（起点压低避开"退货包运费"）
        w, h = self.d.window_size()
        for _ in range(6):
            self.d.swipe(w // 2, int(h * 0.88), w // 2, int(h * 0.50), duration=0.3)
            time.sleep(self.get_sleep_time())
            if self.d.xpath('//*[@text="批准文号"]').exists:
                break

        # 步骤2: 国药准字有就拿，没有就空
        approval = self._safe_get_attr('//*[starts-with(@text, "国药准字")]')
        if approval:
            print(f"  批准文号={approval}")
        else:
            print("  [无] 未找到国药准字")
        return approval

    # ============================================================
    # 数据聚合
    # ============================================================

    def integrate_data(self):
        print(self.SEP)
        print(f"  小红书商品详情页采集")
        print(f"  时间: {self.get_current_date()}")
        print(f"  设备: {self.device_id}")
        print(f"  目标: {self.product_name}  |  品牌: {self.brand or '-'}  |  规格: {self.spec_list or '-'}")
        print(self.SEP)

        title = self.get_title()
        price = self.get_price()
        spec = self.get_spec()
        shop_name = self.get_shop_name()
        company = self.get_company_name()
        link = self.get_product_link()
        approval = self.get_approval_number()

        print(f"\n{self.SEP}")
        print("  采集结果汇总")
        print(self.SEP)
        print(f"  标题:       {title}")
        print(f"  价格:       {price} 元" if price else "  价格:       (未取到)")
        print(f"  规格:       {spec}")
        print(f"  店铺名:     {shop_name}")
        print(f"  公司名:     {company}")
        print(f"  批准文号:   {approval}")
        print(f"  商品链接:   {link}")
        print(self.SEP)

        data = {
            "product_name": title,
            "min_price": price,
            "spec": spec,
            "shop_name": shop_name,
            "company_name": company,
            "approval_number": approval,
            "product_link": link,
            "scrape_date": self.get_current_date(),
        }
        print(f"\n  完整数据:\n  {data}")
        return data

    # ============================================================
    # 进入搜索页
    # ============================================================

    def enter_search_page(self):
        """首页 → 市集 → 搜索框 → 粘贴关键字 → 搜索 → 商品列表页"""
        # 1. 点市集
        el = self.d.xpath('//*[@text="市集"]')
        if not el.exists:
            print("[错误] 未找到「市集」")
            return False
        el.click()
        time.sleep(random.uniform(1.5, 2))
        self._dismiss_popup_if_exists()

        # 2. 点搜索入口（弹窗遮挡时自动重试）
        def _find_search_entry():
            tvs = self.d.xpath('//android.widget.TextView').all()
            for tv in tvs:
                try:
                    b = tv.info.get('bounds', {})
                    if 150 <= b.get('top', 0) <= 180 and 200 <= b.get('bottom', 0) <= 250:
                        tv.click()
                        return True
                except Exception:
                    continue
            return False

        if not self._find_with_popup_retry(_find_search_entry, step_name="搜索入口"):
            print("[错误] 未找到搜索入口")
            return False
        time.sleep(random.uniform(1, 1.5))

        # 3. 输入关键字
        spec_str = " ".join(self.spec_list) if self.spec_list else ""
        search_key = f"{self.brand} {self.product_name} {spec_str}".strip()
        print(f"  搜索词: {search_key}")

        def _find_edit_text():
            edit = self.d.xpath('//android.widget.EditText')
            if edit.exists:
                edit.click()
                return True
            return False

        if not self._find_with_popup_retry(_find_edit_text, step_name="搜索输入框"):
            print("[错误] 未找到搜索输入框")
            return False
        time.sleep(0.5)
        self.d.send_keys(search_key, clear=True)
        time.sleep(random.uniform(0.5, 1))

        # 4. 点搜索按钮
        def _find_search_btn():
            search_btn = self.d.xpath('//*[@text="搜索"]')
            if search_btn.exists:
                search_btn.click()
                return True
            return False

        if not self._find_with_popup_retry(_find_search_btn, step_name="搜索按钮"):
            print("[错误] 未找到搜索按钮")
            return False
        time.sleep(random.uniform(1.5, 2))
        self._dismiss_popup_if_exists()

        print("  已进入商品列表页")
        return True

    # ============================================================
    # 列表遍历主循环
    # ============================================================

    def run(self, max_pages=50):
        """主采集循环：遍历列表页 → 逐个商品 → 详情采集 → 退回列表"""
        if not self.is_on_list_page():
            print("[错误] 当前不在商品列表页，请手动进入列表页后重试")
            return

        collected = 0
        unrelated = 0

        for page in range(max_pages):
            print(f"\n{'=' * 60}")
            print(f"  第 {page + 1} 页")
            print(f"{'=' * 60}")

            cards = self.get_product_cards()
            print(f"  当前页 {len(cards)} 个可见商品")

            for i, card in enumerate(cards):
                # 连续 10 个不达标则暂停
                if unrelated >= 10:
                    print(f"\n  [暂停] 连续 {unrelated} 个商品不达标，停止采集")
                    return

                print(f"\n  [{i + 1}]")

                # 1. 点击卡片进入详情页
                try:
                    card.click()
                except Exception as e:
                    print(f"    [失败] 点击失败: {e}")
                    continue
                time.sleep(self.SLEEP_CLICK)
                self._dismiss_popup_if_exists()

                # 2. 确认已离开列表页
                if self.is_on_list_page():
                    print(f"    [跳过] 点击后仍在列表页")
                    continue

                # 3. 读详情页标题，判断是否匹配
                title = self.get_title()
                if not title:
                    print(f"    [跳过] 未读到标题")
                    unrelated += 1
                    self.back_to_list_page()
                    continue

                if not self.is_title_useful(title):
                    print(f"    [跳过] 标题不匹配: {title[:40]}...")
                    unrelated += 1
                    self.back_to_list_page()
                    continue

                # 4. 采集详情（价格、规格、店铺等）
                unrelated = 0
                try:
                    self.integrate_data()
                    collected += 1
                    # 每采集 5 个重启一次 atx 服务，防止越来越卡
                    if collected % 5 == 0:
                        print("    [维护] 重启自动化服务...")
                        self._restart_uiautomator_services()
                except Exception as e:
                    print(f"    [异常] 采集详情失败: {e}")

                # 5. 退回列表
                if not self.back_to_list_page():
                    print("    [警告] 未能回到列表页，尝试继续...")
                    time.sleep(self.get_sleep_time())

                time.sleep(self.get_sleep_time())

            # 翻到下一页
            print(f"\n  翻到第 {page + 2} 页...")
            self.d.swipe_ext("up", scale=0.5)
            time.sleep(self.get_sleep_time())

        print(f"\n  遍历完成，共 {max_pages} 页")


# ============================================================
# 测试入口
# ============================================================

def main():
    print("=" * 60)
    print("  小红书商品详情页采集器")
    print("=" * 60)

    xhs = XHS(
        device_id=DEVICE_ID,
        product_name=PRODUCT_NAME,
        brand=BRAND,
        spec_list=SPEC_LIST,
    )

    if not xhs.connect_device():
        print("设备连接失败，退出")
        return

    # 确保小红书在前台
    try:
        current_pkg = xhs.d.app_current().get("package", "")
        if current_pkg != XHS_PACKAGE:
            print(f"当前前台非小红书(package={current_pkg})，启动...")
            xhs.start_app()
        else:
            print("小红书已在前台")
    except Exception as e:
        print(f"获取前台失败: {e}，启动小红书...")
        xhs.start_app()

    # 自动导航：市集 → 搜索 → 商品列表
    if not xhs.enter_search_page():
        print("进入搜索页失败，退出")
        return

    xhs.run()
    print("\n采集完成！")


if __name__ == "__main__":
    main()