thinkphp6-template/public/storage/upload/20250826/40a24e98e5badf8809cb281ca9189947.txt

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 selenium.common.exceptions import TimeoutException
import time
import logging
import subprocess
import os

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
logger = logging.getLogger(__name__)

# Desired capabilities
desired_caps = {
    "platformName": "Android",
    "platformVersion": "12",
    "deviceName": "emulator-5558",
    "appPackage": "com.phoenix.read",
    "appActivity": "com.dragon.read.pages.splash.SplashActivity",
    "automationName": "UiAutomator2",
    "noReset": True,
    "fullReset": False,
    "autoGrantPermissions": True,
    "appium:ensureWebviewsHavePages": False,
    "appium:uiautomator2ServerLaunchTimeout": 120000,
    "appium:uiautomator2ServerInstallTimeout": 120000,
    "appium:newCommandTimeout": 14400,
    "appium:adbExecTimeout": 60000
}

# 初始化 driver
driver = None
options = UiAutomator2Options().load_capabilities(desired_caps)
try:
    driver = webdriver.Remote('http://localhost:4723/wd/hub', options=options)
    logger.info("Appium 连接成功")
except Exception as e:
    logger.error(f"连接失败: {str(e)}")
    exit(1)


def check_app_alive(max_attempts=3):
    """检查应用是否存活，若闪退则尝试重启"""
    logger.info("检查应用状态")
    global driver
    if driver is None:
        logger.error("driver 未初始化")
        return False
    for attempt in range(max_attempts):
        try:
            current_activity = driver.current_activity
            if current_activity and (
                    "com.phoenix.read" in current_activity or
                    "com.dragon.read" in current_activity or
                    "com.ss.android.excitingvideo" in current_activity
            ):
                logger.info(f"应用存活，当前 Activity: {current_activity}")
                return True
            else:
                logger.error(f"检测到非应用/广告 Activity: {current_activity}")
                raise Exception("非应用 Activity")
        except Exception as e:
            logger.error(f"应用可能闪退，尝试重启（第 {attempt + 1}/{max_attempts} 次）: {str(e)}")
            try:
                # 停止广告和应用进程
                subprocess.run(["adb", "shell", "am", "force-stop", "com.ss.android.excitingvideo"], check=True)
                subprocess.run(["adb", "shell", "am", "force-stop", "com.phoenix.read"], check=True)
                time.sleep(2)
                try:
                    with open(f"crash_page_source_attempt_{attempt + 1}.xml", "w", encoding="utf-8") as f:
                        f.write(driver.page_source)
                except:
                    logger.warning("无法保存崩溃页面源")
                try:
                    driver.start_activity("com.phoenix.read", "com.dragon.read.pages.splash.SplashActivity")
                except AttributeError:
                    subprocess.run(["adb", "shell", "am", "start", "-n",
                                    "com.phoenix.read/com.dragon.read.pages.splash.SplashActivity"], check=True)
                time.sleep(8)
                logger.info("应用重启成功")
                try:
                    handle_popup(fast=False)
                    if find_and_click_element(
                            (AppiumBy.ID, "com.phoenix.read:id/adx"),
                            "首页按钮",
                            wait_time=10
                    ):
                        logger.info("重启后已点击首页")
                        time.sleep(1)
                        handle_popup(fast=True)
                        return True
                    else:
                        raise Exception("点击首页按钮失败")
                except Exception as ex:
                    logger.error(f"重启后进入首页失败: {str(ex)}")
                    with open(f"home_error_page_source_attempt_{attempt + 1}.xml", "w", encoding="utf-8") as f:
                        f.write(driver.page_source)
            except Exception as ex:
                logger.error(f"应用重启失败: {str(ex)}")
                if attempt < max_attempts - 1:
                    logger.info("尝试重启 Appium 会话")
                    try:
                        driver.quit()
                    except:
                        pass
                    time.sleep(5)
                    try:
                        driver = webdriver.Remote('http://localhost:4723/wd/hub', options=options)
                        logger.info("Appium 会话重启成功")
                        time.sleep(5)
                    except Exception as re:
                        logger.error(f"Appium 会话重启失败: {str(re)}")
                        driver = None
        time.sleep(1)
    logger.error("应用多次重启失败")
    return False


