源码：bit-fist-crawler

docker build -t bit-fist-crawler:0.1 .
docker tag bit-fist-crawler:0.1 harbor.test.com/cztl/bit-fist-crawler:0.1
dokcer push harbor.test.com/cztl/bit-fist-crawler:0.1

version: '2'
services:
  bit-fist-crawler:
    image: harbor.test.com/cztl/bit-fist-crawler:0.1
    container_name: bit-fist-crawler
    volumes:
      - "/data/crawler/config:/home/project/config"
      - "/data/crawler/log:/home/project/logs"
    restart: always
    ports:
      - 11423:11423
    mem_limit: 512m
    networks:
      - chuanzangNetwork
    logging:
      options:
        max-size: "10m"
        max-file: "10" 
networks:
  chuanzangNetwork: 
    external: true

sudo mkdir /data/crawler/config

sudo docker-compose -f docker-compose-crawler.yml up -d

pip install opencv-python
pip install numpy

import cv2
import numpy as np
# 打开图片，image_path 是验证码下载的文件路径：src/img/cztl-web-captcha.jpeg，每次下载重写这个图片
img = cv2.imread(image_path)
# 灰度处理 或者 分离通道，这两个都可以得到灰度图，可根据世界效果选；分离通道返回值依次是 b,g,r，可根据世界效果挑选，本例使用的是 g
# img_gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
b, img_gray, r = cv2.split(img)

# 二值化,大于阈值 80 的都转化成 255（黑），否则是 0（白）
ret, img_inv = cv2.threshold(img_gray, 80, 255, cv2.THRESH_BINARY_INV)

# 透视拉伸
img_dst = img_perspective(img_inv)

def img_perspective(img):
    """
    图片透视拉伸
    :param img: 源图片
    :return: 拉伸后的图片
    """
    pos1 = np.float32([[0, 0], [135, 0], [30, 60], [160, 60]])
    pos2 = np.float32([[25, 0], [160, 0], [30, 60], [160, 60]])
    mm = cv2.getPerspectiveTransform(pos1, pos2)
    return cv2.warpPerspective(img, mm, (160, 60))

# 查找轮廓
contours, hierarchy = cv2.findContours(img_dst, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)

# 画出矩形边界,x、y边框起始点的坐标，w、h为宽高
cv2.rectangle(img_dst, (x, y), (x + w, y + h), (255, 255, 255), thickness=1)

box = np.int0([[x, y], [x + w, y], [x + w, y + h], [x, y + h]])
cv2.drawContours(img_dst, [box], 0, (0, 0, 255), 2)
roi = img_dst[box[0][1]:box[3][1], box[0][0]:box[1][0]]
roi_std = cv2.resize(roi, (30, 30))  # 将字符图片统一调整为30x30的图片大小

def get_rect_contours(img_dst):
    """
    获取矩形边界列表，按照x坐标从左向右排序
    :param img_dst:
    :return:
    """
    contours, hierarchy = cv2.findContours(img_dst, cv2.RETR_EXTERNAL, cv2.CHAIN_APPROX_SIMPLE)
    rects = []
    for contour in contours:
        x, y, w, h = cv2.boundingRect(contour)
        # 排除问号
        if w < 10 and x > 100 or x > 110 and h > 10:
            continue
        # 排除等号
        if 2 < w / h < 6 and x > 90:
            continue
        # 两个字符粘在一起，通常是第二个数字和运算符粘连 或第二个数字和等号粘连 或第一个数字和运算符粘连
        if 28 < w < 50:
            w = 20
            if x < 80:
                rects.append((x + 20, y, w, h))
        # 三个字符粘连，出现在运算符、第二个数字、等号之间 或第一个数字、运算符、第二个数字之间
        if w > 50:
            rects.append((x, y, 20, h))
            rects.append((x + 20, y, 20, h))
            if x < 50:
                rects.append((x + 40, y, w - 40, h))
            continue
        # '*' 被划分太细的情况放弃
        if w < 10 and h < 10:
            continue
        # 矩形切图
        rects.append((x, y, w, h))
    rects.sort(key=None, reverse=False)
    return rects

def mark_img(roi, timestamp):
    """
    人工标注切图
    :param roi:
    :param timestamp:
    :return:
    """
    print("PS:对每张切图输入对应的字符（用于标记切图），回车跳过当前切图，点击关闭退出人工标记切图")
    cv2.imshow("image", roi)
    key = cv2.waitKey(0)
    if key == 27: # 点击关闭，退出
        sys.exit()
    if key == 13: # 回车跳过当前
        return
    char = chr(key)
    print("您输入的key是：", char)
    filename = "%s/%s_%s.jpg" % (img_lib_path, timestamp, char)
    cv2.imwrite(filename, roi)

