安卓刷单脚本开发思路

背景

最近公司叫我们刷单，小程序上租借充电宝, 每人每天60单。

手点到手软，一般60单也需要30分钟左右，太影响心情了。

于是萌生出写个脚本的想法。

思路很明确，租借流程是固定的，所以手机能够自动化点击就行，于是产生了以下方案：

• 使用手机连点器
• 使用scrcpy 加ADB方案, 这里又有两个思路: 1. ocr识别, 2. cv2纯图像识别

1. 方案一, 连点器

手机连点器是最开始尝试的，因为开发最简单，测试部门那边开发的。唯一不好解决的是小程序内部广告，为此我们单独关闭广告白名单功能。 想着目的马上能达到，激动的心，颤抖的手。大规模跑一次有没有！ 接着就是刷了一次后微信居然封号了（需要再认证一遍才能解锁，提示涉嫌刷单），WC？

2. 方案2, scrcpy + ocr

其实这里使用adb 就可以了，思路是远程使用adb 进行截图，然后再使用OCR进行图像识别。 然后根据识别的关键字进行匹配，确定当前页面和流程。

OCR 我们使用的RapidOCR， 本来计划使用PaddleOCR, 但是测试阶段发现，识别一次很久，需要 10s+。 后面找都 papidocr， 识别只需要2-3s。

接着就是很简单的业务流程处理。识别截图，确定当前租借处于那一步。然后就完成了。 至于广告，识别出来类似于(X,x)直接就关闭了。

很快这个方案也就开发完成了，大约一天把。

这个方案能跑起来，但是有以下缺点：

1.ocr 识别太慢了，导致误点广告 某些时候，截图+OCR识别总共需要3S， 但是屏幕上此时弹出广告， 程序再去操作就误点广告了。 2.截图太慢, 导致整体效率很拉跨 因为截图+OCR识别太久了，需要3-4S， 每一步操作后，操作下个页面前也使用 sleep 等待的方式，那么整体就会很拖垮。实测一单需要45s左右。

2. 方案3, scrcpy + cv2， 图形识别

这个方案使用scrcpy 将手机屏幕流式传输到电脑（也就是实时的），电脑端每200ms采集一次图像，然后使用cv2 识别图像中的关键帧。 比如租借按钮, 确定按钮之类的（需要提前将这些icon裁剪出来）。

这个方案由于是图像搜索，每次只需要1s， 而且每200ms就能触发检测一次。反应比较灵活。图像关键帧也还是好使。 实测一单需要30s左右,和手工操作一样了。

附录1： scrcpy-client 安装失败。

Py3.10.10,每次安装scrcpy-client报错：

C:\Users\ray>pip install --upgrade scrcpy-client
Collecting scrcpy-client
  Using cached scrcpy_client-0.4.7-py3-none-any.whl.metadata (2.8 kB)
Collecting adbutils<2.0.0,>=1.0.8 (from scrcpy-client)
  Using cached adbutils-1.2.15-py3-none-win_amd64.whl.metadata (915 bytes)
Collecting av<10.0.0,>=9.0.0 (from scrcpy-client)
  Using cached av-9.2.0.tar.gz (2.4 MB)
  Installing build dependencies ... done
  Getting requirements to build wheel ... error
  error: subprocess-exited-with-error

  × Getting requirements to build wheel did not run successfully.
  │ exit code: 1
  ╰─> [70 lines of output]
      Compiling av\buffer.pyx because it changed.
      [1/1] Cythonizing av\buffer.pyx
      Compiling av\bytesource.pyx because it changed.
      [1/1] Cythonizing av\bytesource.pyx
      Compiling av\descriptor.pyx because it changed.

这个问题可以通过以下步骤解决:

1.Github上下载源码 https://github.com/leng-yue/py-scrcpy-client

2.单独安装符合你版本的av 版本，我这里安装的是17.1.0

3.复制scrcpy,scrcpyui到你开发目录下直接使用就行

附录2： cv2方案源码

from adbutils import adb
import scrcpy

import cv2
from pyzbar.pyzbar import decode
import time
import threading
import cv2
import numpy as np

import cv2
# Setup client
debug = True
bastdir = "D:\\shuadanimgs\\v2\\"

# 步骤截图
im_step1 = cv2.imread(bastdir + "1.png")
im_step2 = cv2.imread(bastdir + "2.png")
im_step3 = cv2.imread(bastdir + "3.png")
im_step4 = cv2.imread(bastdir + "4.png")
im_step6 = cv2.imread(bastdir + "6.png")
im_step7 = cv2.imread(bastdir + "7.png")


im_step8 = cv2.imread(bastdir + "8.png")
im_step9 = cv2.imread(bastdir + "9.png")

im_step_u1 = cv2.imread(bastdir + "13_u.png")
im_step_u2 = cv2.imread(bastdir + "12_u.png")

