zw/iOSAI
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

ai 开始测试

Browse Source
This commit is contained in:
zhangkai
2025-09-01 21:51:36 +08:00
parent e20d81144d
commit 8317470fb5
7 changed files with 416 additions and 322 deletions
Download Patch File Download Diff File Expand all files Collapse all files

15
Utils/AiUtils.py
View File

@@ -1,4 +1,3 @@
import os
import re
import cv2
@@ -9,6 +8,7 @@ import xml.etree.ElementTree as ET
from lxml import etree
from wda import Client
# 工具类
class AiUtils(object):
@@ -17,7 +17,7 @@ class AiUtils(object):
def findImageInScreen(cls, target, udid):
try:
# 加载原始图像和模板图像
image_path = AiUtils.imagePathWithName(udid,"bgv") # 替换为你的图像路径
image_path = AiUtils.imagePathWithName(udid, "bgv") # 替换为你的图像路径
template_path = AiUtils.imagePathWithName("", target) # 替换为你的模板路径
# 读取图像和模板，确保它们都是单通道灰度图
@@ -64,7 +64,6 @@ class AiUtils(object):
return int(match.group()) # 将匹配到的数字转换为整数
return None # 如果没有找到数字，返回 None
# 选择截图
@classmethod
def screenshot(cls):
@@ -125,6 +124,7 @@ class AiUtils(object):
height, width, channels = heart_icon.shape
roi = np.zeros((height, width, channels), dtype=np.uint8)
# 将爱心图标粘贴到透明背景上
for c in range(channels):
roi[:, :, c] = np.where(mask_inv == 255, heart_icon[:, :, c], roi[:, :, c])
@@ -142,7 +142,6 @@ class AiUtils(object):
else:
print("未选择有效区域。")
# 根据名称获取图片完整地址
@classmethod
def imagePathWithName(cls, udid, name):
@@ -166,7 +165,6 @@ class AiUtils(object):
# 返回根目录
return project_root
# 创建一个以udid命名的目录
@classmethod
def makeUdidDir(cls, udid):
@@ -189,6 +187,8 @@ class AiUtils(object):
LogManager.info(f"目录 {udid_dir} 已存在，跳过创建", udid)
print(f"目录 {udid_dir} 已存在，跳过创建")
# 查找首页按钮
# uuid 设备id
# click 是否点击该按钮
@@ -209,10 +209,9 @@ class AiUtils(object):
print(e)
return None
# 查找关闭按钮
@classmethod
def findLiveCloseButton(cls,udid="eca000fcb6f55d7ed9b4c524055214c26a7de7aa"):
def findLiveCloseButton(cls, udid="eca000fcb6f55d7ed9b4c524055214c26a7de7aa"):
client = wda.USBClient(udid)
session = client.session()
session.appium_settings({"snapshotMaxDepth": 10})
@@ -277,7 +276,6 @@ class AiUtils(object):
client = wda.USBClient("eca000fcb6f55d7ed9b4c524055214c26a7de7aa")
print(client.source())
# 查找app主页上的收件箱按钮
@classmethod
def getMsgBoxButton(cls, session: Client):
@@ -317,6 +315,7 @@ class AiUtils(object):
# 屏幕宽度
app = root.xpath('/XCUIElementTypeApplication')
screen_w = cls.parse_float(app[0], 'width', 414.0) if app else 414.0
# 找 Table 的可见范围

226
Utils/LogManager.py
View File

