源碼：bit-fist-crawler

初學(xué) python，還有很多問(wèn)題待優(yōu)化，歡迎和我一樣的小白一起研究，也歡迎大佬路過(guò)指點(diǎn)！

1. 介紹

用于從爬取某系統(tǒng)數(shù)據(jù)，同步到數(shù)據(jù)庫(kù)

• 數(shù)學(xué)驗(yàn)證碼識(shí)別與自動(dòng)計(jì)算并登錄
• 會(huì)話(huà)保持，驗(yàn)證碼識(shí)別錯(cuò)誤、session 過(guò)期，自動(dòng)重試
• 架構(gòu) python、docker、mysql

2. 安裝教程

2.1. docker 打包、推送

確保本地 docker 為啟動(dòng)狀態(tài)

打開(kāi) cmd 或 powershell，cd 到 Dockerfile 同級(jí)目錄下，執(zhí)行 docker 打包：

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

harbor.test.com 替換成自己的 docker 倉(cāng)庫(kù)

2.2. 服務(wù)器配置信息

docker-compose 配置文件 docker-compose-crawler.yml

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

其中 "/data/crawler/config","/data/crawler/log" 分別用于掛載配置文件路徑和日志文件路徑。

在服務(wù)器上創(chuàng)建文件夾，用于存放配置文件覆蓋代碼中的 /config

sudo mkdir /data/crawler/config

將 config.txt 上傳到該文件夾

2.3. 啟動(dòng)

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

啟動(dòng)后，日志掛載在 /data/crawler/log

3. 實(shí)現(xiàn)過(guò)程

3.0. 背景

開(kāi)發(fā)項(xiàng)目時(shí)，遇到了需要用爬蟲(chóng)爬數(shù)據(jù)的需求。目標(biāo)網(wǎng)站是個(gè)需要登錄驗(yàn)證的網(wǎng)站，不能直接獲取接口或頁(yè)面。用戶(hù)名、賬號(hào)對(duì)方已經(jīng)給到我們，剩下的是就是通過(guò)驗(yàn)證，然后獲取數(shù)據(jù)。
用 java 應(yīng)該也是可以實(shí)現(xiàn)的，但是想借這個(gè)契機(jī)學(xué)習(xí)一下 python，所以決定用 python 實(shí)現(xiàn)。于是乎在菜鳥(niǎo)學(xué)了幾天 Python 3 教程，真是入門(mén)級(jí)的好去處^-^。
看了好多文章，最后實(shí)現(xiàn)了這個(gè)從 0 到 1 的過(guò)程，很開(kāi)心。

初學(xué) python，代碼并不是很規(guī)范，文件夾也是隨便建的。還有很多問(wèn)題待優(yōu)化，歡迎和我一樣的小白一起研究，也歡迎大佬路過(guò)指點(diǎn)！

源碼地址：bit-fist-crawler

3.1. 驗(yàn)證碼識(shí)別

識(shí)別驗(yàn)證碼就是這個(gè)項(xiàng)目的核心了，第一次接觸，還是挺費(fèi)勁的。這個(gè)功能是使用 opencv-python 這個(gè)庫(kù)來(lái)做圖形處理的。
代碼主要在 src/utils/captcha_util.py 中。驗(yàn)證碼各式各樣，每種驗(yàn)證碼的圖形處理都會(huì)有不同的處理過(guò)程，所以這部分代碼只能參考思路。

我的驗(yàn)證碼是這樣的：

驗(yàn)證碼

識(shí)別出 5+1，然后計(jì)算結(jié)果 6 就是最終結(jié)果。好在這個(gè)驗(yàn)證碼都是個(gè)位運(yùn)算，而且后面的 “=？”固定，不需要識(shí)別，所以最終只要識(shí)別前三個(gè)字符就行。

這部分用到這兩個(gè)庫(kù)：

pip install opencv-python
pip install numpy

