spider_work/jihuanshe_spider/jhs_app_spider_0720.py

# -*- coding: utf-8 -*-
# Author : Charley
# Python : 3.10.8
# Date   : 2026/07/20
#
# 变更(2026/07/20)：
#   [bugfix]
#   1. 分类下拉选择由「固定滑动 + 硬编码 Button[ca_idx-2] 索引」改为
#      「按分类文字(content-desc)滚动查找并点击」。原索引方案在 category_list
#      增删项时会错位/越界（新增『峡谷争锋』后，index 11『迪士尼洛卡纳』点
#      Button[9] 触发 list index out of range）。
#   2. 每个 (分类, 语言) 迭代整体包 try/except，单项失败只跳过、不再让异常冒泡
#      终止整个任务（修复 20260718.log 里首个分类超时后整轮任务被 kill）。
#   3. 翻页时 offspring('android.widget.Button') 查询包 try/except。
#   [optimize]
#   1. 设备连接/停开 App/poco 初始化从模块顶层挪进 init_env()，避免 import 即重启 App。
#   3. 删除从未被调用的死代码 scrape_detail() / is_page_changed()。
#   4. 顶部『畅销榜』Tab 改为优先按文字点击(poco('畅销榜'))，失败再回退图片模板，
#      免掉模板 resolution=(1080,2340) 与模拟器 900x1600 不一致的隐患。
#   5. 关键固定 time.sleep 改为 wait_for_appearance 事件等待（等到即走，更快更稳）。
#   6. 翻页中内联去重（按 title+编号）；停止条件保持旧版「rank==100 或翻满
#      max_pages」，另加 guard 防死循环。
#      注：曾试过「连续N页无新增即停」，但实测会把尾部截断少抓几条，已回退。
#   注：UI 里『迪士尼洛卡纳』的 content-desc 现已无空格；select_category 会自动
#      对带空格的分类名做无空格候选匹配。
import inspect
import subprocess

from airtest.core.api import *
from poco.drivers.android.uiautomation import AndroidUiautomationPoco
from loguru import logger
from mysql_pool import MySQLConnectionPool
from tenacity import retry, stop_after_attempt, wait_fixed

logger.remove()
logger.add("./logs/{time:YYYYMMDD}.log", encoding='utf-8', rotation="00:00",
           format="[{time:YYYY-MM-DD HH:mm:ss.SSS}] {level} {message}",
           level="DEBUG", retention="7 day")

PACKAGE_NAME = 'com.jihuanshe'
DEVICE_URI = "Android:///"          # 如需固定模拟器可改 "Android://127.0.0.1:5557"
ADB_TARGET = "127.0.0.1:5557"       # adb connect 目标

# 翻页参数（为尽量保证数据完整性：每页滑动步长调小、增加页面重叠，靠去重兜住）
# SCROLLS_PER_PAGE 越小 -> 每页前进越少、重叠越多 -> 越不易跳行，但需要更多页数
SCROLLS_PER_PAGE = 2                # 每页滑动次数（旧版为 4，减小以降低跳行）
MAX_PAGES = 60                      # 最大翻页数（步长变小后相应增大，防止翻不到 rank100）

category_list = [
    {'游戏王': ['日文', '简中']},
    {'宝可梦': ['简中', '日文', '英文']},
    {'航海王': ['简中', '日文']},
    {'数码宝贝': ['简中']},
    {'符文战场': ['简中']},
    {'WS黑白双翼': ['日文', '简中']},
    {'携站之境UA': ['简中', '日文']},
    {'VG卡片战斗先导者': ['简中', '日文']},
    {'高达GCG': ['简中']},
    {'影之诗': ['简中']},
    {'峡谷争锋': ['简中']},
    {'迪士尼洛卡纳': ['简中']}  # 2026/07/20 起统一无空格，与 App 及库中数据对齐
]

# 运行时由 init_env() 初始化，避免 import 即产生副作用
# poco = None
window_width = None
window_height = None


