Mrli
别装作很努力,
因为结局不会陪你演戏。
Contacts:
QQ博客园

每周一个开源项目——ddt-sharp-shooter

2022/05/30 Python 开源项目
Word count: 1,814 | Reading time: 9min

ddt-sharp-shooter

这是一个基于 Pynput 的 DDT 工具。基本原理在于，得知风力、角度、距离的情况下，参考力度表得出发射力度，而后发射。
其中，风力、角度通过 ddddocr（An awesome captcha recognition library）识别，屏距通过标记屏距测量框、敌我位置来推算，力度通过按压时长来体现，具体见这里。

使用到的库:

screeninfo、pillow、ddddocr、pynput、py2app

  • pynput: 控制和监视输入设备；类似的有PyHook3（监视键鼠）、pywin32 （模拟键鼠）
  • ddddocr：识别验证码，这边用来识别数字
  • py2app: 将Python程序打包成MacOS应用程序
  • screeninfo: 获得屏幕显示信息：monitors = screeninfo.get_monitors()、_screen_size = (monitors[0].width, monitors[0].height)
  • pillow: 进行屏幕截图

What’s New:

1.进程间通信

Tkinter界面开启mainloop进程，其中又开辟出一个子线程来侦听其他进程发送的数据消息，然后通过tk.Text控件来展示

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import multiprocessing
import threading
import time
import tkinter
# 声明全局对象及类型
_tk: tkinter.Tk
_text: tkinter.Text
_terminate = False
_queue: multiprocessing.Queue

def update_text():
    """子线程侦听进程消息"""
    while not _terminate:
        if not _queue.empty():
            text = _queue.get(False)
            append_text(text)
        else:
            time.sleep(1)

def append_text(text):
    _text.config(state='normal')
    _text.insert('end', f'\n{text}')
    _text.see('end')
    _text.config(state='disabled')

def run(gui_queue):
    global _tk, _text, _queue, _screen_size
    _queue = gui_queue
    # ...
    threading.Thread(target=update_text).start()
    
    _tk.mainloop()

2.py2app创建MacOS应用

py2app setup，在macos下创建python应用， python setup.py py2app

3.ddddocr识别数字并清晰

对纯数字识别结果进行清洗

1
2
3
4
5
6
7
8
9
10
11
12
13
14
def recognize_digits(image: bytes):
    """进行数字识别"""
    ocr = ddddocr.DdddOcr(show_ad=False)
    result = ocr.classification(image)
    return wash_digits(result)

def wash_digits(digits: str):
    """由于不会出现非数字, 所以对易识别错误的字符进行替换"""
    washed = digits \
        .replace('g', '9').replace('q', '9') \
        .replace('l', '1').replace('i', '1') \
        .replace('z', '2') \
        .replace('o', '0')
    return re.sub(r'\D', '0', washed)

4.pynput处理键鼠事件

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
# km.py
"""
Keyboard and mouse input/output,
and ScreenGrabbing
"""
from pynput import keyboard, mouse

def on_click(x, y, btn_type, down):
    """鼠标点击回调函数"""
    if btn_type == mouse.Button.left and down:
        _queue.put((int(x), int(y)))


def on_press(event):
    """按键回调函数"""
    try:
        # 处理正常的ASCII字符
        if event.char == 'q':
            # 设置毒药, 当queue.get()获得None的时候, 消费端应该被kill
            _queue.put(None)
        else:
            _queue.put(event.char)
    except AttributeError:
        # 处理特殊的按键
        if event == keyboard.Key.esc:
            _queue.put('esc')
        elif event == keyboard.Key.enter:
            _queue.put('enter')
        elif event == keyboard.Key.space:
            _queue.put(' ')
        elif event == keyboard.Key.backspace:
            _queue.put('delete')


def space_press_and_release(duration):
    """Press the key 'space' down for a while, and then release"""
    # @param duration: 持续时间
    k = keyboard.Controller()
    k.press(keyboard.Key.space)
    time.sleep(duration)
    k.release(keyboard.Key.space)


def key_press_and_release(key):
    """Press certain key several times down, and then release immediately"""
    # @param key: 按下按键
    k = keyboard.Controller()
    k.press(key)
    k.release(key)
    time.sleep(0.37)


def run(km_queue):
    global _queue
    _queue = km_queue
	# 注册键盘事件监听与回调函数
    keyboard_listener = keyboard.Listener(on_press=on_press)
    keyboard_listener.start()
    time.sleep(.5)
    # 注册鼠标事件监听与回调函数
    mouse_listener = mouse.Listener(on_click=on_click)
    mouse_listener.start()


if __name__ == '__main__':
    run(None)
    while True:
        time.sleep(1)

on_press函数只是产生了queue的数据，数据具体是在main.py中的handle_inputs中被消费的

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
def km_listen_queue():
    while True:
        inputs = _km_queue.get()
        if inputs is None:  # poison bill
            # 接收毒药, 毒药的设置可见上文Listener中回调函数是何时put None的
            if len(_wind_degree_points) > 0:
                with open(_W_D_POINTS_DUMP_NAME, 'wb') as f:
                    pickle.dump(_wind_degree_points, f)
            _gui_process.terminate()
            break
        else:  # handle inputs
            handle_inputs(inputs)

