#!/usr/bin/env python # -*- coding: utf-8 -*- import base64 import getpass import io import json import os import pathlib import queue import re import subprocess import sys import threading import time import traceback from dataclasses import dataclass from datetime import datetime from typing import Any from urllib.parse import quote import requests import tkinter as tk from PIL import Image, ImageTk from requests.adapters import HTTPAdapter from tkinter import messagebox, ttk from urllib3.util.retry import Retry def now_text() -> str: return datetime.now().strftime("%H:%M:%S") def to_text(value: Any, default: str = "") -> str: if value is None: return default return str(value) def to_int(value: Any, default: int = 0) -> int: try: return int(value) except Exception: try: return int(float(value)) except Exception: return default def to_float(value: Any, default: float = 0.0) -> float: try: return float(value) except Exception: return default def to_bool(value: Any, default: bool = False) -> bool: if value is None: return default if isinstance(value, bool): return value text = str(value).strip().lower() if text in {"1", "true", "yes", "y", "on"}: return True if text in {"0", "false", "no", "n", "off"}: return False return default def is_success_result(result: dict[str, Any] | None) -> bool: return isinstance(result, dict) and result.get("成功") is True def is_found_result(result: dict[str, Any] | None) -> bool: return is_success_result(result) and result.get("找到") is True def build_result( success: bool, message: str, data: dict[str, Any] | None = None, raw: str = "", url: str = "", method: str = "", found: bool | None = None, ) -> dict[str, Any]: result = { "成功": bool(success), "消息": to_text(message), "数据": data if isinstance(data, dict) else {}, "原始": to_text(raw), "地址": to_text(url), "方法": to_text(method), } if found is not None: result["找到"] = bool(found) return result RUNTIME_CLONE_VM_NAME_PATTERN = re.compile(r"^Quark-\d{10,}$", re.IGNORECASE) ORDER_API_URLS: list[str] = [ 'https://quark.daifei.top/api.php', 'https://rt.viphc.cn/api.php', ] def is_runtime_clone_vm_name(name: str) -> bool: return bool(RUNTIME_CLONE_VM_NAME_PATTERN.fullmatch(to_text(name).strip())) def should_verify_vm_running_after_power_on_failure(result: dict[str, Any] | None) -> bool: if not isinstance(result, dict): return False message = to_text(result.get("消息")).strip().lower() raw = to_text(result.get("原始")).strip().lower() combined = f"{message}\n{raw}" patterns = ( "read timed out", "json解析失败", "connection aborted", "remote end closed connection", "connection reset", "timeout", ) return any(pattern in combined for pattern in patterns) class OrderApiClient: REQUEST_RETRY_COUNT = 3 REQUEST_RETRY_BACKOFF = 0.8 STATUS_MAP = { "登录成功": "登录成功", "已登录成功": "登录成功", "成功登录": "登录成功", "领取成功": "领取成功", "兑换成功": "领取成功", "提交成功": "领取成功", "成功": "领取成功", "领取失败": "领取失败", "兑换失败": "领取失败", "提交失败": "领取失败", "失败": "领取失败", "验证码错误": "验证码错误", "验证码有误": "验证码错误", "验证码不正确": "验证码错误", "短信验证码错误": "验证码错误", "登录超时": "登录超时", "超时": "登录超时", "验证码超时": "登录超时", "登录失败": "登录失败", "登录不成功": "登录失败", "验证码发送失败": "验证码发送失败", "发送验证码失败": "验证码发送失败", "发码失败": "验证码发送失败", "短信发送失败": "验证码发送失败", "等待用户填写验证码": "等待用户填写验证码", "等待验证码": "等待用户填写验证码", "等待用户输入验证码": "等待用户填写验证码", "等待填写验证码": "等待用户填写验证码", "等待设备接管": "等待设备接管", "待设备接管": "等待设备接管", "等待设备处理": "等待设备接管", } def __init__(self, base_url: str, timeout_ms: int = 30000): self.base_url = base_url.rstrip("/") self.timeout_ms = timeout_ms self.session = requests.Session() retry = Retry( total=self.REQUEST_RETRY_COUNT, connect=self.REQUEST_RETRY_COUNT, read=self.REQUEST_RETRY_COUNT, backoff_factor=self.REQUEST_RETRY_BACKOFF, allowed_methods=frozenset({"POST"}), status_forcelist=[429, 500, 502, 503, 504], raise_on_status=False, ) adapter = HTTPAdapter(max_retries=retry) self.session.mount("http://", adapter) self.session.mount("https://", adapter) def _action_url(self, action: str) -> str: return f"{self.base_url}?action={quote(action, safe='')}" def _post(self, action: str, payload: dict[str, Any], timeout_ms: int | None = None) -> dict[str, Any]: url = self._action_url(action) timeout = (timeout_ms or self.timeout_ms) / 1000.0 try: response = self.session.post( url, json=payload, timeout=timeout, headers={"Content-Type": "application/json;charset=utf-8"}, ) text = response.text except Exception as exc: return build_result(False, f"请求执行异常: {exc}", {}, str(exc), url, "POST") try: data = response.json() except Exception: return build_result(False, "JSON 解析失败", {}, text, url, "POST") code = to_int(data.get("code"), 0) message = to_text(data.get("msg"), "请求成功" if code == 200 else "请求失败") body = data.get("data") if isinstance(data.get("data"), dict) else {} return build_result(code == 200, message, body, text, url, "POST") @staticmethod def _first_non_empty(data: dict[str, Any] | None, keys: list[str], default: Any = "") -> Any: if not isinstance(data, dict): return default for key in keys: value = data.get(key) if value is not None and to_text(value) != "": return value return default @classmethod def _infer_task_type(cls, order_status: str, verify_code: str) -> str: status = to_text(order_status) code = to_text(verify_code) if status in {"待反馈登录结果", "登录成功"}: return "submit_exchange" if status in {"发送验证码中", "等待设备接管"}: return "submit_exchange" if code else "check_qualification" return "" @classmethod def normalize_order(cls, data: dict[str, Any]) -> dict[str, Any] | None: body = data if isinstance(data, dict) else {} task = body.get("task") if isinstance(body.get("task"), dict) else None record = body.get("record") if isinstance(body.get("record"), dict) else {} source = task or record if task is None and not record and to_text(body.get("status")) == "": return None order = dict(source) mobile = to_text(cls._first_non_empty(order, ["mobile", "phone", "phone_no", "phone_num", "mobile_phone"], "")) verify_code = to_text( cls._first_non_empty( body, ["sms_code", "verify_code"], cls._first_non_empty(order, ["verify_code", "sms_code", "code"], ""), ) ) order_status = to_text( cls._first_non_empty( body, ["status", "exchange_status"], cls._first_non_empty(order, ["exchange_status", "status"], ""), ) ) record_id = to_int(cls._first_non_empty(order, ["record_id", "id"], 0), 0) order_no = to_text(cls._first_non_empty(order, ["record_no", "task_order_no", "order_no"], "")) card_key = to_text(cls._first_non_empty(order, ["card_key", "invite_code", "exchange_code"], "")) task_type = to_text(cls._first_non_empty(order, ["task_type"], cls._infer_task_type(order_status, verify_code))) result_message = to_text(cls._first_non_empty(body, ["message"], cls._first_non_empty(order, ["result_message"], ""))) order["record_id"] = record_id order["mobile"] = mobile order["verify_code"] = verify_code order["exchange_status"] = order_status order["task_type"] = task_type order["record_no"] = order_no if card_key: order["card_key"] = card_key return { "订单": order, "记录ID": record_id, "订单编号": order_no, "手机号": mobile, "验证码": verify_code, "卡密": card_key, "订单状态": order_status, "任务类型": task_type, "结果说明": result_message, } def get_order( self, device_no: str, device_name: str = "", remark: str = "", capabilities: str = "", ) -> dict[str, Any]: result = self._post( "device_pull_task", { "device_no": to_text(device_no), "device_name": to_text(device_name), "remark": to_text(remark), "capabilities": to_text(capabilities), }, 20000, ) if result["成功"] is not True: return result order = self.normalize_order(result["数据"]) if order is None: return build_result(False, "暂无订单", result["数据"], result["原始"], result["地址"], result["方法"]) final_result = build_result(True, result["消息"], order, result["原始"], result["地址"], result["方法"]) final_result.update(order) return final_result @classmethod def _order_record_id(cls, order: dict[str, Any] | None) -> int: if not isinstance(order, dict): return 0 if isinstance(order.get("订单"), dict): inner = order["订单"] return to_int(order.get("记录ID"), to_int(inner.get("record_id"), to_int(inner.get("id"), 0))) return to_int(order.get("record_id"), to_int(order.get("id"), 0)) @classmethod def _order_mobile(cls, order: dict[str, Any] | None) -> str: if not isinstance(order, dict): return "" if isinstance(order.get("订单"), dict): return to_text(order.get("手机号"), to_text(order["订单"].get("mobile"), "")) return to_text(order.get("mobile"), "") @classmethod def _order_code(cls, order: dict[str, Any] | None) -> str: if not isinstance(order, dict): return "" if isinstance(order.get("订单"), dict): return to_text(order.get("验证码"), to_text(order["订单"].get("verify_code"), "")) return to_text(order.get("verify_code"), "") @classmethod def default_status_message(cls, status: str) -> str: return { "登录成功": "设备反馈登录成功", "领取成功": "设备反馈领取成功", "领取失败": "设备反馈领取失败", "验证码错误": "设备反馈验证码错误", "登录超时": "设备反馈登录超时", "登录失败": "设备反馈登录失败", "验证码发送失败": "设备反馈验证码发送失败", "等待用户填写验证码": "设备已进入等待验证码阶段", "等待设备接管": "任务重新等待设备接管", }.get(status, "设备反馈订单状态") def feedback_order_status( self, device_no: str, order: dict[str, Any], status: str, message: str | None = None, payload: dict[str, Any] | None = None, ) -> dict[str, Any]: standard_status = self.STATUS_MAP.get(to_text(status)) url = self._action_url("device_submit_result") if not standard_status: return build_result(False, f"不支持的订单状态：{status}", {}, "", url, "POST") record_id = self._order_record_id(order) if record_id <= 0: return build_result(False, "订单 record_id 无效", {}, "", url, "POST") ext = dict(payload or {}) if isinstance(order.get("订单"), dict): inner = order["订单"] ext["task_type"] = to_text(order.get("任务类型"), to_text(inner.get("task_type"), ext.get("task_type", ""))) ext["record_no"] = to_text(order.get("订单编号"), to_text(inner.get("record_no"), ext.get("record_no", ""))) else: ext["task_type"] = to_text(order.get("task_type"), ext.get("task_type", "")) ext["record_no"] = to_text(order.get("record_no"), ext.get("record_no", "")) ext["mobile"] = to_text(ext.get("mobile"), self._order_mobile(order)) ext["verify_code"] = to_text(ext.get("verify_code"), self._order_code(order)) return self._post( "device_submit_result", { "device_no": to_text(device_no), "record_id": record_id, "status": standard_status, "message": to_text(message, self.default_status_message(standard_status)), "payload": ext, }, 20000, ) class VmwareApiClient: def __init__(self, base_url: str, timeout_ms: int = 60000): self.base_url = base_url.rstrip("/") self.timeout_ms = timeout_ms self.session = requests.Session() def _url(self, path: str) -> str: path = path if path.startswith("/") else f"/{path}" return f"{self.base_url}{path}" def _handle_response(self, response: requests.Response, url: str, method: str) -> dict[str, Any]: text = response.text try: data = response.json() except Exception: return build_result(False, "JSON解析失败", {}, text, url, method) success = True message = "" if data.get("ok") is False: success = False if data.get("detail"): success = False message = to_text(data.get("detail")) elif data.get("error"): success = False message = to_text(data.get("error")) elif data.get("message") and data.get("ok") is False: success = False message = to_text(data.get("message")) if not message: message = "请求成功" if success else "请求失败" result = build_result(success, message, data if isinstance(data, dict) else {}, text, url, method) if isinstance(data, dict) and "exists" in data: result["找到"] = to_bool(data.get("exists"), False) return result def _post(self, path: str, body: dict[str, Any], timeout_ms: int | None = None) -> dict[str, Any]: url = self._url(path) timeout = (timeout_ms or self.timeout_ms) / 1000.0 try: response = self.session.post( url, json=body, timeout=timeout, headers={"Content-Type": "application/json;charset=utf-8"}, ) response.raise_for_status() return self._handle_response(response, url, "POST") except requests.HTTPError: try: return self._handle_response(response, url, "POST") except Exception as exc: return build_result(False, f"HTTP错误: {exc}", {}, getattr(response, "text", ""), url, "POST") except Exception as exc: return build_result(False, f"请求执行异常: {exc}", {}, str(exc), url, "POST") def health(self) -> dict[str, Any]: url = self._url("/health") timeout = 10.0 try: response = self.session.get(url, timeout=timeout) response.raise_for_status() return self._handle_response(response, url, "GET") except requests.HTTPError: try: return self._handle_response(response, url, "GET") except Exception as exc: return build_result(False, f"HTTP错误: {exc}", {}, getattr(response, "text", ""), url, "GET") except Exception as exc: return build_result(False, f"请求执行异常: {exc}", {}, str(exc), url, "GET") def list_vms(self) -> dict[str, Any]: url = self._url("/vms") timeout = 30.0 try: response = self.session.get(url, timeout=timeout) response.raise_for_status() return self._handle_response(response, url, "GET") except requests.HTTPError: try: return self._handle_response(response, url, "GET") except Exception as exc: return build_result(False, f"HTTP错误: {exc}", {}, getattr(response, "text", ""), url, "GET") except Exception as exc: return build_result(False, f"请求执行异常: {exc}", {}, str(exc), url, "GET") def get_status(self, vm_ref: str) -> dict[str, Any]: return self._post("/vm/status", {"vm_ref": to_text(vm_ref)}, 15000) def clone_vm(self, source_vm_ref: str, new_name: str, clone_type: str = "full", destination_root: str = "") -> dict[str, Any]: body = { "source_vm_ref": to_text(source_vm_ref), "new_name": to_text(new_name), "clone_type": to_text(clone_type, "full"), } if destination_root: body["destination_root"] = to_text(destination_root) return self._post("/vm/clone", body, 1800000) def power_on(self, vm_ref: str, gui: bool = True) -> dict[str, Any]: return self._post("/vm/power-on", {"vm_ref": to_text(vm_ref), "gui": bool(gui)}, 420000) def power_off(self, vm_ref: str, soft: bool = True) -> dict[str, Any]: return self._post("/vm/power-off", {"vm_ref": to_text(vm_ref), "soft": bool(soft)}, 240000) def delete_vm(self, vm_ref: str, force: bool = True, delete_disk: bool = True) -> dict[str, Any]: return self._post( "/vm/delete", {"vm_ref": to_text(vm_ref), "force": bool(force), "delete_disk": bool(delete_disk)}, 240000, ) def close_vmware_gui(self) -> dict[str, Any]: return self._post("/vmware/close-gui", {}, 30000) def close_quark_all(self, vm_ref: str) -> dict[str, Any]: return self._post("/vm/quark/close-all", {"vm_ref": to_text(vm_ref)}, 60000) def set_quark_default_browser(self, vm_ref: str) -> dict[str, Any]: return self._post("/vm/quark/set-default-browser", {"vm_ref": to_text(vm_ref)}, 90000) def screenshot(self, vm_ref: str, with_ocr: bool = False) -> dict[str, Any]: return self._post("/vm/screenshot", {"vm_ref": to_text(vm_ref), "with_ocr": bool(with_ocr)}, 10000) def click(self, vm_ref: str, x: int, y: int, button: str = "left", double_click: bool = False) -> dict[str, Any]: return self._post( "/vm/click", { "vm_ref": to_text(vm_ref), "x": to_int(x, 0), "y": to_int(y, 0), "button": to_text(button, "left"), "double_click": bool(double_click), }, 45000, ) def find_text( self, vm_ref: str, text: str, match_mode: str = "contains", ignore_case: bool = True, min_score: float = 0.5, index: int = 0, return_image: bool = False, search_left: int | None = None, search_top: int | None = None, search_width: int | None = None, search_height: int | None = None, ) -> dict[str, Any]: body: dict[str, Any] = { "vm_ref": to_text(vm_ref), "text": to_text(text), "match_mode": to_text(match_mode, "contains"), "ignore_case": bool(ignore_case), "min_score": to_float(min_score, 0.5), "index": to_int(index, 0), "return_image": bool(return_image), } if search_left is not None: body["search_left"] = to_int(search_left, 0) if search_top is not None: body["search_top"] = to_int(search_top, 0) if search_width is not None: body["search_width"] = to_int(search_width, 0) if search_height is not None: body["search_height"] = to_int(search_height, 0) return self._post("/vm/find-node", body, 60000) def click_text( self, vm_ref: str, text: str, match_mode: str = "contains", ignore_case: bool = True, min_score: float = 0.5, index: int = 0, button: str = "left", double_click: bool = False, search_left: int | None = None, search_top: int | None = None, search_width: int | None = None, search_height: int | None = None, ) -> dict[str, Any]: body: dict[str, Any] = { "vm_ref": to_text(vm_ref), "text": to_text(text), "match_mode": to_text(match_mode, "contains"), "ignore_case": bool(ignore_case), "min_score": to_float(min_score, 0.5), "index": to_int(index, 0), "button": to_text(button, "left"), "double_click": bool(double_click), } if search_left is not None: body["search_left"] = to_int(search_left, 0) if search_top is not None: body["search_top"] = to_int(search_top, 0) if search_width is not None: body["search_width"] = to_int(search_width, 0) if search_height is not None: body["search_height"] = to_int(search_height, 0) return self._post("/vm/find-and-click", body, 60000) def type_text(self, vm_ref: str, text: str, interval_ms: int = 20, press_enter: bool = False) -> dict[str, Any]: return self._post( "/vm/type-text", { "vm_ref": to_text(vm_ref), "text": to_text(text), "interval_ms": to_int(interval_ms, 20), "press_enter": bool(press_enter), }, 45000, ) def find_ui_node( self, vm_ref: str, text: str | None = None, automation_id: str | None = None, class_name: str | None = None, control_type: str | None = None, framework_id: str | None = None, scope: str = "foreground", max_depth: int = 8, max_nodes: int = 600, index: int = 0, match_mode: str = "contains", ignore_case: bool = True, include_offscreen: bool = False, include_disabled: bool = True, view: str = "control", ) -> dict[str, Any]: body: dict[str, Any] = { "vm_ref": to_text(vm_ref), "scope": to_text(scope, "foreground"), "max_depth": to_int(max_depth, 8), "max_nodes": to_int(max_nodes, 600), "index": to_int(index, 0), "match_mode": to_text(match_mode, "contains"), "ignore_case": bool(ignore_case), "include_offscreen": bool(include_offscreen), "include_disabled": bool(include_disabled), "view": to_text(view, "control"), } if text: body["text"] = to_text(text) if automation_id: body["automation_id"] = to_text(automation_id) if class_name: body["class_name"] = to_text(class_name) if control_type: body["control_type"] = to_text(control_type) if framework_id: body["framework_id"] = to_text(framework_id) return self._post("/vm/find-ui-node", body, 90000) def click_ui_node( self, vm_ref: str, text: str | None = None, automation_id: str | None = None, class_name: str | None = None, control_type: str | None = None, framework_id: str | None = None, scope: str = "foreground", max_depth: int = 8, max_nodes: int = 600, index: int = 0, match_mode: str = "contains", ignore_case: bool = True, include_offscreen: bool = False, include_disabled: bool = True, view: str = "control", button: str = "left", double_click: bool = False, ) -> dict[str, Any]: body: dict[str, Any] = { "vm_ref": to_text(vm_ref), "scope": to_text(scope, "foreground"), "max_depth": to_int(max_depth, 8), "max_nodes": to_int(max_nodes, 600), "index": to_int(index, 0), "match_mode": to_text(match_mode, "contains"), "ignore_case": bool(ignore_case), "include_offscreen": bool(include_offscreen), "include_disabled": bool(include_disabled), "view": to_text(view, "control"), "button": to_text(button, "left"), "double_click": bool(double_click), } if text: body["text"] = to_text(text) if automation_id: body["automation_id"] = to_text(automation_id) if class_name: body["class_name"] = to_text(class_name) if control_type: body["control_type"] = to_text(control_type) if framework_id: body["framework_id"] = to_text(framework_id) return self._post("/vm/click-ui-node", body, 90000) def wait_vm_running(self, vm_ref: str, timeout_seconds: int = 180, interval_seconds: float = 2.0, stop_event: threading.Event | None = None) -> dict[str, Any]: deadline = time.time() + timeout_seconds while time.time() < deadline: if stop_event and stop_event.is_set(): return build_result(False, "用户已停止", {}, "", self._url("/vm/status"), "POST") result = self.get_status(vm_ref) if is_success_result(result): data = result.get("数据", {}) if to_bool(data.get("exists")) and to_bool(data.get("running")): return build_result(True, "虚拟机已运行", data, "", self._url("/vm/status"), "POST") time.sleep(interval_seconds) return build_result(False, "等待虚拟机运行超时", {}, "", self._url("/vm/status"), "POST") def clone_and_power_on(self, source_vm_ref: str, new_name: str) -> dict[str, Any]: combined_result = self._post("/vm/clone-and-power-on", { "source_vm_ref": to_text(source_vm_ref), "new_name": to_text(new_name), "clone_type": "full", "gui": True, }, 360000) if is_success_result(combined_result): data = combined_result.get("数据", {}) return build_result( True, "复制并开机成功", data, "", self._url("/vm/clone-and-power-on"), "POST", ) status_result = self.get_status(new_name) if is_success_result(status_result): status_data = status_result.get("数据", {}) if to_bool(status_data.get("exists")): if to_bool(status_data.get("running")): return build_result( True, "复制并开机成功(超时后确认VM已运行)", status_data, to_text(combined_result.get("原始")), self._url("/vm/clone-and-power-on"), "POST", ) power_result = self.power_on(new_name, True) if is_success_result(power_result): return build_result( True, "复制并开机成功(超时后补充开机)", {"power_on": power_result.get("数据", {}), "status": status_data}, "", self._url("/vm/power-on"), "POST", ) clone_result = self.clone_vm(source_vm_ref, new_name, "full", "") if not is_success_result(clone_result): return clone_result clone_data = clone_result.get("数据", {}) target_vmx = to_text(clone_data.get("target_vmx_path")) power_ref = target_vmx if target_vmx else new_name power_result = self.power_on(power_ref, True) if not is_success_result(power_result) and should_verify_vm_running_after_power_on_failure(power_result): running_result = self.wait_vm_running(power_ref, 240, 2.0) if is_success_result(running_result): return build_result( True, "复制并开机成功(接口超时后已确认虚拟机运行)", { "clone": clone_data, "power_on": power_result.get("数据", {}), "running_check": running_result.get("数据", {}), "recovered_after_timeout": True, }, to_text(power_result.get("原始")), self._url("/vm/power-on"), "POST", ) if not is_success_result(power_result): return power_result return build_result( True, "复制并开机成功", { "clone": clone_data, "power_on": power_result.get("数据", {}), }, "", self._url("/vm/power-on"), "POST", ) def safe_delete_vm( self, vm_ref: str, force: bool = True, delete_disk: bool = True, retry_count: int = 5, interval_seconds: float = 3.0, ) -> dict[str, Any]: self.power_off(vm_ref, False) for _ in range(max(to_int(retry_count, 5), 1)): status_result = self.get_status(vm_ref) data = status_result.get("数据", {}) if isinstance(status_result, dict) else {} if is_success_result(status_result) and not to_bool(data.get("running"), False): delete_result = self.delete_vm(vm_ref, force, delete_disk) if is_success_result(delete_result): return delete_result time.sleep(max(interval_seconds, 0.5)) return self.delete_vm(vm_ref, force, delete_disk) @dataclass class WorkflowConfig: vm_api_url: str device_no: str device_name: str remark: str capabilities: str source_vm_name: str run_count: int browser_cache_wait_seconds: int system_stabilize_wait_seconds: int auto_cleanup: bool class WorkflowRunner(threading.Thread): FEEDBACK_RETRY_COUNT = 10 FEEDBACK_RETRY_INTERVAL_SECONDS = 3.0 def __init__(self, app: "AutomationApp", config: WorkflowConfig): super().__init__(daemon=True) self.app = app self.config = config self.stop_event = threading.Event() self.order_apis: list[OrderApiClient] = [OrderApiClient(url) for url in ORDER_API_URLS] self.order_api_index: int = 0 self.vm_api = VmwareApiClient(config.vm_api_url) self.current_vm_name = "" self.current_order: dict[str, Any] | None = None self.force_cleanup_current_round = False self.force_cleanup_reason = "" def stop(self) -> None: self.stop_event.set() def log(self, message: str) -> None: self.app.enqueue(("log", f"[{now_text()}] {message}")) def set_claim_status(self, status: str) -> None: self.app.set_claim_status(status) def set_current_vm(self, vm_name: str) -> None: self.current_vm_name = vm_name self.app.enqueue(("current_vm", vm_name)) def request_force_cleanup(self, reason: str) -> None: self.force_cleanup_current_round = True self.force_cleanup_reason = to_text(reason).strip() def sleep_with_stop(self, seconds: float) -> bool: deadline = time.time() + seconds while time.time() < deadline: if self.stop_event.is_set(): return False time.sleep(0.2) return True def wait_with_progress(self, seconds: int, prefix: str) -> bool: total = max(to_int(seconds, 0), 0) for remaining in range(total, 0, -1): self.ensure_running() self.log(f"{prefix}，剩余 {remaining} 秒") if not self.sleep_with_stop(1): return False return True def ensure_running(self) -> None: if self.stop_event.is_set(): raise RuntimeError("用户已停止") @staticmethod def text_matches(candidate: str, query: str, match_mode: str = "contains", ignore_case: bool = True) -> bool: candidate_text = to_text(candidate) query_text = to_text(query) if not query_text: return True if match_mode == "regex": flags = re.IGNORECASE if ignore_case else 0 try: return re.search(query_text, candidate_text, flags) is not None except re.error: return False if ignore_case: candidate_text = candidate_text.lower() query_text = query_text.lower() if match_mode == "exact": return candidate_text == query_text return query_text in candidate_text @staticmethod def clamp_search_region( width: int, height: int, search_left: int | None = None, search_top: int | None = None, search_width: int | None = None, search_height: int | None = None, ) -> tuple[int, int, int, int]: left = max(to_int(search_left, 0), 0) if search_left is not None else 0 top = max(to_int(search_top, 0), 0) if search_top is not None else 0 if left > width: left = width if top > height: top = height if search_width is None: region_width = max(0, width - left) else: region_width = max(0, min(to_int(search_width, 0), width - left)) if search_height is None: region_height = max(0, height - top) else: region_height = max(0, min(to_int(search_height, 0), height - top)) return left, top, region_width, region_height def capture_ocr_snapshot(self, vm_name: str, stage: str = "") -> dict[str, Any] | None: result = self.vm_api.screenshot(vm_name, True) if not is_success_result(result): prefix = f"{stage} | " if stage else "" self.log(f"{prefix}截图OCR失败：{result.get('消息')}") return None data = result.get("数据", {}) if isinstance(result.get("数据"), dict) else {} if data.get("ocr_available") is False: prefix = f"{stage} | " if stage else "" backend = to_text(data.get("ocr_backend")).strip() or "unknown" error_text = to_text(data.get("ocr_error")).strip() message = f"OCR引擎不可用，当前后端={backend}" if error_text: message = f"{message}，原因={error_text}" self.log(f"{prefix}{message}") return None return { "vm_name": vm_name, "width": to_int(data.get("width"), 0), "height": to_int(data.get("height"), 0), "ocr_items": data.get("ocr_items") if isinstance(data.get("ocr_items"), list) else [], } def find_text_in_snapshot( self, snapshot: dict[str, Any] | None, text: str, match_mode: str = "contains", ignore_case: bool = True, min_score: float = 0.5, index: int = 0, search_left: int | None = None, search_top: int | None = None, search_width: int | None = None, search_height: int | None = None, ) -> dict[str, Any] | None: if not isinstance(snapshot, dict): return None width = to_int(snapshot.get("width"), 0) height = to_int(snapshot.get("height"), 0) region_left, region_top, region_width, region_height = self.clamp_search_region( width, height, search_left, search_top, search_width, search_height, ) region_right = region_left + region_width region_bottom = region_top + region_height matches: list[dict[str, Any]] = [] for item in snapshot.get("ocr_items", []): if not isinstance(item, dict): continue score = to_float(item.get("score"), 0.0) if score < min_score: continue bounds = item.get("bounds") if isinstance(item.get("bounds"), dict) else {} center_x = to_int(bounds.get("center_x"), -1) center_y = to_int(bounds.get("center_y"), -1) if center_x < region_left or center_x >= region_right or center_y < region_top or center_y >= region_bottom: continue if not self.text_matches(to_text(item.get("text")), text, match_mode, ignore_case): continue matches.append( { "text": to_text(item.get("text")), "score": score, "points": item.get("points") if isinstance(item.get("points"), list) else [], "bounds": bounds, } ) if 0 <= index < len(matches): return matches[index] return None def click_text_in_snapshot( self, vm_name: str, snapshot: dict[str, Any] | None, text: str, match_mode: str = "contains", ignore_case: bool = True, min_score: float = 0.5, index: int = 0, button: str = "left", double_click: bool = False, search_left: int | None = None, search_top: int | None = None, search_width: int | None = None, search_height: int | None = None, ) -> bool: match = self.find_text_in_snapshot( snapshot, text, match_mode, ignore_case, min_score, index, search_left, search_top, search_width, search_height, ) if not match: return False bounds = match.get("bounds") if isinstance(match.get("bounds"), dict) else {} center_x = to_int(bounds.get("center_x"), -1) center_y = to_int(bounds.get("center_y"), -1) if center_x < 0 or center_y < 0: return False self.vm_api.click(vm_name, center_x, center_y, button, double_click) return True def click_refresh_after_desktop_context_menu(self, vm_name: str) -> bool: snapshot = self.capture_ocr_snapshot(vm_name, "桌面右键菜单") if snapshot is None: return False if self.click_text_in_snapshot(vm_name, snapshot, "刷新", "contains", True, 0.5, 0, "left", False): return True self.log('右键菜单中未识别到"刷新"') return False def vm_exists_for_cleanup(self, vm_name: str) -> bool: if not vm_name.strip(): return False try: status_result = self.vm_api.get_status(vm_name) except Exception as exc: self.log(f"检查虚拟机是否存在失败[{vm_name}]：{exc}") return False data = status_result.get("数据", {}) if isinstance(status_result.get("数据"), dict) else {} return to_bool(data.get("exists"), False) def call_and_log(self, title: str, func, *args, **kwargs) -> dict[str, Any]: result = func(*args, **kwargs) success = to_text(result.get("成功")) message = to_text(result.get("消息")) self.log(f"{title} | 成功={success} | 消息={message}") return result def run(self) -> None: self.app.enqueue(("running", True)) try: index = 0 total = max(self.config.run_count, 1) while index < total and not self.stop_event.is_set(): self.ensure_running() index += 1 self.log(f"========== 第 {index}/{total} 次运行开始 ==========") clone_ok = self.run_once(index) self.log(f"========== 第 {index}/{total} 次运行结束 ==========") if not clone_ok: index -= 1 self.log("复制并发延迟15秒") if not self.sleep_with_stop(15): break continue if index < total and not self.stop_event.is_set(): self.sleep_with_stop(2) except RuntimeError as exc: self.log(str(exc)) except Exception: self.log("脚本执行异常：") self.log(traceback.format_exc()) finally: self.app.enqueue(("running", False)) def run_once(self, loop_index: int) -> bool: self.current_order = None self.force_cleanup_current_round = False self.force_cleanup_reason = "" self._clone_failed = False vm_name = f"Quark-{int(time.time() * 1000)}" self.set_current_vm(vm_name) self.log(f"本次虚拟机名称：{vm_name}") try: self.log("开始复制虚拟机，预计 180 秒完成") result = self.call_and_log("复制并开机", self.vm_api.clone_and_power_on, self.config.source_vm_name, vm_name) if not is_success_result(result): self._clone_failed = True return False self.log("等待虚拟机运行...") result = self.vm_api.wait_vm_running(vm_name, 180, 2, self.stop_event) self.log(f"等待虚拟机运行 | 成功={result.get('成功')} | 消息={result.get('消息')}") if not is_success_result(result): return health_result = self.vm_api.health() health_data = health_result.get("数据", {}) if isinstance(health_result.get("数据"), dict) else {} if is_success_result(health_result) and health_data.get("ocr_available") is False: backend = to_text(health_data.get("ocr_backend")).strip() or "unknown" error_text = to_text(health_data.get("ocr_error")).strip() reason = f"OCR引擎不可用，当前后端={backend}" if error_text: reason = f"{reason}，原因={error_text}" self.request_force_cleanup(f"{reason}，结束本次任务") self.log(reason) return if not self.wait_for_boot_desktop_signal(vm_name): return self.log("执行桌面刷新动作") self.vm_api.click(vm_name, 535, 308, "right", False) if not self.sleep_with_stop(2): return self.click_refresh_after_desktop_context_menu(vm_name) if not self.sleep_with_stop(2): return self.refresh_until_quark_visible(vm_name) if self.stop_event.is_set(): return if not self.phase_open_quark(vm_name): return if not self.phase_get_order(): return try: status_result = self.vm_api.get_status(vm_name) status_data = status_result.get('数据', {}) if isinstance(status_result.get('数据'), dict) else {} vm_running = to_bool(status_data.get('running'), False) except Exception as exc: self.log(f'取单后检查VM运行状态失败：{exc}，直接执行清理') self.request_force_cleanup('取单后VM状态检查异常，结束本次任务') return if not vm_running: self.log('取单后检测到VM未处于开机状态，直接执行清理') self.request_force_cleanup('取单后VM已关机，结束本次任务') return self.phase_login_and_judge(vm_name) finally: vm_exists = self.vm_exists_for_cleanup(vm_name) if self.stop_event.is_set(): if vm_exists: self.log("脚本已停止，保留当前虚拟机，不自动关机或删除") else: self.log("脚本已停止，当前虚拟机尚未创建成功，跳过清理") elif (self.config.auto_cleanup or self.force_cleanup_current_round) and vm_exists: if self.force_cleanup_current_round and self.force_cleanup_reason: self.log(self.force_cleanup_reason) self.log("开始安全删除虚拟机") delete_result = self.vm_api.safe_delete_vm(vm_name, True, True, 5, 3) self.log(f"安全删除虚拟机 | 成功={delete_result.get('成功')} | 消息={delete_result.get('消息')}") try: self.vm_api.close_vmware_gui() except Exception: pass elif (self.config.auto_cleanup or self.force_cleanup_current_round) and not vm_exists: if self.force_cleanup_current_round and self.force_cleanup_reason: self.log(self.force_cleanup_reason) self.log("当前虚拟机未创建成功，跳过自动删除") elif vm_exists: self.log("当前轮次结束，已保留虚拟机，等待手动处理") else: self.log("当前轮次未生成可保留的虚拟机") return not self._clone_failed def wait_for_boot_desktop_signal(self, vm_name: str) -> bool: silent_seconds = 80 detect_seconds = 100 total_seconds = silent_seconds + detect_seconds self.log(f"VM已运行，开机稳定检测共 {total_seconds} 秒(前 {silent_seconds} 秒静默，后 {detect_seconds} 秒检测)") if not self.sleep_with_stop(silent_seconds): return False self.log(f"静默等待结束，开始OCR检测，间隔10秒，最长 {detect_seconds} 秒") deadline = time.time() + detect_seconds attempt = 0 while time.time() < deadline and not self.stop_event.is_set(): attempt += 1 snapshot = self.capture_ocr_snapshot(vm_name, f"开机检测第 {attempt} 次") if snapshot is not None: recycle_bin_match = self.find_text_in_snapshot(snapshot, "回收站", "contains", True, 0.5, 0) if recycle_bin_match: self.log("开机检测已识别到回收站，提前跳出") return self.wait_with_progress(10, "开机成功后额外等待10秒") remaining_seconds = max(0, int(deadline - time.time())) self.log(f"开机检测第 {attempt} 次未识别到回收站，剩余约 {remaining_seconds} 秒") wait_seconds = min(10.0, deadline - time.time()) if wait_seconds > 0 and not self.sleep_with_stop(wait_seconds): return False if self.stop_event.is_set(): return False self.log(f"开机检测在 {detect_seconds} 秒内未识别到回收站，继续后续流程") return True def refresh_until_quark_visible(self, vm_name: str) -> None: for index in range(1, 11): self.ensure_running() self.log(f"桌面刷新检测夸克图标，第 {index}/10 次") self.vm_api.click(vm_name, 535, 308, "right", False) if not self.sleep_with_stop(2): return self.click_refresh_after_desktop_context_menu(vm_name) if not self.sleep_with_stop(2): return snapshot = self.capture_ocr_snapshot(vm_name, "桌面刷新后检测夸克") quark_match = self.find_text_in_snapshot(snapshot, "夸克", "contains", True, 0.5, 0) if quark_match: self.log("刷新后已检测到夸克文字，继续后续流程") return self.log("本次刷新后还未检测到夸克，等待 3 秒再试") if not self.sleep_with_stop(3): return self.log("连续 10 次刷新后仍未检测到夸克，继续尝试打开夸克流程") def phase_open_quark(self, vm_name: str) -> bool: self.log("进入打开夸克和默认浏览器流程") deadline = time.time() + 4 * 60 activity_entry_click_count = 0 activity_clicked = False while time.time() < deadline: self.ensure_running() self.log("打开夸克，处理默认浏览器流程") snapshot = self.capture_ocr_snapshot(vm_name, "打开夸克阶段") if snapshot is None: if not self.sleep_with_stop(0.1): return False continue if self.click_text_in_snapshot(vm_name, snapshot, "登录并领取", "contains", True, 0.5, 0, "left", False, 578, 138, 684, 215): self.log('检测到"登录并领取"，优先尝试点击') if not self.wait_for_phone_dialog_after_login_entry(vm_name): return False continue confirm_match = self.find_text_in_snapshot(snapshot, "确认登录", "contains", True, 0.5, 0, 564, 461, 798, 504) if confirm_match: self.log('已检测到"确认登录"，准备进入取单流程') if not self.sleep_with_stop(0.1): return False return True phone_match = self.find_text_in_snapshot(snapshot, "手机", "contains", True, 0.5, 0, 598, 498, 699, 533) if phone_match: self.log("检测到手机号登录入口，优先点击进入") self.vm_api.click(vm_name, 622, 487, "left", False) if not self.sleep_with_stop(0.1): return False continue desktop_match = self.find_text_in_snapshot(snapshot, "回收站", "contains", True, 0.5, 0) if desktop_match and self.click_text_in_snapshot(vm_name, snapshot, "夸克", "contains", True, 0.5, 0, "left", True): self.log("检测到桌面，尝试双击夸克") if not self.wait_with_progress(self.config.browser_cache_wait_seconds, "等待浏览器加载缓存"): return False continue browser_match = self.find_text_in_snapshot(snapshot, "输入网址", "contains", True, 0.5, 0) if browser_match: if activity_clicked: self.log("已进入浏览器且活动已点击成功，点击个人中心入口") self.vm_api.click(vm_name, 970, 59, "left", False) if not self.sleep_with_stop(0.1): return False continue activity_entry_click_count += 1 self.log("已进入浏览器，点击活动入口") self.log(f"活动入口点击次数：{activity_entry_click_count}/10") if activity_entry_click_count > 10: self.request_force_cleanup("活动入口连续点击超过 10 次，按打开夸克超时处理，结束本次任务") return False self.vm_api.click(vm_name, 999, 60, "left", False) self.log("已点击活动入口第一个坐标，等待 2 秒检查活动是否加载") if not self.sleep_with_stop(2): return False activity_snapshot = self.capture_ocr_snapshot(vm_name, "活动入口点击后检查") if activity_snapshot is None: if not self.sleep_with_stop(0.1): return False continue activity_loaded = ( self.find_text_in_snapshot(activity_snapshot, "7天", "contains", True, 0.5, 0) or self.find_text_in_snapshot(activity_snapshot, "SVIP", "contains", True, 0.5, 0) or self.find_text_in_snapshot(activity_snapshot, "网盘SVIP", "contains", True, 0.5, 0) ) if activity_loaded: self.log("识别到活动入口（7天/SVIP），点击活动入口第二个坐标") self.vm_api.click(vm_name, 993, 109, "left", False) if not self.sleep_with_stop(0.1): return False login_found = False for _ in range(20): if self.stop_event.is_set(): return False poll_snap = self.capture_ocr_snapshot(vm_name, '活动点击后等待登录并领取') if poll_snap is not None and self.find_text_in_snapshot(poll_snap, '登录并领取', 'contains', True, 0.5, 0, 578, 138, 684, 215): self.log('活动点击后检测到"登录并领取"，提前跳出') login_found = True break if not self.sleep_with_stop(1): return False activity_clicked = login_found else: new_tab_match = self.find_text_in_snapshot(activity_snapshot, "打开新的标签页", "contains", True, 0.5, 0) if new_tab_match: self.log('活动未加载完成，检测到"打开新的标签页"，回点第一个坐标并等待 10 秒') self.vm_api.click(vm_name, 999, 60, "left", False) else: self.log('活动未加载完成，未检测到"打开新的标签页"，等待 10 秒') if not self.wait_with_progress(10, "活动未加载完成等待10秒"): return False continue if self.click_text_in_snapshot(vm_name, snapshot, "前往设置", "contains", True, 0.5, 0, "left", False): self.log('检测到"前往设置"，尝试点击') if not self.sleep_with_stop(0.1): return False continue edge_match = self.find_text_in_snapshot(snapshot, "Edge", "contains", True, 0.5, 0, 368, 544, 718, 638) if edge_match: self.log("检测到默认浏览器仍是 Edge，切换到夸克") if self.click_text_in_snapshot(vm_name, snapshot, "Edge", "contains", True, 0.5, 0, "left", False, 368, 544, 718, 638): if not self.sleep_with_stop(1): return False self.vm_api.click(vm_name, 475, 457, "left", False) if not self.sleep_with_stop(1): return False self.vm_api.click(vm_name, 814, 49, "left", False) if not self.sleep_with_stop(0.1): return False continue phone_input_match = self.find_text_in_snapshot(snapshot, '请输入手机号', 'contains', True, 0.5, 0, 559, 358, 696, 387) if phone_input_match: self.log('检测到"请输入手机号"，点击文字返回') self.vm_api.click(vm_name, 561, 135, 'left', False) if not self.sleep_with_stop(0.1): return False continue phone_entry2_match = self.find_text_in_snapshot(snapshot, '手机', 'contains', True, 0.5, 0, 588, 446, 657, 534) if phone_entry2_match: self.log('检测到手机登录入口（手机），点击进入') self.vm_api.click(vm_name, 622, 490, 'left', False) if not self.sleep_with_stop(0.1): return False continue if self.click_text_in_snapshot(vm_name, snapshot, "别的时间", "contains", True, 0.5, 0, "left", False): self.log('检测到"别的时间"，按最低优先级尝试点击') if not self.sleep_with_stop(0.1): return False continue if not self.sleep_with_stop(0.1): return False self.log("设置默认浏览器超时") return False def wait_for_phone_dialog_after_login_entry(self, vm_name: str) -> bool: timeout_seconds = 15 poll_seconds = 3 deadline = time.time() + timeout_seconds attempt = 0 self.log('点击"登录并领取"后，开始等待手机号登录窗口，最长 15 秒') while time.time() < deadline and not self.stop_event.is_set(): attempt += 1 snapshot = self.capture_ocr_snapshot(vm_name, f"等待手机号登录窗口第 {attempt} 次") if snapshot is not None: phone_match = self.find_text_in_snapshot(snapshot, "手机", "contains", True, 0.5, 0, 598, 498, 699, 533) if phone_match: self.log('点击"登录并领取"后已识别到"手机"，继续后续流程') return True remaining_seconds = max(0, int(deadline - time.time())) self.log(f'点击"登录并领取"后第 {attempt} 次仍未识别到"手机"，剩余约 {remaining_seconds} 秒') wait_seconds = min(float(poll_seconds), deadline - time.time()) if wait_seconds > 0 and not self.sleep_with_stop(wait_seconds): return False if self.stop_event.is_set(): return False self.log('点击"登录并领取"后 15 秒内仍未识别到"手机"，继续外层流程重试') return True def handle_profile_member_result(self, vm_name: str) -> bool: self.log('最终结果判断逻辑：立即截图找成功 → 等8秒找成功 → 点左上角等5秒找成功/失败 → 无结果继续循环') L, T, W, H = 296, 325, 711, 69 def check_success(snap: dict[str, Any] | None, stage: str) -> bool: if snap is None: return False vip = self.find_text_in_snapshot(snap, '网盘SVIP会员', 'contains', True, 0.5, 0, L, T, W, H) not_opened = self.find_text_in_snapshot(snap, '尚未开通', 'contains', True, 0.5, 0, L, T, W, H) if vip and not not_opened: self.log(f'{stage}：检测到"网盘SVIP会员"且无"尚未开通"，领取成功') self.feedback('领取成功') return True self.log(f'{stage}：未检测到成功标志，继续等待') return False def check_all(snap: dict[str, Any] | None, stage: str) -> bool: if snap is None: return False vip = self.find_text_in_snapshot(snap, '网盘SVIP会员', 'contains', True, 0.5, 0, L, T, W, H) not_opened = self.find_text_in_snapshot(snap, '尚未开通', 'contains', True, 0.5, 0, L, T, W, H) expired = self.find_text_in_snapshot(snap, '已过期', 'contains', True, 0.5, 0, L, T, W, H) renew = self.find_text_in_snapshot(snap, '续费恢复', 'contains', True, 0.5, 0, L, T, W, H) if not_opened or expired or renew: self.log(f'{stage}：检测到失败标志（尚未开通/已过期/续费恢复），领取失败') self.feedback('领取失败') return True if vip: self.log(f'{stage}：检测到"网盘SVIP会员"且无失败标志，领取成功') self.feedback('领取成功') return True self.log(f'{stage}：未检测到任何判断标志，继续等待') return False snap1 = self.capture_ocr_snapshot(vm_name, '会员判断第1次（立即）') if check_success(snap1, '第1次判断'): return False if not self.wait_with_progress(8, '会员判断等待8秒'): return False snap2 = self.capture_ocr_snapshot(vm_name, '会员判断第2次（等8秒后）') if check_success(snap2, '第2次判断'): return False self.log('点击左上角(86,60)，等待5秒') self.vm_api.click(vm_name, 86, 60, 'left', False) if not self.wait_with_progress(5, '点击左上角后等待5秒'): return False snap3 = self.capture_ocr_snapshot(vm_name, '会员判断第3次（点左上角后）') if check_all(snap3, '第3次判断'): return False self.log('三次判断均无结果，点击左上角继续循环') self.vm_api.click(vm_name, 86, 60, 'left', False) if not self.wait_with_progress(5, '继续循环前等待5秒'): return False return True def phase_get_order(self) -> bool: self.log('开始获取订单') self.set_claim_status('等待接单') api_count = len(self.order_apis) while not self.stop_event.is_set(): for i in range(api_count): if self.stop_event.is_set(): return False idx = i % api_count api = self.order_apis[idx] api_url = ORDER_API_URLS[idx] result = api.get_order( self.config.device_no, self.config.device_name, self.config.remark, self.config.capabilities, ) if is_success_result(result): self.current_order = result self.order_api_index = idx self.set_claim_status('待领取') self.log(f'获取订单成功 | 来源={api_url} | 手机号={result.get("手机号")} | 验证码={result.get("验证码")}') return True message = to_text(result.get('消息')) if '请求执行异常' in message or 'JSON 解析失败' in message: self.log(f'取单接口网络异常[{api_url}]：{message}') else: self.log(f'暂无订单或取单失败[{api_url}]：{message}') if not self.sleep_with_stop(3): return False return False def feedback(self, status: str, message: str | None = None) -> dict[str, Any]: order = self.current_order or {} payload: dict[str, Any] = {} self.set_claim_status(status) screenshot_payload = self.capture_final_screenshot_payload(status) if screenshot_payload: payload.update(screenshot_payload) api = self.order_apis[self.order_api_index] api_url = ORDER_API_URLS[self.order_api_index] last_result = build_result(False, '未开始反馈') for attempt in range(1, self.FEEDBACK_RETRY_COUNT + 1): result = api.feedback_order_status(self.config.device_no, order, status, message, payload or None) last_result = result self.log( f'反馈订单状态[{status}] 第{attempt}/{self.FEEDBACK_RETRY_COUNT}次 | ' f'来源={api_url} | 成功={result.get("成功")} | 消息={result.get("消息")}' ) if is_success_result(result): return result if attempt < self.FEEDBACK_RETRY_COUNT: self.log(f'反馈失败，{int(self.FEEDBACK_RETRY_INTERVAL_SECONDS)} 秒后重试') time.sleep(self.FEEDBACK_RETRY_INTERVAL_SECONDS) self.log(f'反馈订单状态[{status}] 连续失败 {self.FEEDBACK_RETRY_COUNT} 次，停止重试') return last_result def capture_final_screenshot_payload(self, claim_status: str | None = None) -> dict[str, Any]: vm_name = to_text(self.current_vm_name).strip() if not vm_name: self.log("最终截图抓取跳过：当前虚拟机名称为空") return {} self.log(f"开始抓取最终截图并随回传上传：{vm_name}") result = self.vm_api.screenshot(vm_name, False) if not is_success_result(result): self.log(f"最终截图抓取失败：{result.get('消息')}") return {} meta = result.get("数据", {}) if isinstance(result.get("数据"), dict) else {} image_base64 = to_text(meta.get("image_base64")).strip() if not image_base64: self.log("最终截图抓取成功，但接口未返回 image_base64") return {} try: image_bytes = base64.b64decode(image_base64) preview_claim_status = to_text(claim_status).strip() or self.app.claim_status_var.get().strip() self.app.enqueue( ( "image", { "vm_name": vm_name, "image_bytes": image_bytes, "width": meta.get("width"), "height": meta.get("height"), "claim_status": preview_claim_status, "captured_at": now_text(), }, ) ) except Exception as exc: self.log(f"最终截图预览更新失败：{exc}") self.log("最终截图已写入回传 payload") return { "final_screenshot": f"data:image/png;base64,{image_base64}", "vm_name": vm_name, "screen_width": meta.get("width"), "screen_height": meta.get("height"), "captured_at": datetime.now().strftime("%Y-%m-%d %H:%M:%S"), } def phase_login_and_judge(self, vm_name: str) -> None: if not self.current_order: self.log("当前没有订单，终止登录流程") return mobile = to_text(self.current_order.get("手机号")) verify_code = to_text(self.current_order.get("验证码")) self.log("进入登录和会员判断流程") self.set_claim_status("登录中") deadline = time.time() + 3 * 60 while time.time() < deadline and not self.stop_event.is_set(): if not self.sleep_with_stop(0.3): return self.log("登录-判断会员状态") snapshot = self.capture_ocr_snapshot(vm_name, "登录判断阶段") if snapshot is None: if not self.sleep_with_stop(0.1): return continue if self.click_text_in_snapshot(vm_name, snapshot, "别的时间", "contains", True, 0.5, 0, "left", False): self.log('检测到"别的时间"，尝试点击') if not self.sleep_with_stop(0.1): return continue phone_input_match = self.find_text_in_snapshot(snapshot, '请输入手机号', 'contains', True, 0.5, 0, 559, 358, 696, 387) if phone_input_match: self.log('检测到"请输入手机号"，点击返回') self.vm_api.click(vm_name, 561, 135, 'left', False) if not self.sleep_with_stop(0.1): return continue phone_entry_top = self.find_text_in_snapshot(snapshot, '手机', 'contains', True, 0.5, 0, 588, 446, 657, 534) if phone_entry_top: self.log('检测到手机登录入口（高优先级），点击进入') self.vm_api.click(vm_name, 622, 490, 'left', False) if not self.sleep_with_stop(0.1): return continue input_mobile = ( self.find_text_in_snapshot(snapshot, "输入手机号", "contains", True, 0.5, 0, 563, 262, 804, 299) or self.find_text_in_snapshot(snapshot, "输入手机号", "contains", True, 0.5, 0, 561, 266, 809, 297) ) if input_mobile: self.log("输入手机号") self.vm_api.click(vm_name, 602, 281, "left", False) if not self.sleep_with_stop(0.5): return self.vm_api.type_text(vm_name, mobile, 20, False) if not self.sleep_with_stop(0.1): return continue input_code = ( self.find_text_in_snapshot(snapshot, "手机验证码", "contains", True, 0.5, 0, 559, 316, 715, 352) or self.find_text_in_snapshot(snapshot, "手机验证码", "contains", True, 0.5, 0, 565, 316, 731, 350) ) if input_code: self.log("输入验证码") self.vm_api.click(vm_name, 607, 336, "left", False) if not self.sleep_with_stop(0.5): return self.vm_api.type_text(vm_name, verify_code, 20, False) if not self.sleep_with_stop(0.1): return continue wrong_code = self.find_text_in_snapshot(snapshot, "验证码错误", "contains", True, 0.5, 0, 552, 348, 803, 393) if wrong_code: self.log("检测到验证码错误，准备回传") self.feedback("验证码错误") return confirm_match = self.find_text_in_snapshot(snapshot, "确认登录", "contains", True, 0.5, 0, 564, 461, 798, 504) if confirm_match: self.log("检测到确认登录，等2秒后验证输入框是否消失") if not self.sleep_with_stop(2): return verify_snap = self.capture_ocr_snapshot(vm_name, "确认登录前二次验证") has_mobile_input = ( self.find_text_in_snapshot(verify_snap, "输入手机号", "contains", True, 0.5, 0, 563, 262, 804, 299) or self.find_text_in_snapshot(verify_snap, "输入手机号", "contains", True, 0.5, 0, 561, 266, 809, 297) ) if verify_snap else None has_code_input = ( self.find_text_in_snapshot(verify_snap, "手机验证码", "contains", True, 0.5, 0, 559, 316, 715, 352) or self.find_text_in_snapshot(verify_snap, "手机验证码", "contains", True, 0.5, 0, 565, 316, 731, 350) ) if verify_snap else None if not has_mobile_input and not has_code_input: self.log("输入框已消失，点击确认登录") self.vm_api.click(vm_name, to_int(confirm_match["bounds"].get("center_x"), 0), to_int(confirm_match["bounds"].get("center_y"), 0), "left", False) else: self.log("输入框仍存在，跳过确认登录点击，继续循环") if not self.sleep_with_stop(0.1): return continue profile_page = self.find_text_in_snapshot(snapshot, "个人中心", "contains", True, 0.5, 0, 15, 175, 187, 60) if profile_page: self.log("个人中心内，判断会员状态") should_continue = self.handle_profile_member_result(vm_name) if self.stop_event.is_set(): return if should_continue: continue return home_page = self.find_text_in_snapshot(snapshot, "输入网址", "contains", True, 0.5, 0, 173, 360, 308, 389) if home_page: self.log("检测到主页，点击个人中心入口") self.vm_api.click(vm_name, 970, 59, "left", False) if not self.sleep_with_stop(0.1): return continue phone_entry = self.find_text_in_snapshot(snapshot, "手机", "contains", True, 0.5, 0, 598, 498, 699, 533) if phone_entry: self.log("检测到手机号登录入口，点击进入") self.vm_api.click(vm_name, 622, 487, "left", False) if not self.sleep_with_stop(0.5): return continue if self.click_text_in_snapshot(vm_name, snapshot, "登录并领取", "contains", True, 0.5, 0, "left", False, 578, 138, 684, 215): self.log('检测到"登录并领取"，尝试点击') if not self.wait_for_phone_dialog_after_login_entry(vm_name): return continue if not self.sleep_with_stop(0.1): return if not self.stop_event.is_set(): self.log("登录判断超时") self.feedback("登录超时") class ScreenshotWorker(threading.Thread): def __init__(self, app: "AutomationApp", vm_api_url: str, get_vm_name): super().__init__(daemon=True) self.app = app self.vm_api = VmwareApiClient(vm_api_url) self.get_vm_name = get_vm_name self.stop_event = threading.Event() def stop(self) -> None: self.stop_event.set() def run(self) -> None: self.app.enqueue(("screenshot_running", True)) try: while not self.stop_event.is_set(): vm_name = to_text(self.get_vm_name()).strip() interval = max(to_float(self.app.screenshot_interval_var.get(), 2.0), 0.5) if not vm_name: self.app.enqueue(("log", f"[{now_text()}] 截图线程等待虚拟机名称")) self.stop_event.wait(interval) continue result = self.vm_api.screenshot(vm_name, False) if is_success_result(result): image_base64 = to_text(result.get("数据", {}).get("image_base64")) if image_base64: try: image_bytes = base64.b64decode(image_base64) meta = result.get("数据", {}) self.app.enqueue( ( "image", { "vm_name": vm_name, "image_bytes": image_bytes, "width": meta.get("width"), "height": meta.get("height"), "claim_status": self.app.claim_status_var.get().strip(), "captured_at": now_text(), }, ) ) except Exception as exc: self.app.enqueue(("log", f"[{now_text()}] 截图解析失败: {exc}")) else: self.app.enqueue(("log", f"[{now_text()}] 截图返回成功，但没有 image_base64")) else: self.app.enqueue(("log", f"[{now_text()}] 截图失败[{vm_name}]: {result.get('消息')}")) self.stop_event.wait(interval) finally: self.app.enqueue(("screenshot_running", False)) class AutomationApp(tk.Tk): def __init__(self): super().__init__() self.configure(bg="#f4f7fb") self.instance_id = self.build_instance_id() self.title(f"夸克自动化 GUI - 实例 {self.instance_id}") self.event_queue: queue.Queue[tuple[str, Any]] = queue.Queue() self.workflow_thread: WorkflowRunner | None = None self.screenshot_thread: ScreenshotWorker | None = None self.preview_photo: ImageTk.PhotoImage | None = None self.last_capture_time = "--" self.vm_api_var = tk.StringVar(value="http://127.0.0.1:18765") self.device_no_var = tk.StringVar(value=self.build_default_device_no()) self.device_name_var = tk.StringVar(value="") self.remark_var = tk.StringVar(value="") self.capabilities_var = tk.StringVar(value="") self.source_vm_var = tk.StringVar(value="") self.run_count_var = tk.StringVar(value="1000000") self.infinite_loop_var = tk.BooleanVar(value=True) self.browser_cache_wait_var = tk.StringVar(value="15") self.system_stabilize_wait_var = tk.StringVar(value="120") self.current_vm_var = tk.StringVar(value="") self.screenshot_interval_var = tk.StringVar(value="2") self.auto_cleanup_var = tk.BooleanVar(value=True) self.topmost_var = tk.BooleanVar(value=False) self.status_var = tk.StringVar(value="空闲") self.claim_status_var = tk.StringVar(value="空闲") self.image_info_var = tk.StringVar(value="暂无截图") self.source_vm_options: list[str] = [] self.source_vm_combo: ttk.Combobox | None = None self._build_ui() self.refresh_image_info() self._apply_window_size() self.apply_topmost() self.after(200, self.process_events) self.after(500, self.refresh_source_vm_list) self.protocol("WM_DELETE_WINDOW", self.on_close) def build_instance_id(self) -> str: try: handle_value = int(self.winfo_id()) if handle_value > 0: return str(handle_value) except Exception: pass return str(os.getpid()) def build_default_device_no(self) -> str: username = "" for candidate in (os.environ.get("USERNAME"), getpass.getuser()): text = to_text(candidate).strip() if text: username = text break if not username: username = "user" return f"{username}--{self.instance_id}" def _build_ui(self) -> None: style = ttk.Style(self) style.theme_use("clam") style.configure("TFrame", background="#f4f7fb") style.configure("Card.TFrame", background="#ffffff") style.configure("TLabel", background="#f4f7fb", foreground="#213547", font=("Microsoft YaHei UI", 9)) style.configure("Card.TLabel", background="#ffffff", foreground="#213547", font=("Microsoft YaHei UI", 9)) style.configure("Header.TLabel", background="#f4f7fb", foreground="#0f172a", font=("Microsoft YaHei UI", 12, "bold")) style.configure("Hint.TLabel", background="#ffffff", foreground="#5b6472", font=("Microsoft YaHei UI", 8)) style.configure("Primary.TButton", font=("Microsoft YaHei UI", 9, "bold")) style.configure("Card.TCheckbutton", background="#ffffff", foreground="#213547", font=("Microsoft YaHei UI", 9)) root = ttk.Frame(self, padding=8) root.pack(fill="both", expand=True) root.columnconfigure(0, weight=0, minsize=360) root.columnconfigure(1, weight=1) root.rowconfigure(0, weight=1) left = ttk.Frame(root, style="Card.TFrame", padding=10) left.grid(row=0, column=0, sticky="nsew", padx=(0, 8)) left.columnconfigure(0, weight=1) right = ttk.Frame(root) right.grid(row=0, column=1, sticky="nsew") right.columnconfigure(0, weight=1) right.rowconfigure(1, weight=1) right.rowconfigure(3, weight=1) ttk.Label(left, text="运行配置", style="Header.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 8)) fields = [ ("VM API", self.vm_api_var, "normal"), ("设备编号", self.device_no_var, "normal"), ("运行次数", self.run_count_var, "normal"), ("缓存等待(秒)", self.browser_cache_wait_var, "normal"), ("开机稳定(秒)", self.system_stabilize_wait_var, "normal"), ("当前虚拟机", self.current_vm_var, "readonly"), ("截图间隔(秒)", self.screenshot_interval_var, "normal"), ] row_index = 1 for label, variable, state in fields: ttk.Label(left, text=label, style="Card.TLabel").grid(row=row_index, column=0, sticky="w", pady=(2, 1)) row_index += 1 entry = ttk.Entry(left, textvariable=variable, width=28) if state == "readonly": entry.configure(state="readonly") entry.grid(row=row_index, column=0, sticky="ew", pady=(0, 3)) row_index += 1 if label == "设备编号": ttk.Label(left, text="源虚拟机", style="Card.TLabel").grid(row=row_index, column=0, sticky="w", pady=(2, 1)) row_index += 1 source_vm_bar = ttk.Frame(left, style="Card.TFrame") source_vm_bar.grid(row=row_index, column=0, sticky="ew", pady=(0, 3)) source_vm_bar.columnconfigure(0, weight=1) self.source_vm_combo = ttk.Combobox(source_vm_bar, textvariable=self.source_vm_var, state="readonly", width=22) self.source_vm_combo.grid(row=0, column=0, sticky="ew", padx=(0, 4)) ttk.Button(source_vm_bar, text="刷新列表", command=self.refresh_source_vm_list).grid(row=0, column=1, sticky="ew") row_index += 1 ttk.Checkbutton( left, text="无限循环（忽略运行次数）", variable=self.infinite_loop_var, style="Card.TCheckbutton", ).grid(row=row_index, column=0, sticky="w", pady=(4, 0)) row_index += 1 ttk.Checkbutton( left, text="每轮结束自动删除虚拟机", variable=self.auto_cleanup_var, style="Card.TCheckbutton", ).grid(row=row_index, column=0, sticky="w", pady=(4, 4)) row_index += 1 ttk.Checkbutton( left, text="本窗口保持最前面", variable=self.topmost_var, command=self.apply_topmost, style="Card.TCheckbutton", ).grid(row=row_index, column=0, sticky="w", pady=(0, 4)) row_index += 1 button_bar = ttk.Frame(left, style="Card.TFrame") button_bar.grid(row=row_index, column=0, sticky="ew", pady=(8, 6)) for idx in range(2): button_bar.columnconfigure(idx, weight=1) self.start_button = ttk.Button(button_bar, text="开始", style="Primary.TButton", command=self.start_workflow) self.start_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.stop_button = ttk.Button(button_bar, text="结束", command=self.stop_workflow, state="disabled") self.stop_button.grid(row=0, column=1, sticky="ew", padx=(4, 0)) row_index += 1 shot_bar = ttk.Frame(left, style="Card.TFrame") shot_bar.grid(row=row_index, column=0, sticky="ew", pady=(0, 6)) for idx in range(3): shot_bar.columnconfigure(idx, weight=1) self.screenshot_start_button = ttk.Button(shot_bar, text="开始截图", command=self.start_screenshot) self.screenshot_start_button.grid(row=0, column=0, sticky="ew", padx=(0, 4)) self.screenshot_stop_button = ttk.Button(shot_bar, text="停止截图", command=self.stop_screenshot, state="disabled") self.screenshot_stop_button.grid(row=0, column=1, sticky="ew", padx=4) self.screenshot_once_button = ttk.Button(shot_bar, text="截图一次", command=self.capture_once) self.screenshot_once_button.grid(row=0, column=2, sticky="ew", padx=(4, 0)) row_index += 1 utility_bar = ttk.Frame(left, style="Card.TFrame") utility_bar.grid(row=row_index, column=0, sticky="ew", pady=(0, 6)) for idx in range(3): utility_bar.columnconfigure(idx, weight=1) ttk.Button(utility_bar, text="VM 健康检查", command=self.check_vm_health).grid(row=0, column=0, sticky="ew", padx=(0, 4)) ttk.Button(utility_bar, text="清空日志", command=self.clear_log).grid(row=0, column=1, sticky="ew", padx=4) ttk.Button(utility_bar, text="新开窗口", command=self.launch_new_window).grid(row=0, column=2, sticky="ew", padx=(4, 0)) row_index += 1 manual_bar = ttk.Frame(left, style="Card.TFrame") manual_bar.grid(row=row_index, column=0, sticky="ew", pady=(0, 6)) for idx in range(2): manual_bar.columnconfigure(idx, weight=1) ttk.Button(manual_bar, text="关机当前机", command=self.power_off_current_vm).grid(row=0, column=0, sticky="ew", padx=(0, 4)) ttk.Button(manual_bar, text="删除当前机", command=self.delete_current_vm).grid(row=0, column=1, sticky="ew", padx=(4, 0)) row_index += 1 ttk.Label(left, textvariable=self.status_var, style="Card.TLabel", foreground="#0f766e").grid(row=row_index, column=0, sticky="w", pady=(6, 1)) row_index += 1 ttk.Label(left, text=f"实例标识：{self.instance_id}", style="Hint.TLabel").grid(row=row_index, column=0, sticky="w") row_index += 1 ttk.Label(left, text="设备名称/备注/能力固定为空，无需输入。", style="Hint.TLabel").grid(row=row_index, column=0, sticky="w") row_index += 1 ttk.Label(left, text="源虚拟机从本地扫描结果里选择，不需要手输。", style="Hint.TLabel").grid(row=row_index, column=0, sticky="w") row_index += 1 ttk.Label(left, text="结束只停止脚本；停止时保留当前虚拟机。", style="Hint.TLabel").grid(row=row_index, column=0, sticky="w") row_index += 1 ttk.Label(left, text='截图优先使用"当前虚拟机"，为空则回退到源虚拟机。', style="Hint.TLabel").grid(row=row_index, column=0, sticky="w") ttk.Label(right, text="运行日志", style="Header.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 6)) log_frame = ttk.Frame(right, style="Card.TFrame", padding=8) log_frame.grid(row=1, column=0, sticky="nsew") log_frame.rowconfigure(0, weight=1) log_frame.columnconfigure(0, weight=1) self.log_text = tk.Text( log_frame, wrap="word", font=("Consolas", 8), bg="#0b1220", fg="#dbeafe", insertbackground="#dbeafe", relief="flat", ) self.log_text.grid(row=0, column=0, sticky="nsew") log_scroll = ttk.Scrollbar(log_frame, orient="vertical", command=self.log_text.yview) log_scroll.grid(row=0, column=1, sticky="ns") self.log_text.configure(yscrollcommand=log_scroll.set) ttk.Label(right, text="实时截图", style="Header.TLabel").grid(row=2, column=0, sticky="w", pady=(6, 6)) image_card = ttk.Frame(right, style="Card.TFrame", padding=8) image_card.grid(row=3, column=0, sticky="nsew") image_card.rowconfigure(1, weight=1) image_card.columnconfigure(0, weight=1) ttk.Label(image_card, textvariable=self.image_info_var, style="Card.TLabel").grid(row=0, column=0, sticky="w", pady=(0, 8)) self.image_label = ttk.Label(image_card, text="暂无截图", anchor="center", style="Card.TLabel") self.image_label.grid(row=1, column=0, sticky="nsew") def _apply_window_size(self) -> None: self.update_idletasks() required_width = max(self.winfo_reqwidth() + 24, 980) required_height = max(self.winfo_reqheight() + 24, 780) self.geometry(f"{required_width}x{required_height}") self.minsize(required_width, required_height) def set_claim_status(self, status: str) -> None: text = to_text(status).strip() or "空闲" self.enqueue(("claim_status", text)) def refresh_image_info(self, captured_at: str | None = None) -> None: if captured_at: self.last_capture_time = captured_at claim_status = self.claim_status_var.get().strip() or "\u7a7a\u95f2" capture_text = self.last_capture_time or "--" self.image_info_var.set(f"\u9886\u53d6\u72b6\u6001: {claim_status} | \u622a\u56fe\u65f6\u95f4: {capture_text}") def enqueue(self, item: tuple[str, Any]) -> None: self.event_queue.put(item) def process_events(self) -> None: while True: try: event, payload = self.event_queue.get_nowait() except queue.Empty: break if event == "log": self.log_text.insert("end", f"{payload}\n") self.log_text.see("end") elif event == "current_vm": self.current_vm_var.set(to_text(payload)) elif event == "running": running = bool(payload) self.start_button.configure(state="disabled" if running else "normal") self.stop_button.configure(state="normal" if running else "disabled") self.status_var.set("\u8fd0\u884c\u4e2d" if running else "\u7a7a\u95f2") if not running and self.claim_status_var.get().strip() in {"\u51c6\u5907\u4e2d", "\u7b49\u5f85\u63a5\u5355", "\u5f85\u9886\u53d6", "\u767b\u5f55\u4e2d"}: self.claim_status_var.set("\u7a7a\u95f2") self.refresh_image_info() elif event == "claim_status": self.claim_status_var.set(to_text(payload).strip() or "\u7a7a\u95f2") self.refresh_image_info() elif event == "screenshot_running": running = bool(payload) self.screenshot_start_button.configure(state="disabled" if running else "normal") self.screenshot_stop_button.configure(state="normal" if running else "disabled") elif event == "image": self.show_image(payload) elif event == "source_vm_options": self.apply_source_vm_options(payload) self.after(200, self.process_events) def show_image(self, payload: dict[str, Any]) -> None: try: image = Image.open(io.BytesIO(payload["image_bytes"])) self.update_idletasks() max_width = max(self.image_label.winfo_width() - 16, 240) max_height = max(self.image_label.winfo_height() - 16, 160) image.thumbnail((max_width, max_height)) self.preview_photo = ImageTk.PhotoImage(image) self.image_label.configure(image=self.preview_photo, text="") claim_status = to_text(payload.get("claim_status")).strip() or self.claim_status_var.get().strip() or "\u7a7a\u95f2" captured_at = to_text(payload.get("captured_at")).strip() or now_text() self.claim_status_var.set(claim_status) self.refresh_image_info(captured_at) except Exception as exc: self.image_info_var.set(f"\u622a\u56fe\u663e\u793a\u5931\u8d25: {exc}") self.image_label.configure(image="", text="\u622a\u56fe\u663e\u793a\u5931\u8d25") def apply_topmost(self) -> None: try: self.attributes("-topmost", bool(self.topmost_var.get())) except Exception: pass def apply_source_vm_options(self, payload: dict[str, Any]) -> None: options = payload.get("options") if isinstance(payload, dict) else None if not isinstance(options, list): return cleaned_options: list[str] = [] for item in options: text = to_text(item).strip() if text and text not in cleaned_options: cleaned_options.append(text) self.source_vm_options = cleaned_options if self.source_vm_combo is not None: self.source_vm_combo.configure(values=cleaned_options) current = self.source_vm_var.get().strip() preferred = to_text(payload.get("preferred"), "").strip() if isinstance(payload, dict) else "" if preferred and preferred in cleaned_options: self.source_vm_var.set(preferred) elif current and current in cleaned_options: self.source_vm_var.set(current) elif not current and cleaned_options: self.source_vm_var.set(cleaned_options[0]) else: self.source_vm_var.set("") def refresh_source_vm_list(self) -> None: vm_api_url = self.vm_api_var.get().strip() if not vm_api_url: self.enqueue(("log", f"[{now_text()}] 源虚拟机列表刷新失败：VM API 不能为空")) return current_value = self.source_vm_var.get().strip() self.enqueue(("log", f"[{now_text()}] 开始扫描本地源虚拟机列表")) def worker() -> None: api = VmwareApiClient(vm_api_url) result = api.list_vms() if not is_success_result(result): self.enqueue(("log", f"[{now_text()}] 源虚拟机列表刷新失败：{result.get('消息')}")) return data = result.get("数据", {}) if isinstance(result.get("数据"), dict) else {} items = data.get("items") if isinstance(data.get("items"), list) else [] options: list[str] = [] seen: set[str] = set() sorted_items = sorted( items, key=lambda item: ( to_text(item.get("display_name") if isinstance(item, dict) else "").lower(), to_text(item.get("name") if isinstance(item, dict) else "").lower(), ), ) for item in sorted_items: if not isinstance(item, dict): continue candidate = to_text(item.get("display_name")).strip() or to_text(item.get("name")).strip() or to_text(item.get("vmx_path")).strip() if is_runtime_clone_vm_name(candidate): continue if not candidate or candidate in seen: continue seen.add(candidate) options.append(candidate) self.enqueue(("source_vm_options", {"options": options, "preferred": current_value})) self.enqueue(("log", f"[{now_text()}] 本地源虚拟机扫描完成，共 {len(options)} 台")) threading.Thread(target=worker, daemon=True).start() def get_config(self) -> WorkflowConfig: if self.infinite_loop_var.get(): run_count = 1000000 else: run_count = to_int(self.run_count_var.get(), 1) if run_count <= 0: run_count = 1 browser_cache_wait_seconds = to_int(self.browser_cache_wait_var.get(), 15) if browser_cache_wait_seconds <= 0: browser_cache_wait_seconds = 15 system_stabilize_wait_seconds = to_int(self.system_stabilize_wait_var.get(), 50) if system_stabilize_wait_seconds < 0: system_stabilize_wait_seconds = 50 return WorkflowConfig( vm_api_url=self.vm_api_var.get().strip(), device_no=self.device_no_var.get().strip(), device_name="", remark="", capabilities="", source_vm_name=self.source_vm_var.get().strip(), run_count=run_count, browser_cache_wait_seconds=browser_cache_wait_seconds, system_stabilize_wait_seconds=system_stabilize_wait_seconds, auto_cleanup=bool(self.auto_cleanup_var.get()), ) def validate_config(self) -> WorkflowConfig | None: config = self.get_config() if not config.vm_api_url: messagebox.showerror("错误", "VM API 不能为空") return None if not config.device_no: messagebox.showerror("错误", "设备编号不能为空") return None if not config.source_vm_name: messagebox.showerror("错误", "源虚拟机不能为空") return None return config def start_workflow(self) -> None: if self.workflow_thread and self.workflow_thread.is_alive(): messagebox.showinfo("\u63d0\u793a", "\u811a\u672c\u5df2\u7ecf\u5728\u8fd0\u884c\u4e2d") return config = self.validate_config() if config is None: return self.claim_status_var.set("\u51c6\u5907\u4e2d") self.refresh_image_info() self.enqueue(("log", f"[{now_text()}] \u542f\u52a8\u811a\u672c\uff0c\u8ba1\u5212\u8fd0\u884c {config.run_count} \u6b21")) self.workflow_thread = WorkflowRunner(self, config) self.workflow_thread.start() def stop_workflow(self) -> None: if self.workflow_thread and self.workflow_thread.is_alive(): self.claim_status_var.set("\u5df2\u505c\u6b62") self.refresh_image_info() self.enqueue(("log", f"[{now_text()}] \u6536\u5230\u505c\u6b62\u6307\u4ee4\uff0c\u5f53\u524d\u8f6e\u6b21\u4f1a\u505c\u6b62\uff0c\u5e76\u4fdd\u7559\u5f53\u524d\u865a\u62df\u673a")) self.workflow_thread.stop() def _get_manual_vm_name(self) -> str: return self.current_vm_var.get().strip() def power_off_current_vm(self) -> None: vm_name = self._get_manual_vm_name() if not vm_name: messagebox.showerror("错误", "当前虚拟机为空，无法执行关机") return if not messagebox.askyesno("确认", f"确定要关机当前虚拟机吗？\n\n{vm_name}"): return def worker() -> None: api = VmwareApiClient(self.vm_api_var.get().strip()) result = api.power_off(vm_name, False) self.enqueue(("log", f"[{now_text()}] 手动关机[{vm_name}] | 成功={result.get('成功')} | 消息={result.get('消息')}")) threading.Thread(target=worker, daemon=True).start() def delete_current_vm(self) -> None: vm_name = self._get_manual_vm_name() if not vm_name: messagebox.showerror("错误", "当前虚拟机为空，无法执行删除") return if not messagebox.askyesno("确认", f"确定要删除当前虚拟机吗？\n\n{vm_name}\n\n删除后磁盘文件也会一起删除。"): return def worker() -> None: api = VmwareApiClient(self.vm_api_var.get().strip()) result = api.safe_delete_vm(vm_name, True, True, 5, 3) self.enqueue(("log", f"[{now_text()}] 手动删除[{vm_name}] | 成功={result.get('成功')} | 消息={result.get('消息')}")) if result.get("成功") is True and self.current_vm_var.get().strip() == vm_name: self.enqueue(("current_vm", "")) threading.Thread(target=worker, daemon=True).start() def start_screenshot(self) -> None: if self.screenshot_thread and self.screenshot_thread.is_alive(): messagebox.showinfo("提示", "截图线程已经在运行中") return vm_api_url = self.vm_api_var.get().strip() if not vm_api_url: messagebox.showerror("错误", "VM API 不能为空") return self.enqueue(("log", f"[{now_text()}] 启动实时截图线程")) self.screenshot_thread = ScreenshotWorker(self, vm_api_url, lambda: self.current_vm_var.get().strip() or self.source_vm_var.get().strip()) self.screenshot_thread.start() def stop_screenshot(self) -> None: if self.screenshot_thread and self.screenshot_thread.is_alive(): self.enqueue(("log", f"[{now_text()}] 停止实时截图线程")) self.screenshot_thread.stop() def capture_once(self) -> None: vm_name = self.current_vm_var.get().strip() or self.source_vm_var.get().strip() vm_api_url = self.vm_api_var.get().strip() if not vm_name: messagebox.showerror("错误", "当前虚拟机为空，无法截图") return if not vm_api_url: messagebox.showerror("错误", "VM API 不能为空") return def worker() -> None: api = VmwareApiClient(vm_api_url) result = api.screenshot(vm_name, False) if is_success_result(result): image_base64 = to_text(result.get("数据", {}).get("image_base64")) if image_base64: self.enqueue( ( "image", { "vm_name": vm_name, "image_bytes": base64.b64decode(image_base64), "width": result.get("数据", {}).get("width"), "height": result.get("数据", {}).get("height"), "claim_status": self.claim_status_var.get().strip(), "captured_at": now_text(), }, ) ) self.enqueue(("log", f"[{now_text()}] 手动截图成功: {vm_name}")) else: self.enqueue(("log", f"[{now_text()}] 手动截图成功，但返回里没有 image_base64")) else: self.enqueue(("log", f"[{now_text()}] 手动截图失败: {result.get('消息')}")) threading.Thread(target=worker, daemon=True).start() def check_vm_health(self) -> None: vm_api_url = self.vm_api_var.get().strip() if not vm_api_url: messagebox.showerror("错误", "VM API 不能为空") return def worker() -> None: api = VmwareApiClient(vm_api_url) result = api.health() self.enqueue(("log", f"[{now_text()}] VM 健康检查 | 成功={result.get('成功')} | 消息={result.get('消息')}")) threading.Thread(target=worker, daemon=True).start() def launch_new_window(self) -> None: try: if getattr(sys, "frozen", False): subprocess.Popen([str(pathlib.Path(sys.executable).resolve())], close_fds=True) else: script_path = pathlib.Path(__file__).resolve() subprocess.Popen([sys.executable, str(script_path)], cwd=str(script_path.parent), close_fds=True) self.enqueue(("log", f"[{now_text()}] 已启动新的独立窗口实例")) except Exception as exc: messagebox.showerror("错误", f"新开窗口失败：{exc}") def clear_log(self) -> None: self.log_text.delete("1.0", "end") def on_close(self) -> None: if self.workflow_thread and self.workflow_thread.is_alive(): self.workflow_thread.stop() if self.screenshot_thread and self.screenshot_thread.is_alive(): self.screenshot_thread.stop() self.destroy() if __name__ == "__main__": try: app = AutomationApp() app.mainloop() except Exception: error_text = traceback.format_exc() if getattr(sys, "frozen", False): log_path = str(pathlib.Path(sys.executable).resolve().with_name("夸克自动化_gui_error.log")) else: log_path = str(pathlib.Path(__file__).resolve().with_name("夸克自动化_gui_error.log")) with open(log_path, "a", encoding="utf-8") as fp: fp.write(f"\n[{datetime.now().strftime('%Y-%m-%d %H:%M:%S')}]\n") fp.write(error_text) fp.write("\n") try: root = tk.Tk() root.withdraw() messagebox.showerror("夸克自动化 GUI", f"程序启动失败，错误日志已写入：\n{log_path}") root.destroy() except Exception: pass raise