def handle_popup(fast=False):
    """处理弹窗，增强登录处理"""
    logger.info("处理弹窗")
    if driver is None:
        logger.error("driver 未初始化")
        return
    wait_time = 2 if fast else 5
    popups = [
        '//android.widget.Button[contains(@text, "确定") or contains(@text, "关闭") or contains(@text, "取消") or contains(@text, "允许") or contains(@text, "拒绝") or contains(@text, "登录") or contains(@text, "立即登录") or contains(@content-desc, "登录")]',
        '//android.widget.ImageView[@content-desc="关闭" or contains(@content-desc, "close") or contains(@content-desc, "关闭广告")]',
        '//android.view.View[@content-desc="关闭" or contains(@content-desc, "close") or contains(@content-desc, "关闭广告")]',
        '//android.widget.FrameLayout[contains(@content-desc, "关闭") or contains(@content-desc, "close") or contains(@content-desc, "关闭广告")]',
        '//android.widget.TextView[contains(@text, "关闭") or contains(@text, "取消")]',  # 新增：处理可能的 TextView 关闭按钮
    ]
    for xpath in popups:
        try:
            popup_button = WebDriverWait(driver, timeout=wait_time).until(
                EC.element_to_be_clickable((AppiumBy.XPATH, xpath))
            )
            popup_button.click()
            logger.info(f"处理了弹窗: {xpath}")
            time.sleep(0.5 if fast else 1)
        except TimeoutException:
            logger.debug(f"未检测到弹窗: {xpath}")
        except Exception as e:
            logger.error(f"弹窗处理失败: {str(e)}")


