crawlab_feader/FworkSpider/untils/get_imgcode.py

import requests

import feapder.setting as setting

__all__ = [
    "swordfish_platform",
    "chaojiying_platform",
    "chaojiying_report",
    "get_code",
    "get_code_det",
    "arithmetic_captcha",
]

headers = {"accept": "application/json"}


def _pack_file(file):
    """包装验证码格式"""
    if isinstance(file, str) and file.startswith("data:image"):
        img_file = {"file": file}
    elif isinstance(file, bytes):
        img_file = {"file": file}
    else:
        with open(file, "rb") as f:
            img_bytes = f.read()
        img_file = {"file": img_bytes}
    return img_file


def _simple_captcha(file):
    """
    普通验证码

    @param file: 验证码 - 可以是图片或者图片base64编码
    @return:
    """
    url = f"{setting.CAPTCHA_URL}/v1/images/verify"
    files = _pack_file(file)
    r = requests.post(url, headers=headers, files=files, stream=True, timeout=10)
    rp_json = r.json()
    if "msg" in rp_json and "success" == rp_json["msg"]:
        return str(rp_json["r"]["code"])
    return None


def _arithmetic_captcha(file):
    """算术验证码"""
    url = f"{setting.CAPTCHA_URL}/v1/images/arithmetic"
    files = _pack_file(file)
    r = requests.post(url, headers=headers, files=files, stream=True, timeout=10)
    json_resp = r.json()
    if "msg" in json_resp and "success" == json_resp["msg"]:
        return str(json_resp["r"]["code"])
    return None


def _get_click_verify_captcha(file):
    """点触式验证码"""
    url = f"{setting.CAPTCHA_URL}/v1/images/verify_det"
    files = {"image_content": v for k, v in _pack_file(file).items()}
    r = requests.post(url, headers=headers, files=files, stream=True, timeout=10)
    return r.json()


def swordfish_platform(file, mode="simple"):
    """剑鱼验证码识别平台"""
    if mode.lower() == "arithmetic":
        return _arithmetic_captcha(file)
    elif mode.lower() == "det":
        return _get_click_verify_captcha(file)
    else:
        return _simple_captcha(file)


def chaojiying_platform(file, pic_type: int, spidercode=None):
    """
    超级鹰识别平台

    pic_type，详情查询地址: https://www.chaojiying.com/price.html
    @param str spidercode: 爬虫代码
    """
    files = _pack_file(file)
    url = f"{setting.CAPTCHA_URL}/v1/images/discern?pic_type={pic_type}"
    if spidercode is not None:
        url = f"{setting.CAPTCHA_URL}/v1/images/discern?pic_type={pic_type}&jy_code={spidercode}"

    headers = {'accept': 'application/json'}
    data = {
        'grant_type': '',
        'username': 'jianyu001',
        'password': '123qwe!A',
        'scope': '',
        'client_id': '',
        'client_secret': ''
    }
    response = requests.post(url, headers=headers, data=data, files=files, timeout=10)
    json_resp = response.json()
    '''code 返回0时，打码平台正常返回数据'''
    pic_str = json_resp["r"]["pic_str"]
    pic_id = json_resp["r"]["pic_id"]
    if 0 == json_resp["code"]:
        return pic_str, pic_id
    return None, pic_id


def chaojiying_report(pic_id: str):
    """超级鹰平台识别验证码错误时，提交识别错误的验证码pic_id"""
    url = f"{setting.CAPTCHA_URL}/v1/images/report_err?pic_id={pic_id}"
    headers = {
        'accept': 'application/json',
        'Content-Type': 'application/x-www-form-urlencoded'
    }
    data = {
        'grant_type': '',
        'username': 'jianyu001',
        'password': '123qwe!A',
        'scope': '',
        'client_id': '',
        'client_secret': ''
    }
    response = requests.post(url, headers=headers, data=data, timeout=10)
    '''
    回调成功:{'msg': 'OK', 'code': 0}  
    此接口不能随便调用！程序逻辑里要这样判断： 如果 识别结果是错的 再调用 报错返分 接口。 如果没有这个判断或是无法判断，就不要调用！
    '''
    return response.json()


def get_code(file_path: str) -> dict:
    return swordfish_platform(file_path) or {}


def get_code_det(image_bytes) -> dict:
    return swordfish_platform(image_bytes, mode="det")


# 算术
def arithmetic_captcha(image_stream):
    return swordfish_platform(image_stream, mode="arithmetic")