opencv-python 處理圖片的方式，其實(shí)是將圖片數(shù)字化。眾所周知，圖片是由一個(gè)個(gè)像素組成，每個(gè)像素又是由 r，g，b 三種顏色組成，每種顏色從 0-255 代表其亮度。這樣一來(lái)就可以用一個(gè)三維數(shù)組來(lái)表示一張圖片，前面兩維表示坐標(biāo)，第三維表示色道值。

我的驗(yàn)證碼是 60*160 的圖片，所以讀取之后得到的是 60*160*3 的數(shù)組.

識(shí)別出字符，需要排除干擾因素，把最需要并且最簡(jiǎn)單部分交給程序處理。

對(duì)于這個(gè)驗(yàn)證碼，我的處理方式是：

1. 灰度化：排除顏色信息

import cv2
import numpy as np
# 打開(kāi)圖片，image_path 是驗(yàn)證碼下載的文件路徑：src/img/cztl-web-captcha.jpeg，每次下載重寫(xiě)這個(gè)圖片
img = cv2.imread(image_path)
# 灰度處理 或者 分離通道，這兩個(gè)都可以得到灰度圖，可根據(jù)世界效果選；分離通道返回值依次是 b,g,r，可根據(jù)世界效果挑選，本例使用的是 g
# img_gray = cv2.cvtColor(img, cv2.COLOR_BGR2GRAY)
b, img_gray, r = cv2.split(img)

灰度化

灰度化過(guò)后，圖片中只剩下黑白灰，可以理解成只有一個(gè)色道，這樣三維數(shù)組就變成一個(gè) 60*160 的二維數(shù)組了。

2. 二值化：使圖片中只有黑白兩種顏色
   因?yàn)楝F(xiàn)在圖片有很多灰色地帶，會(huì)比較模糊，會(huì)為識(shí)別圖片造成較大影響，為排除這個(gè)因素，需要進(jìn)行二值化，使其邊界清晰。

# 二值化,大于閾值 80 的都轉(zhuǎn)化成 255（白），否則是 0（黑）
ret, img_inv = cv2.threshold(img_gray, 80, 255, cv2.THRESH_BINARY_INV)

二值化

二值化之后的二維數(shù)組中，只有兩種值，0 和 255.

3. 透視拉伸：這該死的驗(yàn)證碼是傾斜的，所以要做一下拉伸，盡量使字符擺正（但是這個(gè)驗(yàn)證碼，每張驗(yàn)證碼的傾斜程度都不同，這是比較蛋疼的部分，也是我的程序會(huì)誤判的元兇之一，也沒(méi)找到很好的解決辦法）

# 透視拉伸
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))

拉伸點(diǎn)位

pos1 是當(dāng)前圖片中的四個(gè)點(diǎn)為像素坐標(biāo)，按照左上，右上，左下，右下順序排列。pos2 是想要調(diào)整到的目標(biāo)點(diǎn)位。

透視拉伸

4. 切割圖片：把驗(yàn)證碼的前三個(gè)字符切割成圖片，由于會(huì)發(fā)生字符粘連的情況，所以需要額外進(jìn)行判斷和切割

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

# 畫(huà)出矩形邊界,x、y邊框起始點(diǎn)的坐標(biāo)，w、h為寬高
cv2.rectangle(img_dst, (x, y), (x + w, y + h), (255, 255, 255), thickness=1)

矩形邊框

畫(huà)出矩形邊界，已辦用于在調(diào)整的時(shí)候用，可以看到邊框，方便修改。二標(biāo)注的時(shí)候需要把字符切成一張張小圖片

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))  # 將字符圖片統(tǒng)一調(diào)整為30x30的圖片大小

但是有時(shí)候會(huì)出現(xiàn)字符粘連的情況，像下圖就是前兩個(gè)字符粘在一起，沒(méi)法按照輪廓切割。

字符粘連

對(duì)于字符粘連的問(wèn)題，我的方式簡(jiǎn)單粗暴，按照寬度平均分割

