xcode-select --install

# 安装 Node.js（Appium 需要）
brew install node

# 安装 Appium 服务器
npm install -g appium

# 安装 iOS 驱动
appium driver install xcuitest

# 安装 ios-deploy
npm install -g ios-deploy

# 安装 ideviceinstaller（用于获取应用信息）
brew install ideviceinstaller

ideviceinstaller -l

#npm默认安装的appium路径
open /Users/imac/.appium/node_modules/appium-xcuitest-driver/node_modules/appium-webdriveragent/WebDriverAgent.xcodeproj

(.venv) imac@home IOS2 % pip list
Package              Version
-------------------- -----------
Appium-Python-Client 5.1.1
attrs                25.3.0
certifi              2025.7.14
charset-normalizer   3.4.2
h11                  0.16.0
idna                 3.10
numpy                2.2.6
opencv-python        4.12.0.88
outcome              1.3.0.post0
pip                  25.1.1
PySocks              1.7.1
requests             2.32.4
selenium             4.34.2
sniffio              1.3.1
sortedcontainers     2.4.0
trio                 0.30.0
trio-websocket       0.12.2
typing_extensions    4.14.1
urllib3              2.5.0
websocket-client     1.8.0
wsproto              1.2.0
(.venv) imac@home IOS2 % 

from appium import webdriver
from appium.webdriver.common.appiumby import AppiumBy
from selenium.webdriver import ActionChains
from selenium.webdriver.common.actions import interaction
from selenium.webdriver.common.actions.action_builder import ActionBuilder
from selenium.webdriver.common.actions.pointer_input import PointerInput
from selenium.webdriver.support.ui import WebDriverWait
from selenium.webdriver.support import expected_conditions as EC
import time
import cv2
import numpy as np
from appium.options.ios import XCUITestOptions
import traceback
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
from datetime import datetime


