880 lines
32 KiB
Python
880 lines
32 KiB
Python
"""
|
||
Utility Functions - 动作解析、坐标转换、图像处理
|
||
|
||
Extracted from GuiAgent/core/utils/common.py for modular use in DroidBot.
|
||
"""
|
||
import re
|
||
import base64
|
||
import io
|
||
import os
|
||
import logging
|
||
from typing import Optional, Dict, Any, Tuple, List
|
||
|
||
try:
|
||
from PIL import Image
|
||
except ImportError:
|
||
Image = None
|
||
|
||
logger = logging.getLogger(__name__)
|
||
|
||
# ==============================================================================
|
||
# 动作解析
|
||
# ==============================================================================
|
||
|
||
# 兼容多种参数写法
|
||
ACTION_PARAM_PATTERN = re.compile(
|
||
r"(\w+)\s*[:=]\s*(?:'([^']*)'|\"([^\"]*)\"|`([^`]*)`|\[([^\]]*)\]|(\d+,\d+)|([^\s,]+))"
|
||
)
|
||
|
||
|
||
def _extract_thought(prediction: str) -> str:
|
||
"""提取 Thought 段落"""
|
||
match = re.search(r'Thought:\s*(.+?)(?=Action:|$)', prediction, re.DOTALL)
|
||
return match.group(1).strip() if match else ''
|
||
|
||
|
||
def _extract_action_and_params(prediction: str) -> Tuple[str, Dict[str, str]]:
|
||
"""
|
||
提取 Action 类型和参数字符串，并解析成字典
|
||
|
||
支持格式：
|
||
1. Action: click(center='[x, y]')
|
||
2. <|begin_of_box|>tap(center='[511, 426]')<|end_of_box|>
|
||
3. 长文本后直接跟 Action: tap(center='[596, 359]')
|
||
|
||
Returns:
|
||
(action_type, inputs)
|
||
"""
|
||
action_content = prediction
|
||
|
||
# 提取 <|begin_of_box|>...<|end_of_box|> 中的内容
|
||
box_match = re.search(r'<\|begin_of_box\|>(.*?)<\|end_of_box\|>', prediction, re.DOTALL)
|
||
if box_match:
|
||
action_content = box_match.group(1).strip()
|
||
|
||
# 定义有效的动作类型列表
|
||
valid_actions = ['tap', 'click', 'long_tap', 'drag', 'swipe', 'type', 'key_press',
|
||
'wait', 'finished', 'report_stuck_reason', 'receive_email', 'double_click',
|
||
'right_click', 'scroll', 'solve_slider_captcha', 'solve_image_captcha',
|
||
'login_ios']
|
||
|
||
action_type: Optional[str] = None
|
||
params_str: Optional[str] = None
|
||
|
||
# 1) 从 "Action: xxx ..." 行中解析
|
||
line_match = re.search(r'Action:\s*(\w+)(.*)', action_content)
|
||
if line_match:
|
||
matched_action_type = line_match.group(1)
|
||
tail = line_match.group(2).strip()
|
||
|
||
if matched_action_type in valid_actions:
|
||
action_type = matched_action_type
|
||
|
||
# 带括号形式
|
||
paren_match = re.match(r'^\(\s*(.*)\s*\)$', tail, re.DOTALL)
|
||
if paren_match:
|
||
params_str = paren_match.group(1).strip()
|
||
else:
|
||
# 无括号形式
|
||
params_str = tail.strip()
|
||
|
||
# 2) 兜底查找 tap(...)/click(...) 等函数调用
|
||
if action_type is None:
|
||
action_pattern = r'\b(' + '|'.join(valid_actions) + r')\s*\(\s*([^)]*?)\s*\)'
|
||
fallback_match = re.search(action_pattern, action_content, re.DOTALL)
|
||
if fallback_match:
|
||
action_type = fallback_match.group(1)
|
||
params_str = fallback_match.group(2).strip()
|
||
|
||
if action_type is None:
|
||
return '', {}
|
||
|
||
inputs: Dict[str, str] = {}
|
||
if params_str:
|
||
for m in re.finditer(ACTION_PARAM_PATTERN, params_str):
|
||
key = m.group(1)
|
||
raw_val = next((g for g in m.groups()[1:] if g is not None), "")
|
||
# 如果是方括号或逗号分隔的坐标，补回括号
|
||
if m.group(5): # 方括号
|
||
value = f"[{raw_val}]"
|
||
elif m.group(6): # 逗号分隔坐标
|
||
value = f"[{raw_val}]"
|
||
else:
|
||
value = raw_val
|
||
inputs[key] = value.strip()
|
||
|
||
return action_type, inputs
|
||
|
||
|
||
def parse_uitars_action(prediction: str) -> Dict[str, Any]:
|
||
"""
|
||
解析 UITars 风格的动作
|
||
|
||
输入格式：
|
||
Thought: ...
|
||
Action: click(center='[x, y]')
|
||
|
||
Returns:
|
||
{'action_type': str, 'inputs': dict, 'thought': str}
|
||
"""
|
||
action_type, inputs = _extract_action_and_params(prediction)
|
||
return {
|
||
'action_type': action_type,
|
||
'inputs': inputs,
|
||
'thought': _extract_thought(prediction)
|
||
}
|
||
|
||
|
||
def convert_to_executor_action(parsed: Dict[str, Any]) -> Dict[str, Any]:
|
||
"""
|
||
将解析后的动作转换为执行器格式
|
||
|
||
保持原始动作名称，不做平台特定的映射。
|
||
坐标转换在各 executor 内部完成。
|
||
"""
|
||
action_type = parsed.get('action_type', '')
|
||
inputs = parsed.get('inputs', {})
|
||
|
||
# 动作映射（主要处理别名）
|
||
action_map = {
|
||
# 通用动作
|
||
'type': 'type',
|
||
'wait': 'wait',
|
||
'finished': 'finished',
|
||
'report_stuck_reason': 'report_stuck_reason',
|
||
'login_ios': 'login_ios',
|
||
'receive_email': 'receive_email',
|
||
'key_press': 'key_press',
|
||
'hold_key': 'key_press',
|
||
|
||
# 平台特定动作（保持原样）
|
||
'click': 'click',
|
||
'double_click': 'double_click',
|
||
'right_click': 'right_click',
|
||
'drag': 'drag',
|
||
'scroll': 'scroll',
|
||
'tap': 'tap',
|
||
'long_tap': 'long_tap',
|
||
'swipe': 'swipe',
|
||
}
|
||
|
||
result = {
|
||
'action': action_map.get(action_type, action_type),
|
||
'thought': parsed.get('thought', '')
|
||
}
|
||
|
||
def _parse_center(box_str: str) -> Tuple[Optional[float], Optional[float]]:
|
||
"""解析坐标中心点，并处理归一化坐标兜底"""
|
||
if not box_str:
|
||
return None, None
|
||
try:
|
||
nums = [float(n.strip()) for n in box_str.strip('[]').split(',') if n.strip()]
|
||
if len(nums) == 2:
|
||
x, y = nums[0], nums[1]
|
||
elif len(nums) >= 4:
|
||
x1, y1, x2, y2 = nums[0], nums[1], nums[2], nums[3]
|
||
x, y = (x1 + x2) / 2, (y1 + y2) / 2
|
||
else:
|
||
return None, None
|
||
|
||
# 坐标兜底：如果x和y都小于1，说明LLM返回的是0-1归一化坐标，需要乘以1000
|
||
if x < 1 and y < 1:
|
||
logger.debug(f"[坐标兜底] 检测到0-1归一化坐标 ({x:.4f}, {y:.4f})，乘以1000转换")
|
||
x = x * 1000
|
||
y = y * 1000
|
||
|
||
return x, y
|
||
except (ValueError, IndexError):
|
||
return None, None
|
||
return None, None
|
||
|
||
# 单点坐标（支持 center 和 point 参数）
|
||
center_key = 'center' if 'center' in inputs else ('point' if 'point' in inputs else None)
|
||
|
||
if center_key:
|
||
x, y = _parse_center(inputs[center_key])
|
||
if x is not None:
|
||
result['target'] = [x, y]
|
||
elif 'start_center' in inputs:
|
||
x, y = _parse_center(inputs['start_center'])
|
||
if x is not None:
|
||
result['target'] = [x, y]
|
||
|
||
# 拖拽终点
|
||
if 'end_center' in inputs:
|
||
ex, ey = _parse_center(inputs['end_center'])
|
||
if ex is not None:
|
||
result['start'] = result.get('target')
|
||
result['end'] = [ex, ey]
|
||
|
||
# 文本
|
||
if 'content' in inputs:
|
||
result['text'] = inputs['content']
|
||
elif 'value' in inputs:
|
||
result['text'] = inputs['value']
|
||
|
||
# 按键
|
||
if 'key' in inputs:
|
||
result['key'] = inputs['key']
|
||
|
||
# 方向
|
||
if 'direction' in inputs:
|
||
result['direction'] = inputs['direction']
|
||
|
||
# 滑动幅度
|
||
if 'amount' in inputs:
|
||
try:
|
||
result['amount'] = float(inputs['amount'])
|
||
except (ValueError, TypeError):
|
||
pass
|
||
|
||
# 验证码技能参数
|
||
if 'captcha_region' in inputs:
|
||
# 解析 [x1, y1, x2, y2] 格式
|
||
try:
|
||
region_str = inputs['captcha_region'].strip('[]')
|
||
region_nums = [float(n.strip()) for n in region_str.split(',') if n.strip()]
|
||
if len(region_nums) == 4:
|
||
result['captcha_region'] = region_nums
|
||
except (ValueError, IndexError):
|
||
pass
|
||
|
||
if 'slider_position' in inputs:
|
||
try:
|
||
pos_str = inputs['slider_position'].strip('[]')
|
||
pos_nums = [float(n.strip()) for n in pos_str.split(',') if n.strip()]
|
||
if len(pos_nums) == 2:
|
||
result['slider_position'] = pos_nums
|
||
except (ValueError, IndexError):
|
||
pass
|
||
|
||
if 'input_field' in inputs:
|
||
try:
|
||
field_str = inputs['input_field'].strip('[]')
|
||
field_nums = [float(n.strip()) for n in field_str.split(',') if n.strip()]
|
||
if len(field_nums) == 2:
|
||
result['input_field'] = field_nums
|
||
except (ValueError, IndexError):
|
||
pass
|
||
|
||
# report_stuck_reason 参数
|
||
if 'reason_code' in inputs:
|
||
result['reason_code'] = inputs['reason_code'].strip()
|
||
if 'message' in inputs:
|
||
result['message'] = inputs['message'].strip()
|
||
|
||
return result
|
||
|
||
|
||
# ==============================================================================
|
||
# 图像处理
|
||
# ==============================================================================
|
||
|
||
def draw_grid_on_image(image, grid_color: Tuple[int, int, int, int] = (255, 0, 0, 128), grid_divisions: int = 10,
|
||
coord_range: Tuple[int, int] = None):
|
||
"""
|
||
在图像上绘制网格线并标注坐标数字，帮助LLM精确理解坐标位置
|
||
|
||
Args:
|
||
image: PIL Image对象
|
||
grid_color: 网格线颜色 (R, G, B, A)，默认为红色半透明
|
||
grid_divisions: 网格分割数量，默认为10，即绘制10x10的网格
|
||
coord_range: 坐标范围 (max_x, max_y)，默认为 (1000, 1000) 即归一化坐标
|
||
如果为None则使用图像实际像素尺寸
|
||
|
||
Returns:
|
||
绘制了网格和坐标标注的图像
|
||
"""
|
||
if Image is None:
|
||
logger.warning("PIL未安装，无法绘制网格")
|
||
return image
|
||
|
||
try:
|
||
from PIL import ImageDraw, ImageFont
|
||
|
||
# 创建一个可以绘制的副本
|
||
img_copy = image.copy()
|
||
|
||
# 获取图像尺寸
|
||
width, height = img_copy.size
|
||
|
||
# 确定坐标范围
|
||
if coord_range is None:
|
||
coord_max_x, coord_max_y = width, height
|
||
else:
|
||
coord_max_x, coord_max_y = coord_range
|
||
|
||
# 计算网格大小（像素）
|
||
grid_width = width // grid_divisions
|
||
grid_height = height // grid_divisions
|
||
|
||
draw = ImageDraw.Draw(img_copy)
|
||
|
||
# 尝试加载字体，动态计算合适的字体大小
|
||
font_size = max(10, min(width, height) // 40)
|
||
# 依次尝试 macOS / Windows / Linux 字体路径
|
||
_font_candidates = [
|
||
"/System/Library/Fonts/Helvetica.ttc", # macOS
|
||
"C:/Windows/Fonts/arial.ttf", # Windows
|
||
"/usr/share/fonts/truetype/dejavu/DejaVuSans.ttf", # Linux (Debian/Ubuntu)
|
||
]
|
||
font = None
|
||
for _font_path in _font_candidates:
|
||
try:
|
||
font = ImageFont.truetype(_font_path, font_size)
|
||
break
|
||
except Exception:
|
||
continue
|
||
if font is None:
|
||
font = ImageFont.load_default()
|
||
|
||
# 标注颜色（与网格线同色但更不透明）
|
||
label_color = (grid_color[0], grid_color[1], grid_color[2], min(255, grid_color[3] + 80))
|
||
# 标注背景色（半透明白色，提高可读性）
|
||
bg_color = (255, 255, 255, 160)
|
||
|
||
# 绘制垂直线 + 顶部X坐标标注
|
||
for i in range(1, grid_divisions):
|
||
x = i * grid_width
|
||
draw.line([(x, 0), (x, height)], fill=grid_color, width=1)
|
||
|
||
# 计算对应的坐标值
|
||
coord_x = int(i * coord_max_x / grid_divisions)
|
||
label = str(coord_x)
|
||
|
||
# 获取文本尺寸
|
||
bbox = draw.textbbox((0, 0), label, font=font)
|
||
text_w = bbox[2] - bbox[0]
|
||
text_h = bbox[3] - bbox[1]
|
||
|
||
# 在网格线顶部标注（带背景）
|
||
label_x = x - text_w // 2
|
||
label_y = 2
|
||
draw.rectangle([label_x - 1, label_y, label_x + text_w + 1, label_y + text_h + 1], fill=bg_color)
|
||
draw.text((label_x, label_y), label, fill=label_color, font=font)
|
||
|
||
# 绘制水平线 + 左侧Y坐标标注
|
||
for i in range(1, grid_divisions):
|
||
y = i * grid_height
|
||
draw.line([(0, y), (width, y)], fill=grid_color, width=1)
|
||
|
||
# 计算对应的坐标值
|
||
coord_y = int(i * coord_max_y / grid_divisions)
|
||
label = str(coord_y)
|
||
|
||
# 获取文本尺寸
|
||
bbox = draw.textbbox((0, 0), label, font=font)
|
||
text_w = bbox[2] - bbox[0]
|
||
text_h = bbox[3] - bbox[1]
|
||
|
||
# 在网格线左侧标注（带背景）
|
||
label_x = 2
|
||
label_y = y - text_h // 2
|
||
draw.rectangle([label_x - 1, label_y, label_x + text_w + 1, label_y + text_h + 1], fill=bg_color)
|
||
draw.text((label_x, label_y), label, fill=label_color, font=font)
|
||
|
||
logger.debug(f"在图像上绘制了 {grid_divisions}x{grid_divisions} 网格（坐标范围: {coord_max_x}x{coord_max_y}）")
|
||
return img_copy
|
||
except Exception as e:
|
||
logger.error(f"绘制网格失败: {e}")
|
||
return image
|
||
|
||
|
||
def draw_last_action_marker(image, x: int, y: int,
|
||
color: Tuple[int, int, int, int] = (0, 255, 0, 200),
|
||
radius: int = 5, width: int = 3):
|
||
"""
|
||
在截图上绘制上次动作位置的绿色圆圈标记
|
||
|
||
Args:
|
||
image: PIL Image对象
|
||
x: 标记的x坐标（绝对像素坐标）
|
||
y: 标记的y坐标（绝对像素坐标）
|
||
color: 圆圈颜色 (R, G, B, A)，默认绿色
|
||
radius: 圆圈半径，默认5像素
|
||
width: 圆圈线宽，默认3像素
|
||
|
||
Returns:
|
||
绘制了标记的图像
|
||
"""
|
||
if Image is None:
|
||
logger.warning("PIL未安装，无法绘制标记")
|
||
return image
|
||
|
||
try:
|
||
from PIL import ImageDraw, ImageFont
|
||
|
||
img_copy = image.copy()
|
||
draw = ImageDraw.Draw(img_copy)
|
||
|
||
# 绘制空心圆圈
|
||
bbox = [x - radius, y - radius, x + radius, y + radius]
|
||
draw.ellipse(bbox, outline=color, width=width)
|
||
# 绘制坐标数字
|
||
draw.text((x, y), f"({x}, {y})", fill=color, font=ImageFont.load_default())
|
||
logger.debug(f"在截图上绘制了上次动作标记: ({x}, {y})")
|
||
return img_copy
|
||
except Exception as e:
|
||
logger.error(f"绘制动作标记失败: {e}")
|
||
return image
|
||
|
||
|
||
def image_to_base64(image) -> str:
|
||
"""
|
||
将 PIL 图像转换为 Base64 编码字符串
|
||
|
||
Args:
|
||
image: PIL Image对象
|
||
|
||
Returns:
|
||
Base64 编码字符串
|
||
"""
|
||
if Image is None:
|
||
return ""
|
||
|
||
try:
|
||
img_byte_arr = io.BytesIO()
|
||
image.save(img_byte_arr, format='PNG')
|
||
img_byte_arr = img_byte_arr.getvalue()
|
||
return base64.b64encode(img_byte_arr).decode('utf-8')
|
||
except Exception as e:
|
||
logger.error(f"图像转Base64失败: {e}")
|
||
return ""
|
||
|
||
|
||
def get_image_size(image_base64: str) -> Tuple[int, int]:
|
||
"""获取图像尺寸"""
|
||
if Image is None:
|
||
return 0, 0
|
||
|
||
try:
|
||
clean = re.sub(r'^data:image/\w+;base64,', '', image_base64)
|
||
img = Image.open(io.BytesIO(base64.b64decode(clean)))
|
||
return img.size
|
||
except Exception:
|
||
return 0, 0
|
||
|
||
|
||
def strip_base64_prefix(base64_str: str) -> str:
|
||
"""移除 Base64 的 data URI 前缀"""
|
||
return re.sub(r'^data:image/\w+;base64,', '', base64_str)
|
||
|
||
|
||
# ==============================================================================
|
||
# Gmail 邮件检查器 (从 GuiAgent/core/utils/gmail_checker.py 迁移)
|
||
# ==============================================================================
|
||
|
||
class GmailChecker:
|
||
"""
|
||
Gmail 邮件检查器，用于 receive_email 动作的邮件获取
|
||
|
||
需要 Google API 相关依赖:
|
||
- google-auth-oauthlib
|
||
- google-api-python-client
|
||
"""
|
||
|
||
SCOPES = ["https://www.googleapis.com/auth/gmail.readonly"]
|
||
|
||
def __init__(self, credentials_path: str = None, token_path: str = None):
|
||
base_dir = os.path.dirname(os.path.abspath(__file__))
|
||
self.credentials_path = credentials_path or os.path.join(base_dir, "credentials.json")
|
||
self.token_path = token_path or os.path.join(base_dir, "token.json")
|
||
self.creds = None
|
||
self.service = None
|
||
self._initialized = False
|
||
|
||
def _ensure_initialized(self) -> bool:
|
||
"""延迟初始化 Gmail 服务"""
|
||
if self._initialized:
|
||
return self.service is not None
|
||
|
||
self._initialized = True
|
||
|
||
try:
|
||
import os.path
|
||
from google.auth.transport.requests import Request
|
||
from google.oauth2.credentials import Credentials
|
||
from google_auth_oauthlib.flow import InstalledAppFlow
|
||
from googleapiclient.discovery import build
|
||
except ImportError as e:
|
||
logger.warning(f"Gmail API 依赖未安装: {e}")
|
||
return False
|
||
|
||
try:
|
||
# 加载存储的凭证
|
||
if os.path.exists(self.token_path):
|
||
self.creds = Credentials.from_authorized_user_file(self.token_path, self.SCOPES)
|
||
|
||
# 如果没有有效凭证，让用户登录
|
||
if not self.creds or not self.creds.valid:
|
||
if self.creds and self.creds.expired and self.creds.refresh_token:
|
||
self.creds.refresh(Request())
|
||
else:
|
||
flow = InstalledAppFlow.from_client_secrets_file(
|
||
self.credentials_path, self.SCOPES
|
||
)
|
||
self.creds = flow.run_local_server(port=0)
|
||
|
||
# 保存凭证以便下次使用
|
||
with open(self.token_path, "w") as token:
|
||
token.write(self.creds.to_json())
|
||
|
||
# 创建 Gmail 服务
|
||
self.service = build("gmail", "v1", credentials=self.creds)
|
||
logger.info("Gmail 服务初始化成功")
|
||
return True
|
||
|
||
except Exception as e:
|
||
logger.error(f"Gmail 服务初始化失败: {e}")
|
||
return False
|
||
|
||
def get_email_content(self, message_id: str) -> Optional[Dict[str, Any]]:
|
||
"""根据邮件ID获取邮件内容"""
|
||
if not self._ensure_initialized():
|
||
return None
|
||
|
||
try:
|
||
from googleapiclient.errors import HttpError
|
||
|
||
message = self.service.users().messages().get(
|
||
userId="me", id=message_id, format="full"
|
||
).execute()
|
||
|
||
# 提取邮件头信息
|
||
headers = message["payload"]["headers"]
|
||
subject = ""
|
||
sender = ""
|
||
date = ""
|
||
|
||
for header in headers:
|
||
if header["name"] == "Subject":
|
||
subject = header["value"]
|
||
elif header["name"] == "From":
|
||
sender = header["value"]
|
||
elif header["name"] == "Date":
|
||
date = header["value"]
|
||
|
||
# 提取邮件正文
|
||
def extract_message_text(part):
|
||
if "parts" in part:
|
||
for subpart in part["parts"]:
|
||
text = extract_message_text(subpart)
|
||
if text:
|
||
return text
|
||
elif part["mimeType"] == "text/plain":
|
||
if "data" in part["body"]:
|
||
return base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8")
|
||
elif part["mimeType"] == "text/html":
|
||
if "data" in part["body"]:
|
||
html_content = base64.urlsafe_b64decode(part["body"]["data"]).decode("utf-8")
|
||
return f"[HTML内容]\n{html_content}"
|
||
return ""
|
||
|
||
message_text = extract_message_text(message["payload"])
|
||
|
||
# 过滤空行
|
||
if message_text:
|
||
lines = message_text.splitlines()
|
||
filtered_lines = [line for line in lines if line.strip()]
|
||
message_text = '\n'.join(filtered_lines)
|
||
|
||
return {
|
||
"id": message_id,
|
||
"sender": sender,
|
||
"subject": subject,
|
||
"date": date,
|
||
"content": message_text
|
||
}
|
||
|
||
except Exception as e:
|
||
logger.error(f"获取邮件内容时出错: {e}")
|
||
return None
|
||
|
||
def is_email_within_timeframe(self, message: Dict, minutes: int = 2) -> bool:
|
||
"""判断邮件是否在指定时间范围内"""
|
||
import time as time_module
|
||
try:
|
||
internal_date = int(message.get('internalDate', 0)) / 1000
|
||
current_time = time_module.time()
|
||
time_diff_minutes = (current_time - internal_date) / 60
|
||
return time_diff_minutes <= minutes
|
||
except Exception as e:
|
||
logger.warning(f"判断邮件时间时出错: {e}")
|
||
return False
|
||
|
||
def wait_for_new_email(
|
||
self,
|
||
polling_interval: int = 10,
|
||
time_window: int = 2,
|
||
max_retries: int = 10
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
阻塞等待新邮件，直到收到指定时间窗口内的新邮件后返回所有符合条件的邮件
|
||
|
||
Args:
|
||
polling_interval: 检查新邮件的时间间隔（秒）
|
||
time_window: 新邮件的时间窗口（分钟），默认2分钟
|
||
max_retries: 最大重试次数
|
||
|
||
Returns:
|
||
符合条件的邮件列表，每封邮件包含 id、sender、subject、date 和 content 字段
|
||
"""
|
||
import time as time_module
|
||
|
||
if not self._ensure_initialized():
|
||
logger.error("Gmail 服务未初始化，无法等待邮件")
|
||
return []
|
||
|
||
logger.info(f"开始监听新邮件，检查间隔: {polling_interval}秒，时间窗口: {time_window}分钟")
|
||
retry_count = 0
|
||
|
||
while retry_count < max_retries:
|
||
try:
|
||
# 检查是否需要刷新凭证
|
||
from google.auth.transport.requests import Request
|
||
if self.creds and self.creds.expired and self.creds.refresh_token:
|
||
self.creds.refresh(Request())
|
||
with open(self.token_path, "w") as token:
|
||
token.write(self.creds.to_json())
|
||
|
||
# 计算时间窗口的起始时间戳
|
||
current_timestamp = int(time_module.time())
|
||
after_timestamp = current_timestamp - (time_window * 60)
|
||
query = f"after:{after_timestamp}"
|
||
|
||
# 获取最新的一批邮件，使用 q 参数过滤时间
|
||
results = self.service.users().messages().list(userId="me", maxResults=10, q=query).execute()
|
||
messages = results.get("messages", [])
|
||
|
||
valid_emails = []
|
||
if messages:
|
||
for msg_summary in messages:
|
||
message_id = msg_summary["id"]
|
||
message = self.service.users().messages().get(
|
||
userId="me", id=message_id, format="metadata"
|
||
).execute()
|
||
|
||
if self.is_email_within_timeframe(message, time_window):
|
||
email_content = self.get_email_content(message_id)
|
||
if email_content:
|
||
valid_emails.append(email_content)
|
||
else:
|
||
# 假设邮件按时间倒序，遇到超时的可以停止检查
|
||
# 但为了保险，可以根据具体需求决定是否 break。
|
||
# 这里简单起见不 break，检查完前10封。
|
||
pass
|
||
|
||
if valid_emails:
|
||
logger.info(f"收到 {len(valid_emails)} 封 {time_window} 分钟内的邮件")
|
||
return valid_emails
|
||
else:
|
||
logger.debug(f"最近的邮件都不在 {time_window} 分钟时间窗口内，继续等待...")
|
||
else:
|
||
logger.debug("没有找到邮件，继续等待...")
|
||
|
||
retry_count += 1
|
||
time_module.sleep(polling_interval)
|
||
|
||
except Exception as e:
|
||
logger.error(f"监听邮件时出错: {e}")
|
||
retry_count += 1
|
||
time_module.sleep(polling_interval)
|
||
|
||
logger.warning(f"已重试 {max_retries} 次，未收到新邮件")
|
||
return []
|
||
|
||
|
||
def receive_email(
|
||
token_path: str = "token.json",
|
||
credentials_path: str = "credentials.json",
|
||
polling_interval: int = 10,
|
||
time_window: int = 2,
|
||
max_retries: int = 10
|
||
) -> List[Dict[str, Any]]:
|
||
"""
|
||
receive_email 动作的便捷函数
|
||
|
||
Args:
|
||
token_path: Gmail token 文件路径
|
||
credentials_path: Gmail credentials 文件路径
|
||
polling_interval: 检查间隔（秒）
|
||
time_window: 时间窗口（分钟）
|
||
max_retries: 最大重试次数
|
||
|
||
Returns:
|
||
邮件内容列表
|
||
"""
|
||
logger.info("开始接收新邮件")
|
||
|
||
try:
|
||
gmail_checker = GmailChecker(
|
||
token_path=token_path,
|
||
credentials_path=credentials_path
|
||
)
|
||
|
||
new_emails = gmail_checker.wait_for_new_email(
|
||
polling_interval=polling_interval,
|
||
time_window=time_window,
|
||
max_retries=max_retries
|
||
)
|
||
|
||
return new_emails if new_emails else []
|
||
|
||
except Exception as e:
|
||
logger.error(f"执行 receive_email 失败: {e}")
|
||
return []
|
||
|
||
def login_ios(device) -> tuple:
|
||
"""
|
||
iOS Apple ID 密码登录脚本
|
||
|
||
在 agent 识别到系统弹出的 Apple ID 登录对话框时调用，
|
||
通过 WDA 直接操作完成密码输入和登录，避免 agent 操作系统级 UI 出错。
|
||
|
||
完整流程：从点击"通过密码登录"按钮开始，覆盖不同页面调用的场景。
|
||
|
||
Args:
|
||
device: IOSDevice 实例，需要有 _wda_client 属性
|
||
Returns:
|
||
(success: bool, message: str)
|
||
"""
|
||
import time as time_module
|
||
|
||
APPLE_PASSWORD = "tsIcmustyy1"
|
||
|
||
# 获取 WDA 客户端
|
||
wda_client = getattr(device, '_wda_client', None)
|
||
if wda_client is None:
|
||
logger.error("[login_ios] 无法获取 WDA 客户端")
|
||
return False, "无法获取 WDA 客户端"
|
||
|
||
try:
|
||
# === 第1步：查找并点击"通过密码登录"相关按钮 ===
|
||
# 覆盖中英文多种表述，确保从不同页面调用都能处理
|
||
password_login_labels = [
|
||
"通过密码登录", "用密码登录", "使用密码登录", "密码登录",
|
||
"Use Password", "Sign In with Password", "Use Password…",
|
||
"Use Password...",
|
||
]
|
||
|
||
clicked_password_btn = False
|
||
for label in password_login_labels:
|
||
try:
|
||
if wda_client(label=label).click_exists(timeout=1.0):
|
||
logger.info(f"[login_ios] 点击了密码登录按钮: {label}")
|
||
clicked_password_btn = True
|
||
time_module.sleep(1.5) # 等待密码输入界面加载
|
||
break
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] 查找按钮 '{label}' 异常: {e}")
|
||
continue
|
||
|
||
if not clicked_password_btn:
|
||
# 可能已经在密码输入页面，尝试直接查找密码输入框
|
||
logger.info("[login_ios] 未找到密码登录按钮，尝试直接查找密码输入框")
|
||
|
||
# === 第2步：查找并操作密码输入框 ===
|
||
password_field = None
|
||
# 尝试多次查找，等待 UI 加载
|
||
for attempt in range(5):
|
||
try:
|
||
# SecureTextField 是 iOS 密码输入框的控件类型
|
||
secure_fields = wda_client(type='SecureTextField')
|
||
if secure_fields.exists:
|
||
password_field = secure_fields
|
||
logger.info(f"[login_ios] 找到密码输入框 (尝试 {attempt + 1})")
|
||
break
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] 查找密码输入框异常 (尝试 {attempt + 1}): {e}")
|
||
time_module.sleep(1.0)
|
||
|
||
if password_field is None:
|
||
logger.warning("[login_ios] 未找到密码输入框")
|
||
return False, "未找到密码输入框，可能当前页面不在密码登录界面"
|
||
|
||
# 点击密码输入框使其获得焦点
|
||
try:
|
||
password_field.tap()
|
||
time_module.sleep(0.5)
|
||
except Exception as e:
|
||
logger.warning(f"[login_ios] 点击密码输入框失败: {e}")
|
||
|
||
# 清除已有内容（避免重复输入导致密码错误）
|
||
try:
|
||
current_value = password_field.get().value
|
||
if current_value:
|
||
# 全选并删除
|
||
password_field.clear_text()
|
||
time_module.sleep(0.3)
|
||
logger.debug("[login_ios] 已清除密码输入框内容")
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] 清除内容时异常（可继续）: {e}")
|
||
|
||
# 输入密码
|
||
try:
|
||
password_field.set_text(APPLE_PASSWORD)
|
||
time_module.sleep(0.5)
|
||
logger.info("[login_ios] 密码输入完成")
|
||
except Exception as e:
|
||
logger.error(f"[login_ios] 密码输入失败: {e}")
|
||
return False, f"密码输入失败: {e}"
|
||
|
||
# === 第3步：点击登录按钮 ===
|
||
sign_in_labels = [
|
||
"登录", "Sign In", "sign in", "登入", "确定", "OK",
|
||
]
|
||
|
||
clicked_sign_in = False
|
||
for label in sign_in_labels:
|
||
try:
|
||
if wda_client(label=label).click_exists(timeout=1.0):
|
||
logger.info(f"[login_ios] 点击了登录按钮: {label}")
|
||
clicked_sign_in = True
|
||
break
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] 查找登录按钮 '{label}' 异常: {e}")
|
||
continue
|
||
|
||
if not clicked_sign_in:
|
||
# 尝试通过 type=Button 查找登录按钮
|
||
try:
|
||
for label in sign_in_labels:
|
||
if wda_client(label=label, type='Button').click_exists(timeout=1.0):
|
||
logger.info(f"[login_ios] 通过 Button 类型点击登录: {label}")
|
||
clicked_sign_in = True
|
||
break
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] Button 类型查找异常: {e}")
|
||
|
||
if not clicked_sign_in:
|
||
logger.warning("[login_ios] 未找到登录按钮，密码已输入但无法提交")
|
||
return False, "密码已输入但未找到登录按钮"
|
||
|
||
# === 第4步：等待并检查登录结果 ===
|
||
time_module.sleep(3) # 等待登录处理
|
||
|
||
# 检查密码输入框是否已消失（消失说明登录成功或进入下一步）
|
||
try:
|
||
if not wda_client(type='SecureTextField').exists:
|
||
logger.info("[login_ios] 登录成功：密码输入框已消失")
|
||
return True, "Apple ID 密码登录成功"
|
||
else:
|
||
# 密码框仍存在，可能密码错误
|
||
logger.warning("[login_ios] 密码通过后输入框仍存在，可能登录需要额外信息填写")
|
||
return False, "登录可能失败：密码通过后输入框仍存在，可能登录需要额外信息填写"
|
||
except Exception as e:
|
||
logger.debug(f"[login_ios] 检查登录结果异常: {e}")
|
||
# 异常时保守认为成功（可能页面已跳转导致元素不存在）
|
||
return True, "登录操作已执行（无法确认结果）"
|
||
|
||
except Exception as e:
|
||
logger.error(f"[login_ios] 执行异常: {e}")
|
||
return False, f"登录脚本异常: {e}"
|
||
|
||
if __name__ == "__main__":
|
||
new_emails = receive_email()
|
||
if new_emails:
|
||
for email in new_emails:
|
||
print(email)
|
||
else:
|
||
print("没有新邮件") |