def find_and_click_element(locator, description, wait_time=10):
    """检查元素存在并点击"""
    logger.info(f"尝试点击: {description}")
    try:
        elements = WebDriverWait(driver, wait_time).until(
            EC.presence_of_all_elements_located(locator)
        )
        if elements:
            elements[0].click()
            logger.info(f"成功点击: {description}")
            return True
        else:
            logger.error(f"未找到: {description}")
            with open(f"find_error_{description}.xml", "w", encoding="utf-8") as f:
                f.write(driver.page_source)
            try:
                driver.get_screenshot_as_file(f"find_error_{description}.png")
                logger.info(f"已保存截图: find_error_{description}.png")
            except Exception as e:
                logger.error(f"保存截图失败: {str(e)}")
            return False
    except TimeoutException as e:
        logger.error(f"查找超时: {description}, 错误: {str(e)}")
        with open(f"find_error_{description}.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file(f"find_error_{description}.png")
            logger.info(f"已保存截图: find_error_{description}.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        return False
    except Exception as e:
        logger.error(f"点击失败: {description}, 错误: {str(e)}")
        with open(f"find_error_{description}.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file(f"find_error_{description}.png")
            logger.info(f"已保存截图: find_error_{description}.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        return False


def click_welfare_button():
    """点击福利按钮"""
    logger.info("点击福利按钮")
    handle_popup(fast=True)
    if not find_and_click_element(
            (AppiumBy.ID, "com.phoenix.read:id/adw"),
            "福利按钮",
            wait_time=8
    ):
        with open("adw_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        raise Exception("点击福利按钮失败")
    time.sleep(1)


def handle_sign_in():
    """处理签到，仅首次"""
    logger.info("处理签到")
    handle_popup(fast=False)  # 慢速清理弹窗，确保页面干净

    # 定义“去签到”按钮的定位器
    go_sign_in_locators = [
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@content-desc="去签到"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@index="353"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@index="354"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@text, "去签到")]'),
        (AppiumBy.XPATH,
         '/hierarchy/android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/android.widget.LinearLayout/android.widget.FrameLayout/android.widget.FrameLayout[1]/android.widget.FrameLayout[2]/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/android.widget.FrameLayout/com.lynx.tasm.behavior.ui.LynxFlattenUI[81]'),
    ]

    # 定义“立即签到”按钮的定位器
    sign_in_locators = [
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.view.UIView[@content-desc="立即签到"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.view.UIView[contains(@content-desc, "立即签到")]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.view.UIView[@index="317"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.view.UIView[contains(@text, "立即签到")]'),
    ]

    def try_click_button(locators, description, wait_time=8):
        """尝试点击按钮，支持多重定位和点击方式"""
        for locator in locators:
            try:
                # 检查元素是否存在
                elements = driver.find_elements(*locator)
                if not elements:
                    logger.debug(f"未找到‘{description}’元素（locator: {locator[1]}）")
                    continue
                button = WebDriverWait(driver, timeout=wait_time).until(
                    EC.element_to_be_clickable(locator)
                )
                # 获取按钮属性
                attributes = {
                    "clickable": button.get_attribute("clickable") or "N/A",
                    "enabled": button.get_attribute("enabled") or "N/A",
                    "bounds": button.get_attribute("bounds") or "N/A",
                    "displayed": button.get_attribute("displayed") or "N/A",
                    "content-desc": button.get_attribute("content-desc") or "N/A",
                    "text": button.get_attribute("text") or "N/A",
                }
                logger.info(f"{description}按钮属性: {attributes}")
                # 尝试标准点击
                button.click()
                logger.info(f"已点击‘{description}’（locator={locator[1]}）")
                return True
            except TimeoutException:
                logger.error(f"未找到或无法点击‘{description}’（locator: {locator[1]}）")
                # 尝试 JavaScript 点击
                try:
                    elements = driver.find_elements(*locator)
                    if elements:
                        driver.execute_script("arguments[0].click();", elements[0])
                        logger.info(f"通过 JavaScript 点击‘{description}’成功（locator: {locator[1]}）")
                        return True
                except Exception as e:
                    logger.error(f"JavaScript 点击‘{description}’失败（locator: {locator[1]}）: {str(e)}")
                # 尝试坐标点击
                try:
                    elements = driver.find_elements(*locator)
                    if elements:
                        bounds = elements[0].get_attribute("bounds")
                        import re
                        coords = re.match(r"\[(\d+),(\d+)\]\[(\d+),(\d+)\]", bounds)
                        if coords:
                            x = (int(coords.group(1)) + int(coords.group(3))) // 2
                            y = (int(coords.group(2)) + int(coords.group(4))) // 2
                            driver.tap([(x, y)])
                            logger.info(f"通过坐标点击 (x={x}, y={y}) 成功（locator: {locator[1]}）")
                            return True
                        else:
                            logger.error(f"无法解析 bounds 属性（locator: {locator[1]}）")
                except Exception as e:
                    logger.error(f"坐标点击‘{description}’失败（locator: {locator[1]}）: {str(e)}")
        # 保存调试信息
        with open(f"click_error_{description}.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file(f"click_error_{description}.png")
            logger.info(f"已保存截图: click_error_{description}.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        return False

    # 刷新页面
    logger.info("尝试刷新页面状态")
    try:
        find_and_click_element(
            (AppiumBy.ID, "com.phoenix.read:id/adx"),
            "首页按钮",
            wait_time=8
        )
        time.sleep(2)
        click_welfare_button()
        handle_popup(fast=False)
    except Exception as e:
        logger.warning(f"页面刷新失败: {str(e)}")

    # 尝试点击“去签到”
    logger.info("尝试点击‘去签到’")
    if try_click_button(go_sign_in_locators, "去签到", wait_time=8):
        logger.info("已点击‘去签到’，检查‘立即签到’")
        time.sleep(2)  # 等待弹窗出现
        handle_popup(fast=True)
        # 点击“去签到”后检查“立即签到”
        if try_click_button(sign_in_locators, "立即签到", wait_time=8):
            logger.info("已点击‘立即签到’，签到完成")
        else:
            with open("claim_error_sign_after_go.xml", "w", encoding="utf-8") as f:
                f.write(driver.page_source)
            try:
                driver.get_screenshot_as_file("claim_error_sign_after_go.png")
                logger.info("已保存截图: claim_error_sign_after_go.png")
            except Exception as e:
                logger.error(f"保存截图失败: {str(e)}")
            logger.error("点击‘去签到’后仍未找到或无法点击‘立即签到’")
            raise Exception("点击‘立即签到’失败")
    else:
        with open("claim_error_sign_no_go.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("claim_error_sign_no_go.png")
            logger.info("已保存截图: claim_error_sign_no_go.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("未找到‘去签到’")
        raise Exception("签到流程失败")


def go_to_home_and_play_video():
    """去首页并刷视频"""
    logger.info("进入首页并刷视频")
    handle_popup(fast=True)
    start_time = time.time()
    if not find_and_click_element(
            (AppiumBy.ID, "com.phoenix.read:id/adx"),
            "首页按钮",
            wait_time=8
    ):
        with open("home_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("home_error.png")
            logger.info("已保存截图: home_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        raise Exception("点击首页按钮失败")
    time.sleep(1)
    logger.info(f"当前 Activity: {driver.current_activity}")

    logger.info("在首页停留22秒...")
    time.sleep(22)
    handle_popup(fast=True)
    current_activity_before = driver.current_activity
    logger.info(f"福利按钮前 Activity: {current_activity_before}")
    if not find_and_click_element(
            (AppiumBy.ID, "com.phoenix.read:id/adw"),
            "再次点击福利",
            wait_time=8
    ):
        with open("adw2_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("adw2_error.png")
            logger.info("已保存截图: adw2_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("再次点击福利失败")
        raise Exception("点击福利按钮失败")

    handle_popup(fast=True)
    # 定义“立即领取”按钮的定位器
    claim_locators = [
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@content-desc="立即领取"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@content-desc, "立即领取")]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@text, "立即领取")]'),
    ]
    # 定义“领取成功”按钮的定位器
    success_locators = [
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@content-desc="领取成功"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@content-desc, "领取成功")]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[@index="18"]'),
        (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@text, "领取成功")]'),
    ]

    if try_click_button(claim_locators, "立即领取", wait_time=10):
        logger.info("已点击‘立即领取’，等待视频播放45秒")
        time.sleep(45)  # 等待视频播放
        handle_popup(fast=True)

        # 检查并点击“领取成功”按钮
        if try_click_button(success_locators, "领取成功", wait_time=8):
            logger.info("已点击‘领取成功’，视频奖励领取完成")
        else:
            with open("claim_success_error.xml", "w", encoding="utf-8") as f:
                f.write(driver.page_source)
            try:
                driver.get_screenshot_as_file("claim_success_error.png")
                logger.info("已保存截图: claim_success_error.png")
            except Exception as e:
                logger.error(f"保存截图失败: {str(e)}")
            logger.warning("未找到或无法点击‘领取成功’按钮，视频可能未成功播放")

        # 检查应用状态
        for _ in range(6):  # 检查6次，每次5秒
            if check_app_alive():
                logger.info("视频播放后应用正常")
                return True
            time.sleep(5)
        logger.error("视频后多次检查闪退")
        return False
    else:
        with open("claim_video_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("claim_video_error.png")
            logger.info("已保存截图: claim_video_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("点击‘立即领取’失败")
        raise Exception("点击‘立即领取’失败")
    logger.info(f"加载耗时: {time.time() - start_time:.2f}秒")
    return True


def switch_to_welfare_page():
    """切换福利页面"""
    logger.info("切换到福利页面")
    handle_popup(fast=True)
    current_activity_before = driver.current_activity
    logger.info(f"切换福利前: {current_activity_before}")
    if find_and_click_element(
            (AppiumBy.ID, "com.phoenix.read:id/adw"),
            "切换福利",
            wait_time=8
    ):
        time.sleep(1)
        logger.info(f"切换后: {driver.current_activity}")
    else:
        with open("adw3_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("adw3_error.png")
            logger.info("已保存截图: adw3_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("切换失败")
        raise Exception("切换福利页面失败")


def claim_treasure():
    """领取宝箱"""
    logger.info("领取宝箱")
    handle_popup(fast=True)
    if find_and_click_element(
            (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.view.UIView[@content-desc="开宝箱得金币"]'),
            "开宝箱",
            wait_time=8
    ):
        logger.info("已打开宝箱")
        time.sleep(1)
    else:
        with open("treasure_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("treasure_error.png")
            logger.info("已保存截图: treasure_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("打开宝箱失败")

    handle_popup()
    if find_and_click_element(
            (AppiumBy.XPATH, '//com.lynx.tasm.behavior.ui.LynxFlattenUI[contains(@content-desc, "看视频最高再领")]'),
            "看视频",
            wait_time=10
    ):
        logger.info("已点击‘看视频’")
        time.sleep(30)  # 初始等待
        for _ in range(6):  # 检查6次，每次5秒
            if check_app_alive():
                logger.info("视频播放正常")
                # 等待关闭按钮
                try:
                    close_ad_button = WebDriverWait(driver, 10).until(
                        EC.element_to_be_clickable(
                            (AppiumBy.XPATH,
                             '//android.widget.ImageView[@content-desc="关闭" or @text="关闭" or contains(@content-desc, "close") or contains(@text, "button") or contains(@content-desc, "关闭广告")]')
                        )
                    )
                    close_ad_button.click()
                    logger.info("已关闭广告")
                    time.sleep(1)
                    return True
                except:
                    logger.info("未检测到关闭按钮，尝试继续")
            time.sleep(5)
        logger.error("视频后多次检查闪退")
        return False
    else:
        with open("watch_video_error.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("watch_video_error.png")
            logger.info("已保存截图: watch_video_error.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")
        logger.error("点击‘看视频’失败")
    return True


try:
    logger.info("开始执行脚本")
    # 优化模拟器
    try:
        subprocess.run(["adb", "shell", "wm", "size", "720x1280"], check=True)
        subprocess.run(["adb", "shell", "wm", "density", "240"], check=True)
        logger.info("已设置分辨率720x1280，密度240")
    except Exception as e:
        logger.error(f"设置模拟器失败: {str(e)}")

    # 初始加载
    logger.info("等待应用加载")
    time.sleep(10)
    logger.info(f"当前 Activity: {driver.current_activity}")

    # 初始弹窗
    handle_popup()

    # 执行功能
    click_welfare_button()
    handle_sign_in()
    if go_to_home_and_play_video():
        switch_to_welfare_page()
        claim_treasure()

    # 等待5小时
    logger.info("开始等待...")
    wait_time = 18000
    check_interval = 120
    start_time = time.time()
    while time.time() - start_time < wait_time:
        if not check_app_alive():
            logger.error("多次重启动失败，退出")
            break
        handle_popup(fast=True)
        time.sleep(check_interval)
        logger.info("应用仍在运行")

except Exception as e:
    logger.error(f"脚本错误: {str(e)}")
    if driver is not None:
        with open("error_page_source.xml", "w", encoding="utf-8") as f:
            f.write(driver.page_source)
        try:
            driver.get_screenshot_as_file("error_page_source.png")
            logger.info("已保存截图: error_page_source.png")
        except Exception as e:
            logger.error(f"保存截图失败: {str(e)}")

finally:
    logger.info("脚本执行结束，保持 driver 活跃")