进阶｜借助 PyTorch 完成复杂验证码的识别

早在新生报到时，我就萌生了识别验证码并模拟登录的想法，无奈当时技术有限，最终未能达到目标。去年用 numpy 手写打卡系统验证码识别器后，自动脚本一直运行得还算稳定，这也给了我信心，让我继续挑战迎新系统的验证码：

前期工作

特征分析

乍一看好像挺吓人，随机字号、随机倾斜、随机模糊（可能这模糊是 jpg 算法搞的）,甚至还有一条横线，但当我们叠图之后，事情开始变得逐渐明朗起来：


import os
import cv2
import numpy as np
import matplotlib.pyplot as plt

overlay = None

files = os.listdir('../data/raw_image')
for file in files:
    image = cv2.imread(f'../data/raw_image/{file}')
    image = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
    if overlay is None:
        overlay = np.zeros(image.shape)
    overlay += image

if overlay is not None:
    overlay = (overlay * 255 / np.max(overlay)).astype(np.uint8)
    print(overlay.shape)
    plt.subplot(2, 1, 1)
    plt.imshow(overlay)
    plt.subplot(2, 1, 2)
    plt.plot(np.sum(255 - overlay, axis=0))
    plt.show()

左侧的随机短横线几乎不会与文字发生接触，那么我们直接将其裁掉即可，同样地，在右侧裁剪掉一块空白区域，以便后续识别。

同时我在下方绘制了一张折线图，为原图反色后沿纵向将各个像素值累加后得到的结果。不难注意到由于背景渐变色的存在，折线整体有一个从左至右由高到低的趋势，修正后，得到了这样的效果：

下面的步骤，可能看起来会比较复杂，实际上原理却很简单：使用一根横向的「扫描线」自下而上扫描整个折线图，考虑所有被折线图和扫描线围成的部分，不妨将其称为「域」，当底面长度不小于 10 的「域」的数量为 4（字符数）时，认为这是一个较好的切分：

import os
import cv2
import numpy as np
import matplotlib.pyplot as plt


files = os.listdir('../data/raw_image')
for file in files:
    image = cv2.imread(f'../data/raw_image/{file}')
    image = cv2.cvtColor(image, cv2.COLOR_RGB2GRAY)
    sumvirt = np.sum(255 - image, axis=0)
    line = 3088 - 12.04 * np.arange(0, 200)
    sumvirt = (sumvirt.astype(np.float64) - line)[52:158]

    for x in np.arange(0, sumvirt.max()):
        norm = np.where(sumvirt > x, 1, 0)
        diff = np.append(norm, 0) - np.append(0, norm)

        last, results = 0, []
        for i in range(0, len(diff)):
            if diff[i] == 1:
                last = i
            elif diff[i] == -1 and i - last >= 10:
                results.append((last, i))

        if len(results) == 4:
            plt.subplot(3, 1, 1)
            plt.title(f'X = {x}')
            plt.imshow(image[:, 52:158])

            plt.subplot(3, 1, 2)
            plt.plot(sumvirt)
            plt.plot(np.full(sumvirt.shape, x))

            print(results)
            for i in range(4):
                plt.subplot(3, 4, 9 + i)
                plt.imshow(image[:, 52:158][:, range(*results[i])])

            plt.show()
            break

简单二值化一下，贴到白色背景图上，这里放一张和原图的对比：

爬取数据集

接下来就是模拟登录获取训练集了，由于网站会将帐号密码用 RSA 算法加密，再提交到服务器。虽然也可以用 Python 来模拟登录，但逆向 js 代码并找出它的操作逻辑未免过于麻烦，于是这里直接使用 puppeteer + express 计算加密：

'use strict'
const path = require('path')
const express = require('express')
const puppeteer = require('puppeteer')

!(async () => {
    const browser = await puppeteer.launch({
        args: [
            `--user-data-dirs=${path.join(__dirname, '.chromium')}`
        ]
    })

    let page = await browser.newPage()
    await page.goto('http://yxxt.hnu.edu.cn/yxxt/web/xsLogin/login.zf')
    await page.waitForFunction(() => RSAKey ?? false)

    let app = express()
    app.use(require('body-parser').urlencoded())
    app.post('/encrypt', async (req, res) => {
        console.info('New encrypt request!')
        let enc = await page.evaluate(({src, modulus, exponent}) => {
            let rsaKey = new RSAKey()
            rsaKey.setPublic(b64tohex(modulus), b64tohex(exponent))
            return hex2b64(rsaKey.encrypt(src))
        }, req.body)
        res.send(enc)
    })
    app.listen(34567, () => {
        console.info('Server started, listening...')
    })
})()