im_step10 = cv2.imread(bastdir + "10.png")


# 广告截图
im_ad1 = cv2.imread(bastdir + "ad1.png") 
im_ad2 = cv2.imread(bastdir + "ad2.png") 


client = None
refreshtime = int(round(time.time() * 1000))
items =  adb.device_list()
print("[已连接设备] ", items)

if(len(items) > 0):
    client = scrcpy.Client(
        device=items[0],
        flip=False,
        bitrate=1000000000,
        encoder_name=None,
        max_fps=10,
    )
else:
    print("[已连接设备] 未检测到连接设备")



def find_qrcode_position_stable_v2(large_frame) -> tuple[int, int] | None:
    if large_frame is None:
        return None

    # ---- 诊断第一步：检查图像格式 ----
    # print(f"[QR Debug] 输入大图尺寸: {large_frame.shape}")

    # ---- 尝试 1：直接使用原始彩色图解码（很多时候最稳） ----
    qr_codes = decode(large_frame)

    # ---- 尝试 2：如果失败，转为标准灰度图解码 ----
    if not qr_codes:
        gray = cv2.cvtColor(large_frame, cv2.COLOR_BGR2GRAY)
        qr_codes = decode(gray)

    # ---- 尝试 3：如果还失败，使用大窗口自适应二值化（保护二维码的静区边缘） ----
    if not qr_codes:
        # 使用自适应阈值，避免固定 127 阈值把白色背景强行抹白导致边缘丢失
        gray_adaptive = cv2.adaptiveThreshold(
            gray, 255, cv2.ADAPTIVE_THRESH_GAUSSIAN_C, cv2.THRESH_BINARY, 51, 10
        )
        qr_codes = decode(gray_adaptive)

    # ---- 结果解析 ----
    if qr_codes:
        qr = qr_codes[0]
        data_text = qr.data.decode("utf-8")
        x, y, w, h = qr.rect

        # 计算几何中心点
        center_x = int(x + w / 2)
        center_y = int(y + h / 2)

        print(
            f"🎯 [QR 成功] 内容: '{data_text}', 中心坐标: ({center_x}, {center_y})"
        )
        return (center_x, center_y)

    print("❌ [QR 彻底失败] 尝试了原图、灰度图、自适应二值化均未能识别。")
    return None


# print(find_qrcode_position_stable_v2(cv2.imread( bastdir + "tmp\\" + "1787559839028.png")))
# print(find_qrcode_position_stable_v2(cv2.imread(bastdir + "tmp\\" + "1787559840761.png")))
# print(find_qrcode_position_stable_v2(cv2.imread(bastdir + "tmp\\" + "1787559865630.png")))
# print(find_qrcode_position_stable_v2(cv2.imread(bastdir + "tmp\\" + "1787559864060.png")))


def timestamp():
    return int(round(time.time() * 1000))

class PageChangeDetector:

    def __init__(self, diff_threshold=0.20, resize_width=320):
        """初始化检测器

        Args:
            diff_threshold: 页面切换的阈值。
                            0.10 表示当 10% 的像素发生明显变化时，判定为页面切换。
                            可根据实际 APP 的动效大小调节（通常在 0.05 ~ 0.15 之间）。
            resize_width: 内部计算时的缩放宽度。
                          将大图缩小到 320 宽再计算，能提升数十倍速度，且不影响大面积切换的准确率。
        """
        self.diff_threshold = diff_threshold
        self.resize_width = resize_width
        self.prev_frame = None  # 用于缓存上一帧

    def check_page_changed(self, current_frame) -> bool:
        """输入当前帧，判断页面是否发生了切换

        Args:
            current_frame: numpy array, scrcpy 回调传过来的原始 BGR 帧

        Returns:
            bool: True 表示页面切换了，False 表示画面基本没变
        """
        if current_frame is None:
            return False
        
        # 1. 降低分辨率（加快计算速度）
        h, w = current_frame.shape[:2]
        ratio = self.resize_width / w
        resized = cv2.resize(
            current_frame,
            (self.resize_width, int(h * ratio)),
            interpolation=cv2.INTER_AREA,
        )

        # 2. 转为灰度图（色彩对于判断页面切换通常是冗余的，灰度足矣）
        gray = cv2.cvtColor(resized, cv2.COLOR_BGR2GRAY)

        # 3. 高斯模糊（消除传感器噪声、小动画或文字微调带来的干扰）
        gray_blurred = cv2.GaussianBlur(gray, (21, 21), 0)

        # 4. 如果是第一帧，先缓存并返回 False
        if self.prev_frame is None:
            self.prev_frame = gray_blurred
            return False

        # 5. 计算当前帧与上一帧的绝对差值
        frame_delta = cv2.absdiff(self.prev_frame, gray_blurred)

        # 6. 二值化：将差值大于 25 (显著变化) 的像素点设为 255 (白色)，其余设为 0 (黑色)
        _, thresh = cv2.threshold(frame_delta, 25, 255, cv2.THRESH_BINARY)

        # 7. 计算发生变化的像素点占总像素点的比例
        total_pixels = thresh.size
        changed_pixels = cv2.countNonZero(thresh)
        change_ratio = changed_pixels / total_pixels

        # 8. 更新上一帧缓存，供下一次对比使用
        self.prev_frame = gray_blurred

        # 9. 判定：如果变化比例大于设定的阈值，则认为页面切换
        if change_ratio > self.diff_threshold:
            # 打印日志方便你调试阈值
            print(
                f"[Page Changed] 画面变化比例: {change_ratio:.2%}, 触发切换！",
                timestamp()
            )
            save(current_frame)
            return True

        return False
