manyi/apilib/img_utils.py

# coding=utf-8
import logging

import simplejson as json

from aip import AipOcr
from django.conf import settings
from django.core.files.uploadedfile import UploadedFile

from apilib.exceptions import BaiDuApiImageError, BaiDuApiNetError, BaiDuApiSysError
from apilib.utils import split_addr
from apilib.utils_datetime import to_datetime
from apps.web.utils import LimitAttemptsManager


logger = logging.getLogger(__name__)





def parse_identify_image(result):
    if not isinstance(result, dict):
        raise BaiDuApiImageError(u"图像识别错误")

    # 状态字段一定会有 一定需要判断图像的状态
    if result["image_status"] != "normal":
        if result["image_status"] == "reversed_side":
            raise BaiDuApiImageError(u"身份证正反面颠倒，请确认身份证的国徽面或人像面")
        if result["image_status"] == "non_idcard":
            raise BaiDuApiImageError(u"上传的图片中不包含身份证，请确认")
        if result["image_status"] == "blurred":
            raise BaiDuApiImageError(u"身份证模糊，请重新上传")
        if result["image_status"] == "other_type_card":
            raise BaiDuApiImageError(u"上传证件类型错误，请重新上传")
        if result["image_status"] == "over_exposure":
            raise BaiDuApiImageError(u"身份证信息识别错误，请重新上传")
        if result["image_status"] == "over_dark":
            raise BaiDuApiImageError(u"身份证图像亮度过低，识别错误，请重新上传")
        if result["image_status"] == "unknown":
            raise BaiDuApiImageError(u"识别错误，请重新上传")

    # 证号状态只有在人像面才有
    if "idcard_number_type" in result and  result["idcard_number_type"] != 1:
        if result["idcard_number_type"] == -1:
            raise BaiDuApiImageError(u"身份证人像面信息识别错误，请检查")
        if result["idcard_number_type"] == 0:
            raise BaiDuApiImageError(u"身份证证号不合法")
        if result["idcard_number_type"] == 2:
            raise BaiDuApiImageError(u"身份证证号和性别、出生信息都不一致，请检查")
        if result["idcard_number_type"] == 3:
            raise BaiDuApiImageError(u"身份证证号和出生信息不一致，请检查")
        if result["idcard_number_type"] == 4:
            raise BaiDuApiImageError(u"身份证证号和性别信息不一致，请检查")

    # 证件方向 选填
    if "direction" in result and result["direction"] != 0:
        raise BaiDuApiImageError(u"请调整上传图像方向，将上传图片摆正（可参考右侧图片示例）。")

    # 证件风险 选填 防止复印件
    if "risk_type" in result and result["risk_type"] != "normal":
        if result["risk_type"] == "copy":
            raise BaiDuApiImageError(u"上传身份证为复印件")
        if result["risk_type"] == "temporary":
            raise BaiDuApiImageError(u"上传身份证为临时件")
        if result["risk_type"] == "screen":
            raise BaiDuApiImageError(u"上传身份证为翻拍件")
        if result["risk_type"] == "unknown":
            raise BaiDuApiImageError(u"上传身份证异常")

    # 证件质量检查
    if "card_quality" in result:
        if result["card_quality"]["IsClear"] != 1:
            raise BaiDuApiImageError(u"身份证图像不清晰")
        if result["card_quality"]["IsComplete"] != 1:
            raise BaiDuApiImageError(u"身份证图像不完整")
        if result["card_quality"]["IsNoCover"] != 1:
            raise BaiDuApiImageError(u"身份证图像被遮挡")

    # 最后处理下识别的字段结果 注意对各种数据进行下转换
    data = dict()
    if u"住址" in result["words_result"]:
        data["province"], data["city"], data["area"], data["addr"] = split_addr(result["words_result"][u"住址"]["words"])
    if u"出生" in result["words_result"]:
        data["birthday"] = result["words_result"][u"出生"]["words"]
    if u"姓名" in result["words_result"]:
        data["name"] = result["words_result"][u"姓名"]["words"]
    if u"公民身份号码" in result["words_result"]:
        data["code"] = result["words_result"][u"公民身份号码"]["words"]
    if u"性别" in result["words_result"]:
        data["sex"] = result["words_result"][u"性别"]["words"]
    if u"民族" in result["words_result"]:
        data["nation"] = result["words_result"][u"民族"]["words"]

    if u"签发日期" in result["words_result"]:
        data["startTime"] = to_datetime(result["words_result"][u"签发日期"]["words"], "%Y%m%d").strftime("%Y-%m-%d")
    if u"失效日期" in result["words_result"]:
        try:
            data["endTime"] = to_datetime(result["words_result"][u"失效日期"]["words"], "%Y%m%d").strftime("%Y-%m-%d")
        except ValueError:
            data["endTime"] = result["words_result"][u"失效日期"]["words"]

    return data


