2026年07月04日 自动化任务与生活洞察报告
📜 日记摘要与心理洞察
日期: 2026年07月04日 星期六
心情: 🙂 (愉悦)
天气: 河南正阳县，阴，27℃，南风4级，湿度84% (报告时间: 09:30:26)
今日活动:
- 09:58 - 优化 抖音商城抽奖
- 18:52 - 优化淘粉吧提现
- 21:52 - 书旗和茄子任务 49
心理洞察：
日记记录了一天中多项优化工作和任务完成，尽管工作内容偏向技术与重复性操作，但心情仍为“愉悦”。这可能表明对工作成果的满意、解决问题的成就感，或者对任务掌控感的积极体现。在忙碌中仍能保持正面情绪，这是一种宝贵的能力。
建议： 保持这份对优化和解决问题乐在其中的心态，这将是持续进步的动力。同时，别忘了在任务间隙给自己短暂的放松，以维持这种积极的能量。
💡 技术创新与优化建议
# coding=utf-8
import datetime
import os
import platform
import random
import time
import cv2 as cv
import numpy as np
import uiautomator2 as u2
# power_switch
from send_emaill import qq_email
import hmac
import hashlib
import base64
import urllib.parse
import requests
import json
import logging
import allure
import re
from lxml import etree
import subprocess
import pytest
import ast
import threading
import redis
import pickle
import sys
from utils.mysql_util import MySQLManager
from utils.redis_util import RedisManager
from utils.common import get_adb_path
from utils.yaml_util import read_yaml_cached
from utils.sw_control_util import switch_2400, set_fan_speed, new_switch_2400_optimized
from utils.xpath_utils import XPathElement
from utils.ocr_util import pytesseract_pic_to_text
import atexit
import socket
# env()
# 解锁密码
separation_phone1 = "1234"
separation_phone2 = "5678"
class UiAutomatr():
def __init__(self):
self.ip = ""
self.canzhan = ''
self.port = ":5555"
self.redis_expire = 60 * 60 # 缓存过期时间 (1 小时，根据需要调整)
self.system = platform.platform()
self.redis_manager = RedisManager() # 创建 RedisManager 实例
# === 新增：暂停检查的冷却控制 ===
self.last_pause_check_time = 0
self.pause_check_interval = 5 # 设置每 5 秒才真正查一次 Redis
self._DIR_INDEX = {} # 文件夹索引缓存
self._REGEX_CACHE = {} # 正则表达式缓存
self.local_imobj_cache = {} # 图片矩阵缓存
self.read_yaml = read_yaml_cached()
# mi_app_list
try:
self.mi_app_list = self.read_yaml['mi_app_list']
except Exception as e:
print(e)
self.mi_app_list = []
try:
self.force_uninstall_app_list = self.read_yaml['force_uninstall_app_list']
except Exception as e:
print(e)
self.force_uninstall_app_list = []
value = self.read_yaml['mobile_ip']
dic_final = {}
for dic in value:
dic_final.update(dic)
print(dic_final)
self.note9_5g_ip = dic_final.get('note9_5g_ip')
self.mobile_dic = self.read_yaml['mobile_model']
self.main_mobile_ip = self.read_yaml['main_mobile_ip']
adb_path, cur_path = get_adb_path(self.read_yaml)
self.adb_path = adb_path
self.screen_path = os.path.join(cur_path, 'screen_picture/')
self.find_pic_path = os.path.join(cur_path, 'find_picture/xiaoyao/')
self.z = 0
self.lis = []
self.install = []
self.china_name = ""
self.app_name = ''
self.activity = ''
self.project = ''
self.main_start = time.time()
self.waiter_times = None
self.text = ""
self.zl_flag = ''
self.width = '' # 手机宽
self.height = '' # 手机高
self.d = None
self.model = None
self.xpath_signal = ""
self.info = ''
self.separation = "" # 解锁密码
self.threshold = 0.9 # 图片相似度
self.tao_user = '' # 用户
self.user_name = '' # 也是用户
# 偏移坐标
self.width_offset = 1
self.height_offset = 1
self.MySQLManager = MySQLManager()
self.XPathElement = XPathElement
# self.redis_client = self.MySQLManager.redis_client
atexit.register(self.cleanup)
def cleanup(self):
try:
self.del_end()
except Exception:
pass
# print("Cleanup failed:", e)
def __del__(self):
pass
def _get_cache_key(self, *args):
"""生成缓存键"""
return args[0] + hashlib.md5(json.dumps(args).encode('utf-8')).hexdigest()
def del_end(self):
end = time.time()
time.sleep(0.4)
try:
power = ''
try:
output = self.console_input('shell dumpsys battery', ret=1)
match = re.search(r'level: (\d+)', output)
power = match.group(1)
int_power = int(power)
if int_power <= 20 and self.ip != self.main_mobile_ip:
if self.d:
self.d.screen_off() # 关闭屏幕
self.create_cron_switch(40, mip=self.ip, this_power=str(int_power))
print("%s电量过低,等待40分钟充电" % self.ip)
# time.sleep(1800)
if self.model == '【MI 8 SE】':
time.sleep(600)
print("mi8 se 休息10分钟")
else:
if self.ip == self.main_mobile_ip:
if int_power <= 20:
print("main_mobile_ip电量过低,充电" )
self.create_cron_switch(40, mip=self.ip, this_power=str(int_power))
set_fan_speed(255)
elif int_power <= 30 and int_power > 20:
set_fan_speed(40)
elif int_power <= 40 and int_power > 30:
set_fan_speed(30)
elif int_power <= 60 and int_power > 40:
set_fan_speed(20)
elif int_power <= 80 and int_power > 60:
set_fan_speed(15)
else:
set_fan_speed(10)
except Exception as e:
print(e)
now = datetime.datetime.now()
now_time = now.strftime("%Y-%m-%d %H:%M:%S")
try:
output = self.console_input('shell dumpsys battery', ret=1)
temperature = re.search(r'temperature: (\d+)', output).group(1)
temperature = int(temperature) / 10
except Exception as e:
print(e)
temperature = ''
# 计算时间差,最小作为应用时间
start_end_diff = end - self.main_start
mobile_ip = self.model + self.ip
ret = self.MySQLManager.sql_search('content', "%s','手机电量'" % mobile_ip)
if ret:
last_sql = ret[-1][0]
# print("last_sql:", last_sql)
r_list = re.compile(r"时间(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", re.S).findall(last_sql)
if r_list:
# 取出最后一条记录的时间
# print("r_list:", r_list)
# 取出最后一条记录的时间
last_sql_time = datetime.datetime.strptime(r_list[0], "%Y-%m-%d %H:%M:%S")
now = datetime.datetime.now()
# 计算时间差
time_diff = (now - last_sql_time).seconds
# print("time_diff:", time_diff)
start_end_diff = min(time_diff, start_end_diff)
print(time_diff, start_end_diff)
print_text = "['%s','手机电量','电量%s|温度=%s|self.z=%s|用时%.2f秒|时间%s|活动%s']" % \
(self.model + self.ip, power, temperature, self.z, start_end_diff, now_time, self.project)
print(print_text)
if power:
# ['%s','趣头条金币','%s']
self.MySQLManager.write_file(print_text)
# 卸载com.baidu.searchbox(百度)
self.uninstall_app("com.baidu.searchbox")
# 重新赋值
# self.main_start = end
# 添加cpu，内存图片
try:
if "Windows" in self.system:
project_path = self.read_yaml['win_computer']['project_path']
report_path = os.path.join(project_path, "report")
else:
share_path = self.read_yaml['main_ubantu_computer']['share_path']
report_path = os.path.join(share_path, "report")
save_path = os.path.join(report_path, "cpu内存剩余率图.png")
allure.attach.file(save_path, name='cpu内存剩余率图', attachment_type=allure.attachment_type.PNG)
except Exception as e:
print(e)
time.sleep(2)
if self.d:
# pass
try:
print("关闭后台大于100MB的app")
self.kill_heavy_memory_apps(threshold_mb=100)
except Exception as e:
print("关闭后台大于100MB的app出错", e)
# 出现无响应
self.multiple_find_picture([], 1, "点击", "不返回", delem=['d(resourceId="android:id/aerr_close")'])
get_appname = self.get_app_name()
app_list = ['com.miui.home', 'com.xiaomi.market', "com.kindle.zhuayumao", "com.tencent.mm", 'com.example.myaccessibility']
if get_appname not in app_list:
self.d.app_stop(get_appname)
# ipv6写入
self.process_ipv6()
have_tj = self.MySQLManager.sql_search("content", "%s已统计手机容量" % (self.model + self.ip))
if not have_tj:
ret = self.console_input('shell df -h /data', ret=1)
if ret:
text = str(ret)
print(text)
r_list = re.compile('(\\d+G).*?(\\d+G).*?(\\d+G).*?(\\d+%)', re.S).findall(text)
if not r_list:
r_list = re.compile('(\\d+G).*?(\\d+G).*?(\\d+M).*?(\\d+%)', re.S).findall(text)
if r_list:
ex_text = ''
try:
rate = r_list[0][3].replace('%', '')
print(rate)
if int(rate) >= 95:
ex_text = ("(容量过小)")
self.clear_app(app_name="com.jifen.qukan")
ip_post = self.model + self.ip
search_text = ip_post + "','应用大小排行"
text = ''
r_ = MySQLManager().sql_search("content", search_text)
if r_:
text = eval(r_[-1][0])[-1]
print(text)
print("容量过小", text)
qq_email('', '%s 手机容量过小' % self.ip, str(rate) + "
" + text)
except Exception as e:
print(e)
text = "使用%s,可用%s,使用率%s(总共%s)" % (r_list[0][1], r_list[0][2], r_list[0][3] + ex_text, r_list[0][0])
print(text)
self.MySQLManager.write_file("['%s', '手机容量情况', '%s']" % ((self.model + self.ip), text))
time.sleep(1)
# ipv6
try:
version = self.set_voice_to_0()
if version:
print("['%s', '安卓版本', '%s']" % ((self.model + self.ip), version))
self.MySQLManager.write_file("['%s', '安卓版本', '%s']" % ((self.model + self.ip), version))
except Exception as e:
logging.error(e)
try:
version = self.console_input('shell getprop ro.build.version.incremental', ret=1)
if version:
version = version.strip()
print("['%s', 'miui版本', '%s']" % ((self.model + self.ip), version))
self.MySQLManager.write_file("['%s', 'miui版本', '%s']" % ((self.model + self.ip), version))
except Exception as e:
logging.error(e)
self.clear_mobile()
self.d.screen_off() # 关闭屏幕
except Exception as e:
print(e)
self.re_install_uiautomator(e)
def is_ip_reachable(self, ip, timeout=1):
system = platform.system().lower()
if system == "windows":
command = ["ping", "-n", "1", "-w", str(timeout * 1000), ip]
try:
result = subprocess.run(command, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
return result.returncode == 0
except Exception as e:
print(f"执行 Ping 出错: {e}")
return False
else:
# 在 Linux 上，优先尝试 socket 方法
try:
sock = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
sock.settimeout(timeout)
# 尝试连接一个常用端口，如 80 或 443
print('sock.connect_ex((ip, 80))=', sock.connect_ex((ip, 80)))
if sock.connect_ex((ip, 80)) == 0 or sock.connect_ex((ip, 80)) == 111: # 0 表示连接成功，111 表示连接被拒绝（端口关闭）
sock.close()
return True
else:
# 如果 80 不通，可以再尝试一个，例如 443
sock.settimeout(timeout)
result = sock.connect_ex((ip, 443)) == 0
sock.close()
return result
except Exception as e:
print(f"Socket 连接出错: {e}")
return False
def process_ipv6(self):
"""获取并处理 IPv6 地址，限制写入 MySQL 的频率"""
redis_key_last_write = f"{self.model}:{self.ip}:ipv6:last_write"
redis_key_ipv6 = f"{self.model}:{self.ip}:ipv6"
cache_expire_time = 10000
try:
# 检查上次写入 MySQL 的时间
last_write_time_bytes = self.redis_manager.redis_search(redis_key_last_write)
if last_write_time_bytes:
last_write_time = int(last_write_time_bytes.decode('utf-8'))
if time.time() - last_write_time < cache_expire_time:
logging.info(f"Too soon to write to MySQL for {self.model} ({self.ip})")
return
# 获取 IPv6 地址
current_ipv6 = self.get_mobile_ipv6()
if not current_ipv6:
logging.info(f"No IPv6 address found for {self.model} ({self.ip})")
return
# 检查 IPv6 是否有变化
cached_ipv6_bytes = self.redis_manager.redis_search(redis_key_ipv6)
if cached_ipv6_bytes:
cached_ipv6 = cached_ipv6_bytes.decode('utf-8')
if current_ipv6 == cached_ipv6:
logging.info(f"IPv6 address for {self.model} ({self.ip}) has not changed")
# 更新上次写入时间
self.redis_manager.redis_set(redis_key_last_write, times=cache_expire_time, value=str(int(time.time())))
return
# 写入 MySQL
log_message = f"['{self.model}{self.ip}', 'ipv6', '{current_ipv6}']"
print(log_message) # 打印到控制台
self.MySQLManager.write_file(log_message)
# 更新 Redis 中的 IPv6 地址和上次写入时间
self.redis_manager.redis_set(redis_key_ipv6, times=cache_expire_time, value=current_ipv6.encode('utf-8'))
self.redis_manager.redis_set(redis_key_last_write, times=cache_expire_time, value=str(int(time.time())))
except redis.RedisError as e:
logging.error(f"Redis error: {e}")
except Exception as e:
logging.exception(f"An unexpected error occurred: {e}")
def create_cron_switch(self, minutes=180, mip='', force = '', this_power = ''):
# ESP_AB3FFD
print("插座关闭时间为%s分钟" % minutes)
import datetime
# 获取当前时间
current_time = datetime.datetime.now()
# 假设要在当前时间上加上30分钟
minutes_to_add = minutes
new_time = (current_time + datetime.timedelta(minutes=minutes_to_add)).strftime("%H:%M:%S")
# 输出结果
print("现在时间", current_time.strftime("%H:%M:%S"))
print("停止充电时间:", new_time)
if mip:
try:
# 定义该设备充电记录的 Redis 缓存 key
redis_charge_key = f"charging_log_cooldown:{self.model}:{self.ip}"
# 查询 Redis，如果不存在，说明 5 分钟内没有记录过
if not self.redis_manager.redis_search(redis_charge_key):
# 1. 写入 MySQL
m = '充电:%s分钟 %s' % (minutes, ",电量:%s" % this_power if this_power else '')
text = '[%s, "手机充电", "%s"]' % ((self.model + self.ip + "_" + str(current_time.strftime("%H:%M"))), minutes)
self.MySQLManager.write_file("['%s', '手机充电', '%s']" % ((self.model + self.ip + "_" + str(current_time.strftime("%H:%M"))), minutes))
# 2. 写入 Redis 并设置 5 分钟 (300秒) 的过期时间（实现 5 分钟内不重复记录）
self.redis_manager.redis_set(redis_charge_key, times=300, value="1")
print(f"[{self.ip}] 已记录充电状态，进入 5 分钟冷却期。")
else:
print(f"[{self.ip}] 5 分钟内已记录过充电，本次跳过写入 MySQL。")
except Exception as e:
print("充电记录写入失败:", e)
# ESP-2222
relay_index = [1, 2]
chage_name = []
if mip :
try:
win_mobile_ips = self.read_yaml['win_mobile_ip']
for dic in win_mobile_ips:
key = list(dic.keys())[0]
# print("key",key)
value = list(dic.values())
if key in ['Redmi K20', 'note11_5g_ip', 'note11_f_5g_ip', 'note9_5g_ip', 'mix3_ip', 'MI_CC9_Pro' ] and mip in value:
chage_name = [relay_index[0]]
break
elif key in ['redmi9_ip', 'note10pro_ip', 'mi8_lite_ip', 'mix2s_ip', 'mi9', 'MI 8 SE', ] and mip in value:
chage_name = [relay_index[1]]
except Exception as e:
print("没有找到要充电的设备", e)
chage_name = [relay_index]
print("要充电的ip为",mip, "打开的设备为", chage_name)
if not chage_name:
chage_name = relay_index
if mip == self.main_mobile_ip:
# 主手机充电
hour = datetime.datetime.now().hour
if 0 <= hour <= 8 or force == 'force':
print("main_mobile_ip打开充电")
switch_2400(ip="192.168.31.162", mode='开', off_list = [str(new_time)], on_list = [])
else:
print("main_mobile_ip白天不充电")
return
for relay in chage_name:
try:
print("2.要充电的ip为",mip, "打开的设备为", relay)
new_switch_2400_optimized(ip=self.read_yaml['switch_ip']["ESP-EBFB77"], relay=relay, mode='开', off_list=[str(new_time)],on_list = [] )
# switch_2400(ip=self.read_yaml['switch_ip'][switch_ip], mode='开', off_list = [str(new_time)], on_list = [])
except Exception as e:
print("没有找到要充电的设备", e)
def set_past_text(self, text):
# 设置剪贴板
try:
# 空格要转义(需要升级)
app_name = self.get_app_name()
text = text.replace(" ", '\\ ')
self.d.app_start("com.github.uiautomator")
for i in range(5):
self.d.set_clipboard(text, 'label')
if self.get_past_text():
break
elem = self.d(text="允许")
if elem:
elem.click()
elem = self.d(text="始终允许")
if elem:
elem.click()
self.d.app_start(app_name)
except Exception as e:
print("手机没安装此应用", e)
def get_past_text(self):
# 读取剪贴板
text = None
try:
app_name = self.get_app_name()
self.d.app_start("com.github.uiautomator")
for i in range(30):
text = self.d.clipboard
if text:
break
else:
time.sleep(0.1)
elem = self.d(text="允许")
if elem:
elem.click()
elem = self.d(text="始终允许")
if elem:
elem.click()
self.d.app_start(app_name)
except Exception as e:
print("手机没安装此应用", e)
return text
def select_mi_brown(self):
[x, y], k = self.multiple_find_picture([], 10, "点击", "不返回", delem=['d(text="浏览器")', 'd(text="小米浏览器")'])
if x > 0 and y > 0:
self.multiple_find_picture([], 10, "点击", "不返回", delem=['d(text="始终")', 'd(text="总是")'])
def open_url_with_browser(self, url):
"""
使用手机默认浏览器打开指定URL
已修复长链接包含 '&' 导致截断的问题
"""
if not url:
print("URL为空，无法打开")
return
try:
self.open_phone()
print(f"正在使用默认浏览器打开: {url}")
# 【核心修复】：将 URL 中的 & 替换为 \&
# 这样 Android Shell 才能正确识别它是 URL 的一部分，而不是命令结束符
safe_url = url.replace("&", r"\&")
# 构造命令
cmd = 'shell am start -a android.intent.action.VIEW -d "%s"' % safe_url
# 执行命令
self.console_input(cmd)
time.sleep(3)
except Exception as e:
print(f"打开浏览器失败: {e}")
@staticmethod
def get_time():
now = datetime.datetime.now()
not_time = "%02d:%02d:%02d" % (now.hour, now.minute, now.second)
return " " + str(not_time) + ""
def zhifubao_good(self):
# 支付宝点击好的，不再询问
elem = self.d(text="总是保持以上选择，不再询问")
if elem:
elem.sibling(index=5)
time.sleep(0.5)
elem2 = elem.sibling(text="好的")
if elem2:
elem2.click()
def menu_start_app(self, app_name):
self.d.press("menu")
time.sleep(0.5)
[x1, y1], k = self.multiple_find_picture([], 20, "点击", "不返回",
delem=['d(resourceId="com.miui.home:id/title", text=%s)' % app_name,
'd(text="%s")' % app_name])
if x1 > 0 and y1 > 0:
return True
else:
return False
def sql_zl(self, zl_flag=''):
if zl_flag:
self.zl_flag = zl_flag
print(self.zl_flag)
for day in range(10):
get_zl = self.MySQLManager.sql_search("content", self.zl_flag, witch_day=-day)
if get_zl:
str_text = get_zl[0][0]
try:
print(str_text)
except Exception as e:
print(e)
try:
zhu_dic = eval(str_text.replace(self.zl_flag, '').replace('\n', ''))
except Exception as e:
print("|%s|字典转换发生错误|需要检查字典是否特殊字符串%s" % (self.project, e))
self.MySQLManager.write_file("|%s|字典转换发生错误|需要检查字典是否特殊字符串%s" % (self.project, e))
zhu_dic = {}
try:
print(zhu_dic)
except Exception as e:
print(e)
if day >= 1:
print("复制上%s天数据" % day)
self.MySQLManager.write_file(str_text)
return zhu_dic
else:
print("不存在助力码")
if self.zl_flag:
self.MySQLManager.write_file(self.zl_flag + "{}")
return {}
def update_zl(self, zl_text, dic='', user='', zl_flag = ''):
# 更新表里
if zl_flag:
self.zl_flag = zl_flag
if zl_text and zl_text != "undefined":
zl_text = zl_text.replace('"', '“')
get_dic = self.sql_zl()
if not dic:
if user:
dic = user
else:
# 如果dic位'字典为用户名
dic = self.tao_user
try:
get_dic[dic] = zl_text
except Exception as e:
print(e)
get_dic = {}
self.MySQLManager.sql_update(self.zl_flag, self.zl_flag + str(get_dic))
print("更新后")
self.sql_zl()
else:
print("剪贴板不存在文字")
def delete_zl(self, zl_flag='', user=''):
# 更新表里
try:
self.zl_flag = zl_flag
get_dic = self.sql_zl()
print(get_dic)
get_dic.pop(user)
self.MySQLManager.sql_update(zl_flag, zl_flag + str(get_dic))
print("删除%s更新后" % user)
self.sql_zl()
except KeyError as e:
print("【%s】用户不存在，无法删除%s" % (user, e))
@staticmethod
def random_dic(dicts):
# 字典随机排序
dict_key_ls = list(dicts.keys())
random.shuffle(dict_key_ls)
new_dic = {}
for key in dict_key_ls:
new_dic[key] = dicts.get(key)
return new_dic
def get_phone_width_height(self):
if not self.width or not self.height:
try:
self.width = self.d.device_info['display']['width']
self.height = self.d.device_info['display']['height']
except:
self.width = self.d.info['displayWidth']
self.height = self.d.info['displayHeight']
return self.width, self.height
def human_swipe(self, d, start_x, start_y, end_x, end_y):
# 滑动的分段数
steps = 15
# 计算每一步的移动距离
dx = (end_x - start_x) / steps
dy = (end_y - start_y) / steps
# 按下起始点
d.touch.down(start_x, start_y)
time.sleep(0.05) # 短暂延迟，模拟人手
# 执行分段滑动
for i in range(steps):
# 计算当前步的位置，并加入随机偏移，模拟人手的抖动
move_x = start_x + dx * i + random.uniform(-1, 1)
move_y = start_y + dy * i + random.uniform(-1, 1)
# 移动到新位置
d.touch.move(move_x, move_y)
time.sleep(random.uniform(0, 1) / 10 ) # 等待，控制滑动的总时间
# 释放手指
d.touch.up(end_x, end_y)
def tao_hua_yan(self):
for i in range(20):
[x, y], k = self.multiple_find_picture(['淘宝滑块按钮'], 3, "不点击", "不返回", xpath=['//*[@text="向右滑动验证"]/preceding-sibling::android.widget.FrameLayout'], delem=['d(resourceId="nc_1_n1t")'])
if x > 0 and y > 0:
width, height = self.get_phone_width_height()
swipe_distance = int((0.7 + random.uniform(0, 0.1)) * width)
print('滑块验证,手机宽度%s, 滑动距离%s' % (width, swipe_distance))
# self.d.swipe_points([[x, y], [x + swipe_distance, y]], random.uniform(0.1, 0.5))
# self.d.swipe(x, y, x + swipe_distance, y, duration=0.5)
self.human_swipe(self.d, x + random.uniform(-0.05, 0.05), y + random.uniform(-0.01, -0.01), x + swipe_distance, y)
# 淘宝验证失败0
time.sleep(0.5)
[x, y], k = self.multiple_find_picture(['淘宝验证失败'], 5, "点击", "不返回")
if x > 0 and y > 0:
print("淘宝验证失败")
time.sleep(1)
self.back()
else:
break
def friends_help(self, active_dic, help_name='', times=3, return_dic=False):
# 助力
try:
active_dic = self.random_dic(active_dic)
except Exception as e:
print("错误%s，原因%s" % (active_dic, e))
return
# print("active_dic=", active_dic)
help_name = help_name.replace("【", '').replace("】", '')
self.d.app_stop(self.app_name)
dic = {}
for user in active_dic:
have_zl = self.MySQLManager.sql_search("content", "%s单项助力一次%s" % ((
self.tao_user.replace("*", "\\\\*"),
user.replace("*", "\\\\*"))))
# print(have_zl)
all_zl = self.MySQLManager.sql_search("content", "%s被助力了一次%s" % (user.replace("*", "\\\\*"), help_name))
if all_zl:
len_all_zl = len(all_zl)
else:
len_all_zl = 0
# 1.不能助力自己, 2.被助力的人不能助力3(times)次, 3.一个人只能助力一个人
# print("助力人=%s, 被助力人=%s,被助力%s次,助力了吗=%s" % (self.tao_user, user, len_all_zl, have_zl))
if self.tao_user != user and len_all_zl <= times and not have_zl:
if type(active_dic) == list:
print(active_dic, ":列表，继续循环")
zl_code = active_dic.get(user)
if zl_code == 'undefined':
continue
# 助力码写入剪贴板
self.set_past_text(zl_code)
# print(self.get_past_text())
print("助力人=%s, 被助力人=%s,被助力%s次,助力码=%s" % (self.tao_user, user, len_all_zl, zl_code))
help_others_text = "%s单项助力一次%s" % (self.tao_user, help_name)
self.MySQLManager.write_file(help_others_text)
self.MySQLManager.write_file("%s被助力了一次%s" % (user, help_name))
if return_dic:
dic['user'] = user
dic['help_others_text'] = help_others_text
return dic
else:
return help_others_text
else:
print("%s没有需要助力的人了" % self.tao_user)
return False
def set_voice_to_0(self):
# 设置音量为0
version = self.console_input('shell getprop ro.build.version.release', ret=1).strip()
print("安卓" + version)
if version == '11':
# adb shell service call audio 10 i32 3 i32 0 i32 1
self.console_input('shell service call audio 10 i32 3 i32 0 i32 1')
else:
self.console_input('shell media volume --set 0')
# adb -s 192.168.31.182 shell cmd media_session volume --stream 3 --set 0
try:
# 大于安卓11适合
self.console_input('shell cmd media_session volume --stream 3 --set 0')
except Exception as e:
print(e)
return version
def connect_adb(self):
for t in range(1):
adb_return = os.popen(self.adb_path + ' devices').read()
adb_return = adb_return.replace(" ", "").replace("\t", "")
print('adb_return=', adb_return)
# 192.168.31.109:5555 offline
if self.ip + self.port + 'device' in adb_return:
print("t=%s,%sadb连接成功" % (t, self.ip + self.port))
return True
elif self.ip + self.port in adb_return:
print("t=%s,%s连接了adb状态不是devices,需要disconnect" % (t, self.ip + self.port))
cmd1 = '%s disconnect %s%s ' % (self.adb_path, self.ip, self.port)
os.system(cmd1)
else:
# 'echo %s | sudo -S %s' % ('123456', 'poweroff')
print("t=%s,%s没连接adb需要连接adb" % (t, self.ip + self.port))
cmd1 = '%s connect %s%s ' % (self.adb_path, self.ip, self.port)
# os.system(cmd1)
try:
adb_return = os.popen(cmd1).read()
print(adb_return)
except Exception as e:
print(e, cmd1)
time.sleep(1)
# 导出app
def export_app(self, app_cn_name='myaccessibility', app_name='com.example.myaccessibility', ip='', port='5555'):
# PS D:\mobile> adb -s 192.168.31.182 shell pm path com.example.myaccessibility
#package:/data/app/~~2-Bi6oaE7mjf490AKiXzDw==/com.example.myaccessibility-Vl8LBabAMSjyAN3dWJEa3A==/base.apk
#PS D:\mobile> adb -s 192.168.31.182 pull /data/app/~~2-Bi6oaE7mjf490AKiXzDw==/com.example.myaccessibility-Vl8LBabAMSjyAN3dWJEa3A==/base.apk D:\android_apps\myaccessibility.apk
if ip:
self.ip = ip
if port:
self.port = ':' + port
self.read()
# if not self.ip:
# print("不存在ip%s" % self.ip)
# return
cmd1 = '%s -s %s%s shell pm path %s' % (self.adb_path, self.ip, self.port, app_name)
print(cmd1)
output = os.popen(cmd1).read()
print(output)
if "Windows" in self.system:
main_dir = 'D:\\'
elif "macOS" in self.system:
print("mac系统")
main_dir = self.read_yaml['mac_computer']['project_path']
else:
main_dir = self.read_yaml['main_ubantu_computer']['share_path']
if not app_cn_name:
app_cn_name = os.path.join(main_dir, 'android_apps', app_name + '.apk')
else:
app_cn_name = os.path.join(main_dir, 'android_apps', app_cn_name + '.apk')
if 'package' in output:
# package:/data/app/~~2-Bi6oaE7mjf490AKiXzDw==/com.example.myaccessibility-Vl8LBabAMSjyAN3dWJEa3A==/base.apk
path = output.split(':')[1].strip()
cmd2 = '%s -s %s%s pull %s %s' % (self.adb_path, self.ip, self.port, path, app_cn_name)
print(cmd2)
os.popen(cmd2)
# adb -s 192.168.31.176 install -t /mnt/nas/share/android_apps/myaccessibility.apk # -t 强制安装
def read(self):
def delete_text():
self.redis_manager.redis_delete('手机离线%s' % self.ip)
# l1.2删除公网ip的手机离线
# host = self.read_yaml['server']['domain']
# port = self.read_yaml['nas_computer']['redis_remote_port']
# manage = RedisManager(host=host, port=port)
# manage.redis_delete('手机离线%s' % self.ip)
self.xpath_signal = 'text'
self.info = 'text'
t1 = time.time()
if not self.ip:
print("不存在ip%s" % self.ip)
return
if not self.port:
self.port = '5555'
if ':' not in self.port:
self.port = ':' + self.port
adb_connect = False
# self.MySQLManager.write_file("%s开了%s" % (self.ip, self.project))
for i in range(4):
print("i=%s， ip=%s, self.port = %s" % (i, self.ip, self.port))
try:
# 若报错可能没连接adb,背其他设置占用,因为多个模拟器使用adb会导致手机没连接,重新wifi连接adb
if i > 0 and not adb_connect:
raise Exception(" %sadb离线。无需u2.connect" % (self.ip + self.port))
self.d = u2.connect(self.ip + self.port)
self.open_phone()
self.model = self.d.device_info.get("model")
self.model = self.mobile_dic.get(self.model)
print("%s的uiautomator2连接成功" % (self.ip + self.port))
print("手机型号=%s" % self.model)
# if self.ip == self.note9_5g_ip:
# power_switch(1)
if not self.model:
print("%s不存在字典中" % self.d.device_info.get("model"))
try:
threading.Thread(target=delete_text).start()
# host
except:
pass
return True
except Exception as e:
print(e)
self.re_install_uiautomator(e)
time.sleep(1)
adb_connect = self.connect_adb()
if not adb_connect:
print("adb连接失败，使用nodejs与java通信进行处理adb事件")
if "Windows" in self.system:
project_path = self.read_yaml['win_computer']['project_path']
else:
project_path = self.read_yaml['main_ubantu_computer']['project_path']
client_file = os.path.join(project_path, 'node', 'client.js')
print(client_file)
# os.system('node %s %s' % (client_file, self.ip))
try:
print("注意：30版本或者以下adb没有adb pair功能")
subprocess.run(["node", client_file, self.ip], timeout=50) # 50秒超时
time.sleep(10)
except subprocess.TimeoutExpired:
print("命令执行超时")
print("nodejs与java通信处理完成")
else:
sent_text = "%s已经发过邮件" % self.ip
have_sent = self.redis_manager.redis_search(sent_text)
print("have_sent=", have_sent, 'sent_text=', sent_text)
# print(not have_sent and self.ip)
if not have_sent and self.ip and self.ip.strip() != ":5555" and self.ip != '192.168.31.42':
t2 = time.time()
# try:
# path = self.cv_record_video(5)
# print(path)
# except Exception as e:
# print(e)
# path = None
print(t2 - t1)
if (t2 - t1) > 80:
# 小于100秒不属于离线
# l2手机离线，方便显示2.8寸屏幕
self.redis_manager.redis_set('手机离线%s' % self.ip)
host = self.read_yaml['server']['domain']
port = self.read_yaml['nas_computer']['redis_remote_port']
try:
manage = RedisManager(host=host, port=port)
manage.redis_set('手机离线%s' % self.ip)
except:
pass
# self.MySQLManager.write_file('手机离线%s' % self.ip)
# 有效期1小时
self.redis_manager.redis_set(sent_text, times=3600, value='发过')
is_reachable = self.is_ip_reachable(self.ip)
ex_text = ''
print('is_reachable=', is_reachable)
if is_reachable:
print("能ping通， 可能授权失败unauthorized")
ex_text = '能ping通， 可能授权失败unauthorized,' + str(is_reachable)
else:
print("手机离线，ping不通,可能关机或卡死")
ex_text = "手机离线，ping【不】通,可能手机关机或卡死，可能root权限丢了没有(概率很小),is_ip_reachable=" + str(is_reachable)
try:
times = datetime.datetime.now().strftime("%H:%M:%S")
# qq_email("534640040@qq.com", '%s adb连接失败' % self.project,
# "%s多次connect失败,用时%.2f秒， %s" % (self.ip, (t2 - t1), len(self.ip.strip())) + str(self.system) + ex_text)
catch_key_camera = "%sip已经触发录像" % self.ip
have_ = self.redis_manager.redis_search(catch_key_camera, fuzzy_key=True)
url = ''
if not have_:
def clean_filename(filename):
# 替换 Windows/Linux 不允许的字符：/ \ : * ? " < > |
invalid_chars = r'[\\/:*?"<>|]'
# 把非法字符替换成空（删掉）
return re.sub(invalid_chars, '', filename)
try:
hour = datetime.datetime.now().hour
date = datetime.datetime.now().strftime("%Y%m%d")
ip_part = self.ip.split('.')[-1] # 获取IP地址的最后一部分
raw_filename = '%s%s_%s_adb连接异常%s' % (date, hour, ip_part, ex_text)
# 清理后再使用
filename = clean_filename(raw_filename)[:30] # 关键在这里
camera_server_url = self.read_yaml['hnnas_computer']['camera_server_url']
response = requests.post(
f"{camera_server_url}/record/start",
params={
"duration": 10,
"filename": filename + ".mp4" # 加不加 .mp4 后缀都可以，服务端会自动处理
}
)
# http://192.168.31.166:8765/v/2026-05-15_13-38-10
url = f"[回放视频]({camera_server_url}/v/{filename})"
print(url, response.text)
self.redis_manager.redis_set(catch_key_camera, times='today')
except Exception as e:
print("录屏报错", e)
text = "%s %s adb连接失败， %s多次connect失败,用时%.2f秒， %s" % (times, self.project , self.ip, (t2 - t1), len(self.ip.strip())) + str(self.system) + ex_text + url
self.MySQLManager.write_file(text )
except Exception as e:
print(e)
return False
def re_install_uiautomator(self, e):
# , 'gateway error
if 'Invalid version' in str(e) or str(e).strip() == 'closed' or 'gateway error' in str(e):
# shell input keyevent KEYCODE_HOME
self.console_input('shell input keyevent KEYCODE_HOME')
print("卸载com.github.uiautomator")
ret = self.console_input('shell dumpsys battery | findstr "temperature"', ret=1)
print(ret)
self.console_input('uninstall com.github.uiautomator')
# if self.d:
# self.d.app_start(self.app_name)
# pylint: disable=no-member
def screen(self, pic_name, region=(), elem=None, offset=[0, 0, 0, 0]):
path = os.path.join(self.screen_path, pic_name)
# print('path=', path)
directory_path = os.path.dirname(path)
# 如果目录不存在，则创建多层目录
if not os.path.exists(directory_path):
os.makedirs(directory_path)
if region:
replace_file = path.replace('\\', '/')
split_file = replace_file.split('/')
template_name = os.path.join('/'.join(split_file[:-1]), self.ip + 'template.png')
self.d.screenshot(template_name)
# pylint: disable=no-member)
image = cv.imdecode(np.fromfile(template_name, dtype=np.uint8), 1)
size = image.shape
pic_x, pic_y = size[1], size[0]
print('图片x=%s， y=%s' % (pic_x, pic_y))
# （y1（上）:y2(下), x1（左）:x2（右））
# 0.11:0.16, 0.12:0.34(282:395, 134:377])
if 0 < region[0] < 1:
crop_img = image[int(region[0] * pic_y):
int(region[1] * pic_y), int(region[2] * pic_x): int(region[3] * pic_x)] # 获取感兴趣区域
else:
crop_img = image[int(region[0]): int(region[1]), int(region[2]): int(region[3])] # 获取感兴趣区域
# print(path)
# 删除原来图片
os.remove(template_name)
# cv.imwrite(path, crop_img) # 保存到指定目录
cv.imencode('.png', crop_img)[1].tofile(path)
elif elem:
print("元素截图")
replace_file = path.replace('\\', '/')
split_file = replace_file.split('/')
template_name = os.path.join('/'.join(split_file[:-1]), self.ip + 'template.png')
self.d.screenshot(template_name)
# pylint: disable=no-member)
image = cv.imdecode(np.fromfile(template_name, dtype=np.uint8), 1)
size = image.shape
pic_x, pic_y = size[1], size[0]
print('图片x=%s， y=%s' % (pic_x, pic_y))
# （y1（上）:y2(下), x1（左）:x2（右））
# 0.11:0.16, 0.12:0.34(282:395, 134:377])
bounds = elem.info['bounds']
region = [bounds['top'], bounds['bottom'], bounds['left'], bounds['right']]
# 处理偏移
if offset != [0, 0, 0, 0]:
print(region)
# 判断是否为小数比例（0-1之间）
if all(0 < abs(o) < 1 for o in offset if o != 0):
# 使用手机尺寸比例
region[0] += int(offset[0] * pic_y) # top
region[1] += int(offset[1] * pic_y) # bottom
region[2] += int(offset[2] * pic_x) # left
region[3] += int(offset[3] * pic_x) # right
else:
# 直接像素偏移
region[0] += int(offset[0]) # top
region[1] += int(offset[1]) # bottom
region[2] += int(offset[2]) # left
region[3] += int(offset[3]) # right
print(region)
crop_img = image[int(region[0]): int(region[1]), int(region[2]): int(region[3])] # 获取感兴趣区域
# print(path)
# 删除原来图片
path = os.path.abspath(path)
os.remove(template_name)
#cv.imwrite(path, crop_img) # 保存到指定目录
cv.imencode('.png', crop_img)[1].tofile(path)
else:
for i in range(3):
try:
self.d.screenshot(path)
break
except Exception as e:
print(e)
# print("截图第三种", path)
try:
if self.model == '【Redmi 8A】':
self.threshold = 0.75
set_width = 1080
set_height = 2400
this_width, this_height = self.get_phone_width_height()
self.width_offset = this_width / set_width
self.height_offset = this_height / set_height
im = cv.imdecode(np.fromfile(path, dtype=np.uint8), 1)
im_resized = cv.resize(im, (set_width, set_height), interpolation=cv.INTER_LINEAR)
cv.imwrite(path, im_resized)
except Exception as e:
print("[%s]截图[%s]错误" % (self.ip, pic_name))
print(e)
return path
# self.console_input("shell /system/bin/screencap -p /sdcard/Pictures/%s" % pic_name)
# self.console_input('push /sdcard/Pictures/%s %s' % (pic_name, self.screen_path))
def uninstall_app(self, app_name):
if '前程无忧' in self.china_name:
app_name = 'com.job.android'
if '咪咕阅读' in self.china_name:
app_name = 'com.ophone.reader.ui'
if app_name == "com.ifreetalk.ftalk":
print("喜马拉雅app软件真实名字")
app_name = "com.ximalaya.ting.android"
# self.console_input('uninstall %s' % app_name)
elif app_name in self.mi_app_list:
ips = self.force_uninstall_app_list.get(app_name, [])
if self.ip in ips:
print("ip满足在self.force_uninstall_app_list，满足卸载%s" % app_name)
else:
print("在self.mi_app_list，不在self.force_uninstall_app_list，无需卸载%s" % app_name)
return
else:
print("%s卸载了%s" % (self.model, app_name))
if app_name != "com.baidu.searchbox":
self.MySQLManager.write_file("%s卸载了%s" % (self.model, app_name))
self.MySQLManager.write_file("['%s','卸载软件','%s']" % (self.ip + self.model + app_name + '_' + self.project, app_name))
try:
self.d.app_uninstall(app_name)
except Exception as e:
print(e)
# self.console_input('uninstall %s' % app_name)
# d.app_uninstall("cn.com.open.mooc")
def click(self, x, y):
# 点击如果x和y小于1按照比例点击
if x < 0 or y < 0:
print("不合格的坐标")
return
if x < 1 and y < 1:
width, height = self.get_phone_width_height()
# print(width * x, height * y)
self.d.click(width * x, height * y)
# self.console_input("shell input tap %s %s" % (width * x, height * y))
else:
self.d.click(x, y)
# self.console_input("shell input tap %s %s" % (x, y))
# 返回
def back(self):
self.console_input('shell input keyevent BACK')
@staticmethod
def dingmessage(content='', is_at_tall=True,
secret='SECc776e7cde09261a7dc7d3a9cca02087a72faad2c8d3df5a728da0bbd563a8adf',
token='1c50defbb72189a36d243840128b1981a5779058d8e22b46aba0b773e634a6ee'):
# 钉钉发送消息
# 加签
timestamp = str(round(time.time() * 1000))
secret = secret
secret_enc = secret.encode('utf-8')
string_to_sign = '{}\n{}'.format(timestamp, 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))
token = token
# 请求的URL，WebHook地址
webhook = f"https://oapi.dingtalk.com/robot/send?access_token={token}×tamp={timestamp}&sign={sign}"
print(webhook)
# 构建请求头部
header = {"Content-Type": "application/json", "Charset": "UTF-8"}
# 循环生成器并发送消息
message = {
"msgtype": "text",
"text": {"content": content},
"at": {
# @ 所有人
"isAtAll": is_at_tall
}
}
message_json = json.dumps(message)
info = requests.post(url=webhook, data=message_json, headers=header, verify=False) # 打印返回的结果
print(info.text)
def get_pic_most_recently_called_sort(self, len_pic):
try:
dic = {}
for sql_pic_name in len_pic:
sql_pic_name = sql_pic_name[0]
search_text = "%s]找到图片\\\\[%s\\\\]" % (self.ip, sql_pic_name)
get_text = self.MySQLManager.sql_search("content", search_text)
if get_text:
dic[sql_pic_name] = len(get_text)
else:
dic[sql_pic_name] = 0
get_dic = sorted(dic.items(), key=lambda item: item[1], reverse=True)
# print(get_dic)
pic_list = []
for d in get_dic:
pic_list.append([d[0]])
return pic_list
except Exception as e:
print(e)
return
def select_text(self, selection, cut_list_text):
import requests
import jieba
headers = {'user-agent': 'Mozilla/5.0 (Macintosh; Intel Mac OS X 10.9; rv:32.0) Gecko/20100101 Firefox/32.0', }
text_lis = []
for text in cut_list_text:
if text:
a = jieba.lcut(text, cut_all=True) # 全模式
text_lis.append(a)
print(text_lis)
url = "https://www.sogou.com/web?query={}".format(selection)
r = requests.get(url, headers=headers)
r.encoding = 'utf-8'
html = r.text
# print(html)
lis_num = []
for t in cut_list_text:
num = 1
for word in t:
num += html.count(word)
lis_num.append(num)
print(lis_num)
# 首先获取列表中的最大值
max_data = max(lis_num)
# 获取最大值的索引
max_solution_index = lis_num.index(max_data)
print(max_solution_index, cut_list_text[max_solution_index])
return cut_list_text[max_solution_index]
def mobile_reboot(self):
# 不能重启手机
if self.ip not in [self.main_mobile_ip]:
self.MySQLManager.write_file("%s重启手机" % self.model)
self.console_input('shell reboot')
for i in range(120):
time.sleep(10)
result = self.is_ip_reachable(self.ip)
if result:
print("连接成功无需再连接")
break
else:
print("重启后无法连接")
else:
# 不能重启,卸载com.github.uiautomator
print("不能重启,卸载com.github.uiautomator")
self.console_input('shell input keyevent KEYCODE_BACK')
time.sleep(2)
self.console_input('shell input keyevent KEYCODE_HOME')
self.console_input('uninstall com.github.uiautomator')
text_search = "%s已经完成com.github.uiautomator" % self.ip
value = self.redis_manager.redis_search(text_search)
if value and int(value.decode('utf-8')) >= 3:
print("已经1小时内卸载多次无需再卸载,直接重启" )
self.mobile_reboot()
self.redis_manager.redis_set(text_search,times=3600, is_countable=True)
def waite_time_dotask(self, model='r',wait_time=10, key_word='', user=''):
if self.project:
key_word += self.project
now = datetime.datetime.now()
text = '%s完成%s' % (user, key_word)
if model == 'r':
read_content = self.redis_manager.redis_search(text)
# print('text=',text, "read_content=", read_content)
if read_content or read_content == b'':
print("不满足大于%s分钟，无需要做完成%s" % (wait_time, key_word))
return False
else:
print("满足做任务大于%s分钟，需做完成%s" % (wait_time, key_word))
return True
else:
# self.MySQLManager.write_file('%s完成%s%s' % (user, key_word, now))
self.redis_manager.redis_set(text, times=wait_time * 60)
def get_imobj_with_cache(self, pic_path):
# 1. 先找本地内存
if pic_path in self.local_imobj_cache:
return self.local_imobj_cache[pic_path]
# 2. 本地没有，找 Redis
# 注意：图片名作为 Key，为了防止冲突加个前缀
redis_key = f"im_cache:{pic_path}"
try:
cached_bytes = self.redis_manager.redis_search(redis_key) # 假设你的方法返回 bytes
if cached_bytes:
imobj = pickle.loads(cached_bytes)
self.local_imobj_cache[pic_path] = imobj # 存入本地内存供下次直接用
return imobj
except Exception as e:
print(f"Error reading image from Redis: {e}")
# 3. 都没有，读磁盘
if os.path.exists(pic_path):
imobj = cv.imdecode(np.fromfile(pic_path, dtype=np.uint8), 1)
if imobj is not None:
# 存入 Redis (过期时间设为 1 天，避免占用太多内存)
self.redis_manager.redis_set(redis_key, value=pickle.dumps(imobj), times=86400)
# 存入本地
self.local_imobj_cache[pic_path] = imobj
return imobj
return None
def file_list(self, base_path, content_prefix):
"""
优化后的文件列表查找：
1. 全量缓存目录结构，消除 os.listdir
2. 缓存正则编译结果
3. 消除重复的磁盘 IO 检查
"""
# --- 步骤 1: 确保目录索引已建立 ---
if base_path not in self._DIR_INDEX:
self._scan_and_index_dir(base_path)
# --- 步骤 2: 确定搜索路径和前缀 ---
sub_folder = ""
if "_" in content_prefix:
sub_folder = content_prefix.split("_")[0]
# --- 步骤 3: 从内存索引中获取匹配项 ---
# 索引结构：_DIR_INDEX[base_path][sub_folder] = [file1, file2...]
file_pool = self._DIR_INDEX[base_path].get(sub_folder, [])
if not file_pool and sub_folder:
# 如果子文件夹没找到，尝试在根目录找
file_pool = self._DIR_INDEX[base_path].get("", [])
# --- 步骤 4: 缓存并使用正则表达式 ---
if content_prefix not in self._REGEX_CACHE:
# 预编译正则，提高匹配速度
self._REGEX_CACHE[content_prefix] = re.compile(
r"^" + re.escape(content_prefix) + r"\d+\.\w+", re.S
)
pattern = self._REGEX_CACHE[content_prefix]
# --- 步骤 5: 过滤文件 ---
lis = []
for filename in file_pool:
if pattern.match(filename):
if sub_folder:
# 保持原有的路径拼接逻辑返回
lis.append([os.path.join(sub_folder, filename)])
else:
lis.append([filename])
return lis
def _scan_and_index_dir(self, base_path):
"""一次性扫描整个目录，建立二级映射索引"""
index = {"": []} # 根目录文件
try:
# 获取根目录下所有内容
for item in os.listdir(base_path):
full_path = os.path.join(base_path, item)
if os.path.isdir(full_path):
# 如果是子文件夹，扫描子文件夹
index[item] = os.listdir(full_path)
else:
# 如果是根目录文件
index[""].append(item)
self._DIR_INDEX[base_path] = index
except Exception:
self._DIR_INDEX[base_path] = {"": []}
def multiple_find_picture(self, img: list = (), times: int = 20, click: str = "点击", back: str = "不返回",
xpath: list = (), delem: list = (), find_pic="one", xy_percentage=False,
region: list = (), is_print=True, cache_location=False):
"""
#循环找图或单次找图
:param img: 图片名字,不带后缀
:param times: 找图次数
:param click:是否需要点击图片
:param back:按返回键一次
:param elem: xpathelem元素查找
:param delem: delem
:param find_pic: 是否找多个坐标
:param xy_percentage:x,y显示小数点返回方式
:param region:找图区域
:cache_location:是缓存固定坐标，只有固定的用到，有效期1天
"""
self.check_pause_status()
x = -1
y = -1
# cache相关
cacheKey = str(xpath) + str(delem) + str(img)
if cache_location:
# print(cacheKey)
xy_percentage = True
cache_value = self.redis_manager.redis_search(cacheKey)
if cache_value:
# print("使用缓存坐标", cache_value)
[x, y], k = pickle.loads(cache_value)
if is_print:
print("使用缓存坐标", [x, y], k)
if click == "点击":
self.click(x, y)
return [x, y], k
try:
# 充电
# have_watch = self.MySQLManager.sql_search("content", "已经%s设置充电" % datetime.datetime.now().hour)
have_watch = self.redis_manager.redis_search("已经设置充电" )
if not have_watch:
# print(self.d.device_info)
output = self.console_input('shell dumpsys battery', ret=1)
match = re.search(r'level: (\d+)', output)
power = match.group(1)
int_power = int(power)
# print(int_power)
if int_power <= 15 and self.ip != self.main_mobile_ip:
if self.d:
self.d.screen_off() # 关闭屏幕
self.create_cron_switch(30, mip=self.ip, this_power=str(int_power))
self.redis_manager.redis_set("已经设置充电", times=3600)
else:
# 5分钟检测一次
self.redis_manager.redis_set("已经设置充电", times=500)
# self.MySQLManager.write_file("已经%s设置充电" % datetime.datetime.now().hour)
except Exception as e:
# print(e)
pass
# """===========================↓↓↓↓判断1点抢金主红包↓↓↓↓================================="""
# have_waite = self.MySQLManager.sql_search("content", "%s已经等待过,无需再等待" % self.ip)
# if not have_waite:
# now = datetime.datetime.now()
# t1 = datetime.datetime(now.year, now.month, now.day, 0, 50, 00)
# t2 = datetime.datetime(now.year, now.month, now.day, 0, 58, 00)
# daley_time = (60 - now.minute) * 60 + 60
# # print(daley_time)
# if t1 <= now <= t2:
# print("%s满足,需要等待%s秒" % (self.ip, daley_time))
# self.MySQLManager.write_file("%s满足淘宝抢红包任务,需要等待%s秒" % (self.ip, daley_time))
# self.MySQLManager.write_file("%s已经等待过,无需再等待" % self.ip)
# time.sleep(daley_time)
#
# """===========================↑↑↑↑判断1点抢金主红包↑↑↑↑================================="""
# 记录超时时间
start_time = time.time()
i = 1
while i <= times:
delay = random.randint(1, 5) / 20
# 方法:用memuc调用adb截图,hostname,电脑名字
if delem or xpath:
time.sleep(delay)
if xpath: # 查找元素
for k in xpath:
# print(k)
# noinspection PyBroadException
if not k:
continue
try:
# x和y传送两个值,所以可以这样写
# a = self.d.xpath('//*[@' + k + ']')
a = self.d.xpath(k)
# b = a.wait(timeout=None)
# print(a)
# a = self.d(text=k).wait(timeout=delay)
if a:
end_time = time.time()
spend_time = end_time - start_time
if spend_time > 20:
# 耗时大于20秒,要唤醒屏幕
self.open_phone()
(x1, y1, x2, y2) = a.match().bounds
x = int((x1 + x2) / 2)
y = int((y1 + y2) / 2)
if xy_percentage:
width, height = self.get_phone_width_height()
x = x / width
y = y / height
if is_print:
print("[小米%s]找到xpath元素[%s]坐标:(%.2f,%.2f)本次找元素花费%s次秒数%.2f秒" % (str(self.model) + str(self.ip),
k, x, y, i, spend_time))
# self.MySQLManager.write_file(
# "[小米%s]找到xpath元素%s坐标:(%.2f,%.2f)本次找元素花费%s次秒数%.2f秒" % (str(self.model) + str(self.ip),
# k, x, y, i, spend_time))
if click == "点击":
a.click()
if cache_location:
ret = [x, y], xpath.index(k)
self.redis_manager.redis_set(cacheKey, times='today', value= pickle.dumps(ret))
return [x, y], xpath.index(k)
except Exception as result:
# print("未知错误%s" % result)
pass
if delem:
for k in delem:
# print(i, k)
# noinspection PyBroadException
try:
# x和y传送两个值,所以可以这样写
a = eval("self." + k)
# a = self.d(text=k).wait(timeout=delay)
if a:
end_time = time.time()
spend_time = end_time - start_time
if spend_time > 20:
# 耗时大于20秒,要唤醒屏幕
self.open_phone()
x, y = a.center()
x = int(x)
y = int(y)
if xy_percentage:
width, height = self.get_phone_width_height()
x = x / width
y = y / height
if is_print:
print("[小米%s]找到d元素[%s]坐标:(%.2f,%.2f)本次找元素花费%s次秒数=%.2f秒" %
(str(self.model) + str(self.ip), k, x, y, i, spend_time))
# self.MySQLManager.write_file(
# "[小米%s]找到d元素[%s]坐标:(%.2f,%.2f)本次找元素花费%s次秒数=%.2f秒" %
# (str(self.model) + str(self.ip), k, x, y, i, spend_time))
if click == "点击":
# noinspection PyBroadException
try:
a.click()
except Exception:
self.click(x, y)
if cache_location:
ret = [x, y], delem.index(k)
self.redis_manager.redis_set(cacheKey, times='today', value= pickle.dumps(ret))
return [x, y], delem.index(k)
except Exception as result:
# print("未知错误%s" % result)
pass
if img: # 查找图片
pian_x = 0
pian_y = 0
# self.screen('phone_%s.jpg' % self.ip)
# screen_image = self.screen_path + 'phone_%s.png' % self.ip
pth = self.screen('Screenshots/phone_%s.png' % self.ip)
try:
if region:
# 上下左右（y1， y2， x1，x2）
image = cv.imdecode(np.fromfile(pth, dtype=np.uint8), 1)
size = image.shape
pic_x, pic_y = size[1], size[0]
if is_print:
print('图片x=%s， y=%s' % (pic_x, pic_y))
# （y1（上）:y2(下), x1（左）:x2（右））
# 0.11:0.16, 0.12:0.34(282:395, 134:377])
crop_img = image[int(region[0] * pic_y):
int(region[1] * pic_y),
int(region[2] * pic_x):
int(region[3] * pic_x)] # 获取感兴趣区域
cv.imwrite(pth, crop_img) # 保存到指定目录
pian_x = region[2] * pic_x
pian_y = region[0] * pic_y
except Exception as e:
print(e)
imsrc = cv.imdecode(np.fromfile(pth, dtype=np.uint8), 1)
for k in range(len(img)):
imgobj = img[k]
len_pic = len_pic = self.file_list(self.find_pic_path, imgobj)# ^是字符串开头
if len(len_pic) > 1:
# 根据最近成功最多次的图片数来排序
len_pic2 = self.get_pic_most_recently_called_sort(len_pic)
if len_pic2:
len_pic = len_pic2
if len_pic: # 正则遍历图片个数,找几张图:图片都是名字+数字组成
# print(len(len_pic))
for j in range(0, len(len_pic)):
pic = len_pic[j][0]
# 读取截图图片
full_pic_path = self.find_pic_path + pic
# noinspection PyBroadException
try:
# x和y传送两个值,所以可以这样写
imobj = self.get_imobj_with_cache(full_pic_path)
if imobj is None:
continue
h, w = imobj.shape[:2] # 取彩色图片的高、宽
end_time = time.time()
spend_time = end_time - start_time
# print("[手机%s]正在查找:[%s],第%s次失败,正在进行%s次,第%d张,花费%s秒" %
# (str(self.model) + str(self.ip), pic, i, i + 1, j + 1, spend_time))
res = cv.matchTemplate(imsrc, imobj, cv.TM_CCOEFF_NORMED)
# 取匹配程度大于%90的坐标
loc = np.where(res >= self.threshold)
# np.where返回的坐标值(x,y)是(h,w)，注意h,w的顺序
if find_pic == 'one':
for pt in zip(*loc[::-1]):
x, y = pt[0] + w / 2, pt[1] + h / 2 # 中心点
if int(x) and int(y) >= 0:
end_time = time.time()
spend_time = end_time - start_time
# 区域截图使用,x1（左）和y1（上）
x += pian_x
y += pian_y
# 坐标偏移,修改图片分辨率会使用
if self.width_offset != 1 and self.height_offset != 1:
x = x * self.width_offset
y = y * self.height_offset
if xy_percentage:
width, height = self.get_phone_width_height()
x = x / width
y = y / height
if is_print:
get_th = cv.minMaxLoc(res)
if get_th and len(get_th) >= 2:
get_th = get_th[1]
log_msg = f"[手机{self.ip}]找到图片[{pic}]坐标:x={x:.2f},y={y:.2f},精确度={get_th:.2f}，花费{spend_time:.2f}秒"
print(log_msg)
# 👉 关键优化：把每一次成功的找图作为轻量级步骤记录到报告里
with allure.step(f"成功找到图片: {pic} (精度:{get_th:.2f})"):
pass
time.sleep(delay)
if click == "点击":
self.d.click(x, y)
if cache_location:
ret = [x, y], k
print('cacheKey=', cacheKey, 'return=', ret)
self.redis_manager.redis_set(cacheKey, times='today', value= pickle.dumps(ret))
return [x, y], k
else:
loc_lis = []
for pt in zip(*loc[::-1]):
x, y = pt[0] + w / 2, pt[1] + h / 2 # 中心点
if int(x) and int(y) >= 0:
x += pian_x
y += pian_y
loc_lis.append((int(x), int(y)))
time.sleep(delay)
end_time = time.time()
spend_time = end_time - start_time
if len(loc_lis) >= 2:
x_lis = np.array(loc_lis)[:, 0][1:].tolist()
y_list = np.array(loc_lis)[:, 1][1:].tolist()
diff_x = np.var(x_lis, ddof=1)
diff_y = np.var(y_list, ddof=1)
if diff_x > 1 or diff_y > 1:
# print("[手机%s]图片[%s]多个坐标:loc_lis=%s,方差(%s, %s), 花费%s秒" % (
# self.ip, pic, loc_lis, diff_x, diff_y, spend_time))
# ['da_coin', 'da_lv']
text = loc_lis
if len(str(loc_lis)) > 10:
text = str(loc_lis[:10]) + '……'
print(text)
# self.MySQLManager.write_file(
# "[手机%s]找到图片[%s]多个坐标:loc_lis=%s,方差(%s, %s), 花费%s秒" % (
# self.ip, pic, text, diff_x, diff_y, spend_time))
return [k, k], loc_lis
except Exception:
# print("未知错误%s" % res)
pass
end_time = time.time()
spend_time = end_time - start_time
# print("[手机%s]正在查找:[%s],第%s次失败,正在进行%s次,第%d张,花费%s秒" %
# (str(self.model) + str(self.ip), pic, i, i + 1, j + 1, spend_time))
if back == "返回" and i % 2 == 0:
if is_print:
print("[%s%s]返回一次" % (self.model, self.ip))
self.back()
# 唤醒屏幕
self.open_phone()
time.sleep(2.5)
if back == "下滑" and i % 2 == 0:
if is_print:
print("%s下滑一次" % self.ip)
# self.d.swipe_points([[0.5, 0.8], [0.5, 0.3]], 0.1)
self.swipe(389, 967 + random.randint(0, 10), 389, 397 + random.randint(0, 10))
time.sleep(1)
if back == "上滑" and i % 2 == 0:
if is_print:
print("%s上滑一次" % self.ip)
# self.d.swipe_points([[0.5, 0.3], [0.5, 0.8]], 0.1)
self.swipe(389, 397 + random.randint(0, 10), 389, 967 + random.randint(0, 10))
time.sleep(1)
i += 1
else:
end_time = time.time()
spend_time = end_time - start_time
try:
img = "_".join(xpath) + "_".join(img) + "_".join(delem)
except Exception as e:
print(e)
img = ""
# 判断大于50秒不正常
text = "手机%s,
第%s次真的找不到%s,
本次找图花费秒数=%.2f秒,
任务=%s" % (str(self.model) + str(self.ip), i, img, spend_time,self.project)
if spend_time > 100 and '今日头条' not in self.project:
self.MySQLManager.write_file("%s找图时间过长 %s" % (self.ip, text))
self.mobile_reboot()
if is_print:
print("|找图超过规定次数|,%s" % text)
if i > 10:
logging.warning(self.get_time() + "|找图超过规定次数|,%s" % text)
# self.MySQLManager.write_file(
# "|找图超过规定次数|,%s" % text)
if i >= 20:
self.ui_layout(name=text,d=self.d, mode='失败， 多次找图失败')
return [-1, -1], -1
def cv_record_video(self, times):
# opencv 调用摄像头录制视频
# times 秒数
cap = cv.VideoCapture(0)
if not cap.isOpened():
print("Cannot open camera")
return
frame_width = int(cap.get(cv.CAP_PROP_FRAME_WIDTH))
frame_height = int(cap.get(cv.CAP_PROP_FRAME_HEIGHT))
fps = int(cap.get(cv.CAP_PROP_FPS))
path = '/mnt/nas/share/report/output.mp4'
# 更改编解码器为H.264并将输出格式更改为MP4
fourcc = cv.VideoWriter_fourcc(*'mp4v')
out = cv.VideoWriter(path, fourcc, fps, (frame_width,frame_height))
count = 0
while count < times * fps:
ret, frame = cap.read()
if not ret:
break
out.write(frame)
count += 1
cap.release()
out.release()
return path
def is_video_valid(self, filename):
try:
cap = cv.VideoCapture(filename)
if not cap.isOpened():
return False
else:
return True
except Exception as e:
print(f"Error: {e}")
return False
def start_sr(self, cost_time='150'):
# 开始录屏
ip = self.ip + self.port
print("开始录屏", ip)
if cost_time:
print("设置时长时候，应该1-180之间")
os.popen(self.adb_path + " -s " + ip +" shell screenrecord --time-limit %s /sdcard/screen.mp4" % cost_time)
else:
os.popen(self.adb_path + " -s " + ip +" shell screenrecord /sdcard/screen.mp4")
def end_sr(self, name='录屏测试'):
# 结束录屏
ip = self.ip + self.port
video_path = os.path.join(self.screen_path, "demo.mp4")
print("结束录屏", ip)
os.system(self.adb_path + " -s " + ip + " shell pkill -l SIGINT screenrecord")
time.sleep(1)
os.system(self.adb_path + ' -s %s pull /sdcard/screen.mp4 %s' %(ip, video_path))
try:
if self.is_video_valid(video_path):
allure.attach.file(video_path,name=name, attachment_type=allure.attachment_type.MP4)
os.remove(video_path)
self.console_input('shell rm /sdcard/screen.mp4')
except Exception as e:
print(e)
try:
# 删除多余视频
# 获取手机根目录下的文件列表
output = self.console_input('shell ls /sdcard/',ret=1).splitlines()
#print(output)
# 遍历文件列表，找到以 "screen" 开头的文件并删除
for file in output:
# print('file=', file)
if file.startswith('screen'):
print(file)
self.console_input('shell rm /sdcard/{}'.format(file))
except Exception as e:
print(e)
def allure_screen(self, name='找图'):
path = self.screen('phonetimeout/' + self.model + '_ocr.png')
try:
allure.attach.file(path, name=name, attachment_type=allure.attachment_type.PNG)
except Exception as e:
print(e)
def swipe(self, x1, y1, x2, y2):
if x1 < 1 and y1 < 1 and x2 < 1 and y2 < 1:
width, height = self.get_phone_width_height()
# self.console_input("shell input tap %s %s" % (width * x, height * y))
self.console_input("shell input swipe %s %s %s %s" % (width * x1, height * y1, width * x2, height * y2))
else:
self.console_input("shell input swipe %s %s %s %s" % (x1, y1, x2, y2))
def tb_timeout_option(self, content):
"""不同脚本,调用不一样的"""
try:
self.z += 1
print("|超时|手机%s,z=%s,运行%s程序" % (self.ip, self.z, content))
logging.error(self.get_time() + "|超时|手机%s,z=%s,运行%s程序" % (self.ip, self.z, content))
self.MySQLManager.write_file("|超时|手机%s,z=%d,运行%s程序" % (self.ip, self.z, content))
self.ui_layout(name=content, d=self.d, mode='失败, 主动触发tb_timeout_option')
except Exception as e:
print(e)
def console_input(self, cmd_console, ret=0):
cmd = f"{self.adb_path} -s {self.ip}{self.port} {cmd_console}"
if ret == 1:
# 需要返回结果时，使用 subprocess 捕获输出
process = subprocess.Popen(cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output, error = process.communicate()
if process.returncode != 0:
print(f"ADB Command Error: {error.decode('utf-8')}")
return ""
return output.decode('utf-8')
else:
# 仅执行命令，不关心输出
subprocess.Popen(cmd, shell=True).wait()
def get_app_name(self):
result = self.d.info['currentPackageName']
return result
def open_phone(self):
k = 0
# print(self.separation)
for i in range(20): # 找不到一直循环
try:
if self.d.info.get("screenOn") is False:
k += 1
self.d.unlock() # 解锁,
else:
# 判断是否密码锁,是红米note8
if self.ip:
if k != 0:
# 不等于0说明参与解锁了,所以要找密码
print("输入密码")
try:
a = self.d.xpath('//*[@text="紧急呼叫"]').wait(2)
if a:
if not self.separation:
print("找到,self.separation为空,让它等于第一个分身密码")
self.separation = separation_phone1
for j in self.separation:
self.d.xpath('//*[@text="%s"]' % j).click()
time.sleep(0.2)
time.sleep(1)
except Exception as e:
print(e)
return True
else:
return True
except Exception as e:
print(e)
self.re_install_uiautomator(e)
def jurisdiction(self):
# 这一个某些软件点统一才计算秒数
# 打开分身的app//*[@
self.multiple_find_picture([], 1, "点击", "不返回", delem=['d(resourceId="android.miui:id/app1")'])
[x1, y1], k = self.multiple_find_picture([], 1, "点击", "不返回",
['//*[@text="仅在使用中允许"]',
'//*[@text="同意"]',
'//*[@text="同意并继续"]',
'//*[@text="本次运行允许"]',
'//*[@text="允许"]',
'//*[@text="始终允许"]',
'//*[@text="仍然支付"]']) # 弹出来权限,第一个红米note8,第二个小米4的
if x1 > 0 and y1 > 0:
# 红米note8也可能检测到
print("点击6次允许")
self.multiple_find_picture([], 1, "点击", "返回", delem=['d(text="确定")'])
for z in range(8):
time.sleep(1)
self.d.click(x1, y1)
def switch_user(self, flag1='//*[@content-desc="我的淘宝"]', flag2='//*[@content-desc="设置"]', flag3='//*[@text="切换账户"]'):
print("准备切换到%s" % self.tao_user)
self.MySQLManager.write_file("准备切换到%s" % self.tao_user)
time.sleep(5)
if self.z >= 3: # 为什么是7因为,上面执行完还没有一轮
return
print("准备查到我的")
[x, y], k = self.multiple_find_picture([], 10, "点击", "返回", [flag1], delem=['d(text="我的")'])
if x > 0 and y > 0:
try:
text = self.d(description="设置").sibling(index=3).info['contentDescription']
if text.endswith(self.tao_user[-1]) and text.startswith(self.tao_user[0]):
print("%s根据首尾字符串判断切换账号为当前登录账号,不用切换" % (self.model + self.tao_user))
self.MySQLManager.write_file("%s根据首尾字符串判断切换账号为当前登录账号,不用切换" % (self.model + self.tao_user))
return True
# print(text.endswith(self.tao_user[-1]))
# print(text.startswith(self.tao_user[0]))
except Exception as e:
print(e)
time.sleep(1)
[x1, y1], k = self.multiple_find_picture([], 20, "点击", "不返回", [flag2])
if x1 > 0 and y1 > 0:
time.sleep(5)
[x2, y2], k = self.multiple_find_picture([], 20, "点击", "下滑",
[
'//*[@resource-id="com.taobao.taobao:id/tv_multi_account"]',
flag3
])
if x2 > 0 and y2 > 0:
time.sleep(1)
[x3, y3], k = self.multiple_find_picture([], 20, "点击", "不返回", ['//*[@text="{}"]'.format(self.tao_user)])
if x3 > 0 and y3 > 0:
[x4, y4], k = self.multiple_find_picture([], 40, "不点击", "不返回",
[' ', flag1],
delem=['d(index=0, '
'className="android.widget.RelativeLayout").'
'child(text="%s")' % self.tao_user])
if x4 > 0 and y4 > 0:
if k == 0:
print("%s需要切换账号为当前登录账号,不用切换" % (self.model + self.tao_user))
self.MySQLManager.write_file("%s需要切换账号为当前登录账号,不用切换" % (self.model + self.tao_user))
if k == 1:
print("%s切换账号后找到我的淘宝,说明切换成功" % (self.model + self.tao_user))
self.MySQLManager.write_file("%s切换账号后找到我的淘宝,说明切换成功" % (self.model + self.tao_user))
return True
self.tb_timeout_option("switch_user")
self.d.app_stop(self.app_name)
time.sleep(2)
self.d.app_start(self.app_name)
self.switch_user(flag1, flag2, flag3)
"""##################QQ聊天#####################"""
def qq(self, qq_num, text):
self.d.app_start("com.tencent.mobileqq")
for i in range(20):
self.d.app_start("com.tencent.mobileqq")
[x, y], k = self.multiple_find_picture([], 3, "点击", "返回",
['//*[@resource-id="com.tencent.mobileqq:id/et_search_keyword"]'])
if x > 0 and y > 0:
time.sleep(0.2)
[x1, y1], k = self.multiple_find_picture([], 3, "点击", "不返回",
[],
delem=[
'd(resourceId="com.tencent.mobileqq:id/et_search_keyword")'
'.send_keys("%s")' % qq_num])
if x1 > 0 and y1 > 0:
print("点击人,进入")
time.sleep(0.2)
click_person = self.d.xpath(
'//*[@resource-id="com.tencent.mobileqq:id/eap"]/android.widget.LinearLayout[2]')
if click_person:
click_person.click()
print("发送文本")
[x2, y2], k = self.multiple_find_picture([], 6, "不点击", "不返回", [],
delem=['d(resourceId="com.tencent.mobileqq:id/input")'])
if x2 > 0 and y2 > 0:
print("准备点击")
self.d(resourceId="com.tencent.mobileqq:id/input").send_keys(text)
time.sleep(0.2)
self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(text="发送")'])
break
else:
self.back()
return
"""##################开启性能模式#####################"""
def up_up(self):
self.open_phone()
try:
print("开启性能模式")
self.d.open_notification()
[x, y], k = self.multiple_find_picture([], 20, "不点击", "不返回", [],
delem=['d(text="已开启性能模式")',
'd(resourceId="android:id/icon", description="设置")'])
if x > 0 and y > 0:
print("不需要打开性能模式")
self.back()
return
print("需要开启性能模式")
# 性能模式,关闭
time.sleep(0.2)
self.multiple_find_picture([], 5, "点击", "不返回", [],
delem=['d(description="性能模式")', 'd(description="性能模式,关闭")'])
self.back()
# self.d.app_start("com.android.settings")
# self.d(resourceId="android:id/input").send_keys("运行模式")
# self.d(text="电量和性能/性能").click()
# self.d(text="运行模式").click()
# self.multiple_find_picture(["xn"], 20, "点击", "不返回")
# time.sleep(1)
self.MySQLManager.write_file("打开性能模式成功")
except Exception as e:
self.MySQLManager.write_file("打开性能模式失败 %s" % e)
"""##################开启和关闭震动#####################"""
def close_shock(self, ip=''):
self.ip = ip
print("准备关闭震动")
self.read()
print("准备关闭震动")
# self.d.open_notification()
self.d.swipe_points([[0.8, 0.0], [0.8, 0.1]], 0.1)
time.sleep(0.5)
[x, y], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
delem=['d(description="振动,开启")'])
if x > 0 and y > 0:
print("关闭震动")
self.back()
def open_shock(self, ip=""):
self.ip = ip
print("准备打开震动")
self.read()
self.open_phone() # 解锁
# self.d.open_notification()
self.d.swipe_points([[0.8, 0.0], [0.8, 0.1]], 0.1)
time.sleep(0.5)
[x, y], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
delem=['d(description="振动,关闭")'])
if x > 0 and y > 0:
print("开启震动")
self.back()
"""##################卸载小米4和红米note2多余app#####################"""
def uninstall_other_app(self, ip='', port=':5555', set_day=2):
if ip:
self.ip = ip
self.port = port
self.read()
# self.open_phone() # 解锁
if self.ip != self.main_mobile_ip:
print("卸载多余的app")
# self.MySQLManager.write_file("%s进入卸载多余app程序" % self.ip)
for package_name_en in self.d.app_list():
# if i in self.mi_app_list:
# pass
# elif i in self.read_yaml['can_uninstall_app']:
# self.uninstall_app(i)
get_str = self.console_input('shell dumpsys package %s' % package_name_en,
ret=1)
r_list = re.compile('firstInstallTime=(\\d{4}-\\d{2}-\\d{2})', re.S).findall(get_str)
if r_list:
try:
first_install_name = r_list[0]
first_install_name = datetime.datetime.strptime(first_install_name, '%Y-%m-%d').date()
today = datetime.datetime.now().date()
diff_day = (today - first_install_name).days
print('%s第一次安装时间%s，已经安装%s天' % (package_name_en, first_install_name, diff_day))
if diff_day <= set_day:
print("=" * 100)
print("%s满足卸载" % package_name_en)
print("=" * 100)
self.uninstall_app(package_name_en)
except Exception as e:
print("错误%s" % e)
print("卸载结束")
# if self.ip == mi_4_ip:
# print("卸载快手")
# self.d.app_uninstall("com.kuaishou.nebula")
# self.d.app_uninstall("com.eg.android.AlipayGphone")
# print("卸载趣头条")
# self.d.app_uninstall("com.jifen.qukan")
# print("卸特价版淘宝")
# self.d.app_uninstall("com.taobao.litetao")
"""##################手机定时关机#####################"""
def on_off_phone(self, z=0):
self.z = z
self.read()
print("self.z=%s" % self.z)
if self.z >= 5:
self.MySQLManager.write_file("%s超过设置开机时间指定次数" % self.ip)
self.z = 0
return
self.open_phone()
print("进入设置关机时间")
now_hour = datetime.datetime.now().hour
a = self.MySQLManager.sql_search('content', "%s%s点已完成%s任务" % (self.ip, now_hour, '开机时间'))
b = self.MySQLManager.sql_search('content', "%s%s点已完成%s任务" % (self.ip, now_hour, '关机时间'))
print(a, b)
if 'MI 4LTE' in self.model:
print("无需设置开关机")
return
if a and b:
print("满足不用设置开关机时间了")
return
try:
self.d.app_stop("com.android.settings")
except Exception as e:
print(e)
self.d.press("home")
self.d.press("search")
time.sleep(0.5)
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", [],
delem=['d(resourceId="com.android.quicksearchbox:id/query_text")',
'd(text="同意并继续")'])
if x > 0 and y > 0:
if k == 1:
self.tb_timeout_option("找到同意并继续,说明第一次搜索")
self.on_off_phone()
return
try:
self.d(resourceId="com.android.quicksearchbox:id/query_text").send_keys("省电与电池")
except Exception as e:
print(e)
print("输入电量和性能")
# self.d(resourceId="com.android.quicksearchbox:id/query_text").send_keys("定时开关机")
[x1, y1], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
delem=['d(description="icon 省电与电池 手机管家 | 应用功能")',
'd(description="电量和性能 设置/电量和性能")',
'd(text="电量和性能 设置/电量和性能")',
"d(text='设置').sibling(className='android.view.View').child(text='省电与电池')",
"d(text='手机管家').sibling(className='android.view.View').child(text='省电与电池')"])
if x1 > 0 and y1 > 0:
print("电量和性能 系统和设备")
# [x2, y2], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
# delem=[
# 'd(className="android.widget.RelativeLayout").child(text="省电优化")'])
# if x2 > 0 and y2 > 0:
# print("省电优化")
[x3, y3], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
delem=['d(text="电池")', 'd(text="更多电池功能")'])
if x3 > 0 and y3 > 0:
print("电池")
else:
print("电池")
elemts = ['d(resourceId="android:id/title", text="定时开关机")',
'd(resourceId="com.miui.securitycenter:id/title")',
'd(resourceId="com.miui.securitycenter:id/pc_activity_textview5")']
[x4, y4], k = self.multiple_find_picture([], 10, "点击", "不返回", [], delem=elemts)
if x4 > 0 and y4 > 0:
print("定时开关机")
else:
print("找不到定时开关机从新搜索")
self.MySQLManager.write_file("找不到定时开关机从新搜索")
self.tb_timeout_option("%s找不到定时开关机从新搜索" % self.ip)
self.on_off_phone(self.z)
return
now_hour = datetime.datetime.now().hour
if 17 < now_hour <= 23:
print("现在%s点改成凌晨的" % now_hour)
self.modify_time('23', "开机时间") # 调节后比目前大于1,比如22就是23
# self.modify_time('19', "关机时间")
elif 0 <= now_hour <= 7:
print("现在%s点改成上午的" % now_hour)
self.modify_time('9', "开机时间") # 调节后比目前大于1,比如22就是23
else:
self.modify_time('17', "开机时间") # 调节后比目前大于1,比如22就是23
# self.modify_time('3', "关机时间")
time.sleep(0.3)
self.screen("Android/%son_off_phone%shour.jpg" % (self.ip, now_hour))
[x5, y5], k = self.multiple_find_picture([], 10, "点击", "不返回", [],
delem=['d(description="确定")'])
if x5 > 0 and y5 > 0:
print("点对号或点正确")
return
pass
def modify_time(self, num1, open_or_off, count=0):
print("%s回到修改时间程序count=%s[目前为%s]" % (self.ip, count, open_or_off))
self.MySQLManager.write_file("%s回到修改时间程序count=%s" % (self.ip, count))
time.sleep(0.3)
if count >= 5 or self.z >= 5:
print("尝试5次失败")
return
[x5, y5], k = self.multiple_find_picture([], 10, "不点击", "不返回", [],
delem=['d(resourceId="android:id/title", text="%s")' % open_or_off])
if x5 > 0 and y5 > 0:
print("找到%s,如果开机instance为0,如果开机instance 等于1" % open_or_off)
try:
self.search_content(num1, open_or_off, count)
except Exception as e:
qq_email("534640040@qq.com", '%s 设置开关机失败' % self.ip, "设置开关机报错")
pass
return
else:
self.tb_timeout_option("modify_tim")
self.on_off_phone()
return
def search_content(self, num, open_or_off, count):
if count >= 5:
print("尝试5次失败")
self.tb_timeout_option("count_is_more_big")
self.on_off_phone()
qq_email("534640040@qq.com", '%s 设置开关机多次失败' % self.ip, "设置开关机多次失败")
return
if open_or_off == "开机时间":
instance = 0
else:
instance = 1
try:
located_time = self.d(textContains=":", instance=instance).info["text"]
except Exception as e:
print(e)
self.tb_timeout_option("located_time_wrong")
self.on_off_phone()
return
if located_time:
setting_time = located_time.split(":")[0]
print(setting_time)
if setting_time == num:
print("%s满足" % open_or_off)
path = self.screen('Android/' + self.model + '_off_on.png')
try:
allure.attach.file(path, name=open_or_off, attachment_type=allure.attachment_type.PNG)
except Exception as e:
print(e)
now_hour = datetime.datetime.now().hour
self.MySQLManager.write_file("%s%s点已完成%s任务" % (self.ip, now_hour, open_or_off))
else:
print(num, setting_time)
try:
adjust_time = (int(num) - int(setting_time)) % 24
except Exception as e:
print(e)
self.tb_timeout_option("adjust_time_wrong")
self.on_off_phone()
return
if adjust_time:
print("%s不满足,相差%s小时" % (open_or_off, adjust_time))
self.MySQLManager.write_file("%s不满足,%s相差%s小时" % (open_or_off, self.ip, adjust_time))
[x1, y1], k = self.multiple_find_picture([], 10, "点击", "不返回", [], delem=[
'd(resourceId="android:id/title", text="%s")' % open_or_off])
if x1 > 0 and y1 > 0:
time.sleep(1)
print("滑动")
width, height = self.get_phone_width_height()
x = width * 0.2
y = height * 0.7
for i in range(adjust_time):
self.d.swipe(x, y, x, y - 200)
time.sleep(0.5)
self.multiple_find_picture([], 5, "点击", "不返回", [],
delem=['d(resourceId="android:id/button1")'])
count += 1
self.search_content(num, open_or_off, count)
return
def elem_width_high_deal(self, times = 30, elem=None, h=None, w = None, click=None):
for i in range(times):
if elem:
if type(elem) == str:
# 字符串才可以
# print(elem % i)
d_elem = eval(elem % i)
else:
d_elem = elem
if d_elem:
bounds = d_elem.info['bounds']
height= bounds['bottom']-bounds['top']
width= bounds['right']-bounds['left']
center = d_elem.center()
# print(height, width, center)
if h and w:
if height == h and width == w:
if click:
d_elem.click()
return center
elif w:
if width == w:
if click:
d_elem.click()
return center
elif h :
if height == h:
if click:
d_elem.click()
return center
def jxsd(self, times=40, text_search="奖励现身，点我领",img_search = [], back_elem=[], break_elemts = []):
# d(text="下滑浏览商品流，可寻找卡片并提升星球探索度")
break_elemts.append('d(text="验证码拦截")')
print(text_search, img_search)
time.sleep(3)
for i in range(times):
# self.d.swipe_ext("up", 0.8)
self.d.swipe_points([[0.5, 0.8], [0.5, 0.3]], 0.1)
# print("滑动")
try:
# elem = self.d(text=text_search)
[x, y], k = self.multiple_find_picture(img_search, 2, "点击", "不返回", [], delem=['d(text="%s")' % text_search], xy_percentage=True, is_print=False)
if x > 0 and y > 0:
# center = elem.center()
if 0.12< y < 0.92:
# print(y_per)
# elem.click()
self.click(x, y)
time.sleep(0.5)
[x, y], k = self.multiple_find_picture([], 8, "不点击", "不返回", [], delem=['d(text="开心收下 ")', 'd(text="我知道了")', 'd(textContains="碎片溜走了")', 'd(textContains="包裹碎片")', 'd(textContains="下滑浏览商品流")'], xy_percentage=True)
if x > 0 and y > 0:
if k == 0 or k == 1:
self.click(x, y)
# elif k == 2:
# text_elem = self.d(textContains="包裹碎片")
# if text_elem:
# text2 = text_elem.info['text']
# print(text2)
else:
if back_elem:
self.multiple_find_picture([], 20, "不点击", "返回", [], delem=back_elem, xy_percentage=True)
else:
[x, y], k = self.multiple_find_picture([], 1, "点击", "不返回", [], delem=['d(text="开心收下 ")'], xy_percentage=True)
if break_elemts:
[x, y], k = self.multiple_find_picture([], 1, "点击", "不返回", [], delem=break_elemts, xy_percentage=True, is_print=False)
if x > 0 and y > 0:
break
except Exception as e:
print(e)
self.d.swipe_ext("up", 0.9)
def wait_time(self):
i = datetime.datetime.now()
minute, second = i.minute, i.second
delay_time = random.uniform(1, 3)
all_second = 60 * 60 - (minute * 60 + second) + delay_time
if minute > 0:
print("等待%s秒去找红包,延迟%s秒" % (all_second, delay_time))
self.MySQLManager.write_file("等待%s秒去找红包,延迟%s秒" % (all_second, delay_time))
time.sleep(all_second)
else:
print("不需要等待")
# pylint: disable=no-member, disable=assignment-from-none
def slider_verification(self):
[loc_x, loc_y], k = self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(description="javascript:;")'])
self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(text="请点击此处重试")',
'd(description="请点击此处重试")'])
print(loc_x, loc_y)
def match_img(img_path1, img_path2):
imgs = []
# 展示
sou_img1 = cv.imread(img_path1)
if sou_img1 is None:
pass
sou_img2 = cv.imread(img_path2)
# 最小阈值100,最大阈值500
# , 0
img1 = cv.imread(img_path1)
blur1 = cv.GaussianBlur(img1, (3, 3), 0)
canny1 = cv.Canny(blur1, 100, 500)
cv.imwrite('temp1.png', canny1)
# , 0
img2 = cv.imread(img_path2)
blur2 = cv.GaussianBlur(img2, (3, 3), 0)
canny2 = cv.Canny(blur2, 100, 500)
cv.imwrite('temp2.png', canny2)
target = cv.imread('temp1.png')
template = cv.imread('temp2.png')
# 调整大小
target_temp = cv.resize(sou_img1, (350, 200))
target_temp = cv.copyMakeBorder(target_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
template_temp = cv.resize(sou_img2, (200, 200))
template_temp = cv.copyMakeBorder(template_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
imgs.append(target_temp)
imgs.append(template_temp)
# theight, twidth = template.shape[:2]
# 匹配跟拼图
result = cv.matchTemplate(target, template, cv.TM_CCOEFF_NORMED)
cv.normalize(result, result, 0, 1, cv.NORM_MINMAX, -1)
min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
# # # 画圈
# cv.rectangle(target, max_loc, (max_loc[0] + twidth, max_loc[1] + theight), (0, 0, 255), 2)
# target_temp_n = cv.resize(target, (350, 200))
# target_temp_n = cv.copyMakeBorder(target_temp_n, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
# imgs.append(target_temp_n)
# imstack = np.hstack(imgs)
#
# cv.imshow('windows' + str(max_loc), imstack)
# cv.waitKey(0)
# cv.destroyAllWindows()
# 计算距离
# print(max_loc)
distance = str(max_loc).split()[0].split('(')[1].split(',')[0]
x_dis = int(distance) + 150 # 150根据自己设备而定
# t(x_dis)
print("距离为%s" % x_dis)
return x_dis
# pylint: disable=unsubscriptable-object
def pic_shear():
# 1029:1162 指的从竖直方向截取 1029~1162(y轴)
# 198:331 指的是从水平方向截取 198~331(x轴)
pic_path = self.screen_path + self.ip + "screen.png"
c1 = self.screen_path + self.ip + "c1.png"
c2 = self.screen_path + self.ip + "c2.png"
img = cv.imread(pic_path)
if loc_y > 1860:
cv.imwrite(c1, img[1200:1564, 333:891])
cv.imwrite(c2, img[1200:1564, 189:339])
dist = match_img(c1, c2)
else:
cv.imwrite(c1, img[853:1277, 333:891])
cv.imwrite(c2, img[855:1276, 189:339])
dist = match_img(c1, c2)
return dist
# print("已截图")
# d(text="拖动滑块完成拼图")
[x, y], k = self.multiple_find_picture([], 20, "不点击", "不返回", [], delem=['d(text="拖动滑块完成拼图")',
'd(description="拖动滑块完成拼图")'])
if x > 0 and y > 0:
# print(x, y)
width, height = self.get_phone_width_height()
x = x - width * 0.247
# print(x, y)
dis = 0
for i in range(15):
self.screen(self.ip + "screen.png")
dis = pic_shear()
if dis < 160:
self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(description="javascript:;")'])
self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(text="请点击此处重试")',
'd(description="请点击此处重试")'])
time.sleep(random.uniform(1, 1.5))
else:
self.d.swipe(x, y, x + dis, y)
time.sleep(random.uniform(1, 1.5))
[x, y], k = self.multiple_find_picture([], 20, "不点击", "不返回", [], delem=['d(text="拖动滑块完成拼图")',
'd(text="向右拖动滑块填充拼图")',
'd(description="拖动滑块完成拼图")'])
if x < 0 and y < 0:
print("识别成功了")
return
else:
dis = 0
print("距离为:%s" % dis)
def locate_x_y(self, screen_img=None, pic=''):
# 默认的截图比较的
def match_img(img_path1, img_path2):
imgs = []
# 展示
sou_img1 = cv.imread(img_path1)
sou_img2 = cv.imread(img_path2)
# 最小阈值100,最大阈值500
# , 0
img1 = cv.imread(img_path1)
blur1 = cv.GaussianBlur(img1, (3, 3), 0)
canny1 = cv.Canny(blur1, 100, 500)
cv.imwrite('temp1.png', canny1)
# , 0
img2 = cv.imread(img_path2)
blur2 = cv.GaussianBlur(img2, (3, 3), 0)
canny2 = cv.Canny(blur2, 100, 500)
cv.imwrite('temp2.png', canny2)
target = cv.imread('temp1.png')
template = cv.imread('temp2.png')
# 调整大小
target_temp = cv.resize(sou_img1, (350, 200))
target_temp = cv.copyMakeBorder(target_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
template_temp = cv.resize(sou_img2, (200, 200))
template_temp = cv.copyMakeBorder(template_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
imgs.append(target_temp)
imgs.append(template_temp)
theight, twidth = template.shape[:2]
# 匹配跟拼图
result = cv.matchTemplate(target, template, cv.TM_CCOEFF_NORMED)
cv.normalize(result, result, 0, 1, cv.NORM_MINMAX, -1)
min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
print("%s, %s, %s, %s" % (min_val, max_val, min_loc, max_loc))
# cv.rectangle(target, max_loc, (max_loc[0] + twidth, max_loc[1] + theight), (0, 0, 255), 2)
# target_temp_n = cv.resize(target, (350, 200))
# target_temp_n = cv.copyMakeBorder(target_temp_n, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
# imgs.append(target_temp_n)
# imstack = np.hstack(imgs)
# # print("imstack=%s" % imstack)
# cv.imshow('windows' + str(max_loc), imstack)
# cv.waitKey(0)
# cv.destroyAllWindows()
# 计算距离
# distance = str(max_loc).split()[0].split('(')[1].split(',')[0]
# x_dis = int(distance) + 150 # 150根据自己设备而定
# t(x_dis)
# print("距离为%s" % x_dis)
# ↓↓↓保存图片↓↓↓
x1 = max_loc[0]
y1 = max_loc[1]
x2 = max_loc[0] + twidth
y2 = max_loc[1] + theight
print("位置为:x1=%s, y1=%s,x2=%s,y2=%s" % (x1, y1, x2, y2))
# print("%s, %s, %s, %s" % (x1, y1, x2, y2))
return x1, y1, x2, y2
# pylint: disable=unsubscriptable-object
def pic_shear():
# 1029:1162 指的从竖直方向截取 1029~1162(y轴)
# 198:331 指的是从水平方向截取 198~331(x轴)
screen_name_path = self.screen_path + 'phone_%s.png' % self.ip
if screen_img:
# 不为空时候,需要截图
self.screen('phone_%s.png' % self.ip)
edge_detection_picture = self.screen_path + pic
x1, y1, x2, y2 = match_img(screen_name_path, edge_detection_picture)
img = cv.imread(screen_name_path)
ocr_pic = self.screen_path + self.ip + "c1.png"
# 保存需要ocr的小图片
cv.imwrite(ocr_pic, img[y1:y2, x1:x2])
# ocr识别图片
content1 = self.orc_l_x(ocr_pic)
print_text = "[%s]pytesseract识别结果为%s,\n\t边缘检测图片[%s], \n\t截图为[%s], \n\tocr识别图片[%s]" \
% (self.ip, content1, edge_detection_picture, screen_name_path, ocr_pic)
print(print_text)
self.MySQLManager.write_file(print_text)
return content1
content = pic_shear()
return content
def toast_get_message(self, max_time=2, ip='192.168.31.161', port='5555', adb_path = '/usr/local/bin/adb'):
# 我发现连接uiautomator2.connect就输出空的,
# import time
# import threading
# import subprocess
# import queue
# import re
# try:
# captured_output = queue.Queue() # 使用队列来存储捕获的输出
# def capture_output(proc, output_queue):
# """
# 捕获并累积子进程的输出，并在结束时返回整个累积的输出。
# """
# while True:
# line = proc.stdout.readline()
# if not line:
# break
# output_queue.put(line)
# # print(line) # 立即打印输出
# # print(adb_path, '-s', f'{ip}:{port.replace(":", "")}', 'shell', 'uiautomator', 'events')
# proc = subprocess.Popen(
# [adb_path, '-s', f'{ip}:{port.replace(":", "")}', 'shell', 'uiautomator', 'events'],
# stdout=subprocess.PIPE,
# stderr=subprocess.STDOUT,
# bufsize=1, # 取消缓冲
# universal_newlines=True
# )
# output_thread = threading.Thread(target=capture_output, args=(proc, captured_output))
# output_thread.start()
# time.sleep(max_time)
# proc.terminate()
# proc.wait()
# output_thread.join()
# # 收集所有输出
# captured_output_str = ''
# while not captured_output.empty():
# captured_output_str += captured_output.get()
# print(captured_output_str)
# pattern = r"android\.widget\.Toast; Text: \[(.*?)\];"
# # Search the text
# matches = re.findall(pattern, captured_output_str)
# # Print the results
# for match in matches:
# if match:
# print(match)
# return captured_output_str
# except:
t1 = time.time()
for i in range(100):
try:
text = self.d.toast.get_message(max_time, 2, "default message")
if text != "default message":
break
except:
text = "default message"
time.sleep(0.1)
t2 = time.time()
if t2 - t1 >= max_time:
break
return text
def kuai_slider_verification(self):
def match_img(img_path1, img_path2):
imgs = []
# 展示
sou_img1 = cv.imread(img_path1)
sou_img2 = cv.imread(img_path2)
# 最小阈值100,最大阈值500
img1 = cv.imread(img_path1)
blur1 = cv.GaussianBlur(img1, (3, 3), 0)
canny1 = cv.Canny(blur1, 100, 500)
cv.imwrite('temp1.png', canny1)
img2 = cv.imread(img_path2)
blur2 = cv.GaussianBlur(img2, (3, 3), 0)
canny2 = cv.Canny(blur2, 100, 500)
cv.imwrite('temp2.png', canny2)
target = cv.imread('temp1.png')
template = cv.imread('temp2.png')
# 调整大小
target_temp = cv.resize(sou_img1, (350, 200))
target_temp = cv.copyMakeBorder(target_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
template_temp = cv.resize(sou_img2, (200, 200))
template_temp = cv.copyMakeBorder(template_temp, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
imgs.append(target_temp)
imgs.append(template_temp)
# theight, twidth = template.shape[:2]
# 匹配跟拼图
result = cv.matchTemplate(target, template, cv.TM_CCOEFF_NORMED)
cv.normalize(result, result, 0, 1, cv.NORM_MINMAX, -1)
min_val, max_val, min_loc, max_loc = cv.minMaxLoc(result)
# # # 画圈
# cv.rectangle(target, max_loc, (max_loc[0] + twidth, max_loc[1] + theight), (0, 0, 255), 2)
# target_temp_n = cv.resize(target, (350, 200))
# target_temp_n = cv.copyMakeBorder(target_temp_n, 5, 5, 5, 5, cv.BORDER_CONSTANT, value=[255, 255, 255])
# imgs.append(target_temp_n)
# imstack = np.hstack(imgs)
#
# cv.imshow('windows' + str(max_loc), imstack)
# cv.waitKey(0)
# cv.destroyAllWindows()
# 计算距离
# print(max_loc)
distance = str(max_loc).split()[0].split('(')[1].split(',')[0]
x_dis = int(distance) + 250 # 150根据自己设备而定
# t(x_dis)
print("距离为%s" % x_dis)
return x_dis
# pylint: disable=unsubscriptable-object
def pic_shear():
# 1029:1162 指的从竖直方向截取 1029~1162(y轴)
# 198:331 指的是从水平方向截取 198~331(x轴)
pic_path = self.screen_path + self.ip + "screen.png"
c1 = self.screen_path + self.ip + "c1.png"
c2 = self.screen_path + self.ip + "c2.png"
img = cv.imread(pic_path)
cv.imwrite(c1, img[273:842, 262:1032])
cv.imwrite(c2, img[278:839, 51:278])
dist = match_img(c1, c2)
return dist
# print("已截图")
# d(text="拖动滑块完成拼图")
[x, y], k = self.multiple_find_picture([], 20, "不点击", "不返回", [], delem=['d(text="拖动滑块完成拼图")',
'd(text="向右拖动滑块填充拼图")',
'd(description="拖动滑块完成拼图")'])
if x > 0 and y > 0:
# print(x, y)
width, height = self.get_phone_width_height()
x = x - width * 0.4
# print(x, y)
dis = 0
for i in range(15):
self.screen(self.ip + "screen.png")
dis = pic_shear()
if dis <= 250:
self.multiple_find_picture([], 5, "点击", "不返回", [], delem=['d(textContains="cutPic").sibling(index=0)'])
time.sleep(random.uniform(1, 1.5))
else:
self.d.swipe(x, y, x + dis, y)
break
if x >= 0 and y >= 0:
print("距离为:%s" % dis)
# time.sleep(1)
def recognize_text_dddocr(self,image_path: str) -> str:
"""
使用 dddocr 识别图片中的文字，返回识别结果字符串。
参数:
image_path: 图片文件路径（支持png/jpg等格式）
返回:
识别出的文字字符串
"""
import ddddocr
# 创建识别器实例，关闭广告打印（verbose=False）
ocr = ddddocr.DdddOcr(show_ad=False)
# 读取图片字节
with open(image_path, 'rb') as f:
img_bytes = f.read()
# 识别文字
result = ocr.classification(img_bytes)
return result
def ddddocr_slider_verification(self, image):
print("验证")
# d(text="请通过以下验证").child(index=0).child(index=1).child(index=0).child(index=0).child(index=0).child(index=0).info
import ddddocr
import time
import os, base64
import random
det = ddddocr.DdddOcr(det=True)
# 滑块
time.sleep(2)
with open(image, 'rb') as f:
image_base64 = str(base64.b64encode(f.read()), encoding='utf-8')
# print(image_base64)
print('截图成功')
poses = det.detection(img=image_base64)
print('len(poses)=%s, poses=%s' % (len(poses),poses))
if len(poses) <= 1:
return False
else:
pass
# [[694, 33, 737, 74]]
# [[691, 29, 740, 80]]
offset_y = - 1
index = 0
source_pose = poses[0][0]
distance = abs(poses[0][0] - poses[1][0])
elem = self.d(text="向右滑动滑块填充拼图").sibling(index=1)
button = elem.info['bounds']
x, y = elem.center()
x = int(x)
y = int(y)
print(button)
py = (button.get('right') - button.get('left')) / 4
print(py)
move = distance + py
print('距离=%s,偏移=%s, 移动距离=%s' % (distance, py, move))
self.d.swipe(x, y, x + distance, y)
return True
def down_android_app(self, apk_name='应用宝', download_flag=0, means = 'go'):
# apk_name = '应用宝'
if "Windows" in self.system:
main_dir = 'D:\\'
project_path = self.read_yaml['win_computer']['project_path']
go_path = self.read_yaml['win_computer']['go_path']
gomodcache_path = self.read_yaml['win_computer']['gomodcache_path']
gocache_path = self.read_yaml['win_computer']['gocache_path']
elif "macOS" in self.system:
print("mac系统")
main_dir = self.read_yaml['mac_computer']['project_path']
project_path = self.read_yaml['mac_computer']['project_path']
go_path = self.read_yaml['main_ubantu_computer']['go_path']
gomodcache_path = self.read_yaml['main_ubantu_computer']['gomodcache_path']
gocache_path = self.read_yaml['main_ubantu_computer']['gocache_path']
else:
main_dir = self.read_yaml['main_ubantu_computer']['share_path']
project_path = self.read_yaml['main_ubantu_computer']['project_path']
go_path = self.read_yaml['main_ubantu_computer']['go_path']
gomodcache_path = self.read_yaml['main_ubantu_computer']['gomodcache_path']
gocache_path = self.read_yaml['main_ubantu_computer']['gocache_path']
path = os.path.join(main_dir, 'android_apps')
if not os.path.exists(path):
os.makedirs(path)
if not os.path.exists(go_path):
means = 'python'
# 方法1：go语言
if means == 'go':
print("go语言")
os.environ['PATH'] = os.path.normpath(go_path)
os.environ['GOMODCACHE'] = gomodcache_path
os.environ['GOCACHE'] = gocache_path
os.environ['GOPROXY'] = 'https://mirrors.aliyun.com/goproxy/'
# 切换到工程目录
os.chdir(project_path)
print(os.getcwd())
def call_go_program(path, app_name):
# go_path
command = ['go', 'run', os.path.join(project_path, 'download_android_app.go'), path, app_name]
result = subprocess.run(command, capture_output=True, text=True, encoding='utf-8')
output = result.stdout.strip()
# console = 'go run %s %s %s' % (os.path.join(project_path, 'download_android_app.go'), path, app_name)
# output = os.popen(console).read()
return output
# 调用 Go 程序，并传递参数
output_data = call_go_program(path, apk_name)
if not output_data:
os.system('go mod init github.com/antchfx')
time.sleep(0.5)
os.system('go get github.com/antchfx/htmlquery')
self.down_android_app(apk_name, means='python')
print('output_data=', output_data)
r_list = re.compile('【(.*?)】').findall(output_data)
if r_list:
path = r_list[0]
if path:
print(path)
return path
else:
return False
else:
return False
else:
print("python语言")
# 方法二：python
file_name = os.path.join(path, apk_name+'.apk')
if os.path.exists(file_name) and download_flag == 0:
print("已下载无需下载%s" % apk_name)
return file_name
else:
print("需要下载%s" % apk_name)
headers = {
"User-Agent": "Mozilla/5.0 (Linux; Android 6.0; Nexus 5 Build/MRA58N) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/113.0.0.0 Mobile Safari/537.36 Edg/113.0.1774.35",
}
params = {
"pageContext": "appdetail",
}
one_url = 'https://sj.qq.com'
search_url = one_url + "/search?q=%s" % apk_name
response = requests.get(search_url, headers=headers)
html = response.text
# print(html)
htmlElement = etree.HTML(html)
result = htmlElement.xpath("//div[text()='"+apk_name+"']/../../..")
if not result:
print("没结果，需要模糊匹配")
elem = "(//div[contains(text(), '"+apk_name+"')])[1]"
result = htmlElement.xpath(elem +"/../../..")
apk_name = htmlElement.xpath(elem)[0].text
print('真实名字=', apk_name)
# 获取文本
if result:
href = result[0].get("href")
one_url += href
print("1.获取app连接%s" % one_url)
try:
response = requests.get(one_url, headers=headers, params=params)
json_data = response.text
except Exception as e:
# print(e)
if 'Expecting value' in str(e) and len(str(e)) < 300:
qq_email("534640040@qq.com", 'reques版本过高' ,"request版本过高，默认2.27.1")
json_data = e
r_list = re.compile('"name":"'+apk_name+'".*?"download_url":"(.*?)"', re.S).findall(str(json_data))
print(r_list)
if r_list:
apk_url = r_list[0]
print("2.获取app下载连接%s" % apk_url)
# 发送GET请求，并获取响应对象
response = requests.get(apk_url, timeout=30)
# 将响应内容写入本地文件
with open(file_name, 'wb') as f:
f.write(response.content)
# 输出下载完成的消息
print(f"APK文件已下载到 {file_name}")
return file_name
return False
def click_element_by_xpath(self, xpath_expression='', d='',click=True, info = ''):
"""
使用 XPath 定位元素并执行点击操作.
Args:
d: uiautomator2 device 对象
xpath_expression: 用于定位元素的 XPath 表达式
"""
if not d:
d = self.d
xml_string = d.dump_hierarchy()
root = etree.fromstring(xml_string.encode('utf-8'))
elements = root.xpath(xpath_expression)
if not elements:
print(f"元素未找到, XPath: {xpath_expression}")
return False
# print(elements[0].get('text'))
if info:
for element in elements:
text = element.get('text')
if text:
bounds_str = element.get('bounds')
match = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds_str)
if not match:
print(f"无法解析 bounds 属性: {bounds_str}")
return False
x1, y1, x2, y2 = map(int, match.groups())
center_x = (x1 + x2) // 2 #计算中心点 x 坐标
center_y = (y1 + y2) // 2 #计算中心点 y 坐标
return {'text': text, 'center': (center_x, center_y)}
element = elements[0]
bounds_str = element.get('bounds') # 获取边界属性，如 [0,0][100,200]
print(f"Element bounds: {bounds_str}")
if click:
# 使用正则表达式从 bounds 字符串中提取坐标
match = re.match(r'\[(\d+),(\d+)\]\[(\d+),(\d+)\]', bounds_str)
if not match:
print(f"无法解析 bounds 属性: {bounds_str}")
return False
x1, y1, x2, y2 = map(int, match.groups())
center_x = (x1 + x2) // 2 #计算中心点 x 坐标
center_y = (y1 + y2) // 2 #计算中心点 y 坐标
print(f"Clicking at coordinates: ({center_x}, {center_y})")
d.click(center_x, center_y)
return True
def get_dump(self, re_text='逛商品赚金币', mode='comp', select=0):
dic = {}
matches = []
# print(self.d.dump_hierarchy())
if mode == 'comp':
pattern = re.compile(' 0 and y > 0:
print("找垃圾清理")
# d(descriptionContains="安装")
# [x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", [], delem=['d(textContains="旧版应用详情页已停止使用")'])
# if x > 0 and y > 0:
# # print(elem.info['text'])
# print("无法安装垃圾清理")
# self.MySQLManager.write_file("%s已统计手机容量" % (self.model + self.ip))
# return
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", [], delem=['d(descriptionContains="安装")', 'd(text="安装")', 'd(text="同意")'])
if x > 0 and y > 0:
if k == 0:
print("没有安装垃圾清理，需要安装")
self.d(description="已安装").exists(timeout=30)
print("已安装")
continue
elif k == 1:
print("外层安装")
self.d(text="同意").exists(timeout=20)
continue
self.d(text="总共扫描出垃圾").exists(timeout=30)
t3 = ''
try:
t1 = self.d(text="总共扫描出垃圾").sibling(index=0).info['text']
t2 = self.d(text="总共扫描出垃圾").sibling(index=1).info['text']
if not t1:
t1 = self.d(resourceId="com.miui.cleanmaster:id/number").info['text']
t3 = t1 + t2
print("共扫描%s垃圾" % t3)
except Exception as e:
print(e)
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", [], delem=['d(resourceId="com.miui.cleanmaster:id/button_clean")', 'd(textContains="清理选中垃圾")'])
if x > 0 and y > 0:
print("找到清理选中垃圾")
try:
self.d(textContains="已用").exists(timeout=30)
t = self.d(textContains="已用")
if not t:
t = self.d(textContains="共")
t = t.info['text']
print(t)
if t3 and t:
text = "['%s', '手机管家清理', '%s']" % ((self.model + self.ip), "共扫描" + t3 + "垃圾" + ' ' + t)
print(text)
self.MySQLManager.write_file(text)
except Exception as e:
print(e)
break
self.multiple_find_picture([], 20, "点击", "不返回", [], delem=['d(text="确定")'])
self.MySQLManager.write_file("%s已统计手机容量" % (self.model + self.ip))
# d(resourceId="com.miui.cleanmaster:id/button_clean")
# 删除adb下安装包：
print('删除adb下安装包')
# self.console_input('shell find /sdcard -name "*.apk" -type f -delete')
ret = self.console_input('shell find /storage/emulated -name "*.apk"', ret=1)
texts = ret.split('\n')
for text in texts:
if 'Permission denied' not in text:
print("删除%s" % text)
self.console_input('shell rm %s' % text)
print('删除结束')
def task_done(self, r='', w='', user=''):
if not user:
user = self.tao_user + self.project
if not user:
return False
# 这个函数可以用来判断任务是否已经完成
try:
if r:
have = self.MySQLManager.sql_search("content", "%s已经完成%s任务" % (user.replace("*", "\\\\*"), r))
if have:
return True # 在这个例子中假设任务已经完成
else:
return False # 在这个例子中假设任务未完成
if w:
self.MySQLManager.write_file("%s已经完成%s任务" % (user, w))
except Exception as e:
logging.error(e)
def ui_layout(self, name='获取当前界面的布局层次结构', d=None, mode='', path=''):
try:
if self.d:
d = self.d
# 获取UI层次结构
hierarchy_xml = d.dump_hierarchy()
# 准备要写入的代码部分
code_intro = '\nhierarchy_xml = """'
code_outro = '"""'
# XML解析部分代码
parse_and_query_code = (
'\ndef convert_xpath_lxml_to_uiautomator2(lxml_xpath):\n'
'\t# 将XPath表达式拆分为多个部分，基于\']\'\n'
'\timport re\n'
'\tdef replacer(match):\n'
'\t\tclass_name = match.group(1)\n'
'\t\treturn class_name\n'
'\t# 替换所有匹配的部分\n'
'\tconverted_xpath = re.sub(r\'\\*\\[@class="([^"]+)"\\]\', replacer, lxml_xpath)\n'
'\treturn converted_xpath\n'
'from lxml import etree\n'
'# 解析XML结构\n'
'tree = etree.XML(hierarchy_xml.encode(\'utf-8\'))\n'
'# 示例用法(不要写具体标签名，用*代替，如//*[])：\n'
'lxml_xpath = \'//*[@class="android.widget.RelativeLayout"]/*\'\n'
'# 使用XPath查询具有特定资源ID的TextView元素\n'
'results = tree.xpath(lxml_xpath)\n'
'for result in results:\n'
'\tprint(\'Text:\', result.attrib[\'text\'])\n'
'\tprint(\'Bounds:\', result.attrib[\'bounds\'])\n'
'uiautomator2_xpath = convert_xpath_lxml_to_uiautomator2(lxml_xpath)\n'
'print(uiautomator2_xpath)\n'
)
# 将所有部分拼接到一起以形成最终的字符串
full_code_string = code_intro + hierarchy_xml + code_outro + parse_and_query_code
html_content = """
""" % full_code_string
with allure.step(name):
# Optional: 将层次结构信息附加到步骤
allure.attach(html_content, name=name, attachment_type=allure.attachment_type.HTML)
if not path:
path = self.screen('phonetimeout/' + self.model + 'ui.png')
allure.attach.file(path,name=name, attachment_type=allure.attachment_type.PNG)
if '失败' in mode:
assert False, mode
elif '跳过' in mode:
pytest.skip(mode)
elif '损坏' in mode:
raise Exception(mode)
except Exception as e:
logging.error(e)
def get_attribute_value(self, obj, attribute):
try:
value = ast.literal_eval(f'obj.{attribute}')
return value
except (SyntaxError, ValueError):
return None
def open_app1(self):
self.multiple_find_picture([], 10, "点击", "不返回", delem=['d(resourceId="android.miui:id/app1")'], xy_percentage=True)
def browse_task(self, key_list=['d(textContains="浏览完成")', 'd(description="任务完成")','d(text="任务完成")'], times=15, img: list = (), gep=0):
for i in range(times):
if i < 5:
[x, y], k = self.multiple_find_picture([], 1, "点击", "不返回", delem=['d(text="打开")'])
if gep:
if i % gep != 0:
# 多少次找一次元素
self.d.swipe_ext("up", random.uniform(0.4, 1))
continue
time.sleep(random.uniform(2, 3))
# 浏览完成全部完成
[x, y], k = self.multiple_find_picture(img, 2, "不点击", "下滑",delem=key_list, is_print=False)
if x > 0 and y > 0:
break
def back_task(self, key_dic={'d(text="每日任务")':"break", 'd(text="饿了么果园")': '点击', 'd(text="领水滴")':"点击"}, app_name=''):
# d(text="每日任务")
keys_list = list(key_dic.keys())
for i in range(6):
[x, y], k = self.multiple_find_picture([], 2, "不点击", "不返回",delem=keys_list)
if x > 0 and y > 0:
value = key_dic.get(keys_list[k])
if value == '点击':
self.click(x,y)
time.sleep(5)
elif value == 'break':
return True
elif value == '下滑':
# 无需返回
# print("下滑")
self.d.swipe_points([[0.8, 0.6], [0.8, 0.1]], 0.1)
time.sleep(1)
continue
elif value == '右滑':
# 无需返回
# print("下滑")
self.d.swipe_points([[0.8, y], [0.1, y]], 0.1)
time.sleep(1)
continue
# print("app_name:", app_name)
else:
self.back()
time.sleep(2)
if not app_name:
app_name = self.app_name
if i >= 4 and app_name:
get_app_name = self.get_app_name()
if get_app_name != app_name:
print("不是本应用%s" % app_name)
self.d.app_start(app_name)
time.sleep(2)
def calculate_sliding_distance(self, elem, y1=0.6, y2=0.55):
# 计算距离确认是否有弹窗
if elem:
center = elem.center()
y1_loc = center[1]
self.d.swipe_points([[0.8, y1], [0.8, y2]], 0.1)
time.sleep(1)
center = elem.center()
y2_loc = center[1]
distance = y1_loc - y2_loc
# print(distance)
if distance == 0:
return False
else:
return distance
else:
return False
def get_mobile_ipv6(self, ipv6_prefix="2409"):
# 获取ipv6地址，已知小米13可以，其他手机没ipv6
# 方法2：adb shell "curl -s https://ipv6.ddnspod.com"， 这个测试可以使用的ipv6
text = self.console_input('shell ip -6 addr show', ret=1)
r_list = re.compile("(%s:\\S+)/64" % ipv6_prefix, re.S).findall(text)
if r_list:
print(r_list[0])
return r_list[0]
def skip(self, z=3):
if self.z >= z:
pytest.skip("超过最大失败次数")
ret = self.console_input('shell pm list packages', ret=1)
if self.app_name not in ret:
pytest.skip("没有安装app" + self.app_name)
if not self.tao_user and not self.user_name:
pytest.skip("没有安装获取到用户名" + self.model)
def open_app(self, access_elem=[], name_elem=None,ad_elem=[], sleep_time = 1):
for i in range(5):
self.d.app_stop(self.app_name)
time.sleep(1)
print(self.app_name, self.activity)
if self.activity:
self.d.app_start(self.app_name, activity=self.activity)
else:
self.d.app_start(self.app_name)
time.sleep(sleep_time)
if not self.tao_user:
print(" %s没有用户名需要获取用户名" % self.model)
# d(resourceId="com.wuba:id/btn_ads")
if ad_elem:
self.multiple_find_picture([], 15, "点击", "不返回", [], delem=ad_elem, xy_percentage=True)
time.sleep(2)
# d.xpath('//*[@resource-id="com.jifen.qukan:id/app"]/android.widget.RelativeLayout[1]/android.widget.ImageView[1]')
# 点击叉
[x, y], k = self.multiple_find_picture([], 20, "点击", "返回", [], delem=access_elem, xy_percentage=True)
if x > 0 and y > 0:
print("找到我的")
for i in range(4):
try:
# d(resourceId="com.xunmeng.pinduoduo:id/pdd", text="yys534640040.club")
if type(name_elem) == list:
index = i % len(name_elem)
# print('index=', index)
elem = name_elem[index]
else:
elem = name_elem
if elem.exists:
text = elem.info['text']
if not text:
# contentDescription
text = elem.info['contentDescription']
self.tao_user = text.strip()
print(self.tao_user)
os.environ[self.ip] = str(self.tao_user)
time.sleep(1)
# self.multiple_find_picture([], 10, "点击", "不返回", [], delem=['d(text="取消")'], xy_percentage=True)
# time.sleep(1)
break
time.sleep(1)
except Exception as e:
print(e)
else:
break
if not self.tao_user:
self.tb_timeout_option("找用户名失败")
return False
else:
break
def do_task_for_list(self,
key_list=['去完成', '去领取'],
title1_elem='sibling(index=0)',
title2_elem = 'sibling(index=1)',
app_name = 'com.eg.android.AlipayGphone',
browse_key = {'关键字':['浏览', '指定动作奖励', '逛'], '成功元素':['d(textContains="浏览完成")', 'd(description="任务完成")','d(text="任务完成")'], '成功图片':[], '浏览次数':10},
back_key_dict = {'d(text="每日任务")':"break", 'd(text="饿了么果园")': '点击', 'd(text="领水滴")':"点击"},
get_rewards=['d(text="领取")', 'd(text="领奖")'],
exclude_list=['多少运动币','外卖实付', '3个店铺'],
watch_vd={},
other_task_dic={},
do_times=1,
y_per_max=0.94,
y_per_min=0.3,
task_name='',
delay = 6
):
# get_rewards： 任务完成后领取奖励按钮
# back_key_dict: 返回按钮
# watch_vd看视频的，如{'关键字':["看广告视频", '每日可完成7次'], "方法":self.watch_tv}
# other_task_dic比如other_task_dic的扩展，多个字典{'惊喜水滴':self.other_task_dic}
text2 = ''
do_task = 0 # 没有一个满足key_list 则跳出循环
return_dic = {} # 统计没有完成的key_list
max_ins = 5
for ins in range(15):
# 初始索引到0-3，如果遇到不做的任务获准，做过的还有+1
for key_word in key_list:
time.sleep(0.5)
if ins <= 0:
# 返回到做任务位置
is_true = self.back_task(key_dic=back_key_dict, app_name=app_name)
if not is_true:
return
ins %= max_ins
print('key_word=%s, ins=%s' % (key_word, ins))
try:
# 计算return_dic
all = len(self.d.xpath('//*[@text="%s"]' % key_word).all())
if all:
return_dic[key_word] = all
else:
if key_word in return_dic:
del return_dic[key_word]
pass
except Exception as e:
pass
# 查看是不是本应用
get_app_name = self.get_app_name()
if get_app_name != app_name:
print("不是本应用%s" % app_name)
self.d.app_start(app_name)
time.sleep(2)
try:
if '[%s]' in title1_elem:
# print('2')
click_elem = self.XPathElement(self.d, title1_elem % (key_word, ins + 1))
else:
click_elem = self.d(text=key_word, instance=ins)
# print(2, click_elem)
if click_elem:
is_numeric = lambda s: s.replace('.', '', 1).isdigit() if s.count('.') <= 1 else False
if is_numeric(str(title1_elem).replace('-', '')) and is_numeric(str(title2_elem).replace('-', '')):
# print('3')
# 根据索引判断，is_numeric检查是否数字
ret = self.get_dump(re_text=key_word, select=ins)
index = ret.get('index') + title1_elem
text1 = self.d(index=index).info['text']
# title2 = eval(f"click_elem.{title2_elem}")
index = ret.get('index') + title2_elem
text2 = self.d(index=index).info['text']
# print(text1, text2)
elif '//*' in title1_elem or '//com' in title1_elem:
# xpath定位的情况
# print(self.d.xpath(title1_elem % key_word).all()[ins].info)
if '[%s]' in title1_elem:
# 带索引的xpath
# self.XPathElement(self.d,
title1 = self.XPathElement(self.d, title1_elem % (key_word, ins + 1))
text1 = title1.info['text'] if title1 else ''
if title1_elem == title2_elem: # 相等就不需要获取标题2了
text2 = ''
else:
title2 = self.XPathElement(self.d, title2_elem % (key_word, ins + 1))
text2 = title2.info['text'] if title2 else ''
else:
title1 = self.d.xpath(title1_elem % key_word).all()[ins * 2]
text1 = title1.info['text'] if title1 else ''
if title1_elem == title2_elem: # 相等就不需要获取标题2了
text2 = ''
else:
title2 = self.d.xpath(title2_elem % key_word).all()[ins * 3]
text2 = title2.info['text'] if title2 else ''
else:
# print("2323")
# 使用getattr解析方法调用表达式字符串，获取兄弟元素
title1 = eval(f"click_elem.{title1_elem}")
text1 = title1.info['text'] if title1 else ''
title2 = eval(f"click_elem.{title2_elem}")
text2 = title2.info['text'] if title2 else ''
# print(text1, text2)
text = text1 + text2
if any(i in text for i in exclude_list):
print("排除任务【%s】" % text)
continue
user_search = self.tao_user
# text_search = text.replace("(", "").replace(")", "").replace("+", "").replace(' ', '').replace('*', '')
# print(text_search)
text_search = "%s已经完成%s任务" % (user_search + self.project, text)
value = self.redis_manager.redis_search(text_search)
print('value:', value,text)
# print()
# value = self.MySQLManager.sql_search("content", "%s已经完成%s任务" % (user_search + self.project, text_search))
if value and int(value.decode('utf-8')) >= do_times:
print("已经做过【%s】无需再做" % text)
continue
width, height = self.get_phone_width_height()
# print(6)
# print(click_elem.info)
y_per = click_elem.center()[1] / height
if y_per < y_per_min:
print(y_per, "下滑")
self.d.swipe_ext("down", 0.5)
time.sleep(1)
continue
elif y_per > y_per_max:
print(y_per, "上滑")
self.d.swipe_ext("up", 0.5)
continue
else:
print("无需滑动")
print(y_per)
do_task = 1
time.sleep(3)
click_elem.click() # 点击词
time.sleep(1)
self.redis_manager.redis_set(text_search,times='today', is_countable=True)
# self.MySQLManager.write_file("%s已经完成%s任务" % (self.tao_user + self.project, text_search))
# 其他任务
# other_task_dic
flag_other = 0
if other_task_dic:
for action, params in other_task_dic.items():
if action in text:
# print( action, params)
try:
if len(params) == 1:
# 没有参数，直接调用函数
params[0]()
else:
# 有参数，解包元组后调用
params[0](*params[1:])
except Exception as e:
params()
flag_other = 1
# for i in other_task_dic:
# if i in text:
# dic = other_task_dic.get(i)
# dic()
if not flag_other:
# 包含这些关键词的不需要再做了
# 浏览任务
if browse_key:
key = browse_key.get('关键字')
if key:
succeed_key = browse_key.get('成功元素')
succeed_pic = browse_key.get('成功图片')
browse_times = browse_key.get('浏览次数')
if key == '无' or any(i in text for i in key):
self.browse_task(key_list=succeed_key, img=succeed_pic, times=browse_times)
# 视频，广告
if watch_vd:
key = watch_vd.get('关键字')
if any(i in text for i in key):
mean = watch_vd.get('方法')
mean()
# 直播
time.sleep(delay)
# 返回
self.back_task(key_dic=back_key_dict, app_name=app_name)
# 领取奖励
# pass
if get_rewards:
num = 4
flge = 0
for i in range(num):
delem_list = []
pic_list = []
for type_elem in get_rewards:
if 'd' in type_elem:
delem_list.append(type_elem)
else:
pic_list.append(type_elem)
[x, y], k = self.multiple_find_picture(pic_list, 3, "不点击", "不返回", delem=delem_list,xy_percentage=True)
if x > 0 and y > 0:
if y > 0.9:
print(y, "上滑")
if flge == 0:
num += 5
flge = 1
self.d.swipe_ext("up", 0.5)
else:
self.click(x, y)
time.sleep(4)
pass
else:
break
if flge == 1:
for i in range(3):
print("下滑")
self.d.swipe_ext("down", 0.5)
except Exception as e:
print(e)
if do_task == 0:
self.d.swipe_ext("up", 0.5)
if do_task == 0 and ins >= max_ins - 1:
break
if return_dic:
try:
# back_key_dict
result_string = ""
for key, value in return_dic.items():
result_string += f"{key}: {value}, "
result_string = result_string.rstrip(', ')
if back_key_dict:
result_string += "_返回关键字：" +str(list(back_key_dict.keys())[0]).replace('"', '').replace("'", '')
print("['%s','剩余任务次数','%s']" % (self.model + self.tao_user + self.project + task_name, result_string))
self.MySQLManager.write_file("['%s','剩余任务次数','%s']" % (self.model + self.tao_user + self.project + task_name, result_string))
except Exception as e:
logging.error(e)
def click_vd_cross(self, className="android.widget.ImageView", min_x=0.8, max_y = 0.1):
for i in range(10):
elem = self.d(resourceId="android:id/content").child(className=className, instance=i)
if elem:
center = elem.center()
width, height = self.get_phone_width_height()
y_per = center[1] / height
x_per = center[0] / width
# print(x_per, y_per)
if x_per > min_x and y_per < max_y:
# 叉子在左上角
print('点击叉', x_per, y_per)
self.click(x_per, y_per)
break
def concatenate_images(self, image_paths, output_path):
from PIL import Image
# 打开所有图片并获取它们的尺寸
images = [Image.open(path) for path in image_paths]
widths, heights = zip(*(img.size for img in images))
# 计算拼接后的总宽度和最大高度
total_width = sum(widths)
max_height = max(heights)
# 创建一个新图像，尺寸为拼接后图像的尺寸
concatenated_image = Image.new('RGB', (total_width, max_height))
# 将每张图片粘贴到新图像上
x_offset = 0
for img in images:
concatenated_image.paste(img, (x_offset, 0))
x_offset += img.width
# 保存拼接后的图片
concatenated_image.save(output_path)
return output_path
def click_ok_button(self):
# git branch -d temp
# '//*[@text="确定"]'
# d.xpath('//*[@resource-id="android:id/button1"]')
self.allure_screen("确定前")
[x, y], k = self.multiple_find_picture([], 5, "点击", "不返回", xpath=['//*[@text="确定"]', '//*[@resource-id="android:id/button1"]'], xy_percentage=True)
if x > 0 and y > 0:
print("找到确定")
self.allure_screen("确定后")
def clear_app(self, app_name="com.jifen.qukan"):
ret = self.console_input('shell pm clear %s' % app_name, ret=1)
print('清除app%s数据' % app_name, ret)
def yzm(self, times=8, swip=False):
# d(text="请按照说明拖动滑块")
mx = 0
i = 0
t1 = time.time()
distance = 0
all_width = 1
for i in range(times + random.randint(0, 5)):
try:
[x, y], k = self.multiple_find_picture([], 10, "不点击", "不返回", delem=['d(resourceId="scratch-captcha-btn")', 'd(text="请按照说明拖动滑块")', 'd(text="验证码拦截")', 'd(text="访问被拒绝")'], xy_percentage=True)
if x > 0 and y > 0:
if k == 3:
# 淘宝解决方案0
self.multiple_find_picture(['淘宝解决方案'], 2, "点击", "不返回")
if swip:
# get_canny
# '淘宝验证码滑块按钮'
[x, y], k = self.multiple_find_picture(['淘宝操作太频繁了','淘宝验证码滑块按钮'], 5, "不点击", "不返回", delem=['d(resourceId="scratch-captcha-btn")'])
if x > 0 and y > 0:
if k == 0:
return False
all_width, all_height = self.get_phone_width_height()
width = all_width * 0.63
self.d.touch.down(x, y)
time.sleep(0.3)
self.d.touch.move(x+ width, y)
time.sleep(0.5)
# self.swipe(x, y, x + width, y)
print("向右滑动总距离(方便截全图):", x + width)
# 0.03 ,0.24
# # （y1（上）:y2(下), x1（左）:x2（右）） 0.37124999999999997 0.5812499999999999 0.1550925925925926 0.892
# print(y / all_height - 0.24, y / all_height - 0.01, x / all_width, 0.892, y / all_height)
pic_path = self.screen(pic_name = self.ip + 'yzm.png', region=(y / all_height - 0.24, y / all_height - 0.02, x / all_width, 0.892))
x_list, path = self.find_and_show_rightmost_points(pic_path)
# print(x_list)
mx = random.choice(x_list)
# print(i, ",回退,", x + mx, '占比', (mx / width) / width, ', mx=', mx)
# 可以+一定距离(如x + mx + 10),我去掉了,看着能用
distance = x + mx
print(i, '.图片选择的相对距离mx(不算按钮距离)',mx, ',绝对距离按钮',distance, ",swipe距离", x + width, y, distance , y, ', swipe_points距离比例:', [[(x + width)/all_width, y / all_height], [distance / all_width, y / all_height]])
# self.swipe(x + width, y, distance , y)
# time.sleep(4)
self.d.swipe_points([[(x + width)/all_width, y / all_height], [distance / all_width, y / all_height]], random.uniform(0.1, 0.8))
self.d.touch.up(width - mx, y)
# self.d.swipe_points([[x, y], [x + mx, y]], random.uniform(0.1, 1))
time.sleep(1)
pic_path = self.screen(pic_name = 'phonetimeout/' +self.ip + 'yzm.png')
ret_path = self.concatenate_images([pic_path, path], self.screen_path + 'phonetimeout/' + self.ip + '合并.png')
self.ui_layout(name='canny' + str(i) + "_SelectX_" + str(mx) + '_cost_%.2f' % (time.time() - t1), path=ret_path)
num = random.randint(0, 2)
# print(num)
if num == 1:
# 验证码点我反馈0
[x, y], k = self.multiple_find_picture(['验证码点我反馈'], 3, "点击", "不返回", xy_percentage=True)
if x > 0 and y > 0:
[x, y], k = self.multiple_find_picture([], 10, "点击", "不返回", delem=[random.choice(['d(text="滑动验证码后提示验证失败")', 'd(text="频繁看到该验证码")'])], xy_percentage=True)
if x > 0 and y > 0:
[x, y], k = self.multiple_find_picture([], 10, "点击", "不返回", delem=['d(text="提交")'], xy_percentage=True)
self.back()
time.sleep(5)
else:
try:
ret = [distance / all_width, distance,i, time.time() - t1]
if ret[0]:
result = '距离比例%s, 距离%s,滑动次数%s,耗时%.2f秒' % (ret[0], ret[1], ret[2], ret[3])
print("['%s','淘宝滑动验证码结果','%s']" % (self.model + self.tao_user + self.project, result))
self.MySQLManager.write_file("['%s','淘宝滑动验证码结果','%s']" % (self.model + self.tao_user + self.project, result))
self.ui_layout(name=result)
except Exception as e:
print(e)
return ret
except Exception as e:
print(e)
def find_and_show_rightmost_points(self,image_path='/Users/yangyongsheng/nas/download/23.png', num_regions=5):
# image_path = '/Users/yangyongsheng/nas/download/23.png'
# 读取图像
img = cv.imread(image_path)
if img is None:
raise ValueError("无法读取图像，请检查图像路径。")
# 转换为灰度图像
gray = cv.cvtColor(img, cv.COLOR_BGR2GRAY)
# 应用高斯模糊
blurred = cv.GaussianBlur(gray, (5, 5), 0)
# 应用Canny边缘检测
edges = cv.Canny(blurred, 50, 150)
# 找到连通区域
num_labels, labels, stats, centroids = cv.connectedComponentsWithStats(edges, connectivity=8)
# 排除背景，按面积排序，选出前num_regions个最大连通区域
largest_regions = np.argsort(stats[1:, cv.CC_STAT_AREA])[::-1][:num_regions] + 1
rightmost_points = []
# 计算每个连通区域的最右边点
for region in largest_regions:
mask = (labels == region).astype(np.uint8)
contours, _ = cv.findContours(mask, cv.RETR_EXTERNAL, cv.CHAIN_APPROX_SIMPLE)
if contours:
rightmost_point = max(contours[0], key=lambda point: point[0][0])
rightmost_points.append(tuple(rightmost_point[0]))
# 打印并标注最右边点
x_lis = []
for i, point in enumerate(rightmost_points):
print(f"连通区域 {i+1} 的最右边点坐标: {point}")
# 在图像上标注最右边点
cv.circle(img, point, 5, (0, 255, 0), -1)
cv.putText(img, f"{point}", (point[0] + 10, point[1]), cv.FONT_HERSHEY_SIMPLEX, 0.5, (0, 255, 0), 1)
x_lis.append(point[0])
# 展示图像
# plt.figure(figsize=(10, 10))
# plt.imshow(cv.cvtColor(img, cv.COLOR_BGR2RGB))
# plt.title("Rightmost Points")
# plt.show()
save_path = image_path.replace('.png', '_marked.png')
cv.imwrite(save_path, img)
print(f"标记后的图像已保存到 {save_path}")
# print(x_lis)
try:
move = max(x_lis)
x_lis.remove(move)
except Exception as e:
print(e)
print(x_lis)
return x_lis, save_path
def get_appsize(self):
# print(self.model, self.ip, "获取应用大小")
catch_key = '%s今天已经统计应用大小' % (self.model + self.ip)
have = self.redis_manager.redis_search(catch_key)
print(catch_key, have)
if have or have == b'':
return
# 小米应用商店
[x, y], k = self.multiple_find_picture([], 10, "点击", "返回", delem=['d(text="我的")'])
if x > 0 and y > 0:
[x, y], k = self.multiple_find_picture([], 6, "点击", "不返回", [], delem=['d(text="应用卸载")'])
if x > 0 and y > 0:
# elem = self.d(text="应用卸载")
self.d(text="应用卸载").exists(timeout=3)
# elem_zise = self.d(text="应用大小")
# if not elem_zise:
time.sleep(2)
# 点击一下有用
try:
self.d(resourceId="android:id/icon1")
except Exception as e:
self.d(resourceId="com.xiaomi.market:id/sort_mode_chooser").click()
time.sleep(2)
try:
self.d(resourceId="android:id/text1", text="应用大小")
except Exception as e:
self.d(text="应用大小").click()
d = self.d
time.sleep(4)
# 获取 hierarchy
# 示例 XPath 查询
# 1. 根据 class 定位所有 TextView 元素
# d.xpath('//*[contains(@text="大小: ")]')
text = ''
for i in range(10):
[x, y], k = self.multiple_find_picture([], 2, "不点击", "不返回", [],
delem=['d(textContains="其他应用（")'], xy_percentage=True)
if x > 0 and y > 0 :
if y < 0.2:
break
print('其他应用（')
self.d.swipe(0.5, 0.3 + random.uniform(-0.05, 0.1), 0.5, 0.2 + random.uniform(-0.05, 0.1))
time.sleep(0.5)
else:
print("没有找到其他应用（，停止上滑")
break
elem = self.d(resourceId="android:id/text1")
if elem:
text = elem.info['text']
print(text)
if text and text != '应用大小':
print("点击应用大小")
elem.click()
elem = self.d(text="应用大小")
if elem:
elem.click()
time.sleep(3)
print("获取 hierarchy")
xml_string = d.dump_hierarchy()
# print()
# 解析 XML
root = etree.fromstring(xml_string.encode('utf-8'))
text_views = root.xpath('//node[contains(@text, "大小: ")]')
for text_view in text_views:
sibings = text_view.getparent().getchildren()
size = text_view.get('text')
# print(f"TextView Text: {size}")
preceding_sibling = text_view.xpath('./preceding-sibling::node[1]')
if preceding_sibling:
app_name = preceding_sibling[0].get('text')
# print(f"{app_name}{size}")
text += f"{app_name} {size}
"
# print(text)
if text:
print("['%s','应用大小排行','%s']" % (self.model + self.ip, text))
self.MySQLManager.write_file("['%s','应用大小排行','%s']" % (self.model + self.ip, text))
self.redis_manager.redis_set(catch_key, times='today')
# 内存占用
# adb -s 192.168.31.182 shell "dumpsys meminfo | head -20" | awk '
# /K:/ {
# gsub(",", "", $1)
# gsub("K:", "", $1)
# mb = int($1 / 1024)
# printf "%6d MB: ", mb
# for(i=2; i<=NF; i++) printf "%s ", $i
# print ""
# next
# }
# { print }
# '
ret_oneline = self.get_phone_meminfo( top_n=20)
if ret_oneline:
print(ret_oneline)
# 写库前压缩成单行，避免换行符破坏SQL
self.MySQLManager.write_file(
"['%s','内存占用','%s']" % (self.model + self.ip, ret_oneline)
)
def get_phone_meminfo(self, top_n=20) -> str:
"""
获取手机内存排行，自动适配 Linux / Windows。
返回已换算为 MB 的字符串，失败返回空字符串。
"""
# Android shell 里的 pipe 由手机端执行，Linux/Win 均可用
raw = self.console_input(f'shell "dumpsys meminfo | head -{top_n}"', ret=1)
# Windows 上 adb 有时返回 bytes，做一次保险解码
if isinstance(raw, bytes):
raw = raw.decode('utf-8', errors='replace')
if not raw or 'Applications Memory' not in raw:
# 降级：不带 head，Python 侧截断（head 在极少数 Android 版本不存在）
raw = self.console_input('shell dumpsys meminfo', ret=1)
if isinstance(raw, bytes):
raw = raw.decode('utf-8', errors='replace')
if raw:
raw = '\n'.join(raw.splitlines()[:top_n])
if not raw:
return ''
lines = []
for line in raw.splitlines():
m = re.match(r'^\s+([\d,]+)K:(.*)', line)
if m:
mb = int(m.group(1).replace(',', '')) // 1024
lines.append('%6d MB:%s' % (mb, m.group(2)))
else:
lines.append(line)
result = '\n'.join(lines)
return result.replace('\n', ' | ')
def kill_heavy_memory_apps(self, top_n=20, threshold_mb=100, whitelist=None):
"""
获取内存排行，强制停止占用超过 threshold_mb 的非系统 App。
whitelist: 额外保护的包名列表（不停止）
"""
# 系统级/必要进程白名单（不可停止）
DEFAULT_WHITELIST = {
'system', 'app_process', 'com.android.systemui',
'com.android.phone', 'com.android.nfc',
'com.android.bluetooth', 'com.miui.home',
'com.miui.daemon', 'com.xiaomi.finddevice', 'com.tencent.mm',
'com.example.myaccessibility', # 你自己的无障碍服务
}
if whitelist:
DEFAULT_WHITELIST.update(whitelist)
raw = self.get_phone_meminfo(top_n=top_n)
if not raw:
print("get_phone_meminfo 返回空，跳过")
return
# 解析 " 621 MB: com.miui.personalassistant (pid 7300)"
MB_RE = re.compile(r'(\d+)\s+MB:\s+(\S+?)(?::[\w.]+)?\s+\(pid')
stopped = []
skipped = []
for seg in raw.split('|'):
m = MB_RE.search(seg)
if not m:
continue
mb = int(m.group(1))
pkg = m.group(2).split(':')[0] # 去掉 :子进程 后缀
if mb < threshold_mb:
continue # 内存低于阈值，不管
if pkg in DEFAULT_WHITELIST:
skipped.append((pkg, mb))
continue
try:
self.console_input(f'shell am force-stop {pkg}')
print(f"已停止: {pkg} ({mb} MB)")
stopped.append((pkg, mb))
except Exception as e:
print(f"停止 {pkg} 失败: {e}")
print(f"\n共停止 {len(stopped)} 个 App，跳过 {len(skipped)} 个系统/白名单进程")
for pkg, mb in stopped:
print(f" ✓ {pkg} {mb} MB")
for pkg, mb in skipped:
print(f" - {pkg} {mb} MB (保留)")
return stopped
def stop_background_apps(self, ip, port):
from concurrent.futures import ThreadPoolExecutor
# 构建adb连接命令
adb_connect_command = f"{self.adb_path} connect {ip}:{port}"
subprocess.run(adb_connect_command, shell=True)
# 获取后台应用列表
get_background_apps_command = self.adb_path + " %s:%s shell dumpsys activity activities" % (ip, port.replace(":", ''))
print(get_background_apps_command)
# result = subprocess.run(get_background_apps_command, shell=True, capture_output=True, text=True)
result = os.popen(get_background_apps_command).read()
print(result)
# 使用正则表达式获取包名
pattern = re.compile(r'ActivityRecord{.*? u0 (.*?)/')
matches = pattern.findall(result.stdout)
print("后台应用列表：", matches)
# 停止所有后台应用
def stop_single_app(package_name):
if "com.miui.home" not in package_name and "com.google.android.systemui" not in package_name:
stop_app_command = f"{self.adb_path} {ip}:{port} shell am force-stop {package_name}"
subprocess.run(stop_app_command, shell=True)
print(f"应用 {package_name} 已停止")
# 使用多线程并行停止应用
with ThreadPoolExecutor() as executor:
executor.map(stop_single_app, matches)
def open_accessibilit(self, ip=''):
if ip:
self.ip = ip
self.port = "5555"
self.read()
# adb -s 192.168.31.182 shell am start -a android.settings.ACCESSIBILITY_SETTINGS
self.console_input("shell am start -a android.settings.ACCESSIBILITY_SETTINGS")
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", delem=['d(text="已下载的应用")', 'd(text="已下载的服务")', 'd(text="更多已下载的服务")'])
if x > 0 and y > 0:
time.sleep(0.5)
Accessibility_name = '我的Accessibility'
text_list = []
elem2 = self.d(text="我的Accessibility").sibling(resourceId="android:id/summary")
if elem2.exists:
text = elem2.info["text"]
# print(text)
text_list.append(text)
elem = self.d.xpath('//*[@text="%s"]/parent::*/following-sibling::*//android.widget.TextView' % Accessibility_name)
if elem.exists:
text = elem.info["text"]
text_list.append(text)
# print(text)
self.allure_screen("是否关闭Accessibility")
print(text_list)
if '关闭' in text_list or "已关闭" in text_list:
print("打开Accessibility")
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", delem=['d(text="%s")' % Accessibility_name])
if x > 0 and y > 0:
time.sleep(1)
[x, y], k = self.multiple_find_picture([], 20, "点击", "不返回", delem=['d(resourceId="android:id/switch_widget")', 'd(resourceId="android:id/widget_frame")'])
if x > 0 and y > 0:
for i in range(12):
[x, y], k = self.multiple_find_picture([], 10, "不点击", "不返回", delem=['d(text="确定")'])
if x > 0 and y > 0:
# d(resourceId="com.miui.securitycenter:id/check_box")
self.d(resourceId="com.miui.securitycenter:id/check_box").click()
self.d(text="确定").click()
time.sleep(1)
self.d.press("home")
break
else:
time.sleep(1)
def inspect_power(self, ip, port, threshold=40):
try:
if ip:
self.ip = ip
if port:
self.port = port
self.read()
output = self.console_input('shell dumpsys battery', ret=1)
match = re.search(r'level: (\d+)', output)
power = match.group(1)
int_power = int(power)
print(self.ip, '电量', int_power)
if int_power <= threshold and self.ip != self.main_mobile_ip:
if self.d:
self.d.screen_off() # 关闭屏幕
self.create_cron_switch(40, mip=self.ip, this_power=str(int_power))
print("%s电量过低,等待40分钟充电" % self.ip)
# time.sleep(1800)
else:
print("不满足充电")
except Exception as e:
print(e)
def close_mobile(self, ip='', port=''):
if ip:
self.ip = ip
if port:
self.port = ':' + port
self.read()
# print(1)
# 1.关闭所有app
# 使用adb命令关闭所有后台进程
try:
self.stop_background_apps(ip, port)
except Exception as e:
print(e)
# print(2)
# 2.回到桌面
self.d.press("home")
# 3.获取手机电量
self.inspect_power(ip, port)
time.sleep(1)
# 4.关闭屏幕
self.d.screen_off()
def open_location(self, app_name):
# 打开定位
self.console_input("shell pm grant %s android.permission.ACCESS_FINE_LOCATION" % app_name)
def china_yzm(self):
for i in range(50):
path = self.screen("yzm/%s.jpg" % (self.ip + 'qm4' + str(i)), elem=self.d(resourceId="captcha").child(className="android.view.View").child(className="android.view.View"))
print(path)
ret = pytesseract_pic_to_text(path, 'chinese')
print(ret)
self.d(text="换一换").click()
time.sleep(2)
# path = '
' % path
# path = self.screen("phonetimeout/%s.jpg" % (self.ip + 'qm'))
# path = '
' % path
# 请依次点击: "虞" "庖" "快" "联"
# d(text="换一换")
# qq_email("", "%s%s七猫%s满足10元提现，请手动提现，有验证码" % (self.model, self.u2c.tao_user, flag), path)
pass
def get_allphone_temperature_max(self):
win_mobile_ips = self.read_yaml['win_mobile_ip']
temperature_list = []
for dic in win_mobile_ips:
mip = list(dic.values())[0]
mport = ':5555'
device_addr = f"{mip}{mport}"
if mip == self.main_mobile_ip:
print("不算主力手机")
continue
try:
# 1. 使用 'grep' 替代 'findstr'
# 2. 将连接和执行命令分开，方便调试
connect_cmd = f"{self.adb_path} connect {device_addr}"
subprocess.run(connect_cmd, shell=True, check=True, capture_output=True, timeout=5)
temp_cmd = f"{self.adb_path} -s {device_addr} shell dumpsys battery"
process = subprocess.Popen(temp_cmd, shell=True, stdout=subprocess.PIPE, stderr=subprocess.PIPE)
output_bytes, error_bytes = process.communicate(timeout=10)
# 2. 将命令输出的字节串(bytes)解码(decode)为字符串(string)
output_str = output_bytes.decode('utf-8')
error_str = error_bytes.decode('utf-8')
if error_str:
print(f"命令执行出错 for {mip}: {error_str}")
continue
# 3. 在搜索前先检查是否匹配成功，避免AttributeError
match = re.search(r'temperature:\s*(\d+)', output_str)
if match:
temperature_raw = match.group(1)
temperature = int(temperature_raw) / 10
temperature_list.append(temperature)
print(f"成功获取 {mip} 的温度: {temperature}°C")
else:
print(f"未能从 {mip} 的输出中找到温度信息")
except subprocess.TimeoutExpired:
print(f"连接或执行命令超时 for {mip}")
except subprocess.CalledProcessError as e:
print(f"ADB连接失败 for {mip}: {e.stderr.decode('utf-8')}")
except Exception as e:
print(f"发生未知错误 for {mip}: {e}")
print("\n所有成功获取的手机温度:", temperature_list)
return temperature_list
def set_fan_speed(self):
temp_list = self.get_allphone_temperature_max()
# ESP32_fan
ESP32_fan = self.read_yaml.get('switch_ip', {}).get('ESP32_fan', '')
# http://ESP32_fan/setSpeed?value=1
max_temp = max(temp_list) if temp_list else 0
speed = 0
div = 1
hour = time.localtime().tm_hour
if hour >= 22 or hour < 7:
div = 2
if max_temp >= 50:
speed = 180 // div
elif max_temp >= 45:
speed = 160 // div
elif max_temp >= 40:
speed = 1400 // div
elif max_temp >= 35:
speed = 120 // div
elif max_temp >= 30:
speed = 100 // div
else:
speed = 0
url = f'http://{ESP32_fan}/setSpeed?value={speed}'
response = requests.get(url)
print(f"设置风扇速度为 {speed}, 温度 {max_temp}, 结果: {response.text}")
# print(f"设置风扇速度为 {speed}, 温度 {max_temp}")
def check_pause_status(self):
"""
带有频率限制的暂停检查
"""
# 新增：检查环境变量，如果为"调试"模式，则跳过所有暂停逻辑
# win: set AUTOMATION_MODE=调试 $env:AUTOMATION_MODE = "调试" linux: export AUTOMATION_MODE=调试
if os.environ.get('AUTOMATION_MODE') == '调试':
# print("[调试模式] 跳过暂停检查。") # 可以取消这行注释来确认是否进入了调试模式
return
# 1. 检查是否在冷却时间内
current_time = time.time()
if current_time - self.last_pause_check_time < self.pause_check_interval:
return # 时间没到，直接跳过，不查 Redis，零消耗
# 2. 更新最后检查时间
self.last_pause_check_time = current_time
# 3. 真正去查 Redis
pause_key = f"STOP_DEVICE_{self.ip}"
try:
# 你的 redis_manager.redis_search 返回的是 value 或 None
# 注意：你的类可能返回 bytes，需要判断
is_paused = self.redis_manager.redis_search(pause_key)
if is_paused:
print(f"🔴 [Redis信号] 检测到暂停指令！设备 {self.ip} 正在暂停...")
self.MySQLManager.write_file(f"设备 {self.ip} 接收暂停指令，进入等待...")
# 进入死循环等待
while True:
time.sleep(5) # 暂停期间每5秒查一次，不消耗资源
# 暂停期间必须强制查 Redis，忽略冷却时间
check_again = self.redis_manager.redis_search(pause_key)
if not check_again:
print(f"🟢 [Redis信号] 暂停解除！设备 {self.ip} 恢复运行。")
self.MySQLManager.write_file(f"设备 {self.ip} 恢复运行。")
break
except Exception as e:
# Redis 连接失败不应该导致脚本崩溃，打印错误并跳过
print(f"检查暂停状态时 Redis 报错: {e}")
def watch_video(self, times=30, watch_time=0, max_watch_time=5, timeout=70,
close_delem=None, continue_delem=None, reward_img=None, reward_delem=None,
exit_delem=None, live_delem=None, back_func=None, timeout_name=None,
inspect_local=None, close_delem_click=True,close_cha_img=[],
inspect_times=5,close_img=[], live_img=[]
):
"""
通用视频观看方法
Args:
times: 循环次数
watch_time: 当前递归次数（内部使用）
max_watch_time: 最大递归次数
timeout: 超时时间（秒）
close_delem: 关闭/完成标志元素列表
continue_delem: 继续观看弹窗元素列表
reward_img: 奖励图片列表
reward_delem: 奖励元素列表
exit_delem: 退出确认元素列表
live_delem: 直播间元素列表
back_func: 返回任务函数（必需）
timeout_name: 超时任务名称
inspect_local: 查看当前不是看视频
close_delem_click: 是否点击关闭/完成标志
close_cha_img: 关闭/完叉图片
close_img: 关闭图片（优先级高于close_cha_img）
live_img: 直播图片
Example:
# 抖音
self.watch_video(
close_delem=['d(resourceId="com.kuaishou.nebula:id/video_close_icon")', 'd(text="领取成功")'],
continue_delem=['d(description="领取奖励")', 'd(description="坚持退出")'],
reward_delem=['d(text="开心收下")'],
exit_delem=['d(text="坚持退出")'],
live_delem=['d(resourceId="com.ss.android.ugc.livelite:id/ctz")'],
back_func=self.back_dy_task,
timeout_name="看精彩视频赚更多"
)
"""
retrun_text = '正常结束'
if not back_func:
raise ValueError("必须提供 back_func 参数")
# 设置默认值
close_delem = close_delem or []
continue_delem = continue_delem or []
reward_img = reward_img or []
reward_delem = reward_delem or []
exit_delem = exit_delem or []
live_delem = live_delem or []
# 递归次数检查
if watch_time >= max_watch_time:
print(f"看视频次数达到{max_watch_time}次，返回")
return
t1 = time.time()
for m in range(times):
time.sleep(3 + random.uniform(1, 2))
span_time = time.time() - t1
if inspect_local:
[x, y], k = self.multiple_find_picture([], 1, "不点击", "不返回", [], delem=inspect_local)
if x > 0 and y > 0:
print("当前不是看视频，返回")
retrun_text = '当前不是看视频界面'
break
# 超时处理
if span_time > timeout:
print(f"用时过多，已超过{timeout}秒")
self.ui_layout("看视频用时过久")
self.back()
time.sleep(4)
if exit_delem:
self.multiple_find_picture([], 10, "点击", "不返回", [], delem=exit_delem, xy_percentage=True)
if reward_img or reward_delem:
self.multiple_find_picture([], 5, "点击", "不返回", reward_img, delem=reward_delem)
back_func()
break
if self.get_app_name() != self.app_name:
print("%s不是做任务的app %s" % (self.get_app_name(), self.app_name))
time.sleep(12)
if self.activity:
self.d.app_start(self.app_name, self.activity)
else:
self.d.app_start(self.app_name)
# 检查视频是否结束
if close_delem:
[x6, y6], k = self.multiple_find_picture(close_img, 1, "不点击", "不返回", [], delem=close_delem)
if x6 > 0 and y6 > 0:
print("检测到视频结束,close_delem_click=", close_delem_click)
if close_delem_click is True:
print("ture")
self.click(x6, y6)
elif close_delem_click and type(close_delem_click) == list:
for i in range(2):
print("点击对应元素")
[x, y], k = self.multiple_find_picture(close_cha_img, 4, "点击", "不返回", [], delem=close_delem_click)
if x > 0 and y > 0:
time.sleep(2)
else:
back_func()
time.sleep(2)
# 处理继续观看弹窗
if continue_delem:
[x4, y4], k_continue = self.multiple_find_picture([], 5, "点击", "不返回", delem=continue_delem)
if x4 > 0 and y4 > 0 and k_continue == 0:
watch_time += 1
print(f"满足第一个元素，再次看视频，次数{watch_time}，耗时{span_time:.2f}秒")
self.watch_video(times, watch_time, max_watch_time, timeout, close_delem,
continue_delem, reward_img, reward_delem, exit_delem,
live_delem, back_func, timeout_name, inspect_local, close_delem_click,close_cha_img,
inspect_times, close_img, live_img)
return retrun_text
if back_func:
back_func()
if reward_img or reward_delem:
self.multiple_find_picture(reward_img, 5, "点击", "不返回", delem=reward_delem)
break
# 直播间检查
if m > inspect_times:
if live_img or live_delem:
[x, y], k = self.multiple_find_picture(live_img, 1, "点击", "不返回", delem=live_delem, xy_percentage=True)
if x > 0 and y > 0:
time.sleep(5)
print("退出直播间")
else:
print(f"循环结束，共{times}次")
if timeout_name:
self.tb_timeout_option(timeout_name)
back_func()
# 最后尝试领取奖励
if reward_img or reward_delem:
self.multiple_find_picture([], 5, "点击", "不返回", reward_img, delem=reward_delem)
return retrun_text
def wx_key_word(self):
y_py = 0
if self.model == '【MI 9 SE】':
y_py = 0.03
print(self.model,"y偏移：", y_py)
self.click(0.464, 0.785 + y_py) # 5
print("点击5")
time.sleep(0.6)
self.click(0.1, 0.712 + y_py) # 1
print("点击1")
time.sleep(0.8)
self.click(0.779, 0.823 + y_py) # 9
print("点击9")
time.sleep(0.5)
self.click(0.779, 0.823 + y_py) # 9
print("点击9")
time.sleep(0.7)
self.click(0.435, 0.824 + y_py) # 8
print("点击8")
time.sleep(0.7)
self.click(0.435, 0.712 + y_py) # 2
print("点击2")
# 微信_支付成功
time.sleep(4)
[x, y], k = self.multiple_find_picture(['微信_支付成功'], 10, "不点击", "不返回", xy_percentage=True)
if x > 0 and y > 0:
pic_path = self.screen("babanc.png")
text = '
' % pic_path
qq_email("", "%s，微信转账成功" % (self.ip + self.model), "[%s]
%s" % (self.model + self.ip, text))
def get_number_by_y(self, target_y, y0=638):
"""
根据传入的 y 坐标返回对应的数字
:param target_y: 需要查询的 y 坐标 (比如传入431，应该返回2)
:param y0: 数字 0 所在的基准 y 坐标，默认为 638
:return: 对应的数字 (int)
"""
# 计算每个数字之间的标准像素间距
# 根据 0(638) 和 5(121) 的距离得出: (638 - 121) / 5 = 103.4
step = 103.4
# 计算与基准坐标(数字0)的距离差
diff = y0 - target_y
# 距离差除以步长，得到对应的数字
# 由于屏幕坐标点击可能存在1-2像素的误差，这里使用 round() 进行四舍五入取整
number = round(diff / step)
return number
class Test(UiAutomatr):
def __init__(self, ip="", this_port=":5555", user='523040'):
super(Test, self).__init__()
self.ip = ip
self.port = this_port
self.z = 0
self.read()
self.tao_user = user
text = ''
# self.wx_key_word()
# time.sleep(10)
# exit()
# self.d.set_fastinput_ime(False)
# self.d(description="请输入登录密码").click()
# time.sleep(1)
# self.d.shell("input text 519yang982")
# print("输入密码结束")
# # time.sleep(3)
# # self.d.send_keys(str(10), clear=True)
# self.d.shell("input text %s" % '10')
# print("输入密码结束")
# ret = self.d.dump_hierarchy()
# print(ret)
self.multiple_find_picture(['测试_小米运动'], 20, "点击", "不返回", delem=['d(textContains="个商品得奖励")'],xy_percentage=True)
time.sleep(20)
exit()
# for i in range(5):
# elem = self.d.xpath('//*[@text="天"]/preceding-sibling::android.view.View/*[%s]' % (i + 1))
# if elem.exists:
# if elem.info.get('text') == '.':
# text += '.'
# else:
# elem = self.d.xpath('//*[@text="天"]/preceding-sibling::android.view.View/*[%s]/android.widget.TextView' % (i + 1))
# if elem.exists:
# top = elem.info.get('bounds', {}).get('top', 0)
# print("top:", top)
# nom = self.get_number_by_y(int(top))
# text += str(nom)
# # text += elem.info.get('text', '')
# else:
# print("没有找到元素")
# print(text)
# screen_path = self.screen(self.ip + "xmshipin_screen.png",
# elem=self.d(text="天").sibling(className="android.view.View"))
# elem = self.d.xpath('//android.widget.LinearLayout/android.widget.TextView')
# elem = self.d(className="android.widget.LinearLayout").child(className="android.widget.TextView")
# print(elem.info)
screen_path = self.screen(self.ip + "xmshipin_screen.png")
# # # , binarize_num=190
ret = pytesseract_pic_to_text(screen_path, pic_cache=False,config='--psm 3', text_padding='\n')
all_text = ret.get('all_text', '')
print(all_text)
if ret:
all_text = ret.get('all_text', '')
# # 0-08余额 0-08余额
# r_list = re.compile('(\\d+\\.\\d+)余额').findall(text.replace('-', '.').replace('00', '.0.0'))
# if r_list:
# money = float(r_list[0])
# for i in ret:
# text = ret.get(i)
# # print(text)
# if "*" in text:
# print(ret)
# ret = self.recognize_text_dddocr(tjb_yzm)
# if ret:
# tao_user = ret
time.sleep(399)
# money = pytesseract_pic_to_text(tjb_yzm).get('all_text')
# if money:
# money = money.replace('#', '')
# print(money)
# tjb_yzm = self.screen('yzm/%s.jpg' % (self.ip + 'weixin2'))
# all_text = pytesseract_pic_to_text(tjb_yzm).get('all_text')
# print("available settings keys:", self.d.settings.available_keys())
# print("available settings keys:", self.d.settings.available_keys())
# # 或者从 __dict__ 里看内部存储
# print("settings __dict__:", self.d.settings.__dict__)
# ret = self.d.dump_hierarchy()
# # 正确调用：加上 self.
# print("验证码是：", ret)
# open(r'F:\备份\mi\hierarchy.xml', 'w', encoding='utf-8').write(ret)
# print()
# self.do_task_for_list()
# self.clear_mobile()
# self.click_vd_cross()
# self.jurisdiction()
# self.click_vd_cross()
# self.get_mobile_ipv6()
# print("打开清理软件")
# d(text="浏览60秒")
# self.open_accessibilit()
# self.export_app()
try:
self.china_yzm()
except Exception as e:
print(e)
# self.multiple_find_picture([], 20, "不点击", "不返回", delem=['d(textContains="个商品得奖励")'],xy_percentage=True)
# self.d.app_start("com.miui.cleanmaster", activity="com.miui.optimizecenter.MainActivity")
# self.d.app_start("com.jifen.qukan")
# self.do_task_for_list(key_list=['去完成', '去领取', ' 去完成', ' 去领取'],
# title1_elem='sibling(index=1).child(index=0)',
# title2_elem = 'sibling(index=1).child(index=1)',
# app_name = 'com.cainiao.wireless',
# # '逛'
# browse_key = {'关键字':['浏览'], '成功元素':['d(textContains="浏览完成")', 'd(description="任务完成")','d(text="任务完成")'], '成功图片':[], '浏览次数':10},
# back_key_dict = {'d(text