mitmproxy 的使用

pip install mitmproxy

安装 https 证书： http://mitm.it/

需将浏览器切换至 8080端口的代理

image-20230525114631624.png

image-20230525114757995.png

设置全局代理

image-20230525114908131.png

1
http=127.0.0.1:8080;https=127.0.0.1:8080

使用脚本拦截解析请求

建议使用mitmweb，关闭浏览器，一边调试脚本

1
mitmweb -s 大众点评自动化.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
# -*- coding: utf-8 -*-
# time: 2023/5/24 17:52
# author: Chen

import json

from pprint import pprint


def response(flow):
    # 检查请求的 URL 是否匹配指定的地址
    urls = (
        "https://ihotel.meituan.com/group/v1/poi/0",  # 详情
        "https://ihotel.meituan.com/api/v2/comments/biz/poiReview",  # 评分
    )
    if flow.request.url.startswith(urls):
        # 解析响应的 JSON 数据
        data = json.loads(flow.response.content.decode("utf-8"))
        pprint(data)

PyAutoGUI

pip install pyautogui

官方文档：https://pyautogui.readthedocs.io/en/latest/index.html?highlight=maximize#

PyAutoGUI——图形用户界面自动化 - 知乎

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
import pyautogui

# 按下Win键
pyautogui.hotkey("win")
# 输入程序名字
pyautogui.typewrite("weixin")
# pyautogui.press("backspace")
# 按下回车键
pyautogui.press("enter")
# 使用快捷键 Win+Up 实现窗口最大化
pyautogui.hotkey("win", "up")
# 鼠标移动
pyautogui.moveTo(1600, 866 - y * 50, duration=1)
# 点击
pyautogui.click()
pyautogui.click(x, y)
解决中文输入
1
2
3
4
import pyperclip

pyperclip.copy("中文")  # 复制到剪贴板
pyautogui.hotkey("ctrl", "v")  # 粘贴

playwright

浏览器初始化：playwright install