def handle_inputs(inputs):
    """To handle inputs"""
    global _command_flag, _direct_force_typing, _wind_direction
    inputs_type = type(inputs)
    # 键盘事件on_press, 返回str
    if inputs_type == str:
        # press ESC to cancel
        if inputs == 'esc':
            reset_inputs()
        elif inputs == '-':
            _wind_direction = -1
        elif inputs == '=':
            _wind_direction = 1
        # press the key 't' twice to enable command mode
        elif inputs == 't':
            _command_flag += 1
            # 最大t的次数为4, 所以4为一个周期
            _command_flag %= 4
            if _command_flag == 2:
                _gui_queue.put("指令输入开启💡")
            elif _command_flag == 0:
                _gui_queue.put("指令输入关闭🔒")
        elif _command_flag == 2:
            # _command_flag == 2此时为命令模式, 输入enter完成设置, reset_inputs
            # press enter to submit command and fire
            if inputs == 'enter':
                direct_force = analyse_direct_force()
                if direct_force > 0:
                    fire(force=direct_force)
                else:
                    wind, degree, distance = analyse_wind(), analyse_degree(), analyse_distance()
                    fire(wind, degree, distance)
                reset_inputs()
            # edit command
            elif inputs == 'delete':
                _direct_force_typing = _direct_force_typing[:-1]
            else:
                _direct_force_typing += inputs
        # when not in command mode
        # any key except 't' will reset mode flag
        # which means only consecutive 't' input can enable command mode
        elif _command_flag == 1:
            reset_inputs()
    # 鼠标事件on_click, 返回(x, y)
    elif inputs_type == tuple:
        if _command_flag == 2:
            _distance_points.append(inputs)
            _gui_queue.put(f'{len(_distance_points)} 个点已标记')
        # 按三次t之后再点击, 进入设置“角度中心位置、风力中心位置”
        if _command_flag == 3:
            if len(_wind_degree_points) == 2:
                # 如果标记过了风力和角度, 则重新标记
                _wind_degree_points.clear()
            _wind_degree_points.append(inputs)
            if len(_wind_degree_points) == 1:
                _gui_queue.put('角度位置已标记📐️')
            else:
                _gui_queue.put('风力位置已标记🌪️')

5.屏幕截图

1
2
3
4
5
6
7
8
9
10

def grab_box(box: tuple) -> bytes:
    # 开启内存IO
    bytes_io = io.BytesIO()
    # 截图
    image = ImageGrab.grab().resize((_screen_size[0], _screen_size[1])).crop(box)
    # 写入内存IO
    image.save(bytes_io, format='png')
    # 获得bytes值
    return bytes_io.getvalue()

执行流程

mian.py为程序入口

  1. 定义全局变量

    1
    2
    3
    4
    5
    6
    7
    8
    9
    10
    11
    12
    13
    14
    _W_D_POINTS_DUMP_NAME = 'wind_degree_points.list'
    _PRESS_DURATION_PER_FORCE = 4.1 / 100
    _screen_size: tuple
    _command_flag = 0
    _direct_force_typing = ''
    _wind_direction = -1
    _distance_unit = 0  # 用于像素与屏距转换
    _distance_points = []  # 用于计算 _distance_unit 以及 屏距
    _wind_degree_points = []  # 用于配置角度、风力 屏幕截取位置的点
    _gui_process: multiprocessing.Process
    # GUI是另外的进程, 所以需要使用进程间的队列
    _gui_queue = multiprocessing.Queue()
    # 相比之下, km是主进程中的, 不涉及进程间数据通信, 所以直接使用普通Queue就行
    _km_queue = Queue()

  2. 获得屏幕信息：_screen_size = (monitors[0].width, monitors[0].height)

  3. 开启GUI进程: _gui_process = multiprocessing.Process(target=gui_run, args=(_gui_queue,))

  4. 开启pynput的键鼠侦听Listener线程

  5. 开启键鼠处理线程

    1
    2
    3
    4
    threading.Thread(target=km_listen_queue).start()
    # km_listen_queue中接收poison bill(即按下"q")时, 关闭进程
    threading.Thread(target=gui_check_alive).start()
    # GUI检测心跳线程, 当GUI的状态不是is_alive时, 向queue中投递毒药, 将terminate设置为False. 从而关闭GUI进程中的update_text线程

  6. 在键鼠处理线程中，根据键鼠输入值，进行功能生效

代码风格特点：

每个模块（py文件）都有自己的全局变量，通过主程序调用模块函数时进行传引用，而变量全在主程序main.py中定义。

总结：

涉及了多线程、多进程、图像识别、屏幕截图识别、键鼠检测、打包应用等内容，其中对于①多线程和多进程使用、②识别结果后的针对纯数字数据进行清洗；③消息队列的使用、毒药设置；都比较让人有收获。

视频思路

  1. 讲解程序功能
  2. 介绍使用到的库
  3. 讲解程序文件结构
  4. 讲解每个文件实现
  5. 介绍优点和有价值的
  6. 介绍缺点
    1. C式的全局变量风格

【PR教程】2分钟学会制作视频内容导航条

Author: Mrli

Link: https://nymrli.top/2022/05/29/每周一个开源项目——ddt-sharp-shoote/

Copyright: All articles in this blog are licensed under CC BY-NC-SA 3.0 unless stating additionally.

< PreviousPost
爬虫——请求参数逆向NextPost >
爬虫App——夜神模拟器xposed+inspeckage
CATALOG
  1. 1. ddt-sharp-shooter
    1. 1.1. 使用到的库:
    2. 1.2. What’s New:
      1. 1.2.1. 1.进程间通信
      2. 1.2.2. 2.py2app创建MacOS应用
      3. 1.2.3. 3.ddddocr识别数字并清晰
      4. 1.2.4. 4.pynput处理键鼠事件
      5. 1.2.5. 5.屏幕截图
    3. 1.3. 执行流程
    4. 1.4. 代码风格特点：
    5. 1.5. 总结：
  2. 2. 视频思路