与之对应的 Python 代码，用于验证 tesseract 的识别结果，这里只需要确认验证码是否正确，所以帐号和密码都可以乱写（拼音缩写变量名全是从网站的 js 抄的，如果看得难受还请放过我 இωஇ）：

import json
import os
import pickle
import re
import subprocess as sp
from base64 import b64encode
from time import time

import cv2
import httpx
import numpy as np
from loguru import logger

from spider import get_uuid
from splitter import split


def timestamp():
    return f'{time() * 1000:.0f}'


def encrypt(src, modulus, exponent):
    resp = httpx.post('http://localhost:34567/encrypt', data={
        'src': src,
        'modulus': modulus,
        'exponent': exponent
    })
    return resp.text


def login(client, zhlx, zh, mm, yzm, pubkey):
    modulus, exponent = pubkey['modulus'], pubkey['exponent']

    result = client.post('http://yxxt.hnu.edu.cn/yxxt/web/xsLogin/checkLogin.zf', data={
        'dldata': b64encode(json.dumps({
            'zhlx': encrypt(zhlx, modulus, exponent),
            'zh': encrypt(zh, modulus, exponent),
            'mm': encrypt(mm, modulus, exponent),
            'yzm': encrypt(yzm, modulus, exponent)
        }).encode()).decode()
    })
    return result.json()


def ocr(image):
    proc = sp.Popen(['tesseract', '-', '-', '--psm', '7'], stdin=sp.PIPE, stdout=sp.PIPE, stderr=sp.PIPE)
    proc.stdin.write(image)
    proc.stdin.close()
    result = re.sub('[^a-z0-9]', '', proc.stdout.read().decode())
    if len(result) != 4:
        return None
    return result


def check():
    client = httpx.Client(headers={
        'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) '
                      'Chrome/96.0.4664.110 Safari/537.36'
    })

    client.get('http://yxxt.hnu.edu.cn/yxxt/web/xsLogin/login.zf')
    pubkey = client.get(f'http://yxxt.hnu.edu.cn/yxxt/xtgl/login/getPublicKey.zf?time={timestamp()}').json()
    image = client.get(f'http://yxxt.hnu.edu.cn/yxxt/kaptcha?time={timestamp()}')
    captcha = cv2.imdecode(np.frombuffer(image.content, np.uint8), cv2.IMREAD_COLOR)

    chars = split(captcha)
    if not chars:
        return None

    stack = np.hstack(chars)
    code = ocr(cv2.imencode('.png', stack)[1].tobytes())
    if not code:
        return None

    # MMCW | YZMCW
    verify = login(client, 'xsxh', 'u$ern@me', 'p@$$w0rd', code, pubkey)
    if verify['status'] != 'MMCW':  # 密码错误
        return None

    for i, char in enumerate(chars):
        with open(f'data/marked/{code[i]}/{get_uuid()}', 'wb') as fp:
            fp.write(pickle.dumps(char))

    return code


def main():
    logger.info(f'cwd: {os.getcwd()}')
    os.chdir('/home/mufanc/JetBrains/PycharmProjects/PyTorch Demo/HNU Captcha Hard/')

    failed = 0
    for i in range(10000 * 50):
        try:
            if code := check():
                logger.info(f'[{code}] Verify success after {failed} failed attempts')
                failed = 0
            else:
                failed += 1
        except Exception as err:
            logger.error(repr(err))


if __name__ == '__main__':
    main()

意外发现

对标记好的验证码简单做个统计，会发现某些字母似乎并不可能出现：

把可能出现的字符集 2345678ABCDEFGMNPWXY 拿去搜索，能得知这种验证码使用 Google 的开源项目 Kaptcha 生成，其实这一点在获取验证码的 URL 种就有体现，但我当时并没有留意，这个项目最后一次更新是在 2010 年 12 月，目前已经 Archive ~~（不愧是我校作风）~~。

那么或许可以参考这个项目，在本地搭建一个服务器，生成样式差不多的验证码，再对学校的验证码进行迁移学习，这样只需很少的训练样本就能达到不错的训练效果🤔