def train_machine():
    """
    机器训练
    :return: id_label_map, model
    """
    # TODO 后续可尝试将返回值缓存和持久化
    filenames = os.listdir(img_lib_path)
    samples = np.empty((0, 900))
    labels = []
    for filename in filenames:
        filepath = "%s/%s" % (img_lib_path, filename)
        label = filename.split(".")[0].split("_")[-1]
        labels.append(label)
        im = cv2.imread(filepath, cv2.IMREAD_GRAYSCALE)
        roi_std = cv2.resize(im, (30, 30))
        sample = roi_std.reshape((1, 900)).astype(np.float32)
        samples = np.append(samples, sample, 0)
    samples = samples.astype(np.float32)
    unique_labels = list(set(labels))
    unique_ids = list(range(len(unique_labels)))
    label_id_map = dict(zip(unique_labels, unique_ids))
    id_label_map = dict(zip(unique_ids, unique_labels))
    label_ids = list(map(lambda x: label_id_map[x], labels))
    label_ids = np.array(label_ids).reshape((-1, 1)).astype(np.float32)
    model = cv2.ml.KNearest_create()
    model.train(samples, cv2.ml.ROW_SAMPLE, label_ids)
    return id_label_map, model

id_label_map, model = train_machine()
for image in images:
    sample = image.reshape((1, 900)).astype(np.float32)
    # 找出最相似的图片
    ret, results, neighbours, distances = model.findNearest(sample, k=3)
    # 找出该图片对应的 label_ids 中的值
    label_id = int(results[0, 0])
    # 找出对应的 label，这就是识别结果
    label = id_label_map[label_id]
    cv2.imshow("image", image)
    key = cv2.waitKey(0)
    if key == 27:
        sys.exit()
    if key == 13:
        return
    correct_char = chr(key)
    print("您输入的key是：%s,机器识别的key是：%s" % (correct_char, label))

import requests
import http.cookiejar

# 设置一个cookie处理器，它负责从服务器下载cookie到本地，并且在发送请求时带上本地的cookie
    cj = http.cookiejar.CookieJar()
    cookie_support = request.HTTPCookieProcessor(cj)
    opener = request.build_opener(cookie_support, request.HTTPHandler)

    request.install_opener(opener)
    raw_data = {"figure": figure, "username": username, "password": password, "imgId": imgId, "code": code}
    post_data = parse.urlencode(raw_data).encode('utf-8')
    cookie = input("输入cookie：")
    headers = {"Cookie": cookie}
    req = request.Request(url=login_url, data=post_data, method='POST')
    # 打开登录主页面（目的是从页面下载cookie，这样我们在再送post数据时就有cookie了，否则发送不成功）
    response = request.urlopen(req)

import requests
session = requests.session()
wrong_title = "铁路工程管理平台--登录"


def request(url, method=request_method.GET, headers=None, data=None):
    """
    请求数据
    :param url:
    :param method:
    :param headers:
    :param data:
    :return:
    """
    response = session.request(method, url, headers=headers, data=data)
    return response

def request_retry(url, method=request_method.GET, headers=None, data=None):
    """
    请求数据
    :param url:
    :param method:
    :param headers:
    :param data:
    :return:
    """
    response = session.request(method, url, headers=headers, data=data)
    if response.headers.get("Content-Type") == "text/html;charset=UTF-8":
        title = get_tile(response.text)
        if title == wrong_title:
            login()
            response = session.request(method, url, headers=headers, data=data)
    return response

from apscheduler.schedulers.blocking import BlockingScheduler

def main():
    # 登录，收集数据
    login.login()
    data_collector()
    # 定时收集数据
    task = BlockingScheduler()
    task.add_job(data_collector, "cron", hour="7-23", minute="*/30")
    task.start()


def data_collector():
    """
    数据收集
    :return:
    """
    logging.info("开始定时任务")
    # 同步进度信息
    progress_collector.get_construct_points()
    logging.info("定时任务完成")

def get_retry_std(url, headers=None, data=None):
    """
    get方法请求数据,返回json，只适合 responses 是标准输出的情况
    :param url:
    :param headers:
    :param data:
    :return:
    """
    response = get_retry(url, headers=headers, data=data)
    if response.headers.get("Content-Type") != "text/plain;charset=UTF-8":
        log.error("Content-Type 必须是text/plain;charset=UTF-8")
    if response.status_code != 200:
        log.error("请求异常，状态码：", response.status_code)
    response_body = json.loads(response.text)
    return response_body["result"]

from lxml import etree
def get_tile(text):
    """
    根据文本获取html页面title
    :param text:
    :return:
    """
    tree = etree.HTML(text)
    return tree.xpath("//title")[0].text