20250904-初步功能已完成
This commit is contained in:
@@ -1,6 +1,8 @@
|
||||
import math
|
||||
import random
|
||||
import re
|
||||
import time
|
||||
from typing import Tuple, List
|
||||
|
||||
import tidevice
|
||||
import wda
|
||||
@@ -70,14 +72,17 @@ class ControlUtils(object):
|
||||
return True
|
||||
elif session.xpath("//*[@name='nav_bar_start_back']").exists:
|
||||
back = session.xpath("//*[@name='nav_bar_start_back']")
|
||||
back.click()
|
||||
if back.exists:
|
||||
back.click()
|
||||
return True
|
||||
elif session.xpath(
|
||||
"//Window[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]").exists:
|
||||
back = session.xpath(
|
||||
"//Window[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]/Other[1]")
|
||||
back.click()
|
||||
return True
|
||||
|
||||
if back.exists:
|
||||
back.click()
|
||||
return True
|
||||
else:
|
||||
return False
|
||||
except Exception as e:
|
||||
@@ -129,6 +134,7 @@ class ControlUtils(object):
|
||||
videoCell = session.xpath(
|
||||
'(//XCUIElementTypeCollectionView//XCUIElementTypeCell[.//XCUIElementTypeImage[@name="profile_video"]])[1]')
|
||||
|
||||
|
||||
tab = session.xpath(
|
||||
'//XCUIElementTypeButton[@name="TTKProfileTabVideoButton_0" or contains(@label,"作品") or contains(@name,"作品")]'
|
||||
).get(timeout=5) # 某些版本 tab.value 可能就是数量；或者 tab.label 类似 “作品 7”
|
||||
@@ -204,33 +210,65 @@ class ControlUtils(object):
|
||||
print(e)
|
||||
return False
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# 随机滑动一点点距离
|
||||
@classmethod
|
||||
def tap_mini_cluster(cls, center_x: int, center_y: int, session, points=5, duration_ms=60):
|
||||
try:
|
||||
response = session.http.post(
|
||||
"touchAndHold",
|
||||
data={
|
||||
"x": 100,
|
||||
"y": 100,+
|
||||
"duration": 0.1
|
||||
}
|
||||
)
|
||||
print(response)
|
||||
return response
|
||||
except Exception as e:
|
||||
print(e)
|
||||
return None
|
||||
|
||||
# 检测五分钟前和当前的状态是否相同
|
||||
# @classmethod
|
||||
# def compareCurrentWithPreviousState(cls,xml):
|
||||
|
||||
def random_micro_swipe(
|
||||
cls,
|
||||
center_x: int,
|
||||
center_y: int,
|
||||
session,
|
||||
points: int = 6,
|
||||
duration_ms: int = 15,
|
||||
) -> None:
|
||||
"""
|
||||
在 (center_x, center_y) 附近做 20px 左右的不规则微滑动。
|
||||
使用 facebook-wda 的 session.swipe(x1, y1, x2, y2, duration) 接口。
|
||||
"""
|
||||
# 1. 随机方向
|
||||
angle = random.uniform(0, 2 * math.pi)
|
||||
length = random.uniform(18, 22) # 20px 左右
|
||||
end_x = center_x + length * math.cos(angle)
|
||||
end_y = center_y + length * math.sin(angle)
|
||||
|
||||
# 2. 限制在 20px 圆内（防止超出）
|
||||
def clamp_to_circle(x, y, cx, cy, r):
|
||||
dx = x - cx
|
||||
dy = y - cy
|
||||
if dx * dx + dy * dy > r * r:
|
||||
scale = r / math.hypot(dx, dy)
|
||||
x = cx + dx * scale
|
||||
y = cy + dy * scale
|
||||
return int(round(x)), int(round(y))
|
||||
|
||||
end_x, end_y = clamp_to_circle(end_x, end_y, center_x, center_y, 20)
|
||||
|
||||
# 3. 加入轻微噪声，制造“不规则”曲线
|
||||
noise = 3 # 最大偏移像素
|
||||
mid_count = points - 2
|
||||
mid_points: List[Tuple[int, int]] = []
|
||||
for i in range(1, mid_count + 1):
|
||||
t = i / (mid_count + 1)
|
||||
# 线性插值 + 垂直方向噪声
|
||||
x = center_x * (1 - t) + end_x * t
|
||||
y = center_y * (1 - t) + end_y * t
|
||||
perp_angle = angle + math.pi / 2 # 垂直方向
|
||||
offset = random.uniform(-noise, noise)
|
||||
x += offset * math.cos(perp_angle)
|
||||
y += offset * math.sin(perp_angle)
|
||||
x, y = clamp_to_circle(x, y, center_x, center_y, 20)
|
||||
mid_points.append((int(round(x)), int(round(y))))
|
||||
|
||||
# 4. 构造完整轨迹
|
||||
trajectory: List[Tuple[int, int]] = (
|
||||
[(center_x, center_y)] + mid_points + [(end_x, end_y)]
|
||||
)
|
||||
|
||||
# 5. 使用 facebook-wda 的 swipe 接口（逐段 swipe）
|
||||
# 由于总时长太短，我们一次性 swipe 到终点，但用多点轨迹模拟
|
||||
# facebook-wda 支持 swipe(x1, y1, x2, y2, duration)
|
||||
# 我们直接用起点 -> 终点，duration 用总时长
|
||||
print("开始微滑动")
|
||||
session.swipe(center_x, center_y, end_x, end_y, duration_ms / 1000)
|
||||
print("随机微滑动：", trajectory)
|
||||
|
||||
|
||||
|
||||
|
||||
@@ -1,217 +1,3 @@
|
||||
#
|
||||
# import datetime
|
||||
# import io
|
||||
# import logging
|
||||
# import os
|
||||
# import re
|
||||
# import sys
|
||||
# import shutil
|
||||
# import zipfile
|
||||
# from pathlib import Path
|
||||
# import requests
|
||||
#
|
||||
#
|
||||
# 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 = {}
|
||||
# _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，并绑定到设备目录下的固定文件（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)
|
||||
#
|
||||
# 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 _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):
|
||||
# """启动时清空 log 目录下所有文件"""
|
||||
#
|
||||
# # 关闭所有 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 目录
|
||||
# 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():
|
||||
# shutil.rmtree(item)
|
||||
#
|
||||
# # 清缓存
|
||||
# cls._method_loggers.clear()
|
||||
#
|
||||
# @classmethod
|
||||
# def upload_all_logs(cls, server_url, token, userId, tenantId):
|
||||
# log_path = Path(cls.logDir)
|
||||
# if not log_path.exists():
|
||||
# return False
|
||||
#
|
||||
# timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||||
# filename = f"{timestamp}_logs.zip"
|
||||
# print(filename)
|
||||
# zip_buf = io.BytesIO()
|
||||
# with zipfile.ZipFile(zip_buf, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
||||
# for p in log_path.rglob("*"):
|
||||
# if p.is_file():
|
||||
# arcname = str(p.relative_to(log_path))
|
||||
# zf.write(p, arcname=arcname)
|
||||
#
|
||||
# zip_bytes = zip_buf.getvalue()
|
||||
#
|
||||
# headers = {"vvtoken": token}
|
||||
# data = {"tenantId": tenantId, "userId": userId}
|
||||
#
|
||||
#
|
||||
# files = {
|
||||
# "file": (filename, io.BytesIO(zip_bytes), "application/zip")
|
||||
# }
|
||||
#
|
||||
# # 3) 上传
|
||||
# resp = requests.post(server_url, headers=headers, data=data, files=files)
|
||||
# if resp.json()['data']:
|
||||
# return True
|
||||
# return False
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
|
||||
# -*- coding: utf-8 -*-
|
||||
import datetime
|
||||
import io
|
||||
|
||||
Reference in New Issue
Block a user