训练网络

我们切分出来的单个字符小图是 50×50 的，经过两次卷积和池化后送入全连接层，最终得到 20 个输出，代表该图片为相应字符的「概率」

class Character(nn.Module):
    def __init__(self):
        super().__init__()
        self.conv1 = nn.Conv2d(1, 10, (5, 5))  # 5 x 5 卷积层 (50 x 50 -> 46 x 46)
        # 2 x 2 最大池化层 (46 x 46 -> 23 x 23)
        self.conv2 = nn.Conv2d(10, 20, (4, 4))  # 5 x 5 卷积层 (23 x 23 -> 20 x 20)
        # 2 x 2 平均池化层 (20 x 20 -> 10 x 10)
        self.fc1 = nn.Linear(2000, 800)  # 全连接层 (2000 -> 800)
        self.fc2 = nn.Linear(800, 20)  # 全连接层 (800 -> 20)

    def forward(self, x):
        input_size = x.size(0)

        # 卷积层 1
        x = F.relu(self.conv1(x))
        x = F.max_pool2d(x, 2, 2)  # 最大池化

        # 卷积层 2
        x = F.relu(self.conv2(x))
        x = F.avg_pool2d(x, 2, 2)  # 平均池化
        x = x.view(input_size, -1)  # 拉伸

        # 全连接层 1
        x = F.relu(self.fc1(x))

        # 全连接层 2
        x = self.fc2(x)

        return F.log_softmax(x, dim=1)

然后开始训练网络，这里我将十分之一的数据用作测试集，其余数据为训练集（不要问我为什么用 CPU 训练，Linux 平台的 ROCm 真的太难装了 555）：

# 加载必要的库
import torch
import torch.nn as nn
import torch.nn.functional as F
import torch.optim as optim
from torch.utils.data import DataLoader

from dataset import MyDataset
from model import Character

# 定义超参数
BATCH_SIZE = 32  # 每次处理的数据数量
DEVICE = torch.device('cpu')  # 设备
EPOCH = 10  # 数据集循环运行的轮次


# 加载数据
train_set = MyDataset('train')
test_set = MyDataset('test')

train_loader = DataLoader(train_set, batch_size=10, shuffle=True, num_workers=8)
test_loader = DataLoader(test_set, batch_size=10, shuffle=True, num_workers=8)

# 定义优化器
model = Character().to(DEVICE)
optimizer = optim.Adam(model.parameters())


# 定义训练方法
# noinspection PyShadowingNames
def train_model(model, device, train_loader, optimizer, epoch):
    # 模型训练
    model.train()
    for batch_index, (data, label) in enumerate(train_loader):
        # 部署到 Device
        data, label = data.to(device), label.to(device)
        # 梯度初始化为 0
        optimizer.zero_grad()
        # 获取训练结果
        output = model(data)
        # 计算损失
        loss = F.cross_entropy(output, label)
        # 找到概率值最大的下标
        pred = output.argmax(dim=1)
        # 反向传播
        loss.backward()
        # 参数优化
        optimizer.step()
        if batch_index % 1000 == 0:
            print(f'Train Epoch: {epoch} \t Loss: {loss.item():.6f}')
            # print(output)


# 定义测试方法
# noinspection PyShadowingNames
def test_model(model, device, test_loader):
    # 模型验证
    model.eval()
    # 计算正确率
    correct = .0
    # 测试损失
    test_loss = .0
    with torch.no_grad():  # 不会计算梯度，也不会进行反向传播
        for data, label in test_loader:
            # 部署到 Device
            data, label = data.to(device), label.to(device)
            # 测试数据
            output = model(data)
            # 计算测试损失
            test_loss += F.cross_entropy(output, label).item()
            # 找到概率值最大的下标
            pred = output.argmax(dim=1)
            # 累计正确率
            correct += pred.eq(label.view_as(pred)).sum().item()
        test_loss /= len(test_loader.dataset)
        print(f'Test -- Average Loss: {test_loss:.4f} \t Accuracy: {100*correct/len(test_loader.dataset):.3f}%')


# 开始训练
if __name__ == '__main__':
    for epoch in range(1, EPOCH + 1):
        train_model(model, DEVICE, train_loader, optimizer, epoch)
        test_model(model, DEVICE, test_loader)

    torch.save(model.state_dict(), '../model.pck')