基本用法：

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
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
class DyLogin:
    def __init__(self, playwright, dir_path, username, password, shop_name):
        self.pw = playwright
        self.home_url = "https://compassbrand.jinritemai.com/cabin"
        self.user_dir = dir_path
        self.browser = None
        self.page = None
        self.frame = None
        self.username = username
        self.password = password
        self.shop_name = shop_name
        # self.chrome = False

    def start(self):
        """
        主调度程序
        """
        self.launch()
        self.login()
        self.save()
        self.browser.close()

    def launch(self):
        self.browser = self.pw.chromium.launch(
            proxy={"server": "http://10.244.11.3:31280"},
            headless=False,
            args=["--start-maximized", "--disable-features=AutomationControlled"],
            # traces_dir=self.user_dir,
            # executable_path="C:\\Program Files (x86)\\Google\\Chrome\\chrome.exe",
        )
        context = self.browser.new_context(
            # viewport={"width": 1920, "height": 1080},
            no_viewport=True,  # 最大化
            # user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
            ignore_https_errors=True,
        )
        self.page = context.new_page()
        code = (
            """Object.defineProperties(navigator, {webdriver:{get:()=>undefined}});"""
        )
        self.page.add_init_script(code)

    def login(self):
        self.page.goto(self.home_url)
        self.page.wait_for_timeout(10000)
        self.page.click("div.switch-switch.email")
        self.page.wait_for_timeout(1000)
        self.page.fill('input[fieldname="item_label_email"]', "")  # 清空
        self.page.type(
            'input[fieldname="item_label_email"]',
            self.username,
            delay=random.randint(100, 150),
        )
        self.page.wait_for_timeout(1000)
        self.page.fill('input[fieldname="item_label_pwd"]', "")
        self.page.type(
            'input[fieldname="item_label_pwd"]',
            self.password,
            delay=random.randint(100, 150),
        )
        self.page.wait_for_timeout(1000)
        self.page.click("input.ecom-checkbox-input")
        self.page.wait_for_timeout(1000)
        self.page.click("button.account-center-action-button")
        self.page.wait_for_timeout(2000)
        # 滑块
        frames = self.page.frames
        if len(frames) > 1:
            logger.info("出现滑块，开始模拟滑动")
            self.frame = frames[1]
            self.mouse_slide()
        # 短信验证码
        while self.page.url.endswith("/login"):
            logger.info("请处理短信验证码")
            self.page.wait_for_timeout(30000)
        self.page.wait_for_timeout(15000)
        # ele = self.page.locator("div.ecom-dropdown-trigger>span:first-of-type")
        # self.shop_name = ele.text_content()

    def mouse_slide(self):
        slide_try_times = 0
        while slide_try_times < 10:
            slide_try_times += 1
            self.download_pic()
            self.puzzle_pic()
            slider = self.frame.locator("img#captcha-verify_img_slide")
            cdn = slider.bounding_box()
            length = cdn["x"] + 27 + 4
            high = cdn["y"]
            move_length = self.get_distance()
            real_length = move_length * 340 / 552
            logger.info(f"开始滑动操作，滑动距离：{move_length}")
            self.frame.hover("img#captcha-verify_img_slide")
            self.page.mouse.down()
            self.page.mouse.move(length, high)
            for x in self.get_track(real_length):
                length = length + x
                self.page.mouse.move(length, high)
                self.page.wait_for_timeout(random.randint(10, 15))
            self.page.mouse.up()
            self.page.wait_for_timeout(5000)
            # 判断是否滑动通过
            frames = self.page.frames
            if len(frames) < 2:
                logger.info("滑动通过".center(150, "-"))
                break
            else:
                logger.info("滑动失败，重新滑动".center(150, "~"))
                self.page.wait_for_timeout(3000)

    @staticmethod
    def puzzle_pic():
        """
        图片拆分重拼
        """
        img = Image.open(BIG_PNG)
        width, height = img.size
        # 横向平均切分成6张
        block_width = width // 6
        img1 = img.crop((0, 0, block_width, height))
        img2 = img.crop((block_width, 0, 2 * block_width, height))
        img3 = img.crop((2 * block_width, 0, 3 * block_width, height))
        img4 = img.crop((3 * block_width, 0, 4 * block_width, height))
        img5 = img.crop((4 * block_width, 0, 5 * block_width, height))
        img6 = img.crop((5 * block_width, 0, width, height))
        # 创建新图
        new_img = Image.new("RGB", (width, height))
        # 指定的拼装顺序
        order = [5, 1, 4, 6, 3, 2]
        for i in range(6):
            img_index = order[i]
            paste_x = i * block_width
            new_img.paste(locals()["img" + str(img_index)], (paste_x, 0))
        new_img.save(NEW_BIG)

    def download_pic(self):
        big_ele = self.frame.locator("img.captcha-verify-image")
        big_img_src = big_ele.get_attribute("src")
        request.urlretrieve(big_img_src, BIG_PNG)
        small_ele = self.frame.locator("img#captcha-verify_img_slide")
        small_img_src = small_ele.get_attribute("src")
        request.urlretrieve(small_img_src, SMALL_PNG)

    @staticmethod
    def get_distance():
        code = cv2.imread(NEW_BIG, 0)
        small_code = cv2.imread(SMALL_PNG, 0)
        code_jpg = BIG_JPG
        small_code_jpg = SMALL_JPG
        cv2.imwrite(small_code_jpg, small_code)
        cv2.imwrite(code_jpg, code)
        target = cv2.imread(code_jpg)
        target = cv2.cvtColor(target, cv2.COLOR_BGR2GRAY)
        target = abs(255 - target)
        cv2.imwrite(code_jpg, target)
        target = cv2.imread(code_jpg)
        template = cv2.imread(small_code_jpg)
        result = cv2.matchTemplate(target, template, cv2.TM_CCOEFF_NORMED)
        x, y = np.unravel_index(result.argmax(), result.shape)
        return y

    @staticmethod
    def get_track(distance):
        """
        根据偏移量和手动操作模拟计算移动轨迹
        """
        tracks = []  # 移动轨迹
        current = 0  # 当前位移
        mid = distance * 4 / 5  # 减速阈值
        t = 0.2  # 时间间隔
        v = 0  # 初始速度

        while current < distance:
            if current < mid:
                a = random.uniform(2, 5)
            else:
                a = -(random.uniform(12.5, 13.5))
            v0 = v
            v = v0 + a * t
            x = v0 * t + 1 / 2 * a * t * t
            current += x
            if 0.6 < current - distance < 1:
                x = x - 0.53
                tracks.append(round(x, 2))
            elif 1 < current - distance < 1.5:
                x = x - 1.4
                tracks.append(round(x, 2))
            elif 1.5 < current - distance < 3:
                x = x - 1.8
                tracks.append(round(x, 2))
            else:
                tracks.append(round(x, 2))
        return tracks

    def save(self):
        if self.page.url.endswith("/login"):
            raise Exception("登录失败，终止程序")
        cookie = ""
        for ck in self.page.context.cookies():
            cookie += f"{ck['name']}={ck['value']};"
        account = CookieTable.get(CookieTable.name == self.shop_name)
        account.cookie = cookie
        account.update_time = datetime.now()
        account.save()
        logger.info(f"{self.shop_name} cookie更新成功".center(100, "-"))
        logger.info(cookie)
        send_dingding("抖音罗盘定时登录", f"###抖音罗盘定时登录\n账号：**{self.shop_name}** 登陆成功")

特征隐藏：

1
2
code = """Object.defineProperties(navigator, {webdriver:{get:()=>undefined}});"""
self.page.add_init_script(code)

设置Cookie：

1
2
3
4
5
6
7
8
9
ck_dct = [
    {"name": k, "value": v, "domain": ".jd.com", "path": "/"}
    for k, v in cookies.items()
]

with sync_playwright() as pw:
    browser = pw.chromium.launch()
    context = browser.new_context()
    context.add_cookies(ck_dct)

使用浏览器缓存：

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
self.browser = self.pw.chromium.launch_persistent_context(
    # proxy={"server": "http://10.20.1.19:31280"},
    headless=False,
    args=[
        "--start-maximized",
        "--disable-features=AutomationControlled",
        "--disable-blink-features",
    ],
    ignore_default_args=["--enable-automation"],
    user_data_dir="C:\\Users\\14276\\Desktop\\btConfig",
    user_agent="Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/122.0.0.0 Safari/537.36",
    no_viewport=True,
    ignore_https_errors=True,
)
self.page = self.browser.new_page()