Немного о проблеме
Что мы знаем о капче? Капча — автоматизированный тест тьюринга, помогающий отсеивать подозрительные действия недобросовестных роботов от реальных людей. Но, к сожалению (или к счастью, смотря для кого), текстовая капча сильно устарела. Если еще 10 лет назад она была более-менее эффективным методом защиты от роботов, то сейчас ее может взломать любой желающий человек, более-менее разбирающийся в компьютере.
В данной статье-мануале я покажу, как создать собственную нейросеть по распознанию текстовых капч, имея под рукой домашний компьютер, базовые знания в python и немножко примеров капч.
Дисклеймер:
Автор данной статьи ни коим образом не призывает вас совершать какие-либо незаконные действия, используя материалы из данной статьи (да и с другими тоже не стоит) и не несет ответственности за вред, причененный ею.
Цель данной статьи — наглядно показать распространённую проблему текстовых капч.
Как всё начиналось
Было время, когда я писал один скрипт по спаму (альтернативному методу продвижения — рассылке) вк. Скрипт должен был уметь обходить капчу через сторониие сервисы (решают 1000 капч за 10-20 руб). И я подумал, почему бы не сохранять уже решенные капчи у себя на сервере? Таким образом я бесплатно накопил 150к примеров решенных капч (правда 5-10% из них были решены неправильно, но бОльшая часть датасета всё-таки валидна).
Сразу скажу, что я — человек не сильно смыслящий в машинном обучении, поэтому подправьте меня в комментариях, если что)
Выбираем нейросеть
Несмотря на то, что на дворе 2023 год, ничего более-менее адекватно работающего сходу я найти не смог. Большинство примеров предлагали перед распознаванием убрать шумы, разделить капчу ПОБУКВЕННО и перевести её в черно-белый формат (типо для облегчения работы).
Но в моём случае были некоторые проблемы:
Как убрать шумы, если при переводе в black and white линии сливаются с буквам
Как разделить капчу побуквенно, если заранее неизвестно, сколько букв она в себе содержит
В теории можно заморочиться с 1 и 2 пунктом, но терять неделю свободного времени ради такого занятия я был не готов, поэтому пошёл искать другие варианты.
Так я нашёл Tensorflow keras OCR model — CNN + RNN модель, реализующую фукнцию ошибки через CTC loss. В ней необязательно вообще что-то делать с картинками — достаточно минимально их обработать ( transpose(perm=[1, 0, 2]) ) и всё. Дальше модель разберётся сама.
Для ворчунов-профессионалов, которые скажут, что без грамотной обработки данных результат получится хуже, отвечаю: да, он получится хуже. Но он получится. И заработает. А это главное. В распознавании капч не нужна 100% вероятность, так что зачем сильно морочиться с preprocessing, если можно не париться?)
Подготавливаем данные
Убираем ошибки из датасета с помощью анализа входных данных
Для начала необходимо понять, какие вообще символы встречаются у нас в капче. Для этого я написал простенький скрипт
Статистика частоты вхождения букв в капчу
import random
import string
from pathlib import Path
data_dir_test = Path("../images/train")
symbols = {i: 0 for i in string.ascii_lowercase+string.digits}
lengths = [0]*100
for i in data_dir_test.glob("**/*"):
name = i.name.split('-')[0].split('.')[0]
for i in name:
symbols[i] += 1
lengths[len(name)] += 1
ans = []
#print char -> count
for i, item in sorted(symbols.items(), key=lambda e:-e[1]):
print(item, i)
if item > 0: ans.append(i)
#print length -> count
for i in filter(length, lambda e: e != 0):
print(ans)
Output:
67063 z
66807 s
66024 h
64136 q
61743 d
60674 v
25280 2
25208 7
25185 8
25107 x
25001 y
24993 5
24956 e
24784 a
24599 u
24492 4
24185 k
24094 n
22953 m
22318 c
18990 p
30 b
27 f
19 g
18 i
11 j
9 l
6 o
2 r
1 t
0 w
0 0
0 1
0 3
0 6
0 9
_________
LEN[3]=23
LEN[4]=61393
LEN[5]=62670
LEN[6]=14902
LEN[7]=14318
LEN[8]=4
['z', 's', 'h', 'q', 'd', 'v', '2', '7', '8', 'x', 'y', '5', 'e', 'a', 'u', '4', 'k', 'n', 'm', 'c', 'p']Эти действия сразу помогают "почистить" примеры: видно, что буквы b, f, g, i, j, l, o, r, t заведомо ошибочные и что капч, состоящих из 3 и 8 символов, не существует.
Отлично, с символами определились
max_length = 7
characters = ['z', 's', 'h', 'q', 'd', 'v', '2', '7', '8', 'x',
'y', '5', 'e', 'a', 'u', '4', 'k', 'n', 'm', 'c', 'p']Подготовка датасета
Перед дальнейшими действиями разобьем датасет с картинками на 2 папки: train и test. Папка test будет использоваться для тестирования готовой модели, train — для обучения (в процессе обучения датасет случайным образом разбивается на train и validate). Я разбил в пропорциях 1/15 ( у меня примеров много, поэтому 10к для теста будет достаточно; если у Вас датасет меньше, чем 20к, то стоит разбивать хотя бы 2/8 )
Теперь надо сделать преобработку картинки. В модели будет использоваться RGB float32 input:
img_width = 130 # ширина input'a
img_height = 50 # высота input'a
def encode_img(img):
# Preprocessing
# We will use 3 channels input (in original code were 1 channel)
if img_type == '*.jpeg':
img = tf.io.decode_jpeg(img, channels=3, dct_method='INTEGER_ACCURATE')
# img = tf.image.rgb_to_grayscale(img)
else:
img = tf.io.decode_png(img, channels=3)
# Convert to float32 in [0, 1] range
img = tf.image.convert_image_dtype(img, tf.float32)
# Resize to the desired size
img = tf.image.resize(img, [img_height, img_width])
# Transpose the image because we want the time
# dimension to correspond to the width of the image.
img = tf.transpose(img, perm=[1, 0, 2])
return img
def encode_single_sample(img_path, label):
# Read image
img = tf.io.read_file(img_path)
# Preprocessing
img = encode_img(img)
# Map the characters in label to numbers
label = char_to_num(tf.strings.unicode_split(label, input_encoding="UTF-8"))
# Return a dict as our model is expecting two inputs
return {"image": img, "label": label}
И запихать это всё в 2 датасета: один для обучения, другой — для проверки работоспособности модели (в процессе обучения)
def split_data(images: 'list', labels: 'list', train_size=0.9, shuffle=True):
size = len(images)
# 1. Make an indices array and shuffle it, if required
indices = np.arange(size)
if shuffle:
np.random.shuffle(indices)
# 2. Get the size of training samples
train_samples = int(size * train_size)
# 3. Split data into training and validation sets
x_train, y_train = images[indices[:train_samples]], labels[indices[:train_samples]]
x_valid, y_valid = images[indices[train_samples:]], labels[indices[train_samples:]]
return x_train, x_valid, y_train, y_valid
x_train, x_valid, y_train, y_valid = split_data(np.array(images), np.array(labels))
train_dataset = tf.data.Dataset.from_tensor_slices((x_train, y_train))
train_dataset = (
train_dataset.map(encode_single_sample, num_parallel_calls=tf.data.AUTOTUNE)
.batch(batch_size)
.prefetch(buffer_size=tf.data.AUTOTUNE)
)
validation_dataset = tf.data.Dataset.from_tensor_slices((x_valid, y_valid))
validation_dataset = (
validation_dataset.map(encode_single_sample, num_parallel_calls=tf.data.AUTOTUNE)
.batch(batch_size)
.prefetch(buffer_size=tf.data.AUTOTUNE)
)
Создаём и обучаем модель
Пришло время для самой модели. Она будет состоять из 2х блоков Conv2D, RNN, (output), CTC loss Layer (для ошибки)
Генерируем модель (многа сложных букф)
# Loss function
class CTCLayer(layers.Layer):
def __init__(self, name=None):
super().__init__(name=name)
self.loss_fn = keras.backend.ctc_batch_cost
def call(self, y_true, y_pred, **kwargs):
# Compute the training-time loss value and add it
# to the layer using `self.add_loss()`.
batch_len = tf.cast(tf.shape(y_true)[0], dtype="int64")
input_length = tf.cast(tf.shape(y_pred)[1], dtype="int64")
label_length = tf.cast(tf.shape(y_true)[1], dtype="int64")
input_length = input_length * tf.ones(shape=(batch_len, 1), dtype="int64")
label_length = label_length * tf.ones(shape=(batch_len, 1), dtype="int64")
loss = self.loss_fn(y_true, y_pred, input_length, label_length)
self.add_loss(loss)
# At test time, just return the computed predictions
return y_pred
def build_model():
# Inputs to the model
input_img = layers.Input(
# Lets use RGB input instead of black and white
shape=(img_width, img_height, 3), name="image", dtype="float32"
)
labels = layers.Input(name="label", shape=(None,), dtype="float32")
# First conv block
x = layers.Conv2D(
32,
(3, 3),
activation="relu",
kernel_initializer="he_normal",
padding="same",
name="Conv1",
)(input_img)
x = layers.MaxPooling2D((2, 2), name="pool1")(x)
# Second conv block
x = layers.Conv2D(
64,
(3, 3),
activation="relu",
kernel_initializer="he_normal",
padding="same",
name="Conv2",
)(x)
x = layers.MaxPooling2D((2, 2), name="pool2")(x)
# We have used two max pool with pool size and strides 2.
# Hence, downsampled feature maps are 4x smaller. The number of
# filters in the last layer is 64. Reshape accordingly before
# passing the output to the RNN part of the model
new_shape = ((img_width // 4), (img_height // 4) * 64)
x = layers.Reshape(target_shape=new_shape, name="reshape")(x)
x = layers.Dense(64, activation="relu", name="dense1")(x)
x = layers.Dropout(0.2)(x)
# RNNs
x = layers.Bidirectional(layers.LSTM(128, return_sequences=True, dropout=0.25))(x)
x = layers.Bidirectional(layers.LSTM(64, return_sequences=True, dropout=0.25))(x)
# Output layer
x = layers.Dense(
len(char_to_num.get_vocabulary()) + 1, activation="softmax", name="dense2"
)(x)
# Add CTC layer for calculating CTC loss at each step
output = CTCLayer(name="ctc_loss")(labels, x)
# Define the model
model = keras.models.Model(
inputs=[input_img, labels], outputs=output, name="ocr_model_v1"
)
# Optimizer
opt = keras.optimizers.Adam()
# Compile the model and return
model.compile(optimizer=opt)
return model
model = build_model()Добавляем защиту от переобучения
early_stopping = keras.callbacks.EarlyStopping(
monitor="val_loss", patience=early_stopping_patience, restore_best_weights=True
)..... Запускаем! .....
history = model.fit(
train_dataset,
validation_data=validation_dataset,
epochs=100,
callbacks=[early_stopping],
)
Обучалась модель у меня на процессоре (AMD RYZEN 5 3600) ~ 10 часов, на видеокарте (GTX 1070 TI) ~ 3 часа.
Если вдруг захотите обучать на видеокарте, то будьте готовы потратить 2-3 часа на уставку CUDA, tensorflow-GPU, и кучу другого гемороя (например, conda, — без неё на windows с обучением всё очень грустно)
Тестируем
Для тестирования нужно прогнать нашу модель через датасет, заранее оставленный в папочке test.
Код тестов
Подгружаем нашу модель
model: Functional = keras.models.load_model(MODEL_FNAME)Загружаем тестовый датасет
images = list(map(str, [i for i in data_dir_test.glob(img_type)]))
images = sorted(images, key=lambda *h: random.random())
labels = [
img.split(os.path.sep)[-1].split('-')[0].split('.jpeg')[0].ljust(max_length)[:max_length]
for img in images
]
test_dataset = tf.data.Dataset.from_tensor_slices((numpy.array(images), numpy.array(labels)))
test_dataset = (
test_dataset.map(encode_single_sample, num_parallel_calls=tf.data.AUTOTUNE)
.batch(batch_size)
.prefetch(buffer_size=tf.data.AUTOTUNE)
)Визуализация (по желанию):
def visualize():
_, ax = plt.subplots(4, 4, figsize=(15, 5))
for i in range(len(pred_texts)):
img = (image[i, :, :, :] * 255).numpy().transpose([1, 0, 2]).astype(np.uint8)
title = f" {pred_texts[i][0]}{pred_texts[i][1]:.3%} {'ok' if success else 'wrong'}"
ax[i // 4, i % 4].imshow(img)
ax[i // 4, i % 4].set_title(title)
ax[i // 4, i % 4].axis("off")
plt.show()И, соответственно, тесты:
good = 0
total = 0
for i in test_dataset.take(1000):
label, image = i['label'], i['image']
preds = model.call(image)
pred_texts = decode_batch_predictions(preds)[0]
orig_texts = [tf.strings.reduce_join(num_to_char(l)).numpy().decode("utf-8").replace('[UNK]', '') for l in label]
if need_visualization and total == 0:
visualize()
for i in range(len(pred_texts)):
success = pred_texts[i] == orig_texts[i]
if success:
good += 1
total += 1
print(f"Progress {total/len(images):.2%}[{total}/{len(images)}]...", end='\r')
print(f"Success: {(good / total):.2%}\n")
У модель выдавала правильный ответ с вероятностью 94.05%.
Ура?
В целом, можно сказать, что модель готова. НО! Тут, как всегда, не без сюрпризов. А сюрпризы в данном случае: 2ГБ библиотеки Tensorflow, 10-15 секунд на её import, долгое время работы и т.п.
В общем, после длительных поисков, было решено переходить на формат onnx. А что, стильно, модно, молодёжно, подумал я. Но, естественно, с 1-го раза ничего не заработало
Конвертируем модель в ONNX
Сначала я попытался использовать keras2onnx, но он у меня не заработал. Погуглив ошибку, я пришёл к выводу, что во всём виновата windows (о великие линукса), перешел на виртуалку, попробовал на ней и.... Опять ничего не заработало (после 4х часов попыток заставить что-либо двигаться).
Далее на очередь пришёл tf2onnx. Но заработал он у меня только с 3-го раза: в первый раз, после установки, оказалось, что python3.10 не поддерживается, затем, на 2й раз — что я обучал модель на слишком новой версии tensorflow, которую tf2onnx пока не поддерживает (вот зараза).
В результате у меня-таки получилось всё провернуть со следующими версиями:
python3.9
tensorflow==2.9.1
tf2onnx==1.11.1
onnxruntime==1.9.0
onnxconverter-common==1.10.0Ну и сам скрипт конвертации:
import pathlib, tf2onnx.convert, sys
from tools import MODEL_FNAME
OUTPUT_ONNX = "../out.model.onnx"
dir = pathlib.Path(__file__).parent
sys.argv.append("--saved-model")
sys.argv.append(str(dir / MODEL_FNAME))
sys.argv.append("--output")
sys.argv.append(str(dir / OUTPUT_ONNX))
tf2onnx.convert.main()UPDATE: вроде как сейчас уже есть поддержка python3.10, но проверять я это не хочу)
Переписываем код с tensorflow на numpy
Теперь, после конвертации в ONNX, возникла необходимость переписать препроцессинг картинок с tensorflow на великий numpy (весит меньше, работает шустрее).
В случае препроцессинга это не составило особого труда:
def decode_img(data_bytes: bytes):
# same actions, as for tensorflow
img = cv2.imdecode(np.asarray(bytearray(data_bytes), dtype=np.uint8), -1)
img: "np.ndarray" = img.astype(np.float32) / 255.
if img.shape != (img_height, img_width, 3):
cv2.resize(img, (img_width, img_height))
img = img.transpose([1, 0, 2])
# Creating tensor ( adding 4d dimension )
img = np.array([img])
return imgНемного о INTEGER_ACCURATE
Тестируем и... Процент верных ответов снижается на 10%. Это стало очень неприятно, и, как оказалось, всё дело в методе преобрабоки картинки.
Существует много методов декодирования JPEG картинок, и, естественно, tensorflow и numpy по умолчанию используют разные. Чтобы они были наиболее cхожи, стоит указать параметр dct_method='INTEGER_ACCURATE'.
img = tf.io.decode_jpeg(img, channels=3, dct_method='INTEGER_ACCURATE')
Немного официальной документации:
dct_method - String specifying a hint about the algorithm used for decompression. Defaults to "" which maps to a system-specific default. Currently valid values are
["INTEGER_FAST", "INTEGER_ACCURATE"]. The hint may be ignored (e.g., the internal jpeg library changes to a version that does not have that specific option.)
Не повторяйте моих ошибок :)
А вот в случае декодирования результа пришлось вникнуть в Beam Search Decoding for ctc alorithm. Если вкратце, то на выход мы получаем 2-мерный массив длиной в 32 элемента, где каждый элемент — массив вероятностей на этом месте каждой буквы.
Код декодирования результата:
def get_result(pred):
"""CTC decoder of the output tensor
https://distill.pub/2017/ctc/
https://en.wikipedia.org/wiki/Connectionist_temporal_classification
:return string, float
"""
accuracy = 1
last = None
ans = []
# pred - 3d tensor, we need 2d array - first element
for item in pred[0]:
# get index of element with max accuracy
char_ind = item.argmax()
# ignore duplicates and special characters
if char_ind != last and char_ind != 0 and char_ind != len(characters)+1:
# this element is a character - append it to answer
ans.append(characters[char_ind - 1])
# Get accuracy for current character and
# multiply global accuracy by it
accuracy *= item[char_ind]
last = char_ind
answ = "".join(ans)[:max_length]
return answ, accuracy
Тестируем ONNX модель
Настало время финальной проверки: необходимо удостовериться, что мы всё сделали правильно. То есть результат работы ONNX модели не должен сильно отличаться от оригинальной. Также необходимо прогнать тесты на "реальных" капчах (генерируемых сервером) и посмотреть, насколько наша модель с ними справляется (а то вдруг всё было зря
Интересные статьи
Интересные статьи
