All/Python/PaddleOCR/blues/bank.py

import re
from time import time
from utils.util import *
from json import load, dump, loads
from requests import request as http_req
from utils.conf import MAX_FILE_SIZE
from flask import Blueprint, views, render_template, request
from utils.logger import Logger

bank = Blueprint("bank", __name__, url_prefix="/bank")

_InfoFile = "utils/bank-info.json"
__fp = open(_InfoFile, "r", encoding="utf-8")
_InfoForShort = load(__fp)
__fp.close()
__InfoForBin = {__v["bin"]: __v for __v in _InfoForShort.values() if __v["bin"]}
__NumPtn = re.compile(r"(?P<number>\d{15,19})")
_UnresolvedInfo = []
__UnresolvedIncludes = []
_ErrorInfo = []
__ErrorIncludes = []
_FailIcon = "A-Fail"


def __get_bank_and_icon(num: "str") -> "tuple[str, str]":
    _bin = num[:6]
    if _bin in __InfoForBin.keys():
        return __InfoForBin[_bin]["name"], __InfoForBin[_bin]["icon"]
    if num in __UnresolvedIncludes or num in __ErrorIncludes:
        return "", _FailIcon

    url = "https://ccdcapi.alipay.com/validateAndCacheCardInfo.json" \
          f"?_input_charset=utf-8&cardBinCheck=true&cardNo={num}"
    resp, bank_type = http_req("GET", url), ""
    if resp.status_code == 200:
        data = resp.json()
        if data["stat"] == "ok":
            bank_type = data["bank"]
    if bank_type:
        if bank_type in _InfoForShort.keys():
            _InfoForShort[bank_type]["bin"] = _bin
            __InfoForBin[_bin] = _InfoForShort[bank_type]
            return __InfoForBin[_bin]["name"], __InfoForBin[_bin]["icon"]
        if num in __UnresolvedIncludes or bank_type in __UnresolvedIncludes:
            return "", _FailIcon
        __UnresolvedIncludes.append(num)
        __UnresolvedIncludes.append(bank_type)
        _UnresolvedInfo.append({"number": num, "short": bank_type, "name": "", "icon": ""})
        return "", _FailIcon
    __ErrorIncludes.append(num)
    _ErrorInfo.append([num, current_time("%Y-%m-%d %H:%M:%S")])
    return "", _FailIcon


def _get_bank_info(full_str: "str") -> "tuple[bool, dict]":
    res = {"number": "", "bank": "", "icon": ""}
    if search := re.search(__NumPtn, full_str):
        res["number"] = search.group("number")
        res["bank"], res["icon"] = __get_bank_and_icon(res["number"])
        if res["bank"]:
            return True, res
    return False, res


class BankView(views.MethodView):
    @staticmethod
    def get():
        return render_template("bank/index.html")

    @staticmethod
    def post():
        start = time()
        pic = request.files.get("picture")
        if pic is None:
            return Response("empty body")
        ext = get_ext_name(pic.filename)
        if not is_image_ext(ext):
            return Response("文件类型错误")
        content = pic.read()
        if len(content) > MAX_FILE_SIZE:
            return Response("文件过大，请重新选择")

        cropped = preprocess_img(read_img(content))  # 预处理，对深色背景的效果很好
        images = rot_img_2(cropped)
        recognizes = Engine.rec_multi(images)
        info, sta = {}, False
        for i, ocr_res in enumerate(recognizes):
            rec_str = "".join([it[0] for it in ocr_res])
            sta, info = _get_bank_info(rec_str)
            if sta:
                info["duration"] = time() - start
                raw_path = f"static/images/{current_time()}_{rand_str()}.{ext}"
                save_img(raw_path, images[i])
                return Response(data=info)
            else:
                Logger.error(rec_str)

        info["duration"] = time() - start
        return Response("识别失败，请重新选择", info)


class BankHtmlView(views.MethodView):
    @staticmethod
    def post():
        start = time()
        pic = request.files.get("picture")
        if pic is None:
            return Response("empty body")
        ext = get_ext_name(pic.filename)
        if not is_image_ext(ext):
            return Response("文件类型错误")
        content = pic.read()
        if len(content) > MAX_FILE_SIZE:
            return Response("文件过大，请重新选择")

        cropped = preprocess_img(read_img(content))
        images = rot_img_2(cropped)
        recognizes = Engine.rec_multi(images)
        info, err_rec, sta, idx = {}, [], False, 0
        msg = "识别失败，请重新选择"
        for i, ocr_res in enumerate(recognizes):
            rec_str = "".join([it[0] for it in ocr_res])
            sta, info = _get_bank_info(rec_str)
            if sta:
                idx, msg = i, "识别成功"
                break
            else:
                Logger.error(rec_str)
                err_rec.append(rec_str)

        file_path = f"static/images/{current_time()}_{rand_str()}.{ext}"
        save_img(file_path, images[idx])

        info["SUCCESS"] = str(sta).upper()
        info["MESSAGE"] = msg
        if not sta:
            info["icon"] = _FailIcon
        if len(err_rec):
            info["MESSAGE"] += "<br>识别结果：<br>" + "<br>".join(err_rec)
        info["DURATION"] = time() - start  # noqa
        return render_template("bank/result.html", raw=file_path, data=info)


class BankManageView(views.MethodView):
    @staticmethod
    def get():
        info = request.args.get("info")
        if info is None:
            return render_template("bank/manage.html")
        return Response(data={"full": _InfoForShort, "unresolved": _UnresolvedInfo, "error": _ErrorInfo})

    @staticmethod
    def post():
        which = request.form.get("which")
        opt = request.form.get("opt")
        if not which or not opt:
            return Response("key param lost")
        which, opt = which.lower(), opt.lower()

        if which == "full":
            if opt == "update":
                try:
                    data = request.form.get("data")
                    global _InfoForShort
                    _InfoForShort = loads(data)
                    return Response()
                except Exception:  # noqa
                    return Response("JSON格式错误")
            elif opt == "persist":
                with open(_InfoFile, "w", encoding="utf-8") as fp:
                    dump(_InfoForShort, fp, ensure_ascii=False, indent=2)  # noqa
                    fp.close()
                return Response()
            return Response(f"unrecognized opt: <{opt}>")
        if which == "unresolved":
            if opt == "update":
                ...  # TODO
            return Response(f"unrecognized opt: <{opt}>")
        if which == "error":
            if opt == "clear":
                _ErrorInfo.clear()
                return Response()
            return Response(f"unrecognized opt: <{opt}>")
        if which == "icon":
            if opt == "upload":
                name = request.form.get("name")
                icon = request.files.get("icon")
                if name is None:
                    return Response("key param lost")
                if icon is None:
                    return Response("empty body")
                ext = get_ext_name(icon.filename)
                if ext != "png":
                    return Response("icon仅支持png格式")
                content = icon.read()
                if len(content) > MAX_FILE_SIZE:
                    return Response("文件过大，请重新选择")
                try:
                    save_img(f"static/bank/{name}.png", content)
                    return Response()
                except Exception:  # noqa
                    return Response("文件名称不合法")
        return Response(f"unrecognized which: <{which}>")


bank.add_url_rule("/", view_func=BankView.as_view("bank"))
bank.add_url_rule("/html/", view_func=BankHtmlView.as_view("bank-html"))
bank.add_url_rule("/manage/", view_func=BankManageView.as_view("bank-manage"))