def init_env():
    """初始化设备连接、重启目标 App、创建 poco 实例。

    放到运行时执行（而非模块顶层），避免任何 import 本模块的行为都触发 App 重启。
    被 @retry 的主函数每次重试都会重新调用，天然带「重连+重启」的恢复能力。
    """
    global poco, window_width, window_height
    subprocess.run(["adb", "connect", ADB_TARGET])
    connect_device(DEVICE_URI)
    stop_app(PACKAGE_NAME)
    start_app(PACKAGE_NAME)
    poco = AndroidUiautomationPoco(use_airtest_input=True, screenshot_each_action=False)
    window_width, window_height = poco.get_screen_size()
    logger.info(f"init_env 完成: 屏幕 {window_width}x{window_height}")


def wait_exists(node, timeout=8):
    """等待 poco 节点出现，返回是否出现（超时不抛异常）。

    Args:
        node: poco UIObjectProxy 节点。
        timeout (int, optional): 最长等待秒数。Defaults to 8。

    Returns:
        bool: True 表示在超时前出现；False 表示超时未出现。
    """
    try:
        node.wait_for_appearance(timeout=timeout)
        return True
    except Exception:
        return False


def after_log(retry_state):
    """
    retry 回调
    :param retry_state: RetryCallState 对象
    """
    # 检查 args 是否存在且不为空
    if retry_state.args and len(retry_state.args) > 0:
        local_logger = retry_state.args[0]  # 获取传入的 logger
    else:
        local_logger = logger  # 使用全局 logger

    if retry_state.outcome.failed:
        local_logger.warning(
            f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} Times")
    else:
        local_logger.info(f"Function '{retry_state.fn.__name__}', Attempt {retry_state.attempt_number} succeeded")


@retry(stop=stop_after_attempt(3), wait=wait_fixed(10), after=after_log)
def scrape_index():
    """进入首页、关闭广告、点击行情入口 imageIv。

    Raises:
        Exception: imageIv 元素始终不可见时抛出，触发上层 retry。
    """
    # 等待 App 启动到首页：优先等广告或行情入口出现，最多 ~15s
    for _ in range(15):
        if poco(f'{PACKAGE_NAME}:id/ivClose').exists() or poco(f'{PACKAGE_NAME}:id/imageIv').exists():
            break
        time.sleep(1)

    # 判断是否有广告  需要点击
    tag_adv = poco(f'{PACKAGE_NAME}:id/ivClose')
    for _ in range(3):
        if tag_adv.exists():
            logger.warning("广告元素已找到")
            tag_adv.click()
            time.sleep(1)
        else:
            logger.debug("广告元素未找到, 退出")
            break

    logger.info("开始查找imageIv元素...")

    if not wait_exists(poco(f'{PACKAGE_NAME}:id/imageIv'), timeout=10):
        logger.warning("imageIv元素不存在，准备重试...")
        try:
            snapshot_path = f"./logs/debug_screenshot_{time.strftime('%Y%m%d_%H%M%S')}.png"
            snapshot(filename=snapshot_path)
            logger.info(f"已保存调试截图到: {snapshot_path}")
        except Exception as snap_error:
            logger.warning(f"截图失败: {snap_error}")

        raise Exception(f"Cannot find visible node: {PACKAGE_NAME}:id/imageIv")

    poco(f'{PACKAGE_NAME}:id/imageIv').click()
    logger.success("成功找到并点击imageIv元素")


def open_category_dropdown():
    """点开顶部「分类/语言」下拉框。

    下拉入口是标题栏里的第 2 个 Button（tag_drop_down），点开后左侧为分类列表、
    右侧为语言列表。点击后等待列表首项『游戏王』出现作为就绪信号。

    Returns:
        bool: True 表示下拉框已点开且列表就绪；False 表示未找到入口。
    """
    # 已经打开（首项『游戏王』可见）则不重复点击，避免把已开的下拉框又关掉
    if poco('游戏王').exists():
        return True

    tag_drop_down = \
        poco("android:id/content").child("android.widget.FrameLayout").child(
            "android.widget.FrameLayout").child(
            "android.view.View").child("android.view.View").child("android.view.View").child(
            "android.widget.Button")[1]
    if not tag_drop_down.exists():
        logger.warning("分类下拉框入口(tag_drop_down)未找到")
        return False

    tag_drop_down.click()
    if wait_exists(poco('游戏王'), timeout=3):
        return True
    # 兜底：刚才可能把已开的关掉了 / 首次没响应，再点一次
    tag_drop_down.click()
    return wait_exists(poco('游戏王'), timeout=3)