class iOSAutomation:
    def __init__(self, udid):
        """初始化iOS自动化驱动"""
        self.udid = udid
        self.driver = self._init_driver()
        self.pointer_input = PointerInput(interaction.POINTER_TOUCH, "touch")
        self.actions = ActionChains(self.driver)
        # 获取设备屏幕尺寸
        self.screen_size = self.driver.get_window_size()
        self.screen_width = self.screen_size['width']
        self.screen_height = self.screen_size['height']

    def _init_driver(self):
        """初始化Appium驱动（使用新的options方式）"""
        options = XCUITestOptions()
        options.platform_name = "iOS"
        options.automation_name = "XCUITest"
        options.device_name = "iPhone SE"
        options.udid = self.udid
        options.platform_version = "15.8"
        options.no_reset = True
        options.wda_local_port = 8100
        return webdriver.Remote("http://localhost:4723", options=options)

    # 回到主屏幕
    def back_to_home(self):
        self.driver.execute_script("mobile: pressButton", {"name": "home"})
        time.sleep(1)

    # 打开指定应用
    def open_app(self, bundle_id):
        self.driver.execute_script("mobile: launchApp", {"bundleId": bundle_id})
        time.sleep(2)

    # 关闭指定应用
    def close_app(self, bundle_id):
        """通过bundle ID关闭应用"""
        try:
            self.driver.execute_script(
                "mobile: terminateApp",
                {"bundleId": bundle_id}
            )
            time.sleep(1)
            print(f"成功关闭应用: {bundle_id}")
            return True
        except Exception as e:
            print(f"关闭应用失败: {str(e)}")
            return False

    # 点击元素(XPath)
    def click(self, xpath, timeout=10):
        try:
            element = WebDriverWait(self.driver, timeout).until(
                EC.presence_of_element_located((AppiumBy.XPATH, xpath))
            )
            element.click()
            time.sleep(0.5)
            return True
        except Exception as e:
            print(f"点击失败: {str(e)}")
            return False

    # 双击元素(XPath)
    def double_click(self, xpath, timeout=10):
        try:
            element = WebDriverWait(self.driver, timeout).until(
                EC.presence_of_element_located((AppiumBy.XPATH, xpath))
            )
            actions = ActionChains(self.driver)
            actions.double_click(element).perform()
            time.sleep(0.5)
            return True
        except Exception as e:
            print(f"双击失败: {str(e)}")
            return False

    # 滑动操作
    def swipe(self, start_x, start_y, end_x, end_y, duration=500):
        try:
            actual_start_x = self.screen_width * start_x
            actual_start_y = self.screen_height * start_y
            actual_end_x = self.screen_width * end_x
            actual_end_y = self.screen_height * end_y

            actions = ActionBuilder(self.driver, mouse=self.pointer_input)
            actions.pointer_action.move_to_location(actual_start_x, actual_start_y)
            actions.pointer_action.pointer_down()
            actions.pointer_action.pause(duration / 1000)
            actions.pointer_action.move_to_location(actual_end_x, actual_end_y)
            actions.pointer_action.pointer_up()
            actions.perform()

            time.sleep(0.5)
            return True
        except Exception as e:
            print(f"滑动失败: {str(e)}")
            return False

    # 增加返回手势功能：从屏幕最左边向右侧滑动70%
    def swipe_back(self, duration=300):
        """
        模拟iOS返回手势：从屏幕左侧向右滑动70%宽度
        :param duration: 滑动持续时间(毫秒)，默认300ms
        :return: 是否成功
        """
        try:
            # 起始点：屏幕左侧中间位置（x=0%，y=50%）
            # 结束点：屏幕右侧70%宽度位置（x=70%，y=50%）
            return self.swipe(0.05, 0.5, 0.8, 0.5, duration)
            # 使用0.05而不是0完美避开边缘可能的无响应区域
        except Exception as e:
            print(f"返回手势失败: {str(e)}")
            return False

    # 增加返回手势功能：从屏幕最下向上华东
    def swipe_up(self, duration=300):
        """
        模拟iOS返回手势：从屏幕底部向上滑动
        :param duration: 滑动持续时间(毫秒)，默认300ms
        :return: 是否成功
        """
        try:
            # 起始点：屏幕下方中间位置（x=50%，y=70%）
            # 结束点：屏幕上方中间位置（x=50%，y=30%）
            return self.swipe(0.5, 0.7, 0.5, 0.3, duration)
        except Exception as e:
            print(f"上滑手势失败: {str(e)}")
            return False

    # 长按元素(XPath)
    def long_press(self, xpath, duration=1000, timeout=10):
        try:
            element = WebDriverWait(self.driver, timeout).until(
                EC.presence_of_element_located((AppiumBy.XPATH, xpath))
            )
            actions = ActionChains(self.driver)
            actions.click_and_hold(element).pause(duration / 1000).release().perform()
            time.sleep(0.5)
            return True
        except Exception as e:
            print(f"长按失败: {str(e)}")
            return False

    # 等待元素出现
    def wait_for_element(self, xpath, timeout=30):
        try:
            return WebDriverWait(self.driver, timeout).until(
                EC.presence_of_element_located((AppiumBy.XPATH, xpath))
            )
        except Exception as e:
            print(f"等待元素失败: {str(e)}")
            return None

    # 坐标百分比点击
    def click_by_percentage(self, x_percent, y_percent):
        """
        按屏幕百分比坐标点击
        :param x_percent: X轴百分比 (0-1)
        :param y_percent: Y轴百分比 (0-1)
        :return: 是否成功
        """
        try:
            # 计算实际坐标
            x = self.screen_width * x_percent
            y = self.screen_height * y_percent

            # 执行点击操作
            actions = ActionBuilder(self.driver, mouse=self.pointer_input)
            actions.pointer_action.move_to_location(x, y)
            actions.pointer_action.pointer_down()
            actions.pointer_action.pause(0.1)  # 短暂按下
            actions.pointer_action.pointer_up()
            actions.perform()

            time.sleep(0.5)
            print(f"成功点击坐标百分比: ({x_percent}, {y_percent})")
            return True
        except Exception as e:
            print(f"坐标点击失败: {str(e)}")
            return False

    # 截屏
    def shot_png(self, shot_path="shot.png"):
        self.driver.save_screenshot(shot_path)
        print(f"截屏：{shot_path}")

    # 图片点击（适配缩放，带超时检测和异常处理）
    def click_image_with_timeout(self, target_image_path, threshold=0.8, timeout=10, max_retries=1):
        """
        在指定时间内等待目标图片出现并点击，超时则返回False但不终止脚本
        :param target_image_path: 目标图片路径
        :param threshold: 匹配阈值，0-1之间
        :param timeout: 单次超时时间（秒）
        :param max_retries: 最大重试次数
        :return: 是否成功
        """
        retries = 0
        last_max_val = 0  # 记录最高匹配度，用于调试

        while retries <= max_retries:
            try:
                start_time = time.time()

                while time.time() - start_time < timeout:
                    try:
                        # 截取当前屏幕
                        screenshot_path = "current_screenshot.png"
                        self.driver.save_screenshot(screenshot_path)

                        # 读取截图和目标图片
                        screen = cv2.imread(screenshot_path)
                        target = cv2.imread(target_image_path)

                        if screen is None:
                            raise Exception(f"无法读取截图: {screenshot_path}")
                        if target is None:
                            raise Exception(f"无法读取目标图片，请检查路径: {target_image_path}")

                        # 使用模板匹配查找目标图片
                        result = cv2.matchTemplate(screen, target, cv2.TM_CCOEFF_NORMED)
                        min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)
                        last_max_val = max_val  # 更新最高匹配度

                        # 检查匹配度
                        if max_val >= threshold:
                            print(f"找到匹配图片，匹配度: {max_val}")

                            # 计算目标图片中心位置（在截图中的坐标）
                            target_height, target_width = target.shape[:2]
                            center_x = max_loc[0] + target_width // 2
                            center_y = max_loc[1] + target_height // 2

                            # 计算相对于屏幕的百分比坐标（适配不同分辨率和缩放）
                            x_percent = center_x / screen.shape[1]
                            y_percent = center_y / screen.shape[0]

                            # 使用百分比坐标点击
                            return self.click_by_percentage(x_percent, y_percent)

                    except Exception as e:
                        print(f"图片识别过程中出错: {str(e)}")
                        print(traceback.format_exc())  # 打印详细错误堆栈

                    # 等待一段时间后重试
                    time.sleep(0.5)

                # 超时未找到图片，准备重试
                retries += 1
                if retries <= max_retries:
                    print(f"第{retries}次重试寻找图片: {target_image_path}")
                    time.sleep(1)  # 重试前等待1秒

            except Exception as e:
                print(f"重试过程中发生错误: {str(e)}")
                retries += 1
                if retries > max_retries:
                    break

        # 所有重试都失败，返回False并打印信息
        print(
            f"未找到图片: {target_image_path}，在{timeout * (max_retries + 1)}秒内（含{max_retries}次重试）"
            f"最高匹配度: {last_max_val}, 阈值: {threshold}"
        )
        return False

    # 判断图片是否存在,用于判断状态
    def image_exist(self, target_image_path, threshold=0.8, max_scale=1.2, min_scale=0.8, step=0.1):
        """
        检查目标图片是否存在于当前屏幕中（支持图片缩放匹配）
        :param target_image_path: 目标图片路径
        :param threshold: 匹配阈值，0-1之间
        :param max_scale: 最大缩放比例
        :param min_scale: 最小缩放比例
        :param step: 缩放步长
        :return: 存在返回True，否则返回False
        """
        try:
            # 截取当前屏幕
            screenshot_path = "current_screenshot.png"
            self.driver.save_screenshot(screenshot_path)

            # 读取截图和目标图片
            screen = cv2.imread(screenshot_path)
            target = cv2.imread(target_image_path)

            if screen is None:
                raise Exception(f"无法读取截图: {screenshot_path}")
            if target is None:
                raise Exception(f"无法读取目标图片，请检查路径: {target_image_path}")

            # 获取目标图片原始尺寸
            target_height, target_width = target.shape[:2]

            # 尝试不同缩放比例的目标图片进行匹配
            scale = max_scale
            while scale >= min_scale:
                # 计算缩放后的尺寸
                scaled_width = int(target_width * scale)
                scaled_height = int(target_height * scale)

                # 跳过尺寸过小的情况
                if scaled_width < 10 or scaled_height < 10:
                    scale -= step
                    continue

                # 缩放目标图片
                scaled_target = cv2.resize(target, (scaled_width, scaled_height), interpolation=cv2.INTER_AREA)

                # 进行模板匹配
                result = cv2.matchTemplate(screen, scaled_target, cv2.TM_CCOEFF_NORMED)
                max_val = np.max(result)

                # 检查是否达到匹配阈值
                if max_val >= threshold:
                    print(f"找到匹配图片，缩放比例: {scale:.2f}, 匹配度: {max_val:.4f}")
                    return True

                scale -= step

            # 所有缩放比例都检查过仍未找到匹配
            print(f"未找到匹配图片，最高匹配度低于阈值 {threshold}")
            return False

        except Exception as e:
            print(f"图片存在性检查出错: {str(e)}")
            return False

    # 发送钉钉消息
    def send_dingtalk_message(self, webhook, secret, content):
        """
        通过钉钉机器人发送消息
        :param webhook: 钉钉机器人webhook地址
        :param secret: 钉钉机器人加签密钥
        :param content: 要发送的消息内容
        :return: 发送是否成功
        """
        try:
            # 计算时间戳和签名
            timestamp = str(round(time.time() * 1000))
            secret_enc = secret.encode('utf-8')
            string_to_sign = f'{timestamp}\n{secret}'
            string_to_sign_enc = string_to_sign.encode('utf-8')
            hmac_code = hmac.new(secret_enc, string_to_sign_enc, digestmod=hashlib.sha256).digest()
            sign = urllib.parse.quote_plus(base64.b64encode(hmac_code))

            # 构造完整的请求URL
            url = f"{webhook}&timestamp={timestamp}&sign={sign}"

            # 构造消息体
            headers = {'Content-Type': 'application/json;charset=utf-8'}
            data = {
                "msgtype": "text",
                "text": {
                    "content": f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}\n{content}"
                }
            }

            # 发送请求
            response = requests.post(url, headers=headers, data=json.dumps(data))
            result = response.json()

            if result.get("errcode") == 0:
                print("钉钉消息发送成功")
                return True
            else:
                print(f"钉钉消息发送失败: {result.get('errmsg')}")
                return False

        except Exception as e:
            print(f"发送钉钉消息时发生错误: {str(e)}")
            return False

    # 退出驱动
    def quit(self):
        self.driver.quit()


