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\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"] += "
识别结果：
" + "
".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"))