From a0272b7f35e590825fd81ba818ef1e4b08f53553 Mon Sep 17 00:00:00 2001 From: xiaowei <1437591651@qq.com> Date: Sun, 14 Sep 2025 01:27:14 +0800 Subject: [PATCH] =?UTF-8?q?feature(=E6=8A=A2=E7=A5=A8=E8=84=9A=E6=9C=AC)1.?= =?UTF-8?q?=E9=80=82=E9=85=8D=E6=9C=80=E6=96=B0=E5=A4=A7=E9=BA=A6APP;2.?= =?UTF-8?q?=E4=BC=98=E5=8C=96=E6=8A=A2=E7=A5=A8=E9=80=9F=E5=BA=A6;3.?= =?UTF-8?q?=E6=B7=BB=E5=8A=A0=E4=BF=AE=E6=94=B9=E4=BF=A1=E6=81=AF=E7=9A=84?= =?UTF-8?q?md?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- damai_appium/app.md | 42 +++++ damai_appium/config.jsonc | 13 ++ damai_appium/config.py | 15 +- damai_appium/damai_app_v2.py | 288 +++++++++++++++++++++++++++++++++++ 4 files changed, 355 insertions(+), 3 deletions(-) create mode 100644 damai_appium/app.md create mode 100644 damai_appium/config.jsonc create mode 100644 damai_appium/damai_app_v2.py diff --git a/damai_appium/app.md b/damai_appium/app.md new file mode 100644 index 0000000..25a99c4 --- /dev/null +++ b/damai_appium/app.md @@ -0,0 +1,42 @@ +# 安卓端V2版本介绍 +## 执行命令 +### 开启appium服务端 +```bash +appium --address 0.0.0.0 --port 4723 --relaxed-security +``` +如果确定某些按钮点击后不会马上有新页面加载，可以加 `--relaxed-security` 启动 Appium，然后用 `mobile: clickGesture` 直接原生点击： +```python +# 这里的target是一个可以执行click()的对象 +driver.execute_script('mobile: clickGesture', {'elementId': target.id}) +``` +### 执行抢票任务 +```bash +cd damai_appium +python damai_app_v2.py +``` + + +## 只处理了抢票的，预约的暂未考虑 + +## 功能 +- 大麦的大部分票**只能在APP端购买**，所以只运行了安卓侧的实现并进行修改 +- APP更新，**界面信息的票价的Text是空串""**，无法再使用之前的方案去找按钮click，V2是通过分析页面信息，使用索引的方式获取，缺点是需要预先手动写进去，不知道后续有没有什么新的方法获取 +- 增加重试机制 + +## 优化： +- 考虑到界面可以先点到搜索列表，移除了键入搜索和点击搜索按钮的步骤 +- 增加了一些加速的配置capabilities，以及一些性能优化的配置 +- 优化了多人勾选的逻辑，收集坐标信息，几乎一次性全部点击 +- 使用`WebDriverWait`替代`driver.implicitly_wait(5)`，大大提升效率 +- 优化了`click()`的方式，使用 +```python +driver.execute_script("mobile: clickGesture", { + "x": x, + "y": y, + "duration": 50 # 极短点击时间 + }) +``` +- 优化显示逻辑，展示执行的进度 + +## 展望 +- 实现预约功能 \ No newline at end of file diff --git a/damai_appium/config.jsonc b/damai_appium/config.jsonc new file mode 100644 index 0000000..2447c57 --- /dev/null +++ b/damai_appium/config.jsonc @@ -0,0 +1,13 @@ +{ + "server_url": "127.0.0.1:4723", + "keyword": "刘若英", + "users": [ + "xx", // 观演人，需要账号里添加过，一个人是最快的 + "yy" + ], + "city": "泉州", + "date": "10.04", // 无效，当只有一个city的时候，日期不重要了 + "price": "799元", // 无效，大麦的页面不显示价格了，用下面的索引 + "price_index": 1, // 对应票价的索引，从0开始，从低到高的价格排序 + "if_commit_order": true // 无效，默认提交订单 +} diff --git a/damai_appium/config.py b/damai_appium/config.py index 5578333..32bbfdb 100644 --- a/damai_appium/config.py +++ b/damai_appium/config.py @@ -6,26 +6,35 @@ __Description__ = "配置类" __Created__ = 2023/10/27 09:54 """ import json +import re class Config: - def __init__(self, server_url, keyword, users, city, date, price, if_commit_order): + def __init__(self, server_url, keyword, users, city, date, price, price_index, if_commit_order): self.server_url = server_url self.keyword = keyword self.users = users self.city = city self.date = date self.price = price + self.price_index = price_index self.if_commit_order = if_commit_order @staticmethod def load_config(): - with open('config.json', 'r', encoding='utf-8') as config_file: - config = json.load(config_file) + with open('config.jsonc', 'r', encoding='utf-8') as config_file: + content = config_file.read() + # 去掉 // 的注释 + content = re.sub(r'//.*', '', content) + # content = re.sub(r'/\*[\s\S]*?\*/', '', content) + print(content) + config = json.loads(content) + # config = json.load(config_file) return Config(config['server_url'], config['keyword'], config['users'], config['city'], config['date'], config['price'], + config['price_index'], config['if_commit_order']) diff --git a/damai_appium/damai_app_v2.py b/damai_appium/damai_app_v2.py new file mode 100644 index 0000000..ba8fa0e --- /dev/null +++ b/damai_appium/damai_app_v2.py @@ -0,0 +1,288 @@ +# -*- coding: UTF-8 -*- +""" +__Author__ = "BlueCestbon" +__Version__ = "2.0.0" +__Description__ = "大麦app抢票自动化 - 优化版" +__Created__ = 2025/09/13 19:27 +""" + +import time +from appium import webdriver +from appium.options.common.base import AppiumOptions +from appium.webdriver.common.appiumby import AppiumBy +from selenium.webdriver.common.by import By +from selenium.webdriver.support.ui import WebDriverWait +from selenium.webdriver.support import expected_conditions as EC +from selenium.common.exceptions import TimeoutException, NoSuchElementException + +from config import Config + + +class DamaiBot: + def __init__(self): + self.config = Config.load_config() + self.driver = None + self.wait = None + self._setup_driver() + + def _setup_driver(self): + """初始化驱动配置""" + capabilities = { + "platformName": "Android", # 操作系统 + "platformVersion": "14", # 系统版本 + "deviceName": "OPPO Find X8 Pro", # 设备名称 + "appPackage": "cn.damai", # app 包名 + "appActivity": ".launcher.splash.SplashMainActivity", # app 启动 Activity + "unicodeKeyboard": True, # 支持 Unicode 输入 + "resetKeyboard": True, # 隐藏键盘 + "noReset": True, # 不重置 app + "newCommandTimeout": 6000, # 超时时间 + "automationName": "UiAutomator2", # 使用 uiautomator2 + "skipServerInstallation": False, # 跳过服务器安装 + "ignoreHiddenApiPolicyError": True, # 忽略隐藏 API 策略错误 + "disableWindowAnimation": True, # 禁用窗口动画 + # 优化性能配置 + "mjpegServerFramerate": 1, # 降低截图帧率 + "shouldTerminateApp": False, + "adbExecTimeout": 20000, + } + + device_app_info = AppiumOptions() + device_app_info.load_capabilities(capabilities) + self.driver = webdriver.Remote(self.config.server_url, options=device_app_info) + + # 更激进的性能优化设置 + self.driver.update_settings({ + "waitForIdleTimeout": 0, # 空闲时间，0 表示不等待，让 UIAutomator2 不等页面“空闲”再返回 + "actionAcknowledgmentTimeout": 0, # 禁止等待动作确认 + "keyInjectionDelay": 0, # 禁止输入延迟 + "waitForSelectorTimeout": 300, # 从500减少到300ms + "ignoreUnimportantViews": False, # 保持false避免元素丢失 + "allowInvisibleElements": True, + "enableNotificationListener": False, # 禁用通知监听 + }) + + # 极短的显式等待，抢票场景下速度优先 + self.wait = WebDriverWait(self.driver, 2) # 从5秒减少到2秒 + + def ultra_fast_click(self, by, value, timeout=1.5): + """超快速点击 - 适合抢票场景""" + try: + # 直接查找并点击，不等待可点击状态 + el = WebDriverWait(self.driver, timeout).until( + EC.presence_of_element_located((by, value)) + ) + # 使用坐标点击更快 + rect = el.rect + x = rect['x'] + rect['width'] // 2 + y = rect['y'] + rect['height'] // 2 + self.driver.execute_script("mobile: clickGesture", { + "x": x, + "y": y, + "duration": 50 # 极短点击时间 + }) + return True + except TimeoutException: + return False + + def batch_click(self, elements_info, delay=0.1): + """批量点击操作""" + for by, value in elements_info: + if self.ultra_fast_click(by, value): + if delay > 0: + time.sleep(delay) + else: + print(f"点击失败: {value}") + + def ultra_batch_click(self, elements_info, timeout=2): + """超快批量点击 - 带等待机制""" + coordinates = [] + # 批量收集坐标，带超时等待 + for by, value in elements_info: + try: + # 等待元素出现 + el = WebDriverWait(self.driver, timeout).until( + EC.presence_of_element_located((by, value)) + ) + rect = el.rect + x = rect['x'] + rect['width'] // 2 + y = rect['y'] + rect['height'] // 2 + coordinates.append((x, y, value)) + except TimeoutException: + print(f"超时未找到用户: {value}") + except Exception as e: + print(f"查找用户失败 {value}: {e}") + print(f"成功找到 {len(coordinates)} 个用户") + # 快速连续点击 + for i, (x, y, value) in enumerate(coordinates): + self.driver.execute_script("mobile: clickGesture", { + "x": x, + "y": y, + "duration": 30 + }) + if i < len(coordinates) - 1: + time.sleep(0.01) + print(f"点击用户: {value}") + + def smart_wait_and_click(self, by, value, backup_selectors=None, timeout=1.5): + """智能等待和点击 - 支持备用选择器""" + selectors = [(by, value)] + if backup_selectors: + selectors.extend(backup_selectors) + + for selector_by, selector_value in selectors: + try: + el = WebDriverWait(self.driver, timeout).until( + EC.presence_of_element_located((selector_by, selector_value)) + ) + rect = el.rect + x = rect['x'] + rect['width'] // 2 + y = rect['y'] + rect['height'] // 2 + self.driver.execute_script("mobile: clickGesture", {"x": x, "y": y, "duration": 50}) + return True + except TimeoutException: + continue + return False + + def run_ticket_grabbing(self): + """执行抢票主流程""" + try: + print("开始抢票流程...") + start_time = time.time() + + # 1. 城市选择 - 准备多个备选方案 + print("选择城市...") + city_selectors = [ + (AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().text("{self.config.city}")'), + (AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().textContains("{self.config.city}")'), + (By.XPATH, f'//*[@text="{self.config.city}"]') + ] + if not self.smart_wait_and_click(*city_selectors[0], city_selectors[1:]): + print("城市选择失败") + return False + + # 2. 点击预约按钮 - 多种可能的按钮文本 + print("点击预约按钮...") + book_selectors = [ + (By.ID, "cn.damai:id/trade_project_detail_purchase_status_bar_container_fl"), + (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().textMatches(".*预约.*|.*购买.*|.*立即.*")'), + (By.XPATH, '//*[contains(@text,"预约") or contains(@text,"购买")]') + ] + if not self.smart_wait_and_click(*book_selectors[0], book_selectors[1:]): + print("预约按钮点击失败") + return False + + # 3. 票价选择 - 优化查找逻辑 + print("选择票价...") + try: + # 直接尝试点击，不等待容器，实际每次都失败，只能等待 + price_container = self.driver.find_element(By.ID, 'cn.damai:id/project_detail_perform_price_flowlayout') + # price_container = self.wait.until( # 等待找到容器 + # EC.presence_of_element_located((By.ID, 'cn.damai:id/project_detail_perform_price_flowlayout'))) + # 在容器内找 index=1 且 clickable="true" 的 FrameLayout【因为799元的票价是排在第二的，但是page里text是空的被隐藏了】 + target_price = price_container.find_element( + AppiumBy.ANDROID_UIAUTOMATOR, + f'new UiSelector().className("android.widget.FrameLayout").index({self.config.price_index}).clickable(true)' + ) + self.driver.execute_script('mobile: clickGesture', {'elementId': target_price.id}) + except Exception as e: + print(f"票价选择失败，启动备用方案: {e}") + # 备用方案 + # 先找到大容器 + price_container = self.wait.until( + EC.presence_of_element_located((By.ID, 'cn.damai:id/project_detail_perform_price_flowlayout'))) + # 在容器内找 index=1 且 clickable="true" 的 FrameLayout【因为799元的票价是排在第二的，但是page里text是空的被隐藏了】 + target_price = price_container.find_element( + AppiumBy.ANDROID_UIAUTOMATOR, + f'new UiSelector().className("android.widget.FrameLayout").index({self.config.price_index}).clickable(true)' + ) + self.driver.execute_script('mobile: clickGesture', {'elementId': target_price.id}) + + # if not self.ultra_fast_click(AppiumBy.ANDROID_UIAUTOMATOR, + # 'new UiSelector().textMatches(".*799.*|.*\\d+元.*")'): + # return False + + # 4. 数量选择 + print("选择数量...") + if self.driver.find_elements(by=By.ID, value='layout_num'): + clicks_needed = len(self.config.users) - 1 + if clicks_needed > 0: + try: + plus_button = self.driver.find_element(By.ID, 'img_jia') + for i in range(clicks_needed): + rect = plus_button.rect + x = rect['x'] + rect['width'] // 2 + y = rect['y'] + rect['height'] // 2 + self.driver.execute_script("mobile: clickGesture", { + "x": x, + "y": y, + "duration": 50 + }) + time.sleep(0.02) + except Exception as e: + print(f"快速点击加号失败: {e}") + + # if self.driver.find_elements(by=By.ID, value='layout_num') and self.config.users is not None: + # for i in range(len(self.config.users) - 1): + # self.driver.find_element(by=By.ID, value='img_jia').click() + + # 5. 确定购买 + print("确定购买...") + if not self.ultra_fast_click(By.ID, "btn_buy_view"): + # 备用按钮文本 + self.ultra_fast_click(AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().textMatches(".*确定.*|.*购买.*")') + + # 6. 批量选择用户 + print("选择用户...") + user_clicks = [(AppiumBy.ANDROID_UIAUTOMATOR, f'new UiSelector().text("{user}")') for user in + self.config.users] + # self.batch_click(user_clicks, delay=0.05) # 极短延迟 + self.ultra_batch_click(user_clicks) + + # 7. 提交订单 + print("提交订单...") + submit_selectors = [ + (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().text("立即提交")'), + (AppiumBy.ANDROID_UIAUTOMATOR, 'new UiSelector().textMatches(".*提交.*|.*确认.*")'), + (By.XPATH, '//*[contains(@text,"提交")]') + ] + self.smart_wait_and_click(*submit_selectors[0], submit_selectors[1:]) + + end_time = time.time() + print(f"抢票流程完成，耗时: {end_time - start_time:.2f}秒") + return True + + except Exception as e: + print(f"抢票过程发生错误: {e}") + return False + finally: + time.sleep(1) # 给最后的操作一点时间 + self.driver.quit() + + def run_with_retry(self, max_retries=3): + """带重试机制的抢票""" + for attempt in range(max_retries): + print(f"第 {attempt + 1} 次尝试...") + if self.run_ticket_grabbing(): + print("抢票成功！") + return True + else: + print(f"第 {attempt + 1} 次尝试失败") + if attempt < max_retries - 1: + print("2秒后重试...") + time.sleep(2) + # 重新初始化驱动 + try: + self.driver.quit() + except: + pass + self._setup_driver() + + print("所有尝试均失败") + return False + + +# 使用示例 +if __name__ == "__main__": + bot = DamaiBot() + bot.run_with_retry(max_retries=3)