# 示例用法
if __name__ == "__main__":
    DEVICE_UDID = "670784213ac36ee6e67042c7c11cfc01c44a75b4"
    ios = iOSAutomation(DEVICE_UDID)

    # 钉钉配置
    DINGTALK_WEBHOOK = "https://oapi.dingtalk.com/robot/send?access_token=cea075108ea3eee904ea591b7a72a2cc9431877fdb141c50d09b22e89d97ff02"
    DINGTALK_SECRET = "SECe04971aa5d5b8afb27ac6f7ad5b7cfc114ef34bdee69597e524b244206b35b5b"

    try:
        # 发送任务开始通知
        ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, "===  自动化开始  ===")

        ##################################################################
        # 微信操作
        OPEN_STAT = "ON"
        APP_NAME = "微信"
        APP_ID = "com.tencent.xin"
        OVER_STAT=False
        if OPEN_STAT == 'ON' or OPEN_STAT == 'on':
            try:
                # 发送任务开始通知
                #messsage_var=f"自动化任务{}"
                ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, f"开始执行:{APP_NAME}")
                print("回到首页")
                ios.back_to_home()

                print(f"关闭应用：{APP_NAME}")
                ios.close_app(APP_ID)

                print(f"打开应用：{APP_NAME}")
                ios.open_app(APP_ID)
                time.sleep(3)

                # 执行操作...
                print("点击：河南电信")
                APP_XPATH = '//XCUIElementTypeStaticText[@name="河南电信"]'
                ios.wait_for_element(APP_XPATH)
                ios.click(APP_XPATH)

                # 打开签到页面
                for i in range(3):
                    print("点击：签到领流量>>")
                    APP_XPATH = '(//XCUIElementTypeOther[@name="河南电信,签到领流量>>"])[4]'
                    ios.wait_for_element(APP_XPATH)
                    ios.click(APP_XPATH)
                    time.sleep(20)
                    # 判断图片是否存在
                    stat = ios.image_exist("./img/微信/签到页面.png")
                    if stat:
                        break
                    else:
                        ios.swipe_back()
                        time.sleep(2)

                # 滑动
                print("滑动")
                ios.swipe(0.5, 0.6, 0.5, 0.4)

                # 截屏
                print("截屏")
                ios.shot_png(f"{APP_NAME}.png")

                # 点击能量
                for i in range(3):
                    # 判断是否签到
                    stat = ios.image_exist("./img/微信/已签到.png")
                    if stat:
                        ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, f"{APP_NAME}签到已完成")
                        raise Exception("签到已完成，结束流程")
                    else:
                        print("点击指定图片")
                        ios.click_image_with_timeout("./img/微信/未签到.png")
                    time.sleep(1)

            except Exception as e:
                error_msg = f"{APP_NAME}操作终止: {str(e)}"
                print(error_msg)
                ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, error_msg)

        ##################################################################
        # 阿里云盘操作
        OPEN_STAT = "ON"
        APP_NAME = "阿里云盘"
        APP_ID = "com.alicloud.smartdrive"
        if OPEN_STAT == 'ON' or OPEN_STAT == 'on':
            try:
                # 发送任务开始通知
                #messsage_var=f"自动化任务{}"
                ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, f"开始执行:{APP_NAME}")
                print("回到首页")
                ios.back_to_home()

                print(f"关闭应用：{APP_NAME}")
                ios.close_app(APP_ID)

                print(f"打开应用：{APP_NAME}")
                ios.open_app(APP_ID)
                time.sleep(3)

                # 截屏
                ios.shot_png(f"{APP_NAME}.png")

                # 执行操作...
                # 点击指定图片 - 现在找不到图片会返回False并继续执行
                # 点击笑脸
                ios.click_image_with_timeout("./img/阿里云盘/笑脸.png")

                # 滑动
                print("滑动")
                ios.swipe(0.5, 0.8, 0.5, 0.2)

                # 点击
                print("点击：去完成")
                APP_XPATH = '//XCUIElementTypeStaticText[@name="去完成"]'
                ios.wait_for_element(APP_XPATH)
                ios.click(APP_XPATH)
                time.sleep(3)

                print("点击屏幕百分比")
                ios.click_by_percentage(0.92, 0.075)
                time.sleep(2)

                # 点击指定图片 - 现在找不到图片会返回False并继续执行
                # 点击能量
                for i in range(5):
                    print("点击指定图片")
                    success = ios.click_image_with_timeout("./img/阿里云盘/能量球.png")  # 替换为你的模板图片路径
                    if success:
                        ios.swipe_back()
                        print("点击屏幕百分比")
                        ios.click_by_percentage(0.92, 0.075)
                        time.sleep(2)
                    else:
                        print("图片点击操作未成功执行，继续后续步骤...")
                        ios.swipe_up()

                    time.sleep(1)

                print("返回")
                ios.swipe_back()
                ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, "阿里云盘操作完成")

            except Exception as e:
                error_msg = f"阿里云盘操作发生错误: {str(e)}"
                print(error_msg)
                ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, error_msg)

    except Exception as e:
        error_msg = f"全局异常: {str(e)}"
        print(error_msg)
        ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, error_msg)
    finally:
        # 发送任务结束通知
        ios.send_dingtalk_message(DINGTALK_WEBHOOK, DINGTALK_SECRET, "自动化任务执行结束")
        # 最终确保资源正确释放
        ios.quit()