import cv2
import numpy as np


def find_template_position(
    large_frame, template_img, threshold=0.90
) -> tuple[int, int] | None:
    """在大图（当前帧）中寻找小图（模板）的位置，并返回其中央点击坐标

    Args:
        large_frame: scrcpy 传过来的当前帧原始大图 (BGR 格式)
        template_img: 你提前截好并保存的小图 (可以是路径字符串，或者是已用 cv2.imread 读取的矩阵)
        threshold: 匹配置信度阈值 (0.0~1.0)。
                   0.85 表示图片相似度达到 85% 才算找到了。如果找不到，可以调低到 0.75。

    Returns:
        (x, y): 小图中心在原始大图中的像素坐标。如果找不到，则返回 None。
    """
    if large_frame is None:
        return None

    # 1. 确保模板图是 numpy 矩阵格式
    if isinstance(template_img, str):
        # 如果传入的是本地图片路径，直接读取
        # 💡 避坑指南：如果路径包含中文，请确保 cv2.imread 不会返回 None
        template = cv2.imread(template_img)
    else:
        template = template_img

    if template is None:
        print("错误: 无法加载模板图片，请检查路径或数据！")
        return None

    # 获取小图的宽高（用于后面计算中心点）
    th, tw = template.shape[:2]

    # 2. 执行模板匹配 (推荐使用 CCOEFF_NORMED，对光照和颜色变化最鲁棒)
    result = cv2.matchTemplate(large_frame, template, cv2.TM_CCOEFF_NORMED)

    # 3. 寻找匹配度最高的位置
    # min_val: 最小匹配度, max_val: 最大匹配度 (置信度), min_loc: 最小值坐标, max_loc: 最大值坐标
    min_val, max_val, min_loc, max_loc = cv2.minMaxLoc(result)

    # 4. 判断最高匹配度是否超过阈值
    if max_val >= threshold:
        # max_loc 是小图在左上角的坐标 (top_left_x, top_left_y)
        top_left_x, top_left_y = max_loc

        # 5. 计算小图的中心点坐标（点击中央比点击左上角更安全，不容易漏点）
        center_x = int(top_left_x + tw / 2)
        center_y = int(top_left_y + th / 2)

        print(
            f"[Match Success] 找到目标！置信度: {max_val:.2f}, 中心点坐标: ({center_x}, {center_y})"
        )
        return (center_x, center_y)

    # print(f"[Match Failed] 未找到目标，最高相似度仅为: {max_val:.2f}")
    return None

def make_template_tool(output_name="template.png"):
    print("正在从手机获取实时截图...")
    # 1. 自动通过 ADB 获取当前手机原生截屏
    device = adb.device()
    pil_img = device.screenshot()
    
    # 2. 转换为 OpenCV 矩阵格式
    frame = cv2.cvtColor(np.array(pil_img), cv2.COLOR_RGB2BGR)
    
    # 3. 弹出 OpenCV 自带的 ROI (感兴趣区域) 选择器
    # 提示：在弹出的窗口中，用鼠标框选目标，确认无误后按 回车键(Enter) 或 空格键
    print("【操作提示】: 在弹出的窗口中用鼠标框选按钮，选好后按 Enter 键保存。")
    roi = cv2.selectROI("Select Target Button", frame, fromCenter=False, showCrosshair=True)
    
    # 4. 解析坐标并保存
    x, y, w, h = roi
    if w > 0 and h > 0:
        crop_img = frame[y:y+h, x:x+w]
        cv2.imwrite(output_name, crop_img)
        print(f"🎉 完美模板制作成功！已无损保存至: {output_name}")
    else:
        print("取消了裁剪")
        
    cv2.destroyAllWindows()

pageChangeDetector = PageChangeDetector()

def waitpagechange(frame, wait=True, sec=12):
    if wait == False:
        return
    for i in range(sec * 5):
        time.sleep(0.2)
        res = pageChangeDetector.check_page_changed(frame)
        if(res == True):
            return res