def parse_bank_image(result):
    if not isinstance(result, dict):
        raise BaiDuApiImageError(u"图像识别错误")

    if "direction" in result and result["direction"] != 0:
        raise BaiDuApiImageError(u"请调整上传图像方向，将上传图片摆正（可参考右侧图片示例）。")

    if result["result"]["bank_card_type"] == 0:
        raise BaiDuApiImageError(u"无法识别您的银行卡类型，请重新上传")

    data = {
        "bankCode": "".join(result["result"]["bank_card_number"].split()),
    }

    return data


def parse_business_image(result):
    if not isinstance(result, dict):
        raise BaiDuApiImageError(u"图像识别错误")

    if "direction" in result and result["direction"] != 0:
        raise BaiDuApiImageError(u"请调整上传图像方向，将上传图片摆正（可参考右侧图片示例）。")
    if "risk_type" in result and result["risk_type"] != "normal":
        raise BaiDuApiImageError(u"营业执照异常")

    wordsResult = result["words_result"]
    data = dict()
    if u"社会信用代码" in wordsResult:
        data["busCode"] = wordsResult[u"社会信用代码"]["words"]
    if u"单位名称" in wordsResult:
        data["busName"] = wordsResult[u"单位名称"]["words"]
    if u"成立日期" in wordsResult:
        data["startTime"] = to_datetime(wordsResult[u"成立日期"]["words"], u"%Y年%m月%d日").strftime("%Y-%m-%d")
    if u"有效期" in wordsResult:
        try:
            data["endTime"] = to_datetime(wordsResult[u"有效期"]["words"], u"%Y年%m月%d日").strftime("%Y-%m-%d")
        except ValueError:
            data["endTime"] = wordsResult[u"有效期"]["words"]
    if u"地址" in wordsResult:
        data["province"], data["city"], data["area"], data["addr"] = split_addr(wordsResult[u"地址"]["words"])

    if u"法人" in wordsResult:
        data["legalName"] = wordsResult[u"法人"]["words"]

    return data


class IDCardSide(object):
    PERSONAL = "front"
    EMBLEM = "back"