@@ -1,9 +1,96 @@
# import logging
# import os
# import sys
# import shutil
# from pathlib import Path
#
#
# class LogManager:
# # 运行根目录：打包后取 exe 目录；源码运行取项目目录
# if getattr(sys, "frozen", False):
# projectRoot = os.path.dirname(sys.executable)
# else:
# projectRoot = os.path.dirname(os.path.dirname(__file__))
#
# logDir = os.path.join(projectRoot, "log")
# _loggers = {}
#
# @classmethod
# def _setupLogger(cls, udid, name, logName, level=logging.INFO):
# """创建或获取 logger，并绑定到文件"""
# deviceLogDir = os.path.join(cls.logDir, udid)
# os.makedirs(deviceLogDir, exist_ok=True)
# logFile = os.path.join(deviceLogDir, logName)
#
# logger_name = f"{udid}_{name}"
# logger = logging.getLogger(logger_name)
# logger.setLevel(level)
#
# # 避免重复添加 handler
# if not any(
# isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
# for h in logger.handlers
# ):
# fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
# formatter = logging.Formatter(
# "%(asctime)s - %(name)s - %(levelname)s - %(message)s",
# datefmt="%Y-%m-%d %H:%M:%S"
# )
# fileHandler.setFormatter(formatter)
# logger.addHandler(fileHandler)
#
# return logger
#
# @classmethod
# def info(cls, text, udid="system"):
# cls._setupLogger(udid, "infoLogger", "info.log", level=logging.INFO).info(f"[{udid}] {text}")
#
# @classmethod
# def warning(cls, text, udid="system"):
# cls._setupLogger(udid, "warningLogger", "warning.log", level=logging.WARNING).warning(f"[{udid}] {text}")
#
# @classmethod
# def error(cls, text, udid="system"):
# cls._setupLogger(udid, "errorLogger", "error.log", level=logging.ERROR).error(f"[{udid}] {text}")
#
# # 清空日志
# @classmethod
# def clearLogs(cls):
# """启动时清空 log 目录下所有文件"""
# print("开始清空日志...")
#
# # 关闭所有 handler
# for name, logger in logging.Logger.manager.loggerDict.items():
# if isinstance(logger, logging.Logger):
# for handler in logger.handlers[:]:
# try:
# handler.close()
# except Exception:
# pass
# logger.removeHandler(handler)
#
# # 仅删除目录里的所有文件和子目录
# log_path = Path(cls.logDir)
# if log_path.exists():
# for item in log_path.iterdir():
# if item.is_file():
# item.unlink()
# elif item.is_dir():
# import shutil
# shutil.rmtree(item)
#
# print("日志清空完成")
import logging
import os
import re
import sys
import shutil
from pathlib import Path
import requests
class LogManager:
# 运行根目录：打包后取 exe 目录；源码运行取项目目录
@@ -14,11 +101,46 @@ class LogManager:
logDir = os.path.join(projectRoot, "log")
_loggers = {}
_method_loggers = {} # 新增：缓存“设备+方法”的 logger
# ---------- 工具函数 ----------
@classmethod
def _safe_filename(cls, name: str, max_len: int = 80) -> str:
"""
将方法名/udid等转成安全文件名：
- 允许字母数字、点、下划线、连字符
- 允许常见 CJK 字符（中日韩）
- 其他非法字符替换为下划线
- 合并多余下划线，裁剪长度
"""
if not name:
return "unknown"
name = str(name).strip()
# 替换 Windows 非法字符和控制符
name = re.sub(r'[\\/:*?"<>|\r\n\t]+', '_', name)
# 只保留 ① 英数._- ② CJK 统一表意文字、日文平/片假名、韩文音节
name = re.sub(rf'[^a-zA-Z0-9_.\-'
r'\u4e00-\u9fff' # 中
r'\u3040-\u30ff' # 日
r'\uac00-\ud7a3' # 韩
r']+', '_', name)
# 合并多余下划线，去两端空白与下划线
name = re.sub(r'_+', '_', name).strip(' _.')
# 避免空
name = name or "unknown"
# Windows 预留名避免（CON/PRN/AUX/NUL/COM1…）
if re.fullmatch(r'(?i)(CON|PRN|AUX|NUL|COM[1-9]|LPT[1-9])', name):
name = f"_{name}"
# 限长
return name[:max_len] or "unknown"
# ---------- 旧的：按级别写固定文件 ----------
@classmethod
def _setupLogger(cls, udid, name, logName, level=logging.INFO):
"""创建或获取 logger，并绑定到文件"""
deviceLogDir = os.path.join(cls.logDir, udid)
"""创建或获取 logger，并绑定到设备目录下的固定文件（info.log / warning.log / error.log）"""
deviceLogDir = os.path.join(cls.logDir, cls._safe_filename(udid))
os.makedirs(deviceLogDir, exist_ok=True)
logFile = os.path.join(deviceLogDir, logName)
@@ -28,8 +150,8 @@ class LogManager:
# 避免重复添加 handler
if not any(
isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
for h in logger.handlers
isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
for h in logger.handlers
):
fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
formatter = logging.Formatter(
@@ -53,6 +175,59 @@ class LogManager:
def error(cls, text, udid="system"):
cls._setupLogger(udid, "errorLogger", "error.log", level=logging.ERROR).error(f"[{udid}] {text}")
# ---------- 新增：按“设备+方法”分别写独立日志文件 ----------
@classmethod
def _setupMethodLogger(cls, udid: str, method: str, level=logging.INFO):
"""
为某设备的某个方法单独创建 logger：
log/<udid>/<method>.log
"""
udid_key = cls._safe_filename(udid or "system")
method_key = cls._safe_filename(method or "general")
cache_key = (udid_key, method_key)
# 命中缓存
if cache_key in cls._method_loggers:
return cls._method_loggers[cache_key]
deviceLogDir = os.path.join(cls.logDir, udid_key)
os.makedirs(deviceLogDir, exist_ok=True)
logFile = os.path.join(deviceLogDir, f"{method_key}.log")
logger_name = f"{udid_key}.{method_key}"
logger = logging.getLogger(logger_name)
logger.setLevel(level)
logger.propagate = False # 避免向根 logger 传播导致控制台重复打印
# 避免重复添加 handler
if not any(
isinstance(h, logging.FileHandler) and h.baseFilename == os.path.abspath(logFile)
for h in logger.handlers
):
fileHandler = logging.FileHandler(logFile, mode="a", encoding="utf-8")
formatter = logging.Formatter(
"%(asctime)s - %(levelname)s - %(name)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
fileHandler.setFormatter(formatter)
logger.addHandler(fileHandler)
cls._method_loggers[cache_key] = logger
return logger
@classmethod
def method_info(cls, text, method, udid="system"):
"""按设备+方法写 INFO 到 log/<udid>/<method>.log"""
cls._setupMethodLogger(udid, method, level=logging.INFO).info(f"[{udid}][{method}] {text}")
@classmethod
def method_warning(cls, text, method, udid="system"):
cls._setupMethodLogger(udid, method, level=logging.WARNING).warning(f"[{udid}][{method}] {text}")
@classmethod
def method_error(cls, text, method, udid="system"):
cls._setupMethodLogger(udid, method, level=logging.ERROR).error(f"[{udid}][{method}] {text}")
# 清空日志
@classmethod
def clearLogs(cls):
@@ -69,14 +244,53 @@ class LogManager:
pass
logger.removeHandler(handler)
# 仅删除目录里的所有文件和子目录
# 删除 log 目录
log_path = Path(cls.logDir)
if log_path.exists():
for item in log_path.iterdir():
if item.is_file():
item.unlink()
elif item.is_dir():
import shutil
shutil.rmtree(item)
# 清缓存
cls._method_loggers.clear()
print("日志清空完成")
@classmethod
def upload_all_logs(cls, server_url: str, extra_data: dict = None):
"""
上传 log/ 目录下所有文件和子目录中的日志
"""
log_path = Path(cls.logDir)
if not log_path.exists():
print("[upload_all_logs] 日志目录不存在")
return False
success_files, failed_files = [], []
for file in log_path.rglob("*.log"): # 递归找到所有 .log 文件
try:
files = {"file": open(file, "rb")}
data = {"relative_path": str(file.relative_to(log_path))}
if extra_data:
data.update(extra_data)
resp = requests.post(server_url, files=files, data=data, timeout=15)
if resp.status_code == 200:
success_files.append(file.name)
else:
failed_files.append((file.name, resp.status_code))
except Exception as e:
failed_files.append((file.name, str(e)))
finally:
try:
files["file"].close()
except:
pass
print(f"[upload_all_logs] 成功上传: {success_files}")
if failed_files:
print(f"[upload_all_logs] 失败文件: {failed_files}")
return len(failed_files) == 0