def get_rect_contours(img_dst):
    """
    獲取矩形邊界列表，按照x坐標(biāo)從左向右排序
    :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)
        # 排除問(wèn)號(hào)
        if w < 10 and x > 100 or x > 110 and h > 10:
            continue
        # 排除等號(hào)
        if 2 < w / h < 6 and x > 90:
            continue
        # 兩個(gè)字符粘在一起，通常是第二個(gè)數(shù)字和運(yùn)算符粘連 或第二個(gè)數(shù)字和等號(hào)粘連 或第一個(gè)數(shù)字和運(yùn)算符粘連
        if 28 < w < 50:
            w = 20
            if x < 80:
                rects.append((x + 20, y, w, h))
        # 三個(gè)字符粘連，出現(xiàn)在運(yùn)算符、第二個(gè)數(shù)字、等號(hào)之間 或第一個(gè)數(shù)字、運(yùn)算符、第二個(gè)數(shù)字之間
        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
        # '*' 被劃分太細(xì)的情況放棄
        if w < 10 and h < 10:
            continue
        # 矩形切圖
        rects.append((x, y, w, h))
    rects.sort(key=None, reverse=False)
    return rects

5. 人工標(biāo)注

def mark_img(roi, timestamp):
    """
    人工標(biāo)注切圖
    :param roi:
    :param timestamp:
    :return:
    """
    print("PS:對(duì)每張切圖輸入對(duì)應(yīng)的字符（用于標(biāo)記切圖），回車(chē)跳過(guò)當(dāng)前切圖，點(diǎn)擊關(guān)閉退出人工標(biāo)記切圖")
    cv2.imshow("image", roi)
    key = cv2.waitKey(0)
    if key == 27: # 點(diǎn)擊關(guān)閉，退出
        sys.exit()
    if key == 13: # 回車(chē)跳過(guò)當(dāng)前
        return
    char = chr(key)
    print("您輸入的key是：", char)
    filename = "%s/%s_%s.jpg" % (img_lib_path, timestamp, char)
    cv2.imwrite(filename, roi)

這個(gè)過(guò)程就是不斷的加載新的驗(yàn)證碼，把圖片切出來(lái)，人工的查看每張切圖，用時(shí)間戳、下劃線(xiàn)、識(shí)別字符命名圖片，保存起來(lái)，用作訓(xùn)練數(shù)據(jù)。這部分比較無(wú)腦，但是要重復(fù)好多次，保存足夠的訓(xùn)練數(shù)據(jù)。

6. 訓(xùn)練

所謂訓(xùn)練就是利用已存在的數(shù)據(jù)（img_lib中的圖片），歸納總結(jié)出一個(gè)規(guī)律性的、可借鑒的模型來(lái)，后面就可以根據(jù)這個(gè)模型判斷識(shí)別新的驗(yàn)證碼。

def train_machine():
    """
    機(jī)器訓(xùn)練
    :return: id_label_map, model
    """
    # TODO 后續(xù)可嘗試將返回值緩存和持久化
    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

這個(gè)過(guò)程是把 img_lib 下的所有圖片都用 cv2.imread() 讀進(jìn)來(lái)，保存在 samples 中，然后切割文件名，把下劃線(xiàn)之后的字符（也就是圖片對(duì)應(yīng)的字符，此處稱(chēng)之為 label）保存在 labels 中。samples 和 labels 是一一對(duì)應(yīng)的關(guān)系。

id_label_map 的 value 是 labels 去重的集合，key 是角標(biāo)，id_label_map 用于后續(xù)對(duì)應(yīng)查找 label。
label_ids 是 img_lib 中所有圖片依次對(duì)應(yīng) id_label_map 的 key。

7. 識(shí)別

識(shí)別的過(guò)程，主要用到的是 model 這個(gè)對(duì)象，拿到新的驗(yàn)證碼圖片之后，依次和 samples 中的圖片（二維數(shù)組）進(jìn)行對(duì)比，然后找到最接近的圖片，返回這張圖片對(duì)應(yīng)的 label_ids 的值。最后用這個(gè)值去 id_label_map 中找出對(duì)應(yīng)的 label，即識(shí)別到的字符。
說(shuō)到底，是數(shù)學(xué)問(wèn)題啊~

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)
    # 找出該圖片對(duì)應(yīng)的 label_ids 中的值
    label_id = int(results[0, 0])
    # 找出對(duì)應(yīng)的 label，這就是識(shí)別結(jié)果
    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,機(jī)器識(shí)別的key是：%s" % (correct_char, label))

這部分代碼是搬磚來(lái)的，個(gè)人覺(jué)得 train_machine() 可以稍微簡(jiǎn)化一下的，訓(xùn)練數(shù)據(jù)時(shí)，label_ids 如果存的是 id_label_map 的值，而不是 key 的話(huà)，后面就返回一個(gè) model 就行了，后面識(shí)別的 results[0, 0] 直接就是我們想要的結(jié)果，也不需要在取 id_label_map 中找了。

3.2. session 保持及自動(dòng)重試

驗(yàn)證碼識(shí)別的最終目的是登陸，獲取校驗(yàn)所需的信息，從而在之后的請(qǐng)求能通過(guò)校驗(yàn)。本例系統(tǒng)使用的是 session、cookie機(jī)制，所以此處最開(kāi)始用到 http.cookiejar 保存 cookie 的方式請(qǐng)求。

import requests
import http.cookiejar

# 設(shè)置一個(gè)cookie處理器，它負(fù)責(zé)從服務(wù)器下載cookie到本地，并且在發(fā)送請(qǐng)求時(shí)帶上本地的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')
    # 打開(kāi)登錄主頁(yè)面（目的是從頁(yè)面下載cookie，這樣我們?cè)谠偎蚿ost數(shù)據(jù)時(shí)就有cookie了，否則發(fā)送不成功）
    response = request.urlopen(req)

但是這種方式行不通，因?yàn)槊看握?qǐng)求都是一個(gè)新的會(huì)話(huà)，導(dǎo)致沒(méi)次都識(shí)別到的驗(yàn)證碼都和后端不對(duì)應(yīng)。

解決辦法是使用 session = requests.session()，requests庫(kù)的session會(huì)話(huà)對(duì)象可以跨請(qǐng)求保持某些參數(shù)，就是比如你使用session成功的登錄了某個(gè)網(wǎng)站，再次使用該session對(duì)象請(qǐng)求該網(wǎng)站的其他網(wǎng)頁(yè)都會(huì)默認(rèn)使用該session之前使用的cookie等參數(shù)。

import requests
session = requests.session()
wrong_title = "鐵路工程管理平臺(tái)--登錄"


def request(url, method=request_method.GET, headers=None, data=None):
    """
    請(qǐng)求數(shù)據(jù)
    :param url:
    :param method:
    :param headers:
    :param data:
    :return:
    """
    response = session.request(method, url, headers=headers, data=data)
    return response

這樣就解決了會(huì)話(huà)保持的問(wèn)題，但是還有個(gè)問(wèn)題，session遲早會(huì)過(guò)期，所以還需要加上 session 過(guò)期自動(dòng)重新登錄的功能。
這個(gè)系統(tǒng) session 過(guò)期或登陸失敗，都會(huì)重定向會(huì)登錄頁(yè)，所以此處用網(wǎng)頁(yè) title 判斷是否 session 過(guò)期。所以增加了以下方法，這樣 session 過(guò)期或者驗(yàn)證碼識(shí)別錯(cuò)了導(dǎo)致登錄失敗，都可以自動(dòng)重新登錄。

def request_retry(url, method=request_method.GET, headers=None, data=None):
    """
    請(qǐng)求數(shù)據(jù)
    :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