class ImageRecognizer(object):
    """
    OCR 识别工具
    helps: https://cloud.baidu.com/doc/OCR/s/7kibizyfm#%E8%BA%AB%E4%BB%BD%E8%AF%81%E8%AF%86%E5%88%AB
    当前版本不支持url的形式上传
    """

    def __init__(self, visitor, appid=None, appKey=None, secretKey=None, imageData=None, inMemFile=None):

        limiter = LimitAttemptsManager(visitor, self.__class__.__name__, maxTimes=30 if not settings.DEBUG else 9999)
        if limiter.is_exceeded_limit():
            raise BaiDuApiNetError(u"访问频率超限")
        limiter.incr()

        self._appid = appid or settings.BAIDU_IMAGE_RECOGNIZE_APP_ID
        self._appKey = appKey or settings.BAIDU_IMAGE_RECOGNIZE_APP_KEY
        self._appSecret = secretKey or settings.BAIDU_IMAGE_RECOGNIZE_APP_SECRET

        if not all([self._appid, self._appKey, self._appSecret]):
            raise BaiDuApiNetError(u"识别初始化失败")

        if imageData and isinstance(imageData, bytes):
            self._content = imageData
        elif inMemFile and isinstance(inMemFile, UploadedFile):
            self._content = b""
            for _image in inMemFile.chunks():
                self._content += _image
        else:
            self._content = None

        if not self._content:
            raise BaiDuApiImageError(u"无效的图像数据")

        self._client = None

    @property
    def client(self):   # type:() -> AipOcr
        if not self._client:
            self._client = AipOcr(
                appId=self._appid, apiKey=self._appKey, secretKey=self._appSecret
            )

        return self._client

    @staticmethod
    def _handle_result(result, callback):
        """
        对错误作出一定的解析
        """
        if not isinstance(result, dict):
            raise BaiDuApiImageError(u"图像识别失败（10002）")
        if "error_code" not in result:
            return callback(result) if callback else result

        error_code = result["error_code"]

        # 此类错误是一般是对方服务器错误，需要通知系统管理员及时提交工单
        if error_code in [
            1,
            100, 110, 111,
            282000,
            216100, 216101, 216102, 216103,
            216110,
            216630, 216634,
            282003,
            282005, 282006,
            282110, 282111, 282112, 282113, 282114,
            282808, 282809, 282810
        ]:
            raise BaiDuApiImageError(u"图像识别错误，请重新上传（{}-{}）。".format(error_code, result["error_msg"]))

        if error_code == 216200:
            raise BaiDuApiImageError(u"未检测到上传图像，请重新上传。")

        if error_code == 216201:
            raise BaiDuApiImageError(u"图片格式上传错误，请重新上传（支持图像格式为 PNG、JPG、JPEG、BMP）。")

        if error_code == 216202:
            raise BaiDuApiImageError(u"图片尺寸过大，请重新上传（图像不得超过4Mb, 分辨率不高于4096*4096）。")

        if error_code == 216631:
            raise BaiDuApiImageError(u"银行卡识别错误，出现此问题的一般原因为：您上传的图片非银行卡正面，上传了异形卡的图片或上传的银行卡正品图片不完整。")

        if error_code == 216633:
            raise BaiDuApiImageError(u"身份证识别错误，出现此问题的一般原因为：您上传了非身份证图片或您上传的身份证图片不完整")

        # 流量或者QPS 超限的错误
        if error_code in [17, 18, 19]:
            raise BaiDuApiSysError(u"额度超出限制。")

        # 其余的错误直接抛出
        raise BaiDuApiImageError(u"图像识别错误，请重新上传（{}-{}）。".format(error_code, result["error_msg"]))

    def recognize_identify_card(self, idCardSide, detectDirection=True, detectRisk=False, detectQuality=False, detectPhoto=False, detectCard=False, callback=None):
        """
        识别身份证 目前后三个付费功能没有开通
        :param idCardSide: 自动检测身份证正反面，如果传参指定方向与图片相反，支持正常识别，返回参数image_status字段为"reversed_side"
        :param detectDirection: 是否检测图像朝向，默认不检测，即：false。朝向是指输入图像是正常方向、逆时针旋转90/180/270度
        :param detectRisk: 是否开启身份证风险类型(身份证复印件、临时身份证、身份证翻拍、修改过的身份证)检测功能
        :param detectQuality: 是否开启身份证质量类型(边框/四角不完整、头像或关键字段被遮挡/马赛克)检测功能
        :param detectPhoto: 是否检测头像内容，默认不检测
        :param detectCard: 是否检测身份证进行裁剪，默认不检测
        :param callback: 回调处理函数
        """
        options = {
            "detect_direction": json.dumps(detectDirection),
            "detect_risk": json.dumps(detectRisk),
            "detect_quality": json.dumps(detectQuality),
            "detect_photo": json.dumps(detectPhoto),
            "detect_card": json.dumps(detectCard)
        }

        result = self.client.idcard(self._content, idCardSide, options)
        return self._handle_result(result, callback)

    def recognize_bank_card(self, detectDirection=True, callback=None):
        """
        识别银行卡
        两个注意：1不能识别银行开户许可证 2识别出来的银行卡号不连续 需要后续存储的时候去掉空格
        :param detectDirection: 检测方向
        :param callback:
        """
        options = {
            "detect_direction": json.dumps(detectDirection)
        }
        result = self.client.bankcard(self._content, options)
        return self._handle_result(result, callback)

    def recognize_business_license(self, detectDirection=True, riskWarn=False, callback=None):
        options = {
            "detect_direction": json.dumps(detectDirection),
            "risk_warn": json.dumps(riskWarn),
        }
        result = self.client.businessLicense(self._content, options)
        return self._handle_result(result, callback)