def dostep(frame, imstep):
    res = False
    for i in range(1):
        p = find_template_position(frame, imstep)
        if(p != None):
            client.control.touch(p[0], p[1],scrcpy.ACTION_DOWN)
            client.control.touch(p[0], p[1],scrcpy.ACTION_UP)
            res = True
            break
        time.sleep(0.1)
    return res
    # waitpagechange(frame)
def dostep_check(frame, *imstep):
    res = False
    p = None
    for istep in imstep:
        p = find_template_position(frame, istep)
        if(p == None):
            return False
        time.sleep(0.1)

    client.control.touch(p[0], p[1],scrcpy.ACTION_DOWN)
    client.control.touch(p[0], p[1],scrcpy.ACTION_UP)
    return True

currentStep=1
currentImg = None

# 具体业务处理，这里处理每个流程
def dobiz(frame):
    global pageChangeDetector
    global bastdir
    global client
    global currentStep
    global statics

    cv2.imwrite(bastdir +"tmp\\" + str(timestamp()) +".png", frame)

    pageChangeDetector.check_page_changed(frame)
    if(r := dostep(frame, im_ad1)):
        statics[2] += 1
        return
    if(r := dostep(frame, im_ad2)):
        return
    p = find_template_position(frame, im_step8)
    if(None != p):
        dostep(frame, im_step9)
    
    # make_template_tool(bastdir + "2.png")
    # 点击租借
    if(currentStep == 1 ):
        if(dostep(frame, im_step1)):
            currentStep = 1.1
        if(dostep_check(frame, im_step_u1, im_step_u2)):
            pass
        return
    if(currentStep == 1.1 and  dostep(frame, im_step2)):
        currentStep = 1.2
        return
    if(currentStep == 1.2):
        p = find_qrcode_position_stable_v2(frame)
        if(p != None):
            client.control.touch(p[0], p[1] + 20, scrcpy.ACTION_DOWN)
            client.control.touch(p[0], p[1] + 20, scrcpy.ACTION_UP)
            currentStep = 5
            return
    if(currentStep == 5 and  dostep(frame, im_step4)):
        currentStep = 6
        time.sleep(1)
        return
    if(currentStep == 6):
        if(dostep(frame, im_step6)):
            currentStep = 7
        if(dostep(frame, im_step10)):
            currentStep = 1
            statics[1] += 1
            return
    if(currentStep == 7 and dostep(frame, im_step7)):
        currentStep = 1
        statics[0] += 1
        return

    
    return
    pass
def biz():
    global currentImg
    try:
        print("后台线程：scrcpy 服务启动成功...")
        while True:
            s1 = timestamp()
            if(currentImg is not None):
                dobiz(currentImg.copy())
                print("total spend", timestamp() - s1)
                time.sleep(0.1)
    except Exception as e:
        print(e)


def save(frame):
    global bastdir, debug
    max_width = 800
    if(debug):
        cv2.imwrite(bastdir + "raw.png", frame)
    # 1. 调整画面大小（对应原代码中的等比例缩放）
    ratio = max_width / max(client.resolution)
    width = int(frame.shape[1] * ratio)
    height = int(frame.shape[0] * ratio)
    resized_frame = cv2.resize(frame, (width, height), interpolation=cv2.INTER_LINEAR)
    
    # 2. 用 OpenCV 窗口直接显示
    cv2.imwrite(bastdir +"ratio.png", resized_frame)
    
bizthread = threading.Thread(target=biz, args=(), daemon=True)
bizthread.start()
# 统计数据 ， 总订单数, 有人租借中提示次数,  首页租借按钮点击次数, 总耗时秒
statics = [0, 0, 0, 0]
start = timestamp()
t3 = timestamp()
# 每 100ms 刷新一次
def on_frame(frame):
    global currentImg
    global refreshtime
    global statics, t3, start
    if frame is not None:
        t2 =   timestamp()
        statics[3] = (t2 - start) / 1000
        if(t2 - t3 > 10000):
            # save(frame)
            t3 = t2
            print("[statics] [总订单数, 提示有人租借中次数,  首页租借按钮点击次数, 总耗时秒数]", statics)
        if(t2 - refreshtime >= 200):

            refreshtime = t2
            currentImg = frame.copy()
            # waitpagechange(frame, wait=False)


client.add_listener(scrcpy.EVENT_FRAME, on_frame)

# 假设这是启动 scrcpy 的函数
def start_scrcpy_client(client_instance):
    if(client is None):
        print("后台线程：未检测到手机连接，检查调试模...")
        return
    print("后台线程：正在启动 scrcpy 服务...")
    
    client_instance.start()

scrcpy_thread = threading.Thread(target=start_scrcpy_client, args=(client,), daemon=True)
scrcpy_thread.start()

while True:
    time.sleep(1)