3.3 定時(shí)任務(wù)和爬取數(shù)據(jù)

定時(shí)任務(wù)用到了 apscheduler 庫(kù)

from apscheduler.schedulers.blocking import BlockingScheduler

創(chuàng)建一個(gè) BlockingScheduler 對(duì)象，然后把需要定時(shí)的方法放入?yún)?shù)中。此處設(shè)置的是在 7 點(diǎn)到 23 點(diǎn)之間，每半小時(shí)更新一次。

def main():
    # 登錄，收集數(shù)據(jù)
    login.login()
    data_collector()
    # 定時(shí)收集數(shù)據(jù)
    task = BlockingScheduler()
    task.add_job(data_collector, "cron", hour="7-23", minute="*/30")
    task.start()


def data_collector():
    """
    數(shù)據(jù)收集
    :return:
    """
    logging.info("開(kāi)始定時(shí)任務(wù)")
    # 同步進(jìn)度信息
    progress_collector.get_construct_points()
    logging.info("定時(shí)任務(wù)完成")

所爬數(shù)據(jù)分兩種，一種是響應(yīng)格式為 json，另一種響應(yīng)格式為 html 頁(yè)面。
json 格式的數(shù)據(jù)可以借助 json 庫(kù)轉(zhuǎn)化下，直接取出想要的字段即可，例如：

