import base64 import io import math import os import random import re import threading import time import cv2 import numpy as np import requests from PIL import Image, ImageDraw import uiautomator2 as u2 API_URL = "http://api.jfbym.com/api/YmServer/customApi" # 验证码平台 token API_TOKEN = "1nDVocTE2mJ0yLEYb2sZJ5uUY2VIEoGTkIpW44X7Kgk" # 原始截图保存路径 SCREENSHOT_PATH = "./a.jpg" # 裁剪后图片保存路径z CROP_PATH = "./b.jpg" # 滑动轨迹图保存目录 TRACK_DIR = "./slider_tracks" # 验证码截图日志根目录 CAPTCHA_LOG_ROOT = "./captcha_logs" UNKNOWN_DEVICE_ID = "unknown_device" _RUNTIME_CONTEXT = threading.local() SLIDER_METHODS = ("track", "bezier") # 图标点选验证码 CAPTCHA_ICON_CLICK = "icon_click" # 空间推理验证码 CAPTCHA_SPACE_REASON = "space_reason" # 文字点选验证码 CAPTCHA_TEXT_CLICK = "text_click" # 滑块验证码 CAPTCHA_SLIDER = "slider" # 图片内容输入验证码 CAPTCHA_TEXT_INPUT = "text_input" # 输入型验证码的裁剪区域 INPUT_CROP = { "x_start": 210, "x_end": 510, "y_start": 666, "y_end": 788, } # 图标点选验证码的裁剪区域 ICON_CLICK_CROP = { "x_start": 40, "x_end": 680, "y_start": 471, "y_end": 1143, } # 不同验证码类型对应的平台参数配置 VERIFY_CONFIG = { CAPTCHA_ICON_CLICK: {"type": "88888", "direction": None, "extra": None, "y_offset": 0, "result": "points"}, CAPTCHA_SPACE_REASON: {"type": "88888", "direction": None, "extra": "请点击", "y_offset": 0, "result": "points"}, CAPTCHA_TEXT_CLICK: {"type": "30114", "direction": None, "extra": "phrase", "y_offset": 0, "result": "points"}, CAPTCHA_SLIDER: {"type": "22222", "direction": None, "extra": None, "y_offset": 0, "result": "text"}, CAPTCHA_TEXT_INPUT: {"type": "10103", "direction": None, "extra": None, "y_offset": 0, "result": "text"}, # lianxian : {"type": "10114", "direction": None, "extra": None, "y_offset": 0, "result": "text"} } def _safe_name(value, default): if value is None: return default text = str(value).strip() if not text: return default text = re.sub(r"[^0-9A-Za-z_\-.]+", "_", text) return text or default def _resolve_device_id(d=None, device_id=None): if device_id: return _safe_name(device_id, UNKNOWN_DEVICE_ID) candidates = [] if d is not None: for attr in ("serial", "_serial", "device_id"): value = getattr(d, attr, None) if value: candidates.append(value) try: info = d.device_info if isinstance(info, dict): for key in ("serial", "serialno", "udid", "deviceId"): value = info.get(key) if value: candidates.append(value) except Exception: pass for value in candidates: normalized = _safe_name(value, "") if normalized: return normalized return UNKNOWN_DEVICE_ID def _set_runtime_device_id(d=None, device_id=None): _RUNTIME_CONTEXT.device_id = _resolve_device_id(d=d, device_id=device_id) def _get_runtime_device_id(d=None): current = getattr(_RUNTIME_CONTEXT, "device_id", None) if current: return current resolved = _resolve_device_id(d=d) _RUNTIME_CONTEXT.device_id = resolved return resolved def _next_slider_method_order(): idx = getattr(_RUNTIME_CONTEXT, "slider_method_idx", 0) first = SLIDER_METHODS[idx % len(SLIDER_METHODS)] second = SLIDER_METHODS[(idx + 1) % len(SLIDER_METHODS)] _RUNTIME_CONTEXT.slider_method_idx = (idx + 1) % len(SLIDER_METHODS) return [first, second] def _ensure_captcha_dir(captcha_type): safe_type = _safe_name(captcha_type, "unknown_captcha") folder = os.path.join(CAPTCHA_LOG_ROOT, safe_type) os.makedirs(folder, exist_ok=True) return folder def _build_captcha_image_path(captcha_type, d=None, ext=".png", tag=None): device_id = _get_runtime_device_id(d=d) random_part = random.randint(10000000, 99999999) safe_tag = _safe_name(tag, "") if tag else "" suffix = f"_{safe_tag}" if safe_tag else "" filename = f"{device_id}_{random_part}{suffix}{ext}" return os.path.join(_ensure_captcha_dir(captcha_type), filename) def _save_debug_screenshot(d, captcha_type, tag="full"): path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag=tag) try: d.screenshot(path) print(f"[captcha-shot] saved: {path}") return path except Exception as e: print(f"[captcha-shot] save failed: {e}") return None def post_api(image_path, captcha_type, extra=None, direction=None, label_image_path=None, timeout=20): with open(image_path, 'rb') as f: image_base64 = base64.b64encode(f.read()).decode() data = { "token": API_TOKEN, "type": captcha_type, "image": image_base64, } if label_image_path: with open(label_image_path, 'rb') as f: data["label_image"] = base64.b64encode(f.read()).decode() if extra is not None: data["extra"] = extra if direction is not None: data["direction"] = direction headers = { "Content-Type": "application/json" } response = requests.post(API_URL, headers=headers, json=data, timeout=timeout).json() print(response) return response def parse_points(response, y_offset=0): tuple_points = [] data = response.get("data", {}).get("data", "") if not data: return tuple_points for s in data.split('|'): x, y = s.split(',') tuple_points.append((int(x), int(y) + y_offset)) return tuple_points def verify(image_path, captcha_type): config = VERIFY_CONFIG.get(captcha_type) if not config: raise ValueError(f"不支持的验证码类型: {captcha_type}") response = post_api( image_path, config["type"], extra=config["extra"], direction=config["direction"] ) if config["result"] == "points": return parse_points(response, y_offset=config["y_offset"]) return response.get("data", {}).get("data") def crop_image_xy( image_path, output_path=None, x_start=None, x_end=None, y_start=471, y_end=1143 ): if output_path is None: dir_name, file_name = os.path.split(image_path) name, ext = os.path.splitext(file_name) output_path = os.path.join(dir_name, f"{name}_cropped{ext}") with Image.open(image_path) as img: width, height = img.size if x_start is None: x_start = 0 if x_end is None: x_end = width - 1 x_start = max(0, min(x_start, width - 1)) x_end = max(x_start, min(x_end, width - 1)) y_start = max(0, min(y_start, height - 1)) y_end = max(y_start, min(y_end, height - 1)) cropped = img.crop((x_start, y_start, x_end + 1, y_end + 1)) cropped.save(output_path) return output_path def _capture_by_bounds(d, xpath_candidates, output_path=None, screenshot_path=None, captcha_type="generic"): """ 按元素 bounds 截图并裁剪。 return: (cropped_path, bounds_dict) or (None, None) """ if isinstance(xpath_candidates, str): xpath_candidates = [xpath_candidates] if screenshot_path is None: screenshot_path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag="full") if output_path is None: output_path = _build_captcha_image_path(captcha_type, d=d, ext=".png", tag="crop") for xpath in xpath_candidates: try: node = d.xpath(xpath) if not node.exists: continue bounds = node.info.get("bounds", {}) if not bounds: continue d.screenshot(screenshot_path) cropped = crop_image_xy( screenshot_path, output_path=output_path, x_start=bounds["left"], x_end=bounds["right"], y_start=bounds["top"], y_end=bounds["bottom"], ) return cropped, bounds except Exception: continue return None, None def srwz(d): captcha_image_xpaths = [ '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.widget.Image[1]', '//*[@resource-id="captchaImg"]', ] image_path, _ = _capture_by_bounds(d, captcha_image_xpaths, captcha_type=CAPTCHA_TEXT_INPUT) if not image_path: return False data = verify(image_path, CAPTCHA_TEXT_INPUT) input_box = d.xpath('//*[@hint="请输入验证码"]') if input_box.exists: input_box.click() time.sleep(0.5) input_box.set_text(data) d.xpath('//*[@text="验证"]').click() return True else: print("未找到输入框") return False def _clamp(value, min_value, max_value): return max(min_value, min(value, max_value)) def _slider_duration(distance): if distance <= 90: return round(random.uniform(0.18, 0.27), 3) if distance <= 160: return round(random.uniform(0.23, 0.34), 3) return round(random.uniform(0.28, 0.42), 3) def _build_human_slider_track(start_x, start_y, distance): # Keep the reference shape: mostly flat first, then a single smooth downward bend. overshoot = random.randint(1, 2) if distance > 140 and random.random() < 0.18 else 0 move_distance = distance + overshoot steps = int(_clamp(move_distance / random.uniform(6.0, 8.0), 16, 32)) base_y = start_y + random.randint(-1, 1) flat_ratio = random.uniform(0.30, 0.42) # front section almost horizontal drop = distance * random.uniform(0.08, 0.14) if distance > 260: drop *= random.uniform(1.35, 1.75) drop = _clamp(drop, 12.0, 58.0) # obvious tail drop noise_amp = random.uniform(0.03, 0.22) points = [(start_x, base_y)] last_x = start_x for i in range(1, steps + 1): t = i / steps progress = 1 - (1 - t) ** 2.0 progress += random.uniform(-0.0018, 0.0018) progress = _clamp(progress, 0.0, 1.0) x = start_x + int(move_distance * progress) if x <= last_x: x = last_x + 1 last_x = x if t < flat_ratio: # slight tiny rise then back, still near flat u = t / max(flat_ratio, 1e-6) y_curve = -0.9 * math.sin(math.pi * u) else: u = (t - flat_ratio) / max(1 - flat_ratio, 1e-6) y_curve = drop * (u ** 1.7) y_noise = random.uniform(-noise_amp, noise_amp) y = int(round(base_y + y_curve + y_noise)) points.append((x, y)) final_x = start_x + distance final_y = int(round(base_y + drop + random.uniform(-0.6, 0.6))) if overshoot > 0: points.append((start_x + distance + overshoot, final_y)) points.append((final_x, final_y + random.choice([0, 0, 1]))) return points def _save_slider_track_image( points, distance=None, duration=None, drag_ok=True, screenshot_path=SCREENSHOT_PATH, captcha_type=CAPTCHA_SLIDER ): if not points: return None output_path = _build_captcha_image_path(captcha_type, ext=".png", tag="track") try: if os.path.exists(screenshot_path): with Image.open(screenshot_path) as img: canvas = img.convert("RGB") else: canvas = Image.new("RGB", (720, 1280), "white") draw = ImageDraw.Draw(canvas) if len(points) >= 2: draw.line(points, fill=(245, 20, 30), width=9) sx, sy = points[0] ex, ey = points[-1] draw.ellipse((sx - 5, sy - 5, sx + 5, sy + 5), fill=(40, 200, 80)) draw.ellipse((ex - 5, ey - 5, ex + 5, ey + 5), fill=(50, 120, 255)) info = f"ok={drag_ok} dist={distance} dur={duration}s points={len(points)}" draw.rectangle((8, 8, min(canvas.size[0] - 8, 520), 42), fill=(0, 0, 0)) draw.text((14, 14), info, fill=(255, 255, 255)) canvas.save(output_path) print(f"[slider-track] saved: {output_path}") return output_path except Exception as e: print(f"[slider-track] save failed: {e}") return None def _downsample_track_points(points, target_count): if not points or len(points) <= target_count: return points[:] if target_count < 2: return [points[0], points[-1]] sampled = [] last_index = len(points) - 1 for i in range(target_count): idx = int(round(i * last_index / (target_count - 1))) sampled.append(points[idx]) return sampled def _execute_track(d, points, total_duration): if not points or len(points) < 2: return False, points duration = max(0.18, float(total_duration)) # Use fewer points to avoid step explosion, keep curve shape. max_points = random.randint(14, 22) exec_points = _downsample_track_points(points, max_points) try: if hasattr(d, "swipe_points"): # u2: duration here means time-per-step; steps = duration/0.005. # To approximate total duration: # total ~= (duration/0.005) * (len(exec_points)-1) * 0.005 = duration * segments seg_count = max(1, len(exec_points) - 1) per_segment = max(0.01, duration / seg_count) d.swipe_points(exec_points, duration=per_segment) return True, exec_points except Exception: pass # fallback try: sx, sy = exec_points[0] ex, ey = exec_points[-1] d.swipe(sx, sy, ex, ey, duration=duration) return False, exec_points except Exception: return False, exec_points def _slider_still_exists(d): xpath_candidates = [ '//*[@text="请拖动下方滑块完成拼图"]', '//*[contains(@text, "拖动下方滑块")]', '//*[@resource-id="puzzleSliderBox"]', '//*[@resource-id="puzzleImageMain"]', ] for xpath in xpath_candidates: try: if d.xpath(xpath).exists: return True except Exception: continue return False def _cubic_bezier(t, p0, p1, p2, p3): one_minus_t = 1 - t x = ( one_minus_t ** 3 * p0[0] + 3 * one_minus_t ** 2 * t * p1[0] + 3 * one_minus_t * t ** 2 * p2[0] + t ** 3 * p3[0] ) y = ( one_minus_t ** 3 * p0[1] + 3 * one_minus_t ** 2 * t * p1[1] + 3 * one_minus_t * t ** 2 * p2[1] + t ** 3 * p3[1] ) return x, y def _generate_bezier_slider_points(start, end, deviation=30, steps=50): sx, sy = start ex, ey = end mid_x = (sx + ex) / 2 mid_y = (sy + ey) / 2 p1 = ( mid_x - (ex - sx) / 4 + random.uniform(-deviation, deviation), mid_y - (ey - sy) / 4 + random.uniform(-deviation / 2, deviation / 2), ) p2 = ( mid_x + (ex - sx) / 4 + random.uniform(-deviation, deviation), mid_y + (ey - sy) / 4 + random.uniform(-deviation / 2, deviation / 2), ) points = [] for i in range(steps + 1): t = i / steps x, y = _cubic_bezier(t, start, p1, p2, end) if 0 < i < steps: x += random.gauss(0, 1.5) y += random.gauss(0, 1.5) points.append((int(round(x)), int(round(y)))) return points def _execute_bezier_slider(d, start_x, start_y, end_x, end_y): points = _generate_bezier_slider_points( (int(round(start_x)), int(round(start_y))), (int(round(end_x)), int(round(end_y))), deviation=random.randint(20, 40), steps=50, ) if len(points) < 2: return False, points try: d.touch.down(points[0][0], points[0][1]) time.sleep(random.uniform(0.1, 0.2)) total = max(1, len(points) - 1) for i, (x, y) in enumerate(points[1:], 1): d.touch.move(x, y) t = i / total if 0.2 < t < 0.8: interval = random.uniform(0.02, 0.04) else: interval = random.uniform(0.04, 0.08) time.sleep(interval) time.sleep(random.uniform(0.05, 0.15)) d.touch.up(points[-1][0], points[-1][1]) return True, points except Exception: try: d.touch.up(points[-1][0], points[-1][1]) except Exception: pass return False, points def _slider_knob_center(d): slider_xpath = ( '//*[@resource-id="puzzleSliderBox"] | ' '//*[@resource-id="yodaBox"] | ' '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]' ) try: slider_node = d.xpath(slider_xpath) if slider_node.exists: bounds = slider_node.info.get("bounds", {}) if bounds: return ( (bounds["left"] + bounds["right"]) / 2 + random.uniform(-4, 4), (bounds["top"] + bounds["bottom"]) / 2 + random.uniform(-3, 3), ) except Exception: pass return None def _first_existing_bounds(d, xpath_candidates): if isinstance(xpath_candidates, str): xpath_candidates = [xpath_candidates] for xpath in xpath_candidates: try: node = d.xpath(xpath) if not node.exists: continue bounds = node.info.get("bounds", {}) if bounds: return xpath, bounds except Exception: continue return None, None def _click_captcha_close(d, captcha_xpath=None): """点击验证码右上角关闭按钮;优先点显式关闭控件,失败后按容器右上角估算点位。""" close_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]', '//*[@resource-id="btn_close_verify"]', '//*[@content-desc="关闭"]', '//*[@text="关闭"]', ] _, close_bounds = _first_existing_bounds(d, close_xpaths) if close_bounds: cx = int((close_bounds["left"] + close_bounds["right"]) / 2) + random.randint(-2, 2) cy = int((close_bounds["top"] + close_bounds["bottom"]) / 2) + random.randint(-2, 2) d.click(cx, cy) print(f"[captcha-close] click explicit close at ({cx}, {cy})") return True popup_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]', '//*[@resource-id="com.sankuai.meituan:id/h5_container"]', '//*[@resource-id="root"]', '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]', '//*[@text="身份核实"]/android.view.View[1]', ] if captcha_xpath: popup_xpaths.append(captcha_xpath) _, popup_bounds = _first_existing_bounds(d, popup_xpaths) if not popup_bounds: return False left = popup_bounds["left"] right = popup_bounds["right"] top = popup_bounds["top"] bottom = popup_bounds["bottom"] width = max(1, right - left) height = max(1, bottom - top) # 参考示例:[40,391][680,1223] -> 右上角叉号中心约(640, 431)。 offset_x = int(_clamp(width * 0.06, 20, 56)) offset_y = int(_clamp(height * 0.05, 20, 56)) click_x = int(right - offset_x) + random.randint(-3, 3) click_y = int(top + offset_y) + random.randint(-3, 3) d.click(click_x, click_y) print(f"[captcha-close] click inferred close at ({click_x}, {click_y})") return True def _build_directional_track(start_x, start_y, end_x, end_y): distance_x = end_x - start_x distance_y = end_y - start_y if abs(distance_x) < 2 and abs(distance_y) < 2: return [(int(start_x), int(start_y)), (int(end_x), int(end_y))] steps = int(_clamp(abs(distance_x) / random.uniform(7.0, 10.0), 22, 48)) points = [(int(start_x), int(start_y))] last_x = float(start_x) for i in range(1, steps + 1): t = i / steps progress = 1 - (1 - t) ** random.uniform(1.8, 2.25) x = start_x + distance_x * progress + random.uniform(-0.9, 0.9) y = start_y + distance_y * progress + random.uniform(-0.8, 0.8) if distance_x >= 0: if x < last_x: x = last_x + random.uniform(0.2, 1.2) else: if x > last_x: x = last_x - random.uniform(0.2, 1.2) last_x = x points.append((int(round(x)), int(round(y)))) points.append((int(round(end_x)), int(round(end_y)))) return points def _move_with_pressed_touch(d, points): if not points: return for x, y in points: d.touch.move(x, y) time.sleep(random.uniform(0.0015, 0.0045)) def _screenshot_to_image(d): shot = d.screenshot() if isinstance(shot, Image.Image): return shot.convert("RGB") if isinstance(shot, bytes): return Image.open(io.BytesIO(shot)).convert("RGB") if isinstance(shot, str) and os.path.exists(shot): return Image.open(shot).convert("RGB") if hasattr(shot, "convert"): return shot.convert("RGB") fallback_path = _build_captcha_image_path("generic", d=d, ext=".png", tag="fallback") d.screenshot(fallback_path) return Image.open(fallback_path).convert("RGB") def hk(d): screenshot_path = _build_captcha_image_path(CAPTCHA_SLIDER, d=d, ext=".png", tag="full") d.screenshot(screenshot_path) data = verify(screenshot_path, CAPTCHA_SLIDER) if not data: return False try: raw_distance = float(data) except (TypeError, ValueError): return False if raw_distance <= 0: return False image_width = 720 try: with Image.open(screenshot_path) as img: image_width = img.size[0] or 720 except Exception: pass try: screen_width = int(d.info.get("displayWidth", image_width)) except Exception: screen_width = image_width scale = screen_width / image_width if image_width else 1.0 slide_distance = int(raw_distance * scale) if slide_distance < 80: slide_distance += random.randint(3, 6) elif slide_distance < 160: slide_distance += random.randint(2, 5) else: slide_distance += random.randint(1, 4) start_x = 84 + random.randint(-1, 1) start_y = 1034 + random.randint(-2, 2) knob_center = _slider_knob_center(d) if knob_center: start_x, start_y = knob_center max_target_x = screen_width - random.randint(26, 42) target_x = _clamp(start_x + slide_distance, start_x + 18, max_target_x) distance = target_x - start_x if distance < 18: return False method_order = _next_slider_method_order() for method in method_order: if method == "track": points = _build_human_slider_track(int(round(start_x)), int(round(start_y)), int(round(distance))) duration = _slider_duration(distance) time.sleep(random.uniform(0.015, 0.05)) drag_ok, exec_points = _execute_track(d, points, duration) _save_slider_track_image( exec_points, distance=distance, duration=duration, drag_ok=drag_ok, screenshot_path=screenshot_path, captcha_type=CAPTCHA_SLIDER ) else: end_x = target_x + random.uniform(-3, 3) end_y = start_y + random.uniform(-1, 1) drag_ok, exec_points = _execute_bezier_slider(d, start_x, start_y, end_x, end_y) _save_slider_track_image( exec_points, distance=distance, duration=None, drag_ok=drag_ok, screenshot_path=screenshot_path, captcha_type=CAPTCHA_SLIDER ) time.sleep(random.uniform(1.0, 1.8)) if not _slider_still_exists(d): return True print(f"[slider] method {method} failed, switch to next") return False def dianxuan(d): click_area_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]', '//*[@resource-id="com.sankuai.meituan:id/h5_container"]', '//*[@resource-id="root"]', ] image_path, bounds = _capture_by_bounds(d, click_area_xpaths, captcha_type=CAPTCHA_ICON_CLICK) if not image_path or not bounds: return False left = bounds["left"] top = bounds["top"] data = verify(image_path, CAPTCHA_ICON_CLICK) if not data: return False for x, y in data: time.sleep(random.randint(1, 2)) d.click(left + x + random.randint(-7, 7), top + y + random.randint(-7, 7)) return True def wenzidianxuan(d): # 文字点选:按验证码容器 bounds 裁剪后,调用 88888,按返回坐标依次点击 click_area_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]', '//*[@resource-id="com.sankuai.meituan:id/h5_container"]', '//*[@resource-id="root"]', ] image_path, bounds = _capture_by_bounds(d, click_area_xpaths, captcha_type=CAPTCHA_TEXT_CLICK) if not image_path or not bounds: return False image_left = bounds["left"] image_top = bounds["top"] result = post_api(image_path, "88888") verify_data = result.get("data", {}) if not (result.get("code") == 10000 and isinstance(verify_data, dict) and verify_data.get("code") == 0): return False coords_str = verify_data.get("data", "") if not coords_str: return False clicked = 0 for coord in coords_str.split("|"): try: x_img_str, y_img_str = coord.split(",") x_img = int(x_img_str.strip()) y_img = int(y_img_str.strip()) x_screen = image_left + x_img + random.randint(-5, 5) y_screen = image_top + y_img + random.randint(-5, 5) d.click(x_screen, y_screen) clicked += 1 time.sleep(random.uniform(0.8, 1.6)) except Exception: continue return clicked > 0 def click_side(d): """空间推理验证码(请点击数字)。""" click_area_xpaths = [ '//*[@resource-id="com.sankuai.meituan:id/titans_main_layout"]', '//*[@resource-id="com.sankuai.meituan:id/h5_container"]', '//*[@resource-id="root"]', ] image_path, bounds = _capture_by_bounds(d, click_area_xpaths, output_path=CROP_PATH) if not image_path or not bounds: return False left = bounds["left"] top = bounds["top"] points = verify(image_path, CAPTCHA_SPACE_REASON) if not points: return False x, y = points[0] d.click(left + x + random.randint(-2, 2), top + y + random.randint(-2, 2)) # d_list = [ # '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]', # '//*[@resource-id="com.sankuai.meituan:id/yoda_toolbar_title"]', # '//*[@resource-id="com.sankuai.meituan:id/btn_close_verify"]' # ] # for i in d_list: # d.xpath(i).click() return True def Swipe_right(d): """向右拖动到最右侧(非拼图滑块)。""" track_xpath = ( '//*[@resource-id="yodaBoxWrapper"] | ' '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]' ) slider_xpath = ( '//*[@resource-id="yodaBox"] | ' '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.view.View[1]' ) if not d.xpath(track_xpath).exists or not d.xpath(slider_xpath).exists: return False track_bounds = d.xpath(track_xpath).info.get("bounds", {}) slider_bounds = d.xpath(slider_xpath).info.get("bounds", {}) if not track_bounds or not slider_bounds: return False start_x = int((slider_bounds["left"] + slider_bounds["right"]) / 2) + random.randint(-2, 2) start_y = int((slider_bounds["top"] + slider_bounds["bottom"]) / 2) + random.randint(-2, 2) right_limit = int(track_bounds["right"]) - random.randint(4, 10) distance = right_limit - start_x if distance <= 10: return False screenshot_path = _save_debug_screenshot(d, "Swipe_right", tag="full") points = _build_human_slider_track(start_x, start_y, distance) duration = _slider_duration(distance) drag_ok, exec_points = _execute_track(d, points, duration) _save_slider_track_image( exec_points, distance=distance, duration=duration, drag_ok=drag_ok, screenshot_path=screenshot_path or SCREENSHOT_PATH, captcha_type="Swipe_right" ) return True def complexs(d): """Complex slider flow: move to far-right, OCR with label image, then drag back to target.""" slider_xpath_candidates = [ '//*[@resource-id="yodaBox"]', '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]', '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]/android.view.View[1]', ] track_xpath_candidates = [ '//*[@resource-id="yodaBoxWrapper"]', '//*[contains(@text, "请按照说明拖动滑块")]', '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]', ] label_xpath_candidates = [ '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.widget.TextView[1]', ] image_xpath_candidates = [ '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]', ] _, slider_bounds = _first_existing_bounds(d, slider_xpath_candidates) _, track_bounds = _first_existing_bounds(d, track_xpath_candidates) if not slider_bounds or not track_bounds: return False slider_left = slider_bounds["left"] slider_top = slider_bounds["top"] slider_right = slider_bounds["right"] slider_bottom = slider_bounds["bottom"] slider_width = slider_right - slider_left slider_center_x = (slider_left + slider_right) / 2 slider_center_y = (slider_top + slider_bottom) / 2 track_left = track_bounds["left"] track_right = track_bounds["right"] right_end_center_x = track_right - slider_width / 2 right_end_center_y = slider_center_y print(f"滑块中心: ({slider_center_x}, {slider_center_y})") print(f"最右端滑块中心坐标: ({right_end_center_x}, {right_end_center_y})") touch_down = False try: d.touch.down(slider_center_x, slider_center_y) touch_down = True time.sleep(random.uniform(0.08, 0.16)) move_right_points = _build_directional_track( slider_center_x, slider_center_y, right_end_center_x, right_end_center_y, ) _move_with_pressed_touch(d, move_right_points[1:]) print("滑块已到达最右端") _, label_bounds = _first_existing_bounds(d, label_xpath_candidates) _, image_bounds = _first_existing_bounds(d, image_xpath_candidates) if not label_bounds or not image_bounds: return False capture_label_left = label_bounds["left"] capture_label_top = label_bounds["top"] capture_label_right = label_bounds["right"] capture_label_bottom = label_bounds["bottom"] capture_left = image_bounds["left"] capture_top = image_bounds["top"] capture_right = image_bounds["right"] capture_bottom = image_bounds["bottom"] print( "截图区域1(提示文本): " f"left={capture_label_left}, top={capture_label_top}, " f"width={capture_label_right - capture_label_left}, " f"height={capture_label_bottom - capture_label_top}" ) print( "截图区域2(图片): " f"left={capture_left}, top={capture_top}, " f"width={capture_right - capture_left}, " f"height={capture_bottom - capture_top}" ) screenshot_label_path = _build_captcha_image_path("complexs", d=d, ext=".png", tag="label") screenshot_image_path = _build_captcha_image_path("complexs", d=d, ext=".png", tag="image") image = _screenshot_to_image(d) image.crop( (capture_label_left, capture_label_top, capture_label_right, capture_label_bottom) ).save(screenshot_label_path) image.crop( (capture_left, capture_top, capture_right, capture_bottom) ).save(screenshot_image_path) print(f"截图1已保存: {screenshot_label_path}") print(f"截图2已保存: {screenshot_image_path}") result = post_api( screenshot_image_path, "29013", label_image_path=screenshot_label_path, timeout=30, ) print(f"API返回结果: {result}") verify_data = result.get("data", {}) print(f"verify_data={verify_data}") if not (result.get("code") == 10000 and isinstance(verify_data, dict) and verify_data.get("code") == 0): return False data_str = verify_data.get("data", "") if not data_str: return False data_value = int(data_str) print(f"云码返回的像素距离: {data_value}") slider_target_center_x = track_left + data_value min_x = track_left + slider_width / 2 max_x = track_right - slider_width / 2 slider_target_center_x = _clamp(slider_target_center_x, min_x, max_x) print(f"滑块中心目标X坐标: {slider_target_center_x}") _, current_slider_bounds = _first_existing_bounds(d, slider_xpath_candidates) if current_slider_bounds: current_slider_center_x = (current_slider_bounds["left"] + current_slider_bounds["right"]) / 2 else: current_slider_center_x = right_end_center_x actual_distance = slider_target_center_x - current_slider_center_x print(f"实际需要滑动的距离: {actual_distance}") back_points = _build_directional_track( current_slider_center_x, right_end_center_y, slider_target_center_x, right_end_center_y, ) _move_with_pressed_touch(d, back_points[1:]) time.sleep(random.uniform(0.2, 0.4)) d.touch.up(slider_target_center_x, right_end_center_y) touch_down = False time.sleep(random.uniform(1.8, 3.2)) return True except Exception as e: print(f"complex captcha failed: {e}") return False finally: if touch_down: try: d.touch.up(right_end_center_x, right_end_center_y) except Exception: pass def Numbers_English_verify(d): return srwz(d) def slider_verify(d): return hk(d) def Click_images(d): # 两种点选入口统一处理 if d.xpath('//*[@text="请按语序依次点击下图文字"]').exists: return wenzidianxuan(d) return dianxuan(d) def Shortest_connection(d): return lianxian(d) def _handle_generic_captcha(d, xpath, timeout=60): """通用验证码处理:等待人工处理完成。""" start = time.time() while time.time() - start < timeout: if xpath and not d.xpath(xpath).exists: return True time.sleep(1) return False def handle_captcha(d, captcha_type, xpath=None, device_id=None): _set_runtime_device_id(d=d, device_id=device_id) handlers = { "Numbers_English": Numbers_English_verify, "Swipe_right": Swipe_right, "Click_images": Click_images, "slider": slider_verify, "complexs": complexs, "Shortest_connection": Shortest_connection, "click_side": click_side, } func = handlers.get(captcha_type) if func is None: return _handle_generic_captcha(d, xpath) return func(d) def _extract_color_name(api_result): if not isinstance(api_result, dict): return "" if api_result.get("code") == 0 and isinstance(api_result.get("data"), str): return api_result.get("data", "").strip() if api_result.get("code") == 10000: inner = api_result.get("data") if isinstance(inner, dict) and inner.get("code") == 0: return str(inner.get("data", "")).strip() if isinstance(inner, str): return inner.strip() return "" def _normalize_color_name(color_name): if not color_name: return "" alias = { "红": "红色", "红的": "红色", "绿": "绿色", "蓝": "蓝色", "黄": "黄色", "橙": "橙色", "紫": "紫色", "黑": "黑色", "白": "白色", "棕": "棕色", "褐": "褐色", } if color_name in alias: return alias[color_name] for k, v in alias.items(): if k in color_name: return v return color_name def _find_color_coordinates(image_path, color_name): color_name = _normalize_color_name(color_name) color_ranges = { "红色": (([0, 120, 70], [10, 255, 255]), ([170, 120, 70], [180, 255, 255])), "绿色": (([35, 50, 50], [85, 255, 255]),), "蓝色": (([90, 50, 50], [130, 255, 255]),), "黄色": (([20, 100, 100], [30, 255, 255]),), "橙色": (([5, 100, 100], [18, 255, 255]),), "紫色": (([130, 50, 50], [165, 255, 255]),), "黑色": (([0, 0, 0], [180, 255, 50]),), "白色": (([0, 0, 200], [180, 35, 255]),), "棕色": (([8, 60, 20], [20, 255, 180]),), "褐色": (([8, 60, 20], [20, 255, 180]),), } if color_name not in color_ranges: return [] image = cv2.imread(image_path) if image is None: return [] hsv = cv2.cvtColor(image, cv2.COLOR_BGR2HSV) ranges = color_ranges[color_name] if len(ranges) == 2: lower1 = np.array(ranges[0][0]) upper1 = np.array(ranges[0][1]) lower2 = np.array(ranges[1][0]) upper2 = np.array(ranges[1][1]) mask = cv2.bitwise_or(cv2.inRange(hsv, lower1, upper1), cv2.inRange(hsv, lower2, upper2)) else: lower = np.array(ranges[0][0]) upper = np.array(ranges[0][1]) mask = cv2.inRange(hsv, lower, upper) kernel = np.ones((3, 3), np.uint8) mask = cv2.morphologyEx(mask, cv2.MORPH_OPEN, kernel) mask = cv2.morphologyEx(mask, cv2.MORPH_CLOSE, kernel) contours, _ = cv2.findContours(mask, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE) coordinates = [] for contour in contours: area = cv2.contourArea(contour) if area < 30: continue m = cv2.moments(contour) if m["m00"] == 0: continue cx = int(m["m10"] / m["m00"]) cy = int(m["m01"] / m["m00"]) coordinates.append((cx, cy)) return coordinates def _nearest_neighbor_path(points): if not points: return [] if len(points) <= 2: return points[:] unvisited = points[:] path = [unvisited.pop(0)] while unvisited: last_x, last_y = path[-1] idx = min( range(len(unvisited)), key=lambda i: math.hypot(last_x - unvisited[i][0], last_y - unvisited[i][1]) ) path.append(unvisited.pop(idx)) return path def _human_like_path(points): if len(points) < 2: return points[:] curved = [] for i in range(len(points) - 1): start = points[i] end = points[i + 1] mid_x = (start[0] + end[0]) / 2 mid_y = (start[1] + end[1]) / 2 if abs(end[0] - start[0]) > abs(end[1] - start[1]): offset_x = 0 offset_y = random.uniform(-15, 15) else: offset_x = random.uniform(-15, 15) offset_y = 0 control_x = mid_x + offset_x control_y = mid_y + offset_y curved.append(start) for t in np.arange(0.1, 1.0, 0.1): x = (1 - t) ** 2 * start[0] + 2 * (1 - t) * t * control_x + t ** 2 * end[0] y = (1 - t) ** 2 * start[1] + 2 * (1 - t) * t * control_y + t ** 2 * end[1] curved.append((int(x), int(y))) curved.append(points[-1]) return curved def _simulate_human_drawing(d, path): if len(path) < 2: return False try: sx, sy = path[0] d.touch.down(sx, sy) time.sleep(random.uniform(0.05, 0.1)) for i in range(1, len(path)): x, y = path[i] d.touch.move(x + random.randint(-2, 2), y + random.randint(-2, 2)) time.sleep(random.uniform(0.01, 0.03)) time.sleep(random.uniform(0.1, 0.2)) d.touch.up(path[-1][0], path[-1][1]) return True except Exception: return False def retry_captcha( d, xpath_text, handle_func, retry_count=5, captcha_name=None, allow_close_on_third_fail=True, fail_limit_before_close=3 ): # 如果当前页面存在对应验证码,就循环重试处理 if d.xpath(xpath_text).exists: current_name = captcha_name or getattr(handle_func, "__name__", "captcha") fail_streak = 0 for _ in range(retry_count): if not d.xpath(xpath_text).exists: break _save_debug_screenshot(d, current_name, tag="full") try: handle_func(d) except Exception as e: print(f"[captcha] {current_name} handler error: {e}") time.sleep(3) # 验证码消失了,说明处理成功,直接退出 if not d.xpath(xpath_text).exists: break fail_streak += 1 if allow_close_on_third_fail and fail_streak >= fail_limit_before_close: closed = _click_captcha_close(d, captcha_xpath=xpath_text) print(f"[captcha] {current_name} failed {fail_streak} times, switch captcha: {closed}") fail_streak = 0 time.sleep(1.2) def yzm(d=None, device_id=None): # 如果没有传设备对象,就默认连接当前设备 if d is None: d = u2.connect() _set_runtime_device_id(d=d, device_id=device_id) # 向右滑动验证码 retry_captcha( d, '//*[contains(@text, "请向右滑动滑块")]', Swipe_right, captcha_name="Swipe_right", allow_close_on_third_fail=False ) # 滑块验证码 retry_captcha( d, '//*[@text="请拖动下方滑块完成拼图"]', hk, captcha_name="slider", ) # 空间推理验证码 retry_captcha( d, '//*[contains(@text, "请点击")]', click_side, captcha_name="click_side", allow_close_on_third_fail = False ) # 复杂拖动滑块验证码 retry_captcha( d, '//*[contains(@text, "拖动滑块")]', complexs, captcha_name="complexs", allow_close_on_third_fail=False ) # 输入型验证码 retry_captcha(d, '//*[@text="请输入图片中的内容"]', srwz, captcha_name="text_input") # 图标点选验证码 retry_captcha(d, '//*[@text="请依次点击下图图标"]', dianxuan, captcha_name="icon_click") # 文字点选验证码 retry_captcha(d, '//*[@text="请按语序依次点击下图文字"]', wenzidianxuan, captcha_name="text_click") # 最短线连接验证码 retry_captcha(d, '//*[contains(@text, "用最短线连接")]', lianxian, captcha_name="Shortest_connection") def lianxian(d): art_text_xpath = '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[1]' color_points_xpath = '//*[@text="身份核实"]/android.view.View[1]/android.view.View[1]/android.view.View[2]/android.view.View[1]/android.widget.Image[1]' art_text_img_path = _build_captcha_image_path("Shortest_connection", d=d, ext=".png", tag="art_text") color_points_img_path = _build_captcha_image_path("Shortest_connection", d=d, ext=".png", tag="color_points") art_text_img_path, _ = _capture_by_bounds( d, art_text_xpath, output_path=art_text_img_path, captcha_type="Shortest_connection" ) color_points_img_path, color_bounds = _capture_by_bounds( d, color_points_xpath, output_path=color_points_img_path, captcha_type="Shortest_connection" ) if not art_text_img_path or not color_points_img_path or not color_bounds: return False element_left = color_bounds["left"] element_top = color_bounds["top"] element_width = color_bounds["right"] - color_bounds["left"] element_height = color_bounds["bottom"] - color_bounds["top"] api_result = post_api(art_text_img_path, "10118") color_name = _extract_color_name(api_result) if not color_name: return False relative_points = _find_color_coordinates(color_points_img_path, color_name) if len(relative_points) < 2: return False color_img = cv2.imread(color_points_img_path) if color_img is None: return False img_h, img_w = color_img.shape[:2] if img_w <= 0 or img_h <= 0: return False screen_points = [] for rx, ry in relative_points: sx = element_left + int(rx * (element_width / img_w)) sy = element_top + int(ry * (element_height / img_h)) screen_points.append((sx, sy)) path = _nearest_neighbor_path(screen_points) curved = _human_like_path(path) return _simulate_human_drawing(d, curved) if __name__ == '__main__': d = u2.connect("GQIRKB7LVOONM7VW") yzm(d)