def select_category(category):
    """在分类下拉里按文字选中分类，可视区找不到就滚动左侧面板直到出现。

    集换社行情页由 Flutter 渲染，分类项为 ImageView，名字挂在 content-desc 上，
    poco 的 name 匹配即 content-desc（与语言按钮同理）。改用文字定位可彻底避免旧版
    「固定滑动 + 硬编码 Button 索引」在分类列表增删时错位/越界的问题。

    Args:
        category (str): category_list 里的分类名，如 '峡谷争锋'、'迪士尼 洛卡纳'。

    Returns:
        bool: True 表示已点中分类；False 表示滚动多次仍未找到。
    """
    # UI 的 content-desc 可能无空格（如 '迪士尼洛卡纳'），额外做无空格候选
    candidates = [category]
    no_space = category.replace(' ', '')
    if no_space != category:
        candidates.append(no_space)

    # 先把分类面板滚回顶部，保证从已知位置开始逐屏往下找。
    # 否则下拉框若残留在底部滚动位置，目标在视区上方，只往下滚会一直漏找。
    # （向下滑动手指 = 内容回到顶部；目标已在当前视区则直接跳过复位）
    for _ in range(4):
        if any(poco(n).exists() for n in candidates):
            break
        swipe(v1=(0.25, 0.25), v2=(0.25, 0.55), duration=0.3)
        time.sleep(0.5)

    for attempt in range(8):
        for name in candidates:
            node = poco(name)
            if node.exists():
                node.click()
                logger.info(f'点击 集换榜分类 -> {category}')
                time.sleep(1)
                return True
        # 未在可视区 -> 向上滑动手指（内容往上走）看后面的分类
        logger.debug(f'分类 [{category}] 不在可视区，滚动下拉菜单 (第 {attempt + 1} 次)')
        swipe(v1=(0.25, 0.53), v2=(0.25, 0.22), duration=0.4)
        time.sleep(1)

    logger.warning(f'滚动多次仍未找到分类 [{category}]，跳过')
    return False


def click_bestseller_tab():
    """点击顶部『畅销榜』Tab，优先按文字(content-desc)，失败回退图片模板。

    Returns:
        bool: True 表示已点击；False 表示文字与模板均失败。
    """
    tab = poco('畅销榜')
    if tab.exists():
        tab.click()
        return True
    logger.warning("畅销榜 文字定位失败, 回退图片模板")
    try:
        touch(Template(r"tpl1761026966680.png", record_pos=(0.048, -0.781), resolution=(1080, 2340)))
        return True
    except Exception as e:
        logger.error(f"畅销榜 模板点击失败: {e}")
        return False


def parse_sales_data(raw_text):
    """
    解析销售数据文本

    Args:
        raw_text (str): 原始文本数据

    Returns:
        dict: 解析后的数据字典
    """
    raw_list = raw_text.split("\n")

    # 确保第一个元素是数字，如果不是则插入空字符串
    if not raw_list[0].isdigit():
        raw_list.insert(0, '')

    # 检查列表长度是否足够
    if len(raw_list) < 11:
        raise ValueError("输入数据不完整")

    tv_rank = raw_list[0]
    title = raw_list[1]
    tv_number = raw_list[2]
    tv_rarity = raw_list[3]
    tv_cur_sales_num = raw_list[5]
    tv_cur_sales_num = tv_cur_sales_num.replace('张', '') if '张' in tv_cur_sales_num else tv_cur_sales_num

    # 增加判断：如果第7行和第8行连续是"本周期销量涨跌幅"和"本周期平均集换价"，
    # 则在它们之间插入'-'，并将tv_last_sales_num设为'-'
    if len(raw_list) >= 8 and raw_list[6] == "本周期销量涨跌幅" and raw_list[7] == "本周期平均集换价":
        raw_list.insert(7, '-')
        tv_last_sales_num = '-'
    else:
        try:
            tv_last_sales_num = raw_list[7]
        except IndexError:
            tv_last_sales_num = ''

    tv_price = raw_list[9]
    tv_price = tv_price.replace('¥', '') if '¥' in tv_price else tv_price
    tv_sales_data_num = raw_list[10]

    data_dict = {
        'tv_rank': tv_rank,
        'title': title,
        'tv_number': tv_number,
        'tv_rarity': tv_rarity,
        'tv_cur_sales_num': tv_cur_sales_num,
        'tv_last_sales_num': tv_last_sales_num,
        'tv_price': tv_price,
        'tv_sales_data_num': tv_sales_data_num
    }

    return data_dict