def get_retry_std(url, headers=None, data=None):
    """
    get方法請(qǐng)求數(shù)據(jù),返回json，只適合 responses 是標(biāo)準(zhǔn)輸出的情況
    :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("請(qǐng)求異常，狀態(tài)碼：", response.status_code)
    response_body = json.loads(response.text)
    return response_body["result"]

html 頁(yè)面的數(shù)據(jù)需要借助 lxml 解析 dom，最后獲取想要的 dom 的數(shù)據(jù)即可，例如：

from lxml import etree
def get_tile(text):
    """
    根據(jù)文本獲取html頁(yè)面title
    :param text:
    :return:
    """
    tree = etree.HTML(text)
    return tree.xpath("http://title")[0].text

4. 使用說(shuō)明

4.1. 服務(wù)器上啟動(dòng)

直接運(yùn)行 docker 鏡像即可

4.2. 本地使用

本地啟動(dòng) starter.py 即可，半小時(shí)刷新一次。

test_main.py 中 test_analyse_accuracy() 方法用于測(cè)試識(shí)別精確度，自定義循環(huán)次數(shù)，需要每次手動(dòng)輸入用于判斷機(jī)器識(shí)別是否正確。輸出結(jié)果：

...
您輸入的key是：-,機(jī)器識(shí)別的key是：-
您輸入的key是：7,機(jī)器識(shí)別的key是：7
您輸入的key是：1,機(jī)器識(shí)別的key是：1


共測(cè)試58次，正確率百分之94.83

test_bound_result() 方法，用于畫(huà)矩形邊框，輔助調(diào)整切割圖片。
test_mark_images() 人工標(biāo)注圖片，標(biāo)注好的切圖存于 img_lib 中，用作訓(xùn)練數(shù)據(jù)。

配置文件中的內(nèi)容需要 [mysql] 自行補(bǔ)充, [crawler] 部分僅適用于本例，需根據(jù)實(shí)際情況定。

遺留問(wèn)題

1. 誤判率較高，原因有三：
   字符傾斜角度不一，拉伸困難
   字符粘連，平均分割圖片并不能準(zhǔn)確分割字符
   ‘*’ 識(shí)別率低，因?yàn)?‘*’ 會(huì)被識(shí)別成幾個(gè)小點(diǎn)
2. TODO：訓(xùn)練結(jié)果可以序列化保存，避免每次重新計(jì)算，從而不用再依賴(lài) img_lib 庫(kù)

參考文章

Python 3 教程

用Python識(shí)別驗(yàn)證碼

opencv-python 入門(mén)篇

opencv-python 指南