def scroll_up():
    """列表向上滑动一屏（用于畅销榜翻页）。"""
    swipe(
        v1=(0.5, 0.9),  # 起始点（底部）
        v2=(0.5, 0.7),  # 结束点（顶部）
        duration=1.0
    )


def _bottom_key(list_view):
    """取当前列表最后一张有效卡的 (title, 编号) 指纹。

    用于判断 App 懒加载（每约 15 条加载下一批）是否已把新内容渲染出来：
    滑动后底部指纹发生变化，才说明新一批卡真正加载完成、可以读取。

    Args:
        list_view: 畅销榜列表的 poco ScrollView 节点。

    Returns:
        tuple | None: (title, 编号)；列表为空或读取失败时返回 None。
    """
    try:
        btns = list(list_view.offspring('android.widget.Button'))
    except Exception:
        return None
    for el in reversed(btns):
        child = el.child()
        if not child.exists():
            continue
        txt = child.attr('name')
        if not txt or '加入心愿单' in txt:
            continue
        try:
            d = parse_sales_data(txt)
            return (d.get('title'), d.get('tv_number'))
        except Exception:
            continue
    return None


def scrape_pages(category, language):
    """在畅销榜列表页翻页抓取当前(分类,语言)的全部卡牌数据。

    翻页按 (title, 编号) 内联去重；停止条件与旧版一致：抓到 rank==100 或翻满
    max_pages 才停（不做「连续无新增即停」——实测那样会把尾部截断、少抓几条）。
    另加 guard 计数防止空翻页/异常反复 continue 导致死循环。

    Args:
        category (str): 当前分类名（用于打标签入库）。
        language (str): 当前语言名（用于打标签入库）。

    Returns:
        list[dict]: 每条为卡牌销量记录（已带 category/language/crawler_date，已去重）。
    """
    stop_page = False
    page_count = 1
    max_pages = MAX_PAGES
    data_list = []
    seen_keys = set()   # (title, tv_number) 内联去重
    guard = 0           # 防止空翻页/异常反复 continue 导致死循环
    guard_max = max_pages * 3
    prev_bottom = None  # 上一页底部卡指纹，用于判断懒加载是否已加载出新内容

    while page_count <= max_pages and not stop_page:
        guard += 1
        if guard > guard_max:
            logger.warning(f'翻页保护触发({guard}次), 停止 (已 {len(data_list)} 条)')
            break

        # 翻页操作：每页滑动 SCROLLS_PER_PAGE 次（步长小、重叠多，降低跳行）
        if page_count == 1:
            logger.debug("第一页开始翻页")
            scroll_up()
        else:
            logger.debug(f"{page_count}翻页.........")
            for _ in range(SCROLLS_PER_PAGE):
                scroll_up()

        tag_list_view = poco('android.widget.ScrollView')
        if not wait_exists(tag_list_view, timeout=15):
            logger.warning('ScrollView 未出现, 重试本次翻页')
            continue

        # 懒加载等待：App 每约 15 条才加载下一批，滑动后需等新内容真正出现再读，
        # 否则"还在加载就翻页"会把这批卡跳过（用底部卡片指纹是否变化来判断）
        if prev_bottom is not None:
            for _ in range(20):  # 最多等 ~10s
                cur_bottom = _bottom_key(tag_list_view)
                if cur_bottom and cur_bottom != prev_bottom:
                    break
                time.sleep(0.5)

        # 取列表 Button；取不到时重试本次翻页而不是抛异常（修复 0718 整轮被 kill）
        try:
            elements = list(tag_list_view.offspring('android.widget.Button'))
        except Exception as e:
            logger.warning(f'获取 Button 列表异常, 重试本次翻页: {e}')
            time.sleep(2)
            continue

        if not elements:
            logger.warning('未找到任何 Button，等待 2s 后重试本次翻页')
            time.sleep(2)
            continue

        page_last_key = prev_bottom
        for element in elements:
            element_title = element.child()
            if not element_title.exists():
                continue

            tag_element_data = element_title.attr('name')

            try:
                if '加入心愿单' in tag_element_data:
                    logger.debug('<加入心愿单> 字样在tag_element_data, 跳过')
                    continue

                element_data = parse_sales_data(tag_element_data)

                key = (element_data.get('title'), element_data.get('tv_number'))
                page_last_key = key  # 记录本页最后一张(不论新旧)，供下页判断加载完成
                if key in seen_keys:
                    continue
                seen_keys.add(key)
                logger.info(element_data)

                filtered_data = {k: v for k, v in element_data.items() if k != 'tv_rank'}
                filtered_data.update({
                    'category': category,
                    'language': language,
                    'crawler_date': time.strftime("%Y-%m-%d", time.localtime())
                })
                logger.debug(f'scraped data: {filtered_data}')
                data_list.append(filtered_data)

                # 排名到 100 则停止翻页
                if element_data.get('tv_rank') == '100':
                    logger.success(
                        f'已获取第 {page_count} 页数据, 第 {element_data["tv_rank"]} 名, 停止翻页....')
                    stop_page = True
                    break

            except ValueError as e:
                logger.error(f"数据解析错误: {e}")

        if stop_page:
            logger.debug(f'已获取 {len(data_list)} 条数据, 停止翻页....')
            break

        prev_bottom = page_last_key  # 记录本页底部, 下页据此等待懒加载完成
        page_count += 1
        time.sleep(1)
        logger.success(
            '---------------------------------------------------------------------------------------------')

    return data_list


def get_data(sql_pool):
    """
    获取数据
    :param sql_pool: MySQL连接池对象
    """
    scrape_index()

    for ca_idx, category_value in enumerate(category_list):
        category = list(category_value.keys())[0]
        lang_list = list(category_value.values())[0]
        logger.debug(
            f'当前分类索引: {ca_idx}, 分类名称: {category}, 语言列表: {lang_list} >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')

        for lang_idx, language in enumerate(lang_list):
            # 单个 (分类, 语言) 整体兜底：任一步失败只跳过该项，不让异常终止整轮任务
            try:
                # 1. 打开下拉框
                if not open_category_dropdown():
                    logger.warning(f'[{category}-{language}] 下拉框打开失败, 跳过')
                    continue

                # 2. 按文字选分类（滚动查找，避免旧版索引错位/越界）
                if not select_category(category):
                    logger.warning(f'[{category}-{language}] 分类选择失败, 跳过')
                    continue

                # 3. 点击语言
                logger.info(f'当前语言名称: {language} >>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>>')
                tag_lang_list = poco(language)
                if not tag_lang_list.exists():
                    logger.warning(f'[{category}-{language}] 语言项 [{language}] 未找到, 跳过')
                    continue
                tag_lang_list.click()

                # 4. 点击畅销榜（优先文字，回退模板）
                if not click_bestseller_tab():
                    logger.warning(f'[{category}-{language}] 畅销榜 Tab 点击失败, 跳过')
                    continue
                wait_exists(poco('android.widget.ScrollView'), timeout=10)

                # 5. 翻页抓取
                data_list = scrape_pages(category, language)

                # 6. 入库（scrape_pages 已内联去重）
                if data_list:
                    sql_pool.insert_many(table='jhs_bestseller_record', data_list=data_list, ignore=True)
                    logger.success(f'[{category}-{language}] 入库 {len(data_list)} 条')
                else:
                    logger.warning(f'[{category}-{language}] 未抓到数据, 不入库')

            except Exception as e:
                logger.error(f'[{category}-{language}] 处理异常, 跳过该项: {e}')
                continue


@retry(stop=stop_after_attempt(100), wait=wait_fixed(3600), after=after_log)
def jhs_app_bestseller_main():
    """
    主函数
    """
    logger.info(
        f'开始运行 {inspect.currentframe().f_code.co_name} 爬虫任务....................................................')

    # 运行时初始化设备/App/poco（放这里而非模块顶层）
    init_env()

    # 配置 MySQL 连接池
    sql_pool = MySQLConnectionPool(log=logger)
    if not sql_pool:
        logger.error("MySQL数据库连接失败")
        raise Exception("MySQL数据库连接失败")

    try:
        get_data(sql_pool)
    except Exception as e:
        logger.error(f'{inspect.currentframe().f_code.co_name} error: {e}')
    finally:
        logger.info(f'爬虫程序 {inspect.currentframe().f_code.co_name} 运行结束,等待下一轮的采集任务............')


if __name__ == '__main__':
    jhs_app_bestseller_main()