Skip to content

周伟奇 / bmw-ocr

Go to a project
Toggle navigation
Toggle navigation pinning
1e50135a by 周伟奇
Options

prese part 1

1 parent f6840d5c
Inline Side-by-side
Showing 11 changed files with 485 additions and 2315 deletions
src/apps/doc/consts.py
......@@ -1234,26 +1234,39 @@ RESULT_MAPPING = {
}
CA_ADD_COMPARE_FIELDS = (IC_OCR_FIELD, BL_OCR_FIELD, BS_FIELD)
CA_ADD_COMPARE_FIELDS_PRE = (IC_OCR_FIELD,)
COMPARE_FIELDS = (
MVI_OCR_FIELD,
IC_OCR_FIELD,
RP_OCR_FIELD,
BC_OCR_FIELD,
BL_OCR_FIELD,
UCI_OCR_FIELD,
EEP_OCR_FIELD,
DL_OCR_FIELD,
PP_OCR_FIELD,
MVC_OCR_FIELD,
DDA_OCR_FIELD,
HMH_OCR_FIELD,
JYPZ_OCR_FIELD,
HT_FIELD,
BD_FIELD,
BS_FIELD,
HIL_CONTRACT_1_FIELD,
HIL_CONTRACT_2_FIELD,
HIL_CONTRACT_3_FIELD,
)
COMPARE_FIELDS = (MVI_OCR_FIELD,
IC_OCR_FIELD,
RP_OCR_FIELD,
BC_OCR_FIELD,
BL_OCR_FIELD,
UCI_OCR_FIELD,
EEP_OCR_FIELD,
DL_OCR_FIELD,
PP_OCR_FIELD,
MVC_OCR_FIELD,
DDA_OCR_FIELD,
HMH_OCR_FIELD,
JYPZ_OCR_FIELD,
HT_FIELD,
BD_FIELD,
BS_FIELD,
HIL_CONTRACT_1_FIELD,
HIL_CONTRACT_2_FIELD,
HIL_CONTRACT_3_FIELD,
PRE_COMPARE_FIELDS = (
MVI_OCR_FIELD,
IC_OCR_FIELD,
BC_OCR_FIELD,
HMH_OCR_FIELD,
HT_FIELD,
BD_FIELD,
HIL_CONTRACT_1_FIELD,
HIL_CONTRACT_2_FIELD,
)
# 身份证
......
src/apps/doc/mixins.py
......@@ -2,8 +2,9 @@ import re
import json
import requests
from .named_enum import DocStatus
from .models import HILDoc, AFCDoc
from .models import HILDoc, AFCDoc, AFCSEOCRResult, AFCOCRResult, HILSEOCRResult, HILOCRResult
from . import consts
from prese.compare import pre_compare
class MPOSHandler:
......@@ -141,3 +142,31 @@ class DocHandler:
else:
return consts.DATA_SOURCE_LIST[0]
class PreSEHandler:
# preSettlement
@staticmethod
def pre_compare_entrance(pos_content):
application_entity = pos_content.get('applicationEntity')
application_id = pos_content.get('applicationId')
# 根据application_id查找OCR累计结果指定license字段，如果没有，结束
result_class = HILSEOCRResult if application_entity in consts.HIL_SET else AFCSEOCRResult
ca_result_class = HILOCRResult if application_entity in consts.HIL_SET else AFCOCRResult
ca_ocr_res_dict = ca_result_class.objects.filter(application_id=application_id).values(
*consts.CA_ADD_COMPARE_FIELDS_PRE).first()
ocr_res_dict = result_class.objects.filter(application_id=application_id).values(
*consts.PRE_COMPARE_FIELDS).first()
if ocr_res_dict is None:
return
id_res_list = []
for field_name in consts.CA_ADD_COMPARE_FIELDS_PRE:
if field_name == consts.IC_OCR_FIELD:
id_res_list.append(ca_ocr_res_dict.get(field_name) if isinstance(ca_ocr_res_dict, dict) else None)
id_res_list.append(ocr_res_dict.get(field_name))
rebuild_compare_result = pre_compare(pos_content, ocr_res_dict, id_res_list)
return rebuild_compare_result
......
src/apps/doc/views.py
......@@ -48,10 +48,10 @@ from .models import (
MposReport,
)
from .named_enum import ErrorType, AutoResult, WholeResult, RPAResult
from .mixins import DocHandler, MPOSHandler
from .mixins import DocHandler, MPOSHandler, PreSEHandler
from . import consts
from apps.account.authentication import OAuth2AuthenticationWithUser
from celery_compare.tasks import compare, pos_compare
from celery_compare.tasks import compare
class CustomDate(fields.Date):
......@@ -87,6 +87,8 @@ se_vehicle_args = {
'vehicleTransactionAmount': CustomDecimal(required=True),
'vinNo': fields.Str(required=True, validate=validate.Length(max=256)),
'dealer': fields.Str(required=True, validate=validate.Length(max=256)),
'showRoom': fields.Str(required=False),
'agencyDealer': fields.Str(required=False),
'option': CustomDecimal(required=False),
'msrp': CustomDecimal(required=False),
'totalAmount': CustomDecimal(required=False),
......@@ -125,6 +127,7 @@ se_quotationt_args = {
'loanTerm': fields.Int(required=True),
'vehiclePrincipal': CustomDecimal(required=True),
'associatedServicePrincipal': CustomDecimal(required=False),
'mortgageType': fields.Str(required=True, validate=validate.Length(max=16)),
'associatedServiceInfo': fields.List(fields.Nested(se_associated_args), required=False),
'monthlyPaymentInfo': fields.List(fields.Nested(se_payment_args), required=True, validate=validate.Length(min=1)),
}
......@@ -224,6 +227,10 @@ se_compare_content = {
'applicationId': fields.Str(required=True, validate=validate.Length(max=64)),
"applicationVersion": fields.Int(required=True),
'applicationEntity': fields.Str(required=True, validate=validate.OneOf(consts.ENTITY)),
'productGroup': fields.Str(required=False),
'productGroupID': fields.Str(required=False),
'customerType': fields.Str(required=True, validate=validate.OneOf(consts.CUSTOMER_TYPE)),
"firstSubmmisonDate": CustomDate(required=True),
'propertyDocumentPolicy': fields.Str(required=False, validate=validate.Length(max=16)),
......@@ -754,34 +761,45 @@ class CompareView(GenericView):
'''
class SECompareView(GenericView):
class SECompareView(GenericView, PreSEHandler):
permission_classes = [IsAuthenticated]
authentication_classes = [OAuth2AuthenticationWithUser]
# pos上传比对信息接口 SE
# SE preSettlement
@use_args(se_compare_args, location='data')
def post(self, request, args):
# 存库
log_base = '[prese]'
content = args.get('content', {})
business_type = content.get('applicationEntity')
application_id = content.get('applicationId')
uniq_seq = content.get('uniqSeq')
bank_verify = content.get('bankInfo', {}).get('bankVerificationStatus', '')
bank_class = HILbankVerification if business_type in consts.HIL_SET else AFCbankVerification
bank_obj = bank_class.objects.filter(application_id=application_id).first()
# 存库, 用于银行卡比对
try:
bank_class = HILbankVerification if business_type in consts.HIL_SET else AFCbankVerification
bank_obj = bank_class.objects.filter(application_id=application_id).first()
if bank_obj is None and bank_verify == 'PASS':
bank_class.objects.create(
application_id=application_id,
)
elif bank_obj is not None and bank_verify == 'PASS' and bank_obj.on_off is False:
bank_obj.on_off = True
bank_obj.save()
elif bank_obj is not None and bank_verify != 'PASS' and bank_obj.on_off is True:
bank_obj.on_off = False
bank_obj.save()
compare_result = pos_compare(content)
if bank_obj is None and bank_verify == 'PASS':
bank_class.objects.create(
application_id=application_id,
)
elif bank_obj is not None and bank_verify == 'PASS' and bank_obj.on_off is False:
bank_obj.on_off = True
bank_obj.save()
elif bank_obj is not None and bank_verify != 'PASS' and bank_obj.on_off is True:
bank_obj.on_off = False
bank_obj.save()
except Exception as e:
self.running_log.info('{0} [bankCard verify save db error] [applicationEntity={1}] '
'[application_id={2}] [bank_status={3}] [error={4}]'.format(
log_base, business_type, application_id, bank_verify, traceback.format_exc()))
# preSettlement比对
compare_result = self.pre_compare_entrance(content)
self.running_log.info('{0} [prese completed] [applicationEntity={1}] [application_id={2}] [uniq_seq={3}] '
'[result={4}]'.format(log_base, business_type, application_id, uniq_seq, compare_result))
return response.ok(data=compare_result)
......
src/celery_compare/tasks.py
......@@ -39,7 +39,6 @@ from apps.doc.exceptions import GCAPException
from apps.doc.named_enum import RequestTeam, RequestTrigger, ProcessName, ErrorType
from common.tools.comparison import cp
from common.tools.des import decode_des
from pos import pos
compare_log = logging.getLogger('compare')
log_base = '[Compare]'
......@@ -2414,6 +2413,7 @@ def se_compare_license_id(license_en, id_res_list, field_list):
if is_find:
break
result_field_list.clear()
for idx, (name, value) in enumerate(field_list):
# if ocr_field == consts.MVI_OCR_FIELD and name == consts.SE_NEW_ADD_FIELD[9]:
# ocr_str = getattr(cp, consts.ZW_METHOD)(
......@@ -2493,6 +2493,7 @@ def se_compare_license_id(license_en, id_res_list, field_list):
result_field_list.append((name, value, result, ocr_str, img_path, error_type, compare_logic[name][3]))
if not is_find:
result_field_list.clear()
for name, value in field_list:
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
......@@ -3200,135 +3201,3 @@ def compare(application_id, application_entity, uniq_seq, ocr_res_id, is_ca=True
'[error={4}]'.format(log_base, application_entity, application_id, ocr_res_id,
traceback.format_exc()))
# pos接口
def pos_compare(pos_content, application_id, application_entity, uniq_seq, ocr_res_id=None, is_ca=True, is_cms=False):
# POS: application_id, application_entity, uniq_seq, None
# OCR: application_id, business_type(application_entity), None, ocr_res_id
compare_log.info('{0} [pos_compare] [entity={1}] [id={2}] [uniq_seq={3}] [is_ca={4}] '
'[is_cms={5}]'.format(log_base, application_entity, application_id, uniq_seq,
is_ca, is_cms))
# 根据application_id查找最新的比对信息，如果没有，结束
# if is_ca:
# comparison_class = HILComparisonInfo if application_entity == consts.HIL_PREFIX else AFCComparisonInfo
# else:
# if application_entity == consts.HIL_PREFIX:
# comparison_class = HILSECMSInfo if is_cms else HILSEComparisonInfo
# else:
# comparison_class = AFCSECMSInfo if is_cms else AFCSEComparisonInfo
# last_obj = comparison_class.objects.filter(application_id=application_id).last()
# if last_obj is None:
# compare_log.info('{0} [comparison info empty] [entity={1}] [id={2}] [uniq_seq={3}] '
# '[is_ca={4}] [is_cms]={5}'.format(log_base, application_entity, application_id, uniq_seq,
# is_ca, is_cms))
# return
# 根据application_id查找OCR累计结果指定license字段，如果没有，结束
if is_ca:
result_class = HILOCRResult if application_entity == consts.HIL_PREFIX else AFCOCRResult
ca_ocr_res_dict = dict()
else:
result_class = HILSEOCRResult if application_entity == consts.HIL_PREFIX else AFCSEOCRResult
ca_result_class = HILOCRResult if application_entity == consts.HIL_PREFIX else AFCOCRResult
# if ocr_res_id is None:
ca_ocr_res_dict = ca_result_class.objects.filter(application_id=application_id).values(
*consts.CA_ADD_COMPARE_FIELDS).first()
# else:
# ca_ocr_res_dict = ca_result_class.objects.filter(id=ocr_res_id).values(
# *consts.CA_ADD_COMPARE_FIELDS).first()
ocr_res_dict = result_class.objects.filter(application_id=application_id).values(*consts.COMPARE_FIELDS).first()
if ocr_res_dict is None:
compare_log.info('{0} [ocr info empty] [entity={1}] [id={2}] [uniq_seq={3}] [ocr_res_id={4}] '
'[is_ca={5}] [is_cms]={6}'.format(log_base, application_entity, application_id,
uniq_seq, ocr_res_id, is_ca, is_cms))
return
pos_content_obj = json.loads(pos_content)
if is_ca:
# 比对逻辑
# compare_info = get_ca_compare_info(pos_content_obj)
# compare_result, total_fields, failed_count = ca_compare_process(compare_info, ocr_res_dict)
# compare_log.info('{0} [CA] [compare success] [entity={1}] [id={2}] [ocr_res_id={3}] [result={4}]'.format(
# log_base, application_entity, application_id, ocr_res_id, compare_result))
# ca_compare(application_id, application_entity, ocr_res_id, pos_content_obj, ocr_res_dict)
compare_result = pos.PosCompare.ca_compare(pos_content_obj, ocr_res_dict)
else:
id_res_list = []
for field_name in consts.CA_ADD_COMPARE_FIELDS:
if field_name == consts.IC_OCR_FIELD:
id_res_list.append(ocr_res_dict.get(field_name))
id_res_list.append(ca_ocr_res_dict.get(field_name) if isinstance(ca_ocr_res_dict, dict) else None)
if isinstance(ca_ocr_res_dict, dict) and isinstance(ca_ocr_res_dict.get(field_name), str):
tmp_ca_result = json.loads(ca_ocr_res_dict.get(field_name))
if isinstance(ocr_res_dict.get(field_name), str):
tmp_se_result = json.loads(ocr_res_dict.get(field_name))
tmp_ca_result.extend(tmp_se_result)
ocr_res_dict[field_name] = json.dumps(tmp_ca_result)
# auto settlement
# auto_class = HILAutoSettlement if application_entity == consts.HIL_PREFIX else AFCAutoSettlement
# auto_obj = auto_class.objects.filter(application_id=application_id, on_off=True).first()
# bank_class = HILbankVerification if application_entity == consts.HIL_PREFIX else AFCbankVerification
# ignore_bank = bank_class.objects.filter(application_id=application_id, on_off=True).exists()
# if auto_obj is not None:
# auto_result = se_compare_auto(application_id, application_entity, ocr_res_id, pos_content_obj, ocr_res_dict, auto_obj, ignore_bank, id_res_list)
# else:
# auto_result = None
#
# full_result = se_compare(application_id, application_entity, ocr_res_id, pos_content_obj, ocr_res_dict, is_cms, auto_result, ignore_bank, id_res_list)
#
# if auto_obj is not None:
# try:
# auto_obj.ocr_whole_result_pass = full_result
# auto_obj.save()
# compare_log.info('{0} [Auto SE] [result save success] [entity={1}] [id={2}] [ocr_res_id={3}]'.format(
# log_base, application_entity, application_id, ocr_res_id))
# except Exception as e:
# compare_log.error('{0} [Auto SE] [result save error] [entity={1}] [id={2}] [ocr_res_id={3}] '
# '[error={4}]'.format(log_base, application_entity, application_id, ocr_res_id,
# traceback.format_exc()))
# result_table = HILCACompareResult if application_entity == consts.HIL_PREFIX else AFCCACompareResult
# res_obj = result_table.objects.filter(application_id=application_id).first()
compare_result = pos.PosCompare.se_compare(pos_content_obj, ocr_res_dict, id_res_list=id_res_list)
return pos_result_output(compare_result)
# pos输出解析
def pos_result_output(compare_result):
result_obj = json.loads(compare_result)
license_map = dict()
for item in result_obj:
license_en = item.get("License")
if license_map.get(license_en):
license_map.get(license_en).append(item)
else:
info_items = [item]
license_map[license_en] = info_items
is_pass = True
particulars = []
for license, license_items in license_map.items():
particular = {"object_name": license}
# license_comments = OCR_COMPARE_COMMENT.get(license)
# if not license_comments:
# continue
fields = []
particular["fields"] = fields
for license_item in license_items:
field = dict()
field["input"] = license_item.get("Input")
field["ocr"] = license_item.get("OCR")
field["field_is_pass"] = license_item.get("Result") == 'Y'
if not field["field_is_pass"]:
is_pass = False
field["comments"] = license_item.get('comments', '')
fields.append(field)
particular["fields"] = fields
particulars.append(particular)
return {
"is_pass": is_pass,
"particulars": particulars
}
......
src/common/tools/comparison.py
......@@ -29,8 +29,6 @@ class Comparison:
self.RESULT_Y = 'Y'
self.RESULT_N = 'N'
self.RESULT_N1 = 'N1'
self.RESULT_N2 = 'N2'
self.RESULT_NA = 'NA'
self.TRANS_MAP = {
......@@ -610,19 +608,19 @@ class Comparison:
return self.RESULT_N
return self.se_date_compare_2(input_str, ocr_date_set.pop(), three_month=True)
def se_date_compare_pos(self, input_str, ocr_str, **kwargs):
try:
ocr_date = datetime.strptime(ocr_str, "%Y-%m-%d").date()
today = datetime.now().date()
if ocr_date < today:
return self.RESULT_N1
elif today <= ocr_date <= (today + timedelta(days=8)):
return self.RESULT_N2
elif ocr_date > (today + timedelta(days=8)):
return self.RESULT_Y
except Exception as e:
return self.RESULT_N
return self.RESULT_N
# def se_date_compare_pos(self, input_str, ocr_str, **kwargs):
# try:
# ocr_date = datetime.strptime(ocr_str, "%Y-%m-%d").date()
# today = datetime.now().date()
# if ocr_date < today:
# return self.RESULT_N1
# elif today <= ocr_date <= (today + timedelta(days=8)):
# return self.RESULT_N2
# elif ocr_date > (today + timedelta(days=8)):
# return self.RESULT_Y
# except Exception as e:
# return self.RESULT_N
# return self.RESULT_N
cp = Comparison()
......
src/pos/pos.py deleted 100644 → 0
from apps.doc import consts
import json
from common.tools.des import decode_des
from pos import pos_consts
from settings import conf
from pos_compare import SECompare, CACompare
empty_str = ''
des_key = conf.CMS_DES_KEY
empty_error_type = 1000
def ca_compare_process(compare_info, ocr_res_dict):
# individualCusInfo
# corporateCusInfo
# usedCarInfo
compare_result = []
total_fields = 0
failed_count = 0
for info_key, info_value in compare_info.items():
if info_key == 'individualCusInfo':
for idx, license_list in info_value.items():
for license_dict in license_list:
for license_en, field_list in license_dict.items():
result_field_list, field_img_path_dict = CACompare.ca_compare_license(license_en, ocr_res_dict, field_list)
for name, value, result, ocr_str, img_path, error_type, reason in result_field_list:
total_fields += 1
if result == consts.RESULT_N:
failed_count += 1
compare_result.append(
{
consts.HEAD_LIST[0]: info_key,
consts.HEAD_LIST[1]: idx,
consts.HEAD_LIST[2]: license_en,
consts.HEAD_LIST[3]: name,
consts.HEAD_LIST[4]: value,
consts.HEAD_LIST[5]: ocr_str,
consts.HEAD_LIST[6]: result,
consts.HEAD_LIST[7]: field_img_path_dict.get(name, empty_str),
consts.HEAD_LIST[8]: img_path,
consts.HEAD_LIST[9]: error_type,
"comments": reason if result == consts.RESULT_N else ''
}
)
else:
for license_en, field_list in info_value.items():
result_field_list, field_img_path_dict = CACompare.ca_compare_license(license_en, ocr_res_dict, field_list)
for name, value, result, ocr_str, img_path, error_type, reason in result_field_list:
total_fields += 1
if result == consts.RESULT_N:
failed_count += 1
compare_result.append(
{
consts.HEAD_LIST[0]: info_key,
consts.HEAD_LIST[1]: "0",
consts.HEAD_LIST[2]: license_en,
consts.HEAD_LIST[3]: name,
consts.HEAD_LIST[4]: value,
consts.HEAD_LIST[5]: ocr_str,
consts.HEAD_LIST[6]: result,
consts.HEAD_LIST[7]: field_img_path_dict.get(name, empty_str),
consts.HEAD_LIST[8]: img_path,
consts.HEAD_LIST[9]: error_type,
"comments": reason if result == consts.RESULT_N else ''
}
)
return compare_result, total_fields, failed_count
def se_compare_process(compare_info, ocr_res_dict, is_gsyh, is_auto, id_res_list):
# individualCusInfo
# corporateCusInfo
# vehicleInfo
# bankInfo
compare_result = []
total_fields = 0
failed_count = 0
successful_at_this_level = True
failure_reason = {}
cn_reason_list = []
rpa_failure_reason = {}
for info_key, info_value in compare_info.items():
if info_key in ['individualCusInfo', 'applicantInformation']:
for idx, license_list in info_value.items():
for license_dict in license_list:
for license_en, field_list in license_dict.items():
strip_list = []
for a, b in field_list:
if isinstance(b, str):
strip_list.append((a, b.strip()))
elif isinstance(b, list):
c = []
for i in b:
if isinstance(i, str):
c.append(i.strip())
else:
c.append(i)
strip_list.append((a, c))
else:
strip_list.append((a, b))
failure_field = []
# 身份证先SE正反面，后CA正反面
if license_en == consts.ID_EN:
result_field_list, no_ocr_result, field_img_path_dict = SECompare.se_compare_license_id(
license_en, id_res_list, strip_list)
else:
result_field_list, no_ocr_result, field_img_path_dict = SECompare.se_compare_license(
license_en, ocr_res_dict, strip_list)
for name, value, result, ocr_str, img_path, error_type, cn_reason in result_field_list:
if license_en not in consts.SKIP_CARD or not no_ocr_result:
total_fields += 1
if result == consts.RESULT_N:
failed_count += 1
successful_at_this_level = False
failure_field.append(name)
cn_reason_list.append(cn_reason)
rpa_failure_reason.setdefault(cn_reason, []).append(value)
compare_result.append(
{
consts.HEAD_LIST[0]: info_key,
consts.HEAD_LIST[1]: idx,
consts.HEAD_LIST[2]: license_en,
consts.HEAD_LIST[3]: name,
consts.HEAD_LIST[4]: value,
consts.HEAD_LIST[5]: ocr_str,
consts.HEAD_LIST[6]: result,
consts.HEAD_LIST[7]: field_img_path_dict.get(name, empty_str),
consts.HEAD_LIST[8]: img_path,
consts.HEAD_LIST[9]: error_type,
"comments": cn_reason if result == consts.RESULT_N else ''
}
)
if len(failure_field) > 0:
failure_reason.setdefault(info_key, []).append(';'.join(failure_field))
else:
for license_en, field_list in info_value.items():
strip_list = []
for a, b in field_list:
if isinstance(b, str):
strip_list.append((a, b.strip()))
elif isinstance(b, list):
c = []
for i in b:
if isinstance(i, str):
c.append(i.strip())
else:
c.append(i)
strip_list.append((a, c))
else:
strip_list.append((a, b))
failure_field = []
if license_en == consts.MVC34_EN:
result_field_list, field_img_path_dict = SECompare.se_mvc34_compare(license_en, ocr_res_dict, strip_list)
elif license_en in [consts.HIL_CONTRACT_1_EN, consts.HIL_CONTRACT_2_EN, consts.HIL_CONTRACT_3_EN, consts.AFC_CONTRACT_EN]:
result_field_list, field_img_path_dict = SECompare.se_contract_compare(license_en, ocr_res_dict, strip_list, is_gsyh)
elif license_en == consts.BS_EN:
result_field_list, field_img_path_dict = SECompare.se_bs_compare(license_en, ocr_res_dict, strip_list, is_auto)
else:
result_field_list, _, field_img_path_dict = SECompare.se_compare_license(license_en, ocr_res_dict, strip_list)
for name, value, result, ocr_str, img_path, error_type, cn_reason in result_field_list:
total_fields += 1
if result == consts.RESULT_N:
# if license_en != consts.MVI_EN or name != consts.SE_NEW_ADD_FIELD[9]:
successful_at_this_level = False
failed_count += 1
failure_field.append(name)
if isinstance(cn_reason, str):
cn_reason_list.append(cn_reason)
rpa_failure_reason.setdefault(cn_reason, []).append(value)
elif isinstance(cn_reason, list):
cn_reason_list.extend(cn_reason)
rpa_failure_reason.setdefault('、'.join(cn_reason), []).append(value)
compare_result.append(
{
consts.HEAD_LIST[0]: info_key,
consts.HEAD_LIST[1]: "0",
consts.HEAD_LIST[2]: license_en,
consts.HEAD_LIST[3]: name,
consts.HEAD_LIST[4]: value,
consts.HEAD_LIST[5]: ocr_str,
consts.HEAD_LIST[6]: result,
consts.HEAD_LIST[7]: field_img_path_dict.get(name, empty_str),
consts.HEAD_LIST[8]: img_path,
consts.HEAD_LIST[9]: error_type,
"comments": cn_reason if result == consts.RESULT_N else ''
}
)
if len(failure_field) > 0:
failure_reason.setdefault(info_key, []).append(';'.join(failure_field))
if failed_count == 0:
failure_reason_str = ''
cn_failure_reason_str = ''
bs_failure_reason_str = ''
else:
reason_list = []
for key, value in failure_reason.items():
if len(value) > 0:
value_str = json.dumps(value)
reason_list.append('{0}: {1}'.format(key, value_str))
failure_reason_str = '、'.join(reason_list)
tmp_set = set()
last_cn_reason_list = []
bs_cn_reason_list = []
for i in cn_reason_list:
if i in tmp_set:
continue
# elif i in consts.BS_REASON:
# tmp_set.add(i)
# bs_cn_reason_list.append(i)
else:
tmp_set.add(i)
last_cn_reason_list.append(i)
cn_failure_reason_str = '\n'.join(last_cn_reason_list)
bs_failure_reason_str = '\n'.join(bs_cn_reason_list)
return compare_result, total_fields, failed_count, successful_at_this_level, failure_reason_str, cn_failure_reason_str, bs_failure_reason_str, rpa_failure_reason
class CorporateCusInfo:
def __init__(self):
self.customerType = ''
self.companyName = ''
self.firstIdType = ''
self.firstIdNo = ''
self.businessLicenseNo = ''
self.organizationCreditCode = ''
self.taxRegistrationCertificateNo = ''
self.establishmentDate = ''
self.incorporationDate = ''
self.businessLicenseDueDate = ''
self.legalRepName = ''
self.organizationType = ''
self.fleetCustomer = False
self.beneficialOwnerName = ''
self.beneficialOwnerIdType = ''
self.beneficialOwnerIdNo = ''
self.beneficialOwnerIdExpiryDate = ''
class VehicleInfo:
def __int__(self):
self.vehicleStatus = ''
self.vehicleTransactionAmount = 0.0
self.vinNo = ''
self.dealer = ''
self.showRoom = ''
self.agencyDealer = ''
self.option = 0.0
self.msrp = 0.0
self.totalAmount = 0.0
class InsuranceInfo:
def __init__(self):
self.insuredAmount = 0.0
self.insuranceType = ''
self.startDate = ''
self.endDate = ''
class BankInfo:
def __init__(self):
self.bankName = ''
self.branchName = ''
self.applicantType = ''
self.accountHolderName = ''
self.accountNo = ''
self.accountNo = 'bankVerificationStatus'
self.isAllDocUploaded = False
class QuotationtInfo:
def __init__(self):
self.totalLoanAmount = 0.0
self.loanTerm = 0
self.vehiclePrincipal = 0.0
self.associatedServicePrincipal = 0.0
self.mortgageType = ''
self.associatedServiceInfo = []
self.monthlyPaymentInfo = []
class CompareInfoResult:
def __init__(self):
self.compare_info = None
self.is_cdfl_bo = False # 车贷分离，主借
self.is_cdfl_co = False # 车贷分离，共借
self.is_gsyh = False # 是否工商银行
class CompareInfo:
def __init__(self, pos_content):
self.pos_content = pos_content
self.compare_info = {}
self.main_role_info = {}
self.company_info_list = []
self.company_info = None
self.vehicleTransactionAmount = 0.0
self.customer_name = ''
self.id_num = None
self.application_id = pos_content.get('applicationId', '')
self.application_version = pos_content.get('applicationVersion', 0)
self.application_entity = pos_content.get('applicationEntity', '')
self.is_auto_settlement = pos_content.get('isAutoSettlement', '')
self.first_submmison_date = pos_content.get('firstSubmmisonDate', '')
self.product_group_name = pos_content.get('productGroup', '')
self.account_holder_name = ''
self.is_cdfl_bo = False # 车贷分离，主借
self.is_cdfl_co = False # 车贷分离，共借
self.is_gsyh = False # 是否工商银行
self.vin_no = ''
self.bank_name = ''
self.account_no = ''
def get(self, ignore_bank=False):
individual_cus_info = self.pos_content.get('individualCusInfo', [])
corporate_cus_info = self.pos_content.get('corporateCusInfo', {})
vehicle_info = self.pos_content.get('vehicleInfo', {})
insurance_info = self.pos_content.get('insuranceInfo', {})
bank_info = self.pos_content.get('bankInfo', {})
quotationt_info = self.pos_content.get('quotationtInfo', {})
self.getApplicantInformation(individual_cus_info, corporate_cus_info)
self.getVehicleInfo(vehicle_info)
self.getBankInfo(bank_info, ignore_bank)
self.getContract(quotationt_info)
self.getOther(quotationt_info, insurance_info)
compare_result = CompareInfoResult()
compare_result.compare_info = self.compare_info
compare_result.is_cdfl_co = self.is_cdfl_co
compare_result.is_cdfl_bo = self.is_cdfl_bo
compare_result.is_gsyh = self.is_gsyh
return compare_result
def getApplicantInformation(self, individual_cus_info, corporate_cus_info):
individual_info_dict = {}
for individual_info in individual_cus_info:
all_id_num = []
license_dict = {}
customer_name = individual_info.get('customerName', '').strip()
self.customer_name = customer_name
legal_name = corporate_cus_info.get('legalRepName', '')
establishment_date = corporate_cus_info.get('establishmentDate', '')
# 车贷分离判断
is_corporate = individual_info.get('selfEmployedSubType', '') == 'Corporate'
if individual_info['applicantType'] == consts.APPLICANT_TYPE_ORDER[1] and is_corporate:
self.is_cdfl_co = True
if individual_info['applicantType'] == consts.APPLICANT_TYPE_ORDER[0] and not is_corporate:
self.is_cdfl_bo = True
if individual_info.get('idType') in consts.SE_CMS_FIRST_ID_FIELD_MAPPING:
license_en, is_prc = consts.SE_CMS_FIRST_ID_FIELD_MAPPING[individual_info['idType']]
# ['customerName', 'idNum', 'dateOfBirth', 'idExpiryDate', 'hukouProvince']
id_num = decode_des(individual_info.get('idNum', ''), des_key)
self.id_num = id_num
field_input = [('customerName', customer_name),
('idNum', id_num),
('idExpiryDate', individual_info.get('idExpiryDate', ''))]
# if is_prc:
# field_input.append(('hukouProvince', province))
# field_input.append(('真伪', consts.IC_RES_MAPPING.get(1)))
license_dict[license_en] = field_input
all_id_num.append(id_num)
if len(all_id_num) > 0:
self.main_role_info.setdefault(individual_info['applicantType'], []).append(
(customer_name, '、'.join(all_id_num), all_id_num[0])
)
if len(license_dict) > 0:
individual_info_dict.setdefault(individual_info['applicantType'], []).append(license_dict)
self.compare_info['applicantInformation'] = individual_info_dict
def getVehicleInfo(self, vehicleInfo):
main_name = main_id_all = main_id = ''
for applicant_type in consts.APPLICANT_TYPE_ORDER:
if applicant_type in self.main_role_info:
main_name, main_id_all, main_id = self.main_role_info[applicant_type][0]
# hmh_name, _, hmh_id = main_role_info[applicant_type][0]
break
# co_name = co_id = bo_name = bo_id = ''
# if is_cdfl:
# co_name, _, co_id = main_role_info[consts.APPLICANT_TYPE_ORDER[1]][0]
# bo_name, _, bo_id = main_role_info[consts.APPLICANT_TYPE_ORDER[0]][0]
co_name = co_id = bo_name = bo_id = ''
is_cdfl = self.is_cdfl_bo and self.is_cdfl_co
if is_cdfl:
if len(self.main_role_info.get(consts.APPLICANT_TYPE_ORDER[1], [])) > 0:
co_name, _, co_id = self.main_role_info[consts.APPLICANT_TYPE_ORDER[1]][0]
else:
co_name = co_id = ''
if len(self.main_role_info.get(consts.APPLICANT_TYPE_ORDER[0], [])) > 0:
bo_name, _, bo_id = self.main_role_info[consts.APPLICANT_TYPE_ORDER[0]][0]
else:
bo_name = bo_id = ''
# dda_name_list = []
# dda_num_list = []
if len(self.company_info_list) > 0:
# tmp_idx = 1
self.company_info = self.company_info_list[0]
else:
# tmp_idx = 0
self.company_info = None
vehicle_info = {}
vehicle_field_input = []
vehicle_status = vehicleInfo.get('vehicleStatus', '')
first_submission_date = self.first_submmison_date
vin_no = vehicleInfo.get('vinNo', '')
self.vin_no = vin_no
# amount = str(vehicleInfo.get('vehiclePrice', '0.0'))
amount = str(vehicleInfo.get('vehicleTransactionAmount', '0.0'))
self.vehicleTransactionAmount = amount
dealer_name_list = vehicleInfo.get('dealer', '').split()
dealer_name = '' if len(dealer_name_list) == 0 else dealer_name_list[-1]
self.dealer_name = dealer_name
issuer_dealer = vehicleInfo.get('fapiaoIssuerDealer', '').strip()
# 新车发票----------------------------------------------------------------------------------------------------------
if vehicle_status == 'New':
vehicle_field_input.append(('vinNo', vin_no))
vehicle_field_input.append(('dealer', dealer_name if len(issuer_dealer) == 0 else issuer_dealer))
vehicle_field_input.append(('vehicleTransactionAmount', amount))
if isinstance(self.company_info, tuple):
vehicle_field_input.append(
(consts.SE_NEW_ADD_FIELD[0], co_name if self.is_cdfl else self.company_info[0])) # 车贷分离
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[1], self.co_id if self.is_cdfl else self.company_info[1])) # 车贷分离
else:
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[0], co_name if is_cdfl else main_name)) # 车贷分离
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[1], co_id if is_cdfl else main_id)) # 车贷分离
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[2], first_submission_date))
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[3], consts.SE_STAMP_VALUE))
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[4], consts.SE_FPL_VALUE))
bhsj = float(amount) / 1.13
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[5], consts.SPLIT_STR.join([
# format(bhsj, '.2f'),
# format(float(amount) - bhsj, '.2f'),
# consts.RESULT_Y
# ])))
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[7], format(bhsj, '.2f')))
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[8], format(float(amount) - bhsj, '.2f')))
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[9], consts.RESULT_Y))
vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[6], consts.SE_LAYOUT_VALUE))
vehicle_field_input.append(("firstSubmmisonDate", self.first_submmison_date))
vehicle_field_input.append(("productGroupName", self.product_group_name))
if self.product_group_name.find('官方认证二手车') == -1 and self.product_group_name.find('非官方认证二手车') == -1:
vehicle_info[consts.MVI_EN] = vehicle_field_input
# 二手车发票、交易凭证、绿本------------------------------------------------------------------------------------------
# else:
# gb_field_input = [
# ('vinNo', vin_no),
# ]
# gb34_field_input = []
# jypz_field_input = []
# vehicle_field_input.append(('vinNo', vin_no))
# vehicle_field_input.append(('vehicleTransactionAmount', amount))
# if isinstance(company_info, tuple):
# vehicle_field_input.append(
# (consts.SE_NEW_ADD_FIELD[0], co_name if is_cdfl else company_info[0])) # 车贷分离
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[1], co_id if is_cdfl else company_info[1])) # 车贷分离
# jypz_field_input.append((consts.SE_NEW_ADD_FIELD[0], co_name if is_cdfl else company_info[0])) # 车贷分离
# jypz_field_input.append((consts.SE_NEW_ADD_FIELD[1], co_id if is_cdfl else company_info[1])) # 车贷分离
# gb34_field_input.append((consts.SE_GB_USED_FIELD[0], co_name if is_cdfl else company_info[0])) # 车贷分离
# gb34_field_input.append((consts.SE_GB_USED_FIELD[1], co_id if is_cdfl else company_info[1])) # 车贷分离
# else:
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[0], co_name if is_cdfl else main_name)) # 车贷分离
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[1], co_id if is_cdfl else main_id_all)) # 车贷分离
# jypz_field_input.append((consts.SE_NEW_ADD_FIELD[0], co_name if is_cdfl else main_name)) # 车贷分离
# jypz_field_input.append((consts.SE_NEW_ADD_FIELD[1], co_id if is_cdfl else main_id_all)) # 车贷分离
# gb34_field_input.append((consts.SE_GB_USED_FIELD[0], co_name if is_cdfl else main_name)) # 车贷分离
# gb34_field_input.append((consts.SE_GB_USED_FIELD[1], co_id if is_cdfl else main_id_all)) # 车贷分离
# gb34_field_input.append((consts.SE_GB_USED_FIELD[2], first_submission_date))
# vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[2], first_submission_date))
# # vehicle_field_input.append((consts.SE_NEW_ADD_FIELD[3], consts.SE_STAMP_VALUE))
# jypz_field_input.append(('dealerName', dealer_name))
# jypz_field_input.append(('vinNo', vin_no))
# jypz_field_input.append(('vehicleTransactionAmount', amount))
# jypz_field_input.append((consts.SE_GB_USED_FIELD[2], first_submission_date))
# if fp_group.find('Non OCU Product Group') != -1:
# jypz_field_input.append(('type', consts.JYPZ_TYPE_1))
# elif fp_group.find('OCU Product Group') != -1:
# jypz_field_input.append(('type', consts.JYPZ_TYPE_2))
#
# vehicle_info[consts.MVC_EN] = gb_field_input
# vehicle_info[consts.MVC34_EN] = gb34_field_input
# if not detect_list[0]:
# vehicle_info[consts.UCI_EN] = vehicle_field_input
# if not detect_list[1]:
# vehicle_info[consts.JYPZ_EN] = jypz_field_input
# if detect_list[0] and detect_list[1]:
# vehicle_info[consts.UCI_EN] = vehicle_field_input
self.compare_info['vehicleInfo'] = vehicle_info
def getBankInfo(self, bankInfo, ignore_bank):
bank_info = {}
bank_name = bankInfo.get('bankName', '')
self.bank_name = bank_name
account_no = decode_des(bankInfo.get('accountNo', ''), des_key)
self.account_no = account_no
account_holder_name = bankInfo.get('accountHolderName', '')
self.account_holder_name = account_holder_name
self.is_gsyh = True if '工商' in bank_name else False
bank_verification_status = bankInfo.get('bankVerificationStatus', '')
if isinstance(self.company_info, tuple) and self.company_info[0] == account_holder_name:
pass
elif not ignore_bank:
bank_field_input = [
('accountNo', account_no),
('bankName', bank_name),
('type', consts.BC_TYPE_VALUE) # 是否为借记卡
]
if bank_verification_status != 'PASS':
# pos 银行卡校验失败后需要走ocr校验
if bank_verification_status == 'FAIL' or bank_verification_status == 'N/A':
bank_field_input.append(('bankVerificationStatus', ''))
bank_info[consts.BC_EN] = bank_field_input
# DDA------------------------------------------------------------------------------------------------------------
# if is_gsyh or not detect_list[-1]:
# dda_field_input = [
# ('applicationId(1)', last_obj.application_id),
# ('applicationId(2)', last_obj.application_id),
# ('bankName', bank_name),
# ('companyName', consts.HIL_COMPANY_NAME if application_entity in consts.HIL_SET else consts.AFC_COMPANY_NAME),
# ('customerName', dda_name),
# ('idNum', dda_num),
# ('accountHolderName', account_holder_name),
# ('accountNo', account_no),
# ]
# bank_info[consts.DDA_EN] = dda_field_input
if len(bank_info) > 0:
self.compare_info['bankInfo'] = bank_info
def getContract(self, quotationtInfo):
# ASP -------------------------------------------------------------------------------------------------------
asp_list = []
is_asp = False
insurance_price = None
gzs_price = None
have_other_asp = False
fin_total = 0
if str(quotationtInfo.get('associatedServicePrincipal', '0.00')) != '0.00':
is_asp = True
# for asp_info in cms_info.get('associatedServices', []):
for asp_info in quotationtInfo.get('associatedServiceInfo', []):
tmp_asp_name = asp_info.get('service')
if isinstance(tmp_asp_name, str) and len(tmp_asp_name) > 0:
asp_list.append(
(
tmp_asp_name,
asp_info.get('amount', '0.00'),
asp_info.get('financedAmount', '0.00')
)
)
fin_total += float(asp_info.get('financedAmount', '0.00'))
# 购置税
if tmp_asp_name.find(consts.GZS_NAME) != -1:
gzs_price = asp_info.get('amount', '0.00')
# 保单费合计
elif tmp_asp_name.find('机动车辆保险') != -1:
insurance_price = asp_info.get('amount', '0.00')
else:
have_other_asp = True
asp_list.append(
(
consts.ASP_SUM_NAME,
'',
# fin_total,
format(fin_total, '.2f'),
)
)
# CMS Vehicle Price / 1.13 * 10 %
if isinstance(gzs_price, str):
try:
tmp_gzs_list = [float(self.vehicleTransactionAmount) * 0.1 / 1.13, float(gzs_price)]
except Exception as e:
tmp_gzs_list = [self.vehicleTransactionAmount, gzs_price]
else:
tmp_gzs_list = [self.vehicleTransactionAmount, ]
main_name = main_id_all = main_id = ''
for applicant_type in consts.APPLICANT_TYPE_ORDER:
if applicant_type in self.main_role_info:
main_name, main_id_all, main_id = self.main_role_info[applicant_type][0]
# hmh_name, _, hmh_id = main_role_info[applicant_type][0]
break
contract_info = {}
if self.application_entity in consts.HIL_SET:
# HIL合同 售后回租合同 --------------------------------------------------------------------------------------
hil_contract_1_input = [
# (pos_consts.SE_HIL_CON_1_FIELD[0], [full_no] if online_sign else full_no),
# (pos_consts.SE_HIL_CON_1_FIELD[1], full_no),
# (pos_consts.SE_HIL_CON_1_FIELD[2], self.vin_no),
# (pos_consts.SE_HIL_CON_1_FIELD[3], self.dealer_name),
# (pos_consts.SE_HIL_CON_1_FIELD[4], self.vehicleTransactionAmount),
# (pos_consts.SE_HIL_CON_1_FIELD[5],
# str(cms_info.get('financialInformation', {}).get('originationPrincipal', '0.0'))),
# (pos_consts.SE_HIL_CON_1_FIELD[6], str(quotationtInfo.get('loanTerm', '0'))),
# (pos_consts.SE_HIL_CON_1_FIELD[7], schedule_list_str),
# (pos_consts.SE_HIL_CON_1_FIELD[11], self.account_no),
# (pos_consts.SE_HIL_CON_1_FIELD[12], self.account_holder_name),
# (pos_consts.SE_HIL_CON_1_FIELD[13], self.bank_name),
]
# if is_asp:
# # asp各项
# hil_contract_1_input.append((consts.SE_HIL_CON_1_FIELD[8], asp_list))
# # 购置税校验
# if isinstance(gzs_price, str):
# hil_contract_1_input.append(
# (consts.SE_HIL_CON_1_FIELD[9], tmp_gzs_list))
# # 非购置税非车辆保险的其他asp
# if have_other_asp:
# hil_contract_1_input.append((consts.SE_HIL_CON_1_FIELD[15], 'N'))
#
# if isinstance(self.company_info, tuple):
# if self.is_cdfl:
# hil_contract_1_input.append((consts.SE_HIL_CON_1_FIELD[14], self.company_info[2]))
# else:
# hil_contract_1_input.append((consts.SE_HIL_CON_1_FIELD[10], self.company_info[2]))
# contract_info[consts.HIL_CONTRACT_1_EN] = hil_contract_1_input
# HIL合同 车辆租赁抵押合同 --------------------------------------------------------------------------------------
hil_contract_2_input = [
(pos_consts.SE_HIL_CON_2_FIELD[0], self.application_id),
(pos_consts.SE_HIL_CON_2_FIELD[1], self.vin_no),
('文档', '文档'),
]
contract_info[consts.HIL_CONTRACT_2_EN] = hil_contract_2_input
self.compare_info['contract'] = contract_info
else:
# AFC合同------------------------------------------------------------------------------------------------------
vehicle_principal_str = str(quotationtInfo.get('vehiclePrincipal', '0.0'))
afc_contract_input = []
# if is_asp:
# afc_contract_input = [
# (consts.SE_AFC_CON_FIELD[0], full_no),
# ]
# else:
# afc_contract_input = [
# (consts.SE_AFC_CON_FIELD[23], full_no),
# ]
# afc_contract_input.extend([
# (consts.SE_AFC_CON_FIELD[1], self.vehicleTransactionAmount),
# (consts.SE_AFC_CON_FIELD[2], self.vin_no),
# (consts.SE_AFC_CON_FIELD[3],
# str(cms_info.get('financialInformation', {}).get('originationPrincipal', '0.0'))),
# (consts.SE_AFC_CON_FIELD[4], str(cms_info.get('terms', '0'))),
# (consts.SE_AFC_CON_FIELD[5], vehicle_principal_str),
# (consts.SE_AFC_CON_FIELD[6],
# str(quotationtInfo.get('associatedServicePrincipal', '0.0'))),
# (consts.SE_AFC_CON_FIELD[7], self.vehicleTransactionAmount),
# (consts.SE_AFC_CON_FIELD[8], self.vin_no),
# (consts.SE_AFC_CON_FIELD[9], self.dealer_name),
# (consts.SE_AFC_CON_FIELD[10],
# str(cms_info.get('financialInformation', {}).get('originationPrincipal', '0.0'))),
# (consts.SE_AFC_CON_FIELD[11], vehicle_principal_str),
# (consts.SE_AFC_CON_FIELD[12],
# str(cms_info.get('financialInformation', {}).get('associatedServicePrincipal', '0.0'))),
# (consts.SE_AFC_CON_FIELD[13], str(cms_info.get('terms', '0'))),
# (consts.SE_AFC_CON_FIELD[14], self.account_no),
# (consts.SE_AFC_CON_FIELD[15], self.account_holder_name),
# (consts.SE_AFC_CON_FIELD[16], self.bank_name),
# (consts.SE_AFC_CON_FIELD[17], schedule_list_str),
# ])
# if is_asp:
# afc_contract_input.append((consts.SE_AFC_CON_FIELD[20], asp_list))
# afc_contract_input.append((consts.SE_AFC_CON_FIELD[22], asp_list))
# # 购置税校验
# if isinstance(gzs_price, str):
# afc_contract_input.append(
# (consts.SE_AFC_CON_FIELD[21], tmp_gzs_list))
# # 非购置税非车辆保险的其他asp
# if have_other_asp:
# afc_contract_input.append((consts.SE_AFC_CON_FIELD[24], 'N'))
# else:
# afc_contract_input.pop(5)
# afc_contract_input.pop(5)
# afc_contract_input.pop(9)
# afc_contract_input.pop(9)
# contract_info[consts.AFC_CONTRACT_EN] = afc_contract_input
self.compare_info['contract'] = contract_info
def getOther(self, quotationtInfo, insurance_info):
main_name = main_id_all = main_id = ''
for applicant_type in consts.APPLICANT_TYPE_ORDER:
if applicant_type in self.main_role_info:
main_name, main_id_all, main_id = self.main_role_info[applicant_type][0]
# hmh_name, _, hmh_id = main_role_info[applicant_type][0]
break
other_info = {}
if quotationtInfo.get('mortgageType', '') == 'MOTGF':
hmh_field_input = [
(consts.SE_HMH_FIELD[0], self.customer_name),
(consts.SE_HMH_FIELD[1], self.id_num),
(consts.SE_HMH_FIELD[2], self.application_id),
(consts.SE_HMH_FIELD[3], self.application_entity),
(consts.SE_HMH_FIELD[4], consts.SE_STAMP_VALUE),
('文档', '')
]
other_info[consts.HMH_EN] = hmh_field_input
# 保险
if insurance_info.get('insuranceType') != 'Waive Insurance' and insurance_info.get('insuranceType') != 'Insurance After Check':
if insurance_info.get('insuranceType') == 'Comprehensive Insurance':
other_info[consts.BD_EN] = [
(consts.SE_BD_FIELD[0], self.customer_name),
(consts.SE_BD_FIELD[1], self.id_num)
(consts.SE_BD_FIELD[2], self.vin_no)
('文档', '')
]
if len(other_info) > 0:
self.compare_info['other'] = other_info
class PosCompare:
@staticmethod
def ca_compare(pos_content, ocr_res_dict):
compare_info_result = CompareInfo(pos_content).get()
compare_result, total_fields, failed_count = ca_compare_process(compare_info_result, ocr_res_dict)
return compare_result
@staticmethod
def se_compare(pos_content, ocr_res_dict, id_res_list=None):
compare_info_result = CompareInfo(pos_content).get()
compare_result, total_fields, failed_count, successful_at_this_level, failure_reason_str, \
cn_failure_reason_str, bs_failure_reason_str, rpa_failure_reason = se_compare_process(
compare_info_result.compare_info, ocr_res_dict, compare_info_result.is_gsyh, False, id_res_list)
return compare_result
src/pos/pos_compare.py deleted 100644 → 0
from common.tools.comparison import cp
import os
from pos import pos_consts
from apps.doc import consts
import json
from apps.doc.named_enum import RequestTeam, RequestTrigger, ProcessName, ErrorType
import time
empty_str = ''
empty_error_type = 1000
class SECompare:
@staticmethod
def se_mvc34_compare(license_en, ocr_res_dict, field_list):
ocr_field, compare_logic, _ = pos_consts.SE_COMPARE_FIELD[license_en]
ocr_res_str = ocr_res_dict.get(ocr_field)
is_find = False
result_field_list = []
field_img_path_dict = dict()
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
length = len(ocr_res_list)
page34_date_dict = dict()
first_res = None
for res_idx in range(length - 1, -1, -1):
if consts.TRANSFER_DATE in ocr_res_list[res_idx]:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY_2, '')
section_img_path = ocr_res_list[res_idx].get(consts.SECTION_IMG_PATH_KEY_2, '')
for idx, transfer_date in enumerate(ocr_res_list[res_idx].get(consts.TRANSFER_DATE, [])):
try:
transfer_name = ocr_res_list[res_idx].get(consts.TRANSFER_NAME, [])[idx]
except Exception as e:
transfer_name = empty_str
try:
transfer_num = ocr_res_list[res_idx].get(consts.TRANSFER_NUM, [])[idx]
except Exception as e:
transfer_num = empty_str
try:
position_info_date = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY_2, dict()).get(
consts.TRANSFER_DATE, [])[idx]
except Exception as e:
position_info_date = {}
try:
position_info_name = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY_2, dict()).get(
consts.TRANSFER_NAME, [])[idx]
except Exception as e:
position_info_name = {}
try:
position_info_num = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY_2, dict()).get(
consts.TRANSFER_NUM, [])[idx]
except Exception as e:
position_info_num = {}
core_info = {
consts.TRANSFER_NAME: transfer_name,
consts.TRANSFER_NUM: transfer_num,
consts.TRANSFER_DATE: transfer_date,
consts.IMG_PATH_KEY_2: img_path,
consts.SECTION_IMG_PATH_KEY_2: section_img_path,
consts.ALL_POSITION_KEY: {
consts.TRANSFER_NAME: position_info_name,
consts.TRANSFER_NUM: position_info_num,
consts.TRANSFER_DATE: position_info_date,
},
}
page34_date_dict.setdefault(transfer_date, []).append(core_info)
if first_res is None:
first_res = core_info
max_date = None
for date_tmp in page34_date_dict.keys():
try:
max_date_part = time.strptime(date_tmp, "%Y-%m-%d")
except Exception as e:
pass
else:
if max_date is None or max_date_part > max_date:
max_date = max_date_part
if max_date is not None or first_res is not None:
is_find = True
ocr_res = first_res if max_date is None else page34_date_dict[time.strftime('%Y-%m-%d', max_date)][0]
failed_field = []
base_img_path = ocr_res.get(consts.IMG_PATH_KEY_2, '')
for name, value in field_list:
ocr_str = ocr_res.get(compare_logic[name][0])
if not isinstance(ocr_str, str):
result = consts.RESULT_N
ocr_str = empty_str
else:
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
img_path = base_img_path if result == consts.RESULT_N else empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
result_field_list.append(
(name, value, result, ocr_str, img_path, error_type, compare_logic[name][3]))
if result == consts.RESULT_N:
failed_field.append(name)
# section_img_path = ocr_res.get(consts.SECTION_IMG_PATH_KEY_2, '')
# if len(failed_field) > 0 and os.path.exists(section_img_path):
# info = ocr_res.get(consts.ALL_POSITION_KEY, {})
# try:
# last_img = img_process(section_img_path, {}, 0)
# except Exception as e:
# for field in failed_field:
# field_img_path_dict[field] = base_img_path
# else:
# pre, suf = os.path.splitext(section_img_path)
# for field in failed_field:
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3],
# :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = base_img_path
# except Exception as e:
# field_img_path_dict[field] = base_img_path
if not is_find:
for name, value in field_list:
result_field_list.append((name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value,
'{0}未找到'.format(license_en)))
return result_field_list, field_img_path_dict
@staticmethod
def se_contract_compare(license_en, ocr_res_dict, strip_list, is_gsyh):
ocr_field, compare_logic, _ = pos_consts.SE_COMPARE_FIELD[license_en]
ocr_res_str = ocr_res_dict.get(ocr_field)
result_field_list = []
field_img_path_dict = dict()
ocr_res = dict()
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
ocr_res = ocr_res_list.pop()
for name, value in strip_list:
# 购置税校验
# if name == consts.SE_AFC_CON_FIELD[21]:
# if len(value) == 3:
# reason = []
# gzs_verify = value[1] >= value[2]
# if gzs_verify:
# if value[0] == consts.GZS_STATUS[0]:
# reason.append(consts.GZS_REASON_1)
# result = consts.RESULT_N
# else:
# result = consts.RESULT_Y
# else:
# if value[0] == consts.GZS_STATUS[0]:
# reason.append(consts.GZS_REASON_1)
# result = consts.RESULT_N
# reason.append(consts.GZS_REASON_2)
# else:
# result = consts.RESULT_N
# reason = consts.GZS_REASON_1
# ocr_str = empty_str
# else:
if name == consts.SE_HIL_CON_1_FIELD[9] or name == consts.SE_HIL_CON_1_FIELD[15] or \
name == consts.SE_AFC_CON_FIELD[21] or name == consts.SE_AFC_CON_FIELD[24]:
ocr_str_or_list = ''
else:
ocr_str_or_list = ocr_res.get(compare_logic[name][0])
# 招商银行特殊
# if ocr_str_or_list is None and license_en == consts.AFC_CONTRACT_EN \
# and is_gsyh is True and name in consts.CON_BANK_FIELD:
# result = consts.RESULT_Y
# ocr_str = empty_str
# reason = compare_logic[name][3]
# 见证人日期
if name == consts.SE_AFC_CON_FIELD[19]:
if not isinstance(ocr_str_or_list, str) or len(ocr_str_or_list) == 0:
result = consts.RESULT_N
ocr_str = empty_str
else:
is_find_date = False
all_date_list = [ocr_str_or_list]
for date_name in consts.AFC_HT_DATE_FIELDS:
all_date_list.append(ocr_res.get(compare_logic[date_name][0], ''))
if not is_find_date and ocr_str_or_list == ocr_res.get(compare_logic[date_name][0], ''):
is_find_date = True
result = consts.RESULT_Y if is_find_date else consts.RESULT_N
ocr_str = json.dumps(all_date_list, ensure_ascii=False)
reason = compare_logic[name][3]
elif isinstance(ocr_str_or_list, str) or isinstance(ocr_str_or_list, list):
# if is_gsyh is True and name in consts.CON_BANK_FIELD:
# update_args = {'is_gsyh': is_gsyh}
# for k, v in compare_logic[name][2].items():
# update_args[k] = v
# else:
# update_args = compare_logic[name][2]
if isinstance(ocr_str_or_list, list):
# no-asp 合同编号-每页(no-asp)
if name == consts.SE_AFC_CON_FIELD[23]:
ocr_str_or_list.pop()
ocr_str = json.dumps(ocr_str_or_list, ensure_ascii=False)
else:
ocr_str_or_list = ocr_str_or_list.strip()
ocr_str = ocr_str_or_list
result = getattr(cp, compare_logic[name][1])(value, ocr_str_or_list, **compare_logic[name][2])
reason = compare_logic[name][3]
else:
result = consts.RESULT_N
ocr_str = empty_str
reason = compare_logic[name][3]
# img_path = empty_str
if name not in compare_logic:
img_path = empty_str
else:
img_path = ocr_res.get(consts.IMG_PATH_KEY, {}).get(compare_logic[name][0],
empty_str) if result == consts.RESULT_N else empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, result, ocr_str, img_path, error_type, reason))
else:
for name, value in strip_list:
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value,
'{0}未找到'.format(license_en)))
# if ocr_res_str is not None:
# img_map = {}
# for name, _, result, _, img_path, _, _ in result_field_list:
# if result == consts.RESULT_N:
# img_map.setdefault(img_path, []).append(name)
# for path, field_list in img_map.items():
# if os.path.exists(path):
# pre, suf = os.path.splitext(path)
# last_img = cv2.imread(path)
# for field_idx, field in enumerate(field_list):
# try:
# save_path = '{0}_{1}{2}'.format(pre, str(field_idx), suf)
# section_position_list = ocr_res.get(consts.ALL_POSITION_KEY, {}).get(field, [])
# if isinstance(section_position_list, list) and len(section_position_list) == 4:
# field_img = last_img[section_position_list[1]: section_position_list[3],
# section_position_list[0]: section_position_list[2], :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = path
# except Exception as e:
# field_img_path_dict[field] = path
return result_field_list, field_img_path_dict
@staticmethod
def se_bs_compare(license_en, ocr_res_dict, strip_list, is_auto):
# 主共借至少提供一个
# 有担保人，担保人必须提供。主共借没有时，修改comment：人工查看担保人亲属关系
if is_auto:
ocr_field, compare_logic, _ = consts.SE_COMPARE_FIELD_AUTO[license_en]
else:
ocr_field, compare_logic, _ = pos_consts.SE_COMPARE_FIELD[license_en]
ocr_res_str = ocr_res_dict.get(ocr_field)
result_field_list = []
field_img_path_dict = dict()
if ocr_res_str is not None:
pre_field_list = strip_list[:3]
dbr1_field_list = strip_list[3:6]
dbr2_field_list = strip_list[6:]
ocr_res_list = json.loads(ocr_res_str)
# length = len(ocr_res_list)
# 主共借人
pre_best_res = {}
max_correct_count = 0
verify_list = []
verify_false_idx_list = []
for tmp_idx, ocr_res in enumerate(ocr_res_list):
correct_count = 0
pre_tmp_res_part = {}
verify_bool = ocr_res.get('verify', True)
verify_list.append(verify_bool)
if not verify_bool:
verify_false_idx_list.append(str(tmp_idx + 1))
for idx, (name, value) in enumerate(pre_field_list):
ocr_str_or_list = ocr_res.get(compare_logic[name][0])
if isinstance(ocr_str_or_list, str) or isinstance(ocr_str_or_list, list) \
or isinstance(ocr_str_or_list, int):
result = getattr(cp, compare_logic[name][1])(value, ocr_str_or_list, **compare_logic[name][2])
if isinstance(ocr_str_or_list, list):
ocr_str = json.dumps(ocr_str_or_list, ensure_ascii=False)
else:
ocr_str = ocr_str_or_list
reason = compare_logic[name][3]
else:
result = consts.RESULT_N
ocr_str = empty_str
reason = compare_logic[name][3]
if idx == 0 and result == consts.RESULT_N:
break
if result == consts.RESULT_Y:
correct_count += 1
pre_tmp_res_part[name] = (result, ocr_str, reason)
if correct_count > 0 and correct_count >= max_correct_count:
max_correct_count = correct_count
pre_best_res = pre_tmp_res_part
# 真伪
if not is_auto:
name = '真伪'
result = consts.RESULT_Y if all(verify_list) else consts.RESULT_N
reason = '第{0}份银行流水疑似造假，需人工核查'.format('、'.join(verify_false_idx_list))
result_field_list.append((name, empty_str, result, json.dumps(verify_list, ensure_ascii=False),
empty_str, empty_error_type, reason))
# 担保人1
dbr1_best_res = {}
if len(dbr1_field_list) > 0:
max_correct_count = 0
for ocr_res in ocr_res_list:
correct_count = 0
dbr1_tmp_res_part = {}
for idx, (name, value) in enumerate(dbr1_field_list):
ocr_str_or_list = ocr_res.get(compare_logic[name][0])
if isinstance(ocr_str_or_list, str) or isinstance(ocr_str_or_list, list) or isinstance(
ocr_str_or_list, int):
result = getattr(cp, compare_logic[name][1])(value, ocr_str_or_list,
**compare_logic[name][2])
if isinstance(ocr_str_or_list, list):
ocr_str = json.dumps(ocr_str_or_list, ensure_ascii=False)
else:
ocr_str = ocr_str_or_list
reason = compare_logic[name][3]
else:
result = consts.RESULT_N
ocr_str = empty_str
reason = compare_logic[name][3]
if idx == 0 and result == consts.RESULT_N:
break
if result == consts.RESULT_Y:
correct_count += 1
dbr1_tmp_res_part[name] = (result, ocr_str, reason)
if correct_count > 0 and correct_count >= max_correct_count:
max_correct_count = correct_count
dbr1_best_res = dbr1_tmp_res_part
# 担保人2
dbr2_best_res = {}
if len(dbr1_field_list) > 0:
max_correct_count = 0
for ocr_res in ocr_res_list:
correct_count = 0
dbr2_tmp_res_part = {}
for idx, (name, value) in enumerate(dbr2_field_list):
ocr_str_or_list = ocr_res.get(compare_logic[name][0])
if isinstance(ocr_str_or_list, str) or isinstance(ocr_str_or_list, list) or isinstance(
ocr_str_or_list, int):
result = getattr(cp, compare_logic[name][1])(value, ocr_str_or_list,
**compare_logic[name][2])
if isinstance(ocr_str_or_list, list):
ocr_str = json.dumps(ocr_str_or_list, ensure_ascii=False)
else:
ocr_str = ocr_str_or_list
reason = compare_logic[name][3]
else:
result = consts.RESULT_N
ocr_str = empty_str
reason = compare_logic[name][3]
if idx == 0 and result == consts.RESULT_N:
break
if result == consts.RESULT_Y:
correct_count += 1
dbr2_tmp_res_part[name] = (result, ocr_str, reason)
if correct_count > 0 and correct_count >= max_correct_count:
max_correct_count = correct_count
dbr2_best_res = dbr2_tmp_res_part
dbr_ok = False
# 有担保人
if len(dbr1_field_list) > 0:
# 有担保人12
if len(dbr2_field_list) > 0:
if len(dbr1_best_res) > 0 and len(dbr2_best_res) > 0:
dbr_ok = True
# 有担保人1
else:
if len(dbr1_best_res) > 0:
dbr_ok = True
# 无担保人
# else:
# pass
best_res_empty = False
if len(pre_best_res) == 0 and len(dbr1_best_res) == 0 and len(dbr2_best_res) == 0:
best_res_empty = True
for name, value in pre_field_list:
if len(pre_best_res) > 0:
result, ocr_str, reason = pre_best_res[name]
else:
result = consts.RESULT_N
ocr_str = empty_str
if best_res_empty:
reason = consts.SPECIAL_REASON_3
elif dbr_ok: # 有担保人且担保人都提供了流水
reason = consts.SPECIAL_REASON
else:
reason = compare_logic[pre_field_list[0][0]][3]
img_path = empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, result, ocr_str, img_path, error_type, reason))
if len(dbr1_field_list) > 0:
for name, value in dbr1_field_list:
if len(dbr1_best_res) > 0:
result, ocr_str, reason = dbr1_best_res[name]
else:
result = consts.RESULT_N
ocr_str = empty_str
if best_res_empty:
reason = consts.SPECIAL_REASON_3
elif len(pre_best_res) > 0:
reason = consts.SPECIAL_REASON_2
else:
reason = compare_logic[dbr1_field_list[0][0]][3]
img_path = empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, result, ocr_str, img_path, error_type, reason))
if len(dbr2_field_list) > 0:
for name, value in dbr2_field_list:
if len(dbr2_best_res) > 0:
result, ocr_str, reason = dbr2_best_res[name]
else:
result = consts.RESULT_N
ocr_str = empty_str
if best_res_empty:
reason = consts.SPECIAL_REASON_3
elif len(pre_best_res) > 0:
reason = consts.SPECIAL_REASON_2
else:
reason = compare_logic[dbr2_field_list[0][0]][3]
img_path = empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, result, ocr_str, img_path, error_type, reason))
else:
for name, value in strip_list:
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value,
consts.SPECIAL_REASON_3))
return result_field_list, field_img_path_dict
@staticmethod
def se_compare_license_id(license_en, id_res_list, field_list):
ocr_field, compare_logic, special_expiry_date = pos_consts.SE_COMPARE_FIELD[license_en]
is_find = False
no_ocr_result = True
special_expiry_date_slice = False
result_field_list = []
section_img_info = dict()
field_img_path_dict = dict()
# ocr_res_str = ocr_res_dict.get(ocr_field)
for ocr_res_str in id_res_list:
if is_find:
break
if ocr_res_str is not None:
no_ocr_result = False
ocr_res_list = json.loads(ocr_res_str)
# 3/4页去除
# if ocr_field == consts.MVC_OCR_FIELD:
# tmp_list = []
# for res in ocr_res_list:
# if compare_logic['vinNo'][0] in res:
# tmp_list.append(res)
# ocr_res_list = tmp_list
length = len(ocr_res_list)
# 身份证、居住证 过期期限特殊处理
if special_expiry_date:
expiry_dates = dict()
key = compare_logic.get('idExpiryDate')[0]
for date_tmp_idx, ocr_res in enumerate(ocr_res_list):
if key in ocr_res:
expiry_dates[ocr_res[key]] = (ocr_res.get(consts.IMG_PATH_KEY_2, ''), date_tmp_idx)
else:
expiry_dates = dict()
for res_idx in range(length - 1, -1, -1):
if is_find:
break
for idx, (name, value) in enumerate(field_list):
# if ocr_field == consts.MVI_OCR_FIELD and name == consts.SE_NEW_ADD_FIELD[9]:
# ocr_str = getattr(cp, consts.ZW_METHOD)(
# ocr_res_list[res_idx].get(consts.LOWER_AMOUNT_FIELD, ''),
# ocr_res_list[res_idx].get(consts.UPPER_AMOUNT_FIELD, ''),
# )
# else:
ocr_str = ocr_res_list[res_idx].get(compare_logic[name][0])
if not isinstance(ocr_str, str):
result = consts.RESULT_N
ocr_str = empty_str
no_key = True
else:
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
no_key = False
if idx == 0 and result == consts.RESULT_N and length > 1:
break
is_find = True
section_img_info[consts.SECTION_IMG_PATH_KEY] = ocr_res_list[res_idx].get(
consts.SECTION_IMG_PATH_KEY, '')
section_img_info[consts.ALL_POSITION_KEY] = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY,
{})
if special_expiry_date:
section_img_info[consts.SECTION_IMG_PATH_KEY_2] = ocr_res_list[res_idx].get(
consts.SECTION_IMG_PATH_KEY_2, '')
section_img_info[consts.ALL_POSITION_KEY_2] = ocr_res_list[res_idx].get(
consts.ALL_POSITION_KEY_2, {})
# 过期期限特殊处理
if special_expiry_date and name == 'idExpiryDate' and result == consts.RESULT_N:
if no_key:
if len(expiry_dates) == 0:
ocr_str = empty_str
result = consts.RESULT_N
img_path = empty_str
else:
for expiry_date, (date_img_path, date_res_idx) in expiry_dates.items():
expiry_date_res = getattr(cp, compare_logic[name][1])(value, expiry_date,
**compare_logic[name][2])
if expiry_date_res == consts.RESULT_N:
ocr_str = expiry_date
img_path = date_img_path
special_expiry_date_slice = True
section_img_info[consts.SECTION_IMG_PATH_KEY_2] = ocr_res_list[
date_res_idx].get(
consts.SECTION_IMG_PATH_KEY_2, '')
section_img_info[consts.ALL_POSITION_KEY_2] = ocr_res_list[
date_res_idx].get(
consts.ALL_POSITION_KEY_2, {})
break
else:
ocr_str = empty_str
result = consts.RESULT_Y
img_path = empty_str
else:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY_2, '')
special_expiry_date_slice = True
else:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY,
'') if result == consts.RESULT_N else empty_str
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
result_field_list.append(
(name, value, result, ocr_str, img_path, error_type, compare_logic[name][3]))
if not is_find:
for name, value in field_list:
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
no_find_str = consts.DDA_NO_FIND if license_en == consts.DDA_EN else '{0}未找到'.format(license_en)
result_field_list.append(
(name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value, no_find_str))
# if is_find:
# if special_expiry_date_slice:
# special_section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY_2, '')
# if os.path.exists(special_section_img_path):
# field = 'idExpiryDate'
# special_info = section_img_info.get(consts.ALL_POSITION_KEY_2, {})
# special_section_position = special_info.get(consts.POSITION_KEY, {})
# special_section_angle = special_info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(special_section_img_path, special_section_position,
# special_section_angle)
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
# else:
# pre, suf = os.path.splitext(special_section_img_path)
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(special_info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3], :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = special_section_img_path
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
#
# section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY, '')
# if os.path.exists(section_img_path):
# failed_field = []
# base_img_path = empty_str
# for name, _, result, _, img_path, _, _ in result_field_list:
# if result == consts.RESULT_N:
# if special_expiry_date_slice and name == 'idExpiryDate':
# continue
# failed_field.append(name)
# if base_img_path == empty_str:
# base_img_path = img_path
# if len(failed_field) > 0:
# info = section_img_info.get(consts.ALL_POSITION_KEY, {})
# section_position = info.get(consts.POSITION_KEY, {})
# section_angle = info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(section_img_path, section_position, section_angle)
# except Exception as e:
# for field in failed_field:
# field_img_path_dict[field] = base_img_path
# else:
# pre, suf = os.path.splitext(section_img_path)
# for field in failed_field:
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3],
# :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = base_img_path
# except Exception as e:
# field_img_path_dict[field] = base_img_path
return result_field_list, no_ocr_result, field_img_path_dict
@staticmethod
def se_compare_license(license_en, ocr_res_dict, field_list):
ocr_field, compare_logic, special_expiry_date = pos_consts.SE_COMPARE_FIELD[license_en]
is_find = False
no_ocr_result = False
special_expiry_date_slice = False
result_field_list = []
section_img_info = dict()
field_img_path_dict = dict()
ocr_res_str = ocr_res_dict.get(ocr_field)
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
# 3/4页去除
if ocr_field == consts.MVC_OCR_FIELD:
tmp_list = []
for res in ocr_res_list:
if compare_logic['vinNo'][0] in res:
tmp_list.append(res)
ocr_res_list = tmp_list
length = len(ocr_res_list)
# 身份证、居住证 过期期限特殊处理
if special_expiry_date:
expiry_dates = dict()
key = compare_logic.get('idExpiryDate')[0]
for date_tmp_idx, ocr_res in enumerate(ocr_res_list):
if key in ocr_res:
expiry_dates[ocr_res[key]] = (ocr_res.get(consts.IMG_PATH_KEY_2, ''), date_tmp_idx)
else:
expiry_dates = dict()
for res_idx in range(length - 1, -1, -1):
if is_find:
break
new_invoice_result = None
for idx, (name, value) in enumerate(field_list):
# 二手车交易凭证 日期
if ocr_field == consts.JYPZ_OCR_FIELD and name == consts.SE_GB_USED_FIELD[2]:
date_1 = ocr_res_list[res_idx].get(consts.JYPZ_DATE_FIELD_1, '')
if len(date_1) > 0:
date_2 = date_3 = ''
else:
date_1 = ocr_res_list[res_idx].get(consts.JYPZ_DATE_FIELD_2, '')
date_2 = ocr_res_list[res_idx].get(consts.JYPZ_DATE_FIELD_3, '')
date_3 = ocr_res_list[res_idx].get(consts.JYPZ_DATE_FIELD_4, '')
ocr_str = [date_1, date_2, date_3]
# 购车发票 价税合计大小写检验
elif ocr_field == consts.MVI_OCR_FIELD and name == consts.SE_NEW_ADD_FIELD[9]:
ocr_str = getattr(cp, consts.ZW_METHOD)(
ocr_res_list[res_idx].get(consts.LOWER_AMOUNT_FIELD, ''),
ocr_res_list[res_idx].get(consts.UPPER_AMOUNT_FIELD, ''),
)
else:
ocr_str = ocr_res_list[res_idx].get(compare_logic[name][0])
if isinstance(ocr_str, str):
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
no_key = False
# 二手车交易凭证 日期
elif ocr_field == consts.JYPZ_OCR_FIELD and name == consts.SE_GB_USED_FIELD[2]:
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
no_key = False
else:
result = consts.RESULT_N
ocr_str = empty_str
no_key = True
if name == '发票联':
new_invoice_result = result
if name == 'productGroupName' and (name.find('官方认证二手车') > -1 or name.find('官方认证二手车') > -1):
# 发票联必须
if new_invoice_result and new_invoice_result == consts.RESULT_Y:
result = consts.RESULT_Y
else:
result = consts.RESULT_N
if idx == 0 and result == consts.RESULT_N and length > 1:
break
is_find = True
# section_img_info[consts.SECTION_IMG_PATH_KEY] = ocr_res_list[res_idx].get(
# consts.SECTION_IMG_PATH_KEY, '')
# section_img_info[consts.ALL_POSITION_KEY] = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY, {})
# if special_expiry_date:
# section_img_info[consts.SECTION_IMG_PATH_KEY_2] = ocr_res_list[res_idx].get(
# consts.SECTION_IMG_PATH_KEY_2, '')
# section_img_info[consts.ALL_POSITION_KEY_2] = ocr_res_list[res_idx].get(
# consts.ALL_POSITION_KEY_2, {})
# 过期期限特殊处理
if special_expiry_date and name == 'idExpiryDate' and result == consts.RESULT_N:
if no_key:
if len(expiry_dates) == 0:
ocr_str = empty_str
result = consts.RESULT_N
img_path = empty_str
else:
for expiry_date, (date_img_path, date_res_idx) in expiry_dates.items():
expiry_date_res = getattr(cp, compare_logic[name][1])(value, expiry_date,
**compare_logic[name][2])
if expiry_date_res == consts.RESULT_N:
ocr_str = expiry_date
img_path = date_img_path
special_expiry_date_slice = True
section_img_info[consts.SECTION_IMG_PATH_KEY_2] = ocr_res_list[
date_res_idx].get(
consts.SECTION_IMG_PATH_KEY_2, '')
section_img_info[consts.ALL_POSITION_KEY_2] = ocr_res_list[date_res_idx].get(
consts.ALL_POSITION_KEY_2, {})
break
else:
ocr_str = empty_str
result = consts.RESULT_Y
img_path = empty_str
else:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY_2, '')
special_expiry_date_slice = True
else:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY,
'') if result == consts.RESULT_N else empty_str
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
if ocr_field == pos_consts.IC_OCR_FIELD and name == 'idExpiryDate':
result_field_list.append(
(name, value, result, ocr_str, img_path, error_type, compare_logic[name][-1][result]))
else:
result_field_list.append((name, value, result, ocr_str, img_path, error_type, compare_logic[name][-1]))
else:
no_ocr_result = True
# 针对license文档类型输出
if no_ocr_result:
# 保险
if license_en == pos_consts.BD_EN:
# 整体文档相关
pos_field, value = field_list[0]
if pos_field == 'insuranceType' and value == 'Comprehensive Insurance':
if no_ocr_result:
result_field_list.append((pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.HMH_EN: # 抵押登记豁免函
pos_field, value = field_list[0]
if pos_field == 'mortgageType' and value == 'MOTGF':
if no_ocr_result:
result_field_list.append((pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.MVI_EN: # 新车发票
pos_field, value = field_list[0]
if pos_field == 'productGroupName' and no_ocr_result:
result_field_list.append(
(pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.BC_EN: # 银行卡
pos_field, value = field_list[0]
if pos_field == 'bankVerificationStatus' and (value == 'N/A' or value == 'FAIL'):
if no_ocr_result:
result_field_list.append(
(pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.AFC_CONTRACT_EN: # AFC 车辆抵押贷款合同（E-Contract-Non ASP/ASP）
pos_field, value = field_list[0]
if pos_field == 'applicationEntity' and value == 'AFC':
if no_ocr_result:
result_field_list.append((pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.HIL_CONTRACT_1_EN: # 售后回租合同（E-Sign-SLB / OC）
pos_field, value = field_list[0]
if pos_field == 'applicationEntity' and value == 'HIL':
if no_ocr_result:
result_field_list.append((pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
elif license_en == pos_consts.HIL_CONTRACT_2_EN: # 车辆租赁抵押合同（E-Sign-SLB / OC）
pos_field, value = field_list[0]
if pos_field == 'applicationEntity' and value == 'HIL':
if no_ocr_result:
result_field_list.append(
(pos_field, value, consts.RESULT_N, ocr_str, img_path, error_type,
compare_logic['文档'][-1]))
if not is_find:
for name, value in field_list:
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
no_find_str = consts.DDA_NO_FIND if license_en == consts.DDA_EN else '{0}未找到'.format(license_en)
result_field_list.append(
(name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value, no_find_str))
# if is_find:
# if special_expiry_date_slice:
# special_section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY_2, '')
# if os.path.exists(special_section_img_path):
# field = 'idExpiryDate'
# special_info = section_img_info.get(consts.ALL_POSITION_KEY_2, {})
# special_section_position = special_info.get(consts.POSITION_KEY, {})
# special_section_angle = special_info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(special_section_img_path, special_section_position,
# special_section_angle)
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
# else:
# pre, suf = os.path.splitext(special_section_img_path)
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(special_info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3], :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = special_section_img_path
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
#
# section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY, '')
# if os.path.exists(section_img_path):
# failed_field = []
# base_img_path = empty_str
# for name, _, result, _, img_path, _, _ in result_field_list:
# if result == consts.RESULT_N:
# if special_expiry_date_slice and name == 'idExpiryDate':
# continue
# failed_field.append(name)
# if base_img_path == empty_str:
# base_img_path = img_path
# if len(failed_field) > 0:
# info = section_img_info.get(consts.ALL_POSITION_KEY, {})
# section_position = info.get(consts.POSITION_KEY, {})
# section_angle = info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(section_img_path, section_position, section_angle)
# except Exception as e:
# for field in failed_field:
# field_img_path_dict[field] = base_img_path
# else:
# pre, suf = os.path.splitext(section_img_path)
# for field in failed_field:
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3],
# :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = base_img_path
# except Exception as e:
# field_img_path_dict[field] = base_img_path
return result_field_list, no_ocr_result, field_img_path_dict
class CACompare:
@staticmethod
def ca_compare_license(license_en, ocr_res_dict, field_list):
ocr_field, compare_logic, special_expiry_date = consts.CA_COMPARE_FIELD[license_en]
is_find = False
special_expiry_date_slice = False
result_field_list = []
section_img_info = dict()
field_img_path_dict = dict()
ocr_res_str = ocr_res_dict.get(ocr_field)
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
# 副页去除 3/4页去除
if ocr_field == consts.DL_OCR_FIELD or ocr_field == consts.MVC_OCR_FIELD:
tmp_list = []
for res in ocr_res_list:
if compare_logic['vinNo'][0] in res:
tmp_list.append(res)
ocr_res_list = tmp_list
length = len(ocr_res_list)
# 身份证、居住证 过期期限特殊处理
if special_expiry_date:
expiry_dates = set()
expiry_dates_img_path = set()
key = compare_logic.get('idExpiryDate')[0]
for ocr_res in ocr_res_list:
if key in ocr_res:
expiry_dates.add(ocr_res[key])
expiry_dates_img_path.add(ocr_res.get(consts.IMG_PATH_KEY_2, ''))
else:
expiry_dates = set()
expiry_dates_img_path = set()
for res_idx in range(length - 1, -1, -1):
if is_find:
break
for idx, (name, value) in enumerate(field_list):
ocr_str = ocr_res_list[res_idx].get(compare_logic[name][0])
if not isinstance(ocr_str, str):
result = consts.RESULT_N
ocr_str = empty_str
else:
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
if idx == 0 and result == consts.RESULT_N and length > 1:
break
is_find = True
# section_img_info[consts.SECTION_IMG_PATH_KEY] = ocr_res_list[res_idx].get(
# consts.SECTION_IMG_PATH_KEY, '')
# section_img_info[consts.ALL_POSITION_KEY] = ocr_res_list[res_idx].get(consts.ALL_POSITION_KEY, {})
# if special_expiry_date:
# section_img_info[consts.SECTION_IMG_PATH_KEY_2] = ocr_res_list[res_idx].get(
# consts.SECTION_IMG_PATH_KEY_2, '')
# section_img_info[consts.ALL_POSITION_KEY_2] = ocr_res_list[res_idx].get(
# consts.ALL_POSITION_KEY_2, {})
# 过期期限特殊处理
if special_expiry_date and name == 'idExpiryDate' and result == consts.RESULT_N:
for expiry_date in expiry_dates:
expiry_date_res = getattr(cp, compare_logic[name][1])(value, expiry_date,
**compare_logic[name][2])
if expiry_date_res == consts.RESULT_Y:
ocr_str = expiry_date
result = expiry_date_res
break
if result == consts.RESULT_N:
if consts.IMG_PATH_KEY_2 in ocr_res_list[res_idx]:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY_2, '')
special_expiry_date_slice = True
else:
img_path = expiry_dates_img_path.pop() if len(expiry_dates_img_path) > 0 else empty_str
else:
img_path = empty_str
else:
img_path = ocr_res_list[res_idx].get(consts.IMG_PATH_KEY,
'') if result == consts.RESULT_N else empty_str
error_type = empty_error_type if result == consts.RESULT_Y else ErrorType.OCR.value
result_field_list.append((name, value, result, ocr_str, img_path, error_type, compare_logic[name][-1]))
if not is_find:
for name, value in field_list:
result_field_list.append((name, value, consts.RESULT_N, empty_str, empty_str, ErrorType.NF.value, compare_logic[name][-1]))
# if is_find:
# if special_expiry_date_slice:
# special_section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY_2, '')
# if os.path.exists(special_section_img_path):
# field = 'idExpiryDate'
# special_info = section_img_info.get(consts.ALL_POSITION_KEY_2, {})
# special_section_position = special_info.get(consts.POSITION_KEY, {})
# special_section_angle = special_info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(special_section_img_path, special_section_position,
# special_section_angle)
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
# else:
# pre, suf = os.path.splitext(special_section_img_path)
# try:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(special_info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3], :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = special_section_img_path
# except Exception as e:
# field_img_path_dict[field] = special_section_img_path
#
# section_img_path = section_img_info.get(consts.SECTION_IMG_PATH_KEY, '')
# if os.path.exists(section_img_path):
# failed_field = []
# base_img_path = empty_str
# for name, _, result, _, img_path, _ in result_field_list:
# if result == consts.RESULT_N:
# if special_expiry_date_slice and name == 'idExpiryDate':
# continue
# failed_field.append(name)
# if base_img_path == empty_str:
# base_img_path = img_path
# if len(failed_field) > 0:
# info = section_img_info.get(consts.ALL_POSITION_KEY, {})
# section_position = info.get(consts.POSITION_KEY, {})
# section_angle = info.get(consts.ANGLE_KEY, 0)
# try:
# last_img = img_process(section_img_path, section_position, section_angle)
# except Exception as e:
# for field in failed_field:
# field_img_path_dict[field] = base_img_path
# else:
# pre, suf = os.path.splitext(section_img_path)
# for field in failed_field:
# try:
# if license_en == consts.PP_EN:
# res_field = consts.PP_SLICE_MAP[field]
# else:
# res_field = compare_logic[field][0]
# is_valid, coord_tuple = field_build_coordinates(info.get(res_field, {}))
# if is_valid:
# save_path = '{0}_{1}{2}'.format(pre, field, suf)
# field_img = last_img[coord_tuple[0]:coord_tuple[1], coord_tuple[2]:coord_tuple[3],
# :]
# cv2.imwrite(save_path, field_img)
# field_img_path_dict[field] = save_path
# else:
# field_img_path_dict[field] = base_img_path
# except Exception as e:
# field_img_path_dict[field] = base_img_path
return result_field_list, field_img_path_dict
\ No newline at end of file
src/pos/pos_consts.py deleted 100644 → 0
GZS_REASON_1 = '此申请有ASP产品，需人工核查'
GZS_REASON_2 = 'ASP购置税金额小于系统金额'
# license (license_en)
ID_EN = 'PRC ID'
PP_EN = 'Passport'
EEP_EN = 'Resident Permit to Mainland'
RP_EN = 'Resident ID'
BL_EN = 'Business permit'
SME_BL_EN = 'SME Business permit'
MVI_EN = 'newCar Invoice'
UCI_EN = 'usedCar Invoice'
MVC_EN = 'Green Book(1/2)'
MVC34_EN = 'Green Book(3/4)'
BC_EN = 'Bank Card'
DDA_EN = 'DDA'
HMH_EN = 'Mortgage Waiver Letter'
JYPZ_EN = 'Used Car Document'
AFC_CONTRACT_EN = 'AFC Contract'
BD_EN = 'Insurance'
BS_EN = 'Bank Statement'
HIL_CONTRACT_1_EN = '售后回租合同'
HIL_CONTRACT_2_EN = '车辆租赁抵押合同'
HIL_CONTRACT_3_EN = '车辆处置协议'
# 对应ocr_result 表中的字段
IC_OCR_FIELD = 'ic_ocr'
RP_OCR_FIELD = 'rp_ocr'
BL_OCR_FIELD = 'bl_ocr'
EEP_OCR_FIELD = 'eep_ocr'
DL_OCR_FIELD = 'dl_ocr'
PP_OCR_FIELD = 'pp_ocr'
MVC_OCR_FIELD = 'mvc_ocr'
MVI_OCR_FIELD = 'mvi_ocr'
BC_OCR_FIELD = 'bc_ocr'
UCI_OCR_FIELD = 'uci_ocr'
DDA_OCR_FIELD = 'bs_ocr'
HMH_OCR_FIELD = 'hmh_ocr'
JYPZ_OCR_FIELD = 'jypz_ocr'
HT_FIELD = 'ht_ocr'
BD_FIELD = 'bd_ocr'
BS_FIELD = 'bss_ocr'
HIL_CONTRACT_1_FIELD = 'hil_contract_1_ocr'
HIL_CONTRACT_2_FIELD = 'hil_contract_2_ocr'
HIL_CONTRACT_3_FIELD = 'hil_contract_3_ocr'
SE_AFC_CON_FIELD = ['合同编号-每页', '所购车辆价格-小写-重要条款', '车架号-重要条款', '贷款本金金额-重要条款', '贷款期限-重要条款',
'车辆贷款本金金额-重要条款', '附加产品融资贷款本金总额-重要条款', '所购车辆价格', '车架号', '经销商',
'贷款本金金额', '车辆贷款本金金额', '附加产品融资贷款本金总额', '贷款期限', '还款账号', '户名', '开户行',
'还款计划表', '见证人签字', '见证人日期', 'ASP项目详情-重要条款', '购置税校验', 'ASP项目详情',
'合同编号-每页(no-asp)', '无ASP产品']
# '承租人姓名', '承租人证件号码', '承租人法定代表人或授权代表'
SE_HIL_CON_1_FIELD = ['合同编号-每页', '车辆识别代码', '车辆卖方', '车辆原始销售价格', '融资成本总额', '租期',
'还款计划表', 'ASP项目详情', '购置税校验', '承租人法定代表人或授权代表', '还款账号', '户名', '开户行',
'共同承租人法定代表人或授权代表', '无ASP产品']
SE_HIL_CON_2_FIELD = ['合同编号', '车辆识别代码']
# 大陆二代身份证
ID_COMPARE_LOGIC = {
'customerName': ('姓名', 'se_name_compare', {}, '身份证姓名与系统不一致'),
'idNum': ('公民身份号码', 'se_common_compare', {}, '身份证号码与系统不一致'),
'dateOfBirth': ('出生年月', 'se_date_compare', {'input_replace': ''}),
'idExpiryDate': ('有效期限', 'se_date_compare_pos', {'long': True, 'ocr_split': True, 'input_replace': '', 'today': True}, {
'N1': '身份证疑似过期',
'N2': '身份证即将过期，请尽快提交放款申请'
})
}
# 新车发票
# key: pos field
# value: (ocr_result中的映射key，比较方法， 选项， comments/reason)
MVI_COMPARE_LOGIC = {
'firstSubmmisonDate': ('开票日期', 'se_date_compare_2', {'three_month': True}, '请确认发票开票日期，若发票开票日期早于首次提交审批日期则无法受理放款申请'),
'productGroupName': ('productGroupName', 'se_common_compare', {}, '请确认是否提供发票'),
'vinNo': ('车辆识别代码', 'se_common_compare', {}, '发票车架号与系统不一致'),
'vehicleTransactionAmount': ('价税合计小写', 'se_amount_compare', {}, '发票车辆价格与系统不一致'),
'customerName': ('购方名称', 'se_name_compare', {'is_passport': True, 'replace_kuohao': True}, '发票购买方姓名与系统不一致'),
'idNum': ('购买方身份证号或组织机构代码', 'se_common_compare', {}, '发票购买方证件号码与系统不一致'),
'发票联': ('发票联', 'se_common_compare', {}, '发票疑似非发票联'),
'文档': ('发票联', 'se_have_compare', {}, '请确认是否提供发票'),
}
# 银行卡
BC_COMPARE_LOGIC = {
'accountNo': ('CardNum', 'se_common_compare', {'remove_space': True}, '银行卡卡号与系统不一致'),
'bankName': ('BankName', 'se_both_contain_compare', {}, '银行卡开户行与系统不一致'),
'type': ('CardType', 'se_common_compare', {}, '银行卡非借记卡'),
'bankVerificationStatus': ('CardType', 'se_have_compare', {}, '请确认是否提供银行卡'),
'文档': ('文档', 'se_common_compare', {}, '请确认是否提供银行卡'),
}
# AFC 车辆抵押贷款合同（E-Contract-Non ASP/ASP）
HT_COMPARE_LOGIC = {
'合同编号-每页': ('合同编号-每页', 'se_list_compare', {}, '合同编号与系统不一致'),
# '合同编号-每页(no-asp)': ('合同编号-每页', 'se_list_compare', {'pop_last': True}, '合同编号与系统不一致'),
'合同编号-每页(no-asp)': ('合同编号-每页', 'se_list_compare', {}, '合同编号与系统不一致'),
'所购车辆价格-小写-重要条款': ('所购车辆价格-小写-重要条款', 'se_amount_str_compare', {}, '合同首页中车辆价格与系统不一致'),
'车架号-重要条款': ('车架号-重要条款', 'se_common_compare', {}, '合同首页中车架号与系统不一致'),
'贷款本金金额-重要条款': ('贷款本金金额-重要条款', 'se_amount_str_compare', {}, '合同首页中贷款本金与系统不一致'),
'贷款期限-重要条款': ('贷款期限-重要条款', 'se_common_compare', {}, '合同首页中贷款期限与系统不一致'),
'车辆贷款本金金额-重要条款': ('车辆贷款本金金额-重要条款', 'se_amount_str_compare', {}, '合同车辆贷款本金金额-重要条款'),
'附加产品融资贷款本金总额-重要条款': ('附加产品融资贷款本金总额-重要条款', 'se_amount_str_compare', {}, '合同首页中附加产品融资金额与系统不一致'),
'所购车辆价格': ('所购车辆价格', 'se_amount_str_compare', {}, '主合同页中车辆价格与系统不一致'),
'车架号': ('车架号', 'se_common_compare', {}, '主合同页中车架号与系统不一致'),
'经销商': ('经销商', 'se_common_compare', {}, '主合同页中经销商与系统不一致'),
'贷款本金金额': ('贷款本金金额', 'se_amount_str_compare', {}, '主合同页中贷款本金金额与系统不一致'),
'车辆贷款本金金额': ('车辆贷款本金金额', 'se_amount_str_compare', {}, '主合同页中车辆贷款本金金额与系统不一致'),
'附加产品融资贷款本金总额': ('附加产品融资贷款本金总额', 'se_amount_str_compare', {}, '主合同页中附加产品融资贷款本金总额与系统不一致'),
'贷款期限': ('贷款期限', 'se_common_compare', {}, '主合同页中贷款期限与系统不一致'),
'还款账号': ('还款账号', 'se_common_compare', {'remove_space': True}, '主合同页中还款账号与系统不一致'),
'户名': ('户名', 'se_common_compare', {}, '主合同页中户名与系统不一致'),
'开户行': ('开户行', 'se_both_contain_compare', {}, '主合同页中开户行与系统不一致'),
'借款人签字及时间': ('借款人签字及时间', 'se_both_contain_compare', {}, '合同首页签字项与系统不一致'),
'借款人姓名': ('借款人姓名', 'se_common_compare', {}, '主合同主借人姓名与系统不一致'),
'借款人证件号': ('借款人证件号', 'se_common_compare', {}, '主合同主借人证件号码与系统不一致'),
'共借人姓名': ('共借人姓名', 'se_common_compare', {}, '主合同共借人姓名与系统不一致'),
'共借人&抵押人姓名': ('共借人姓名', 'se_common_compare', {}, '主合同共借人&抵押人姓名与系统不一致'),
'共借人证件号': ('共借人证件号', 'se_common_compare', {}, '主合同共借人证件号码与系统不一致'),
'共借人&抵押人证件号': ('共借人证件号', 'se_common_compare', {}, '主合同共借人&抵押人证件号与系统不一致'),
'保证人姓名1': ('保证人姓名1', 'se_common_compare', {}, '主合同担保人1姓名与系统不一致'),
'保证人证件号1': ('保证人证件号1', 'se_common_compare', {}, '主合同担保人1证件号码与系统不一致'),
'保证人姓名2': ('保证人姓名2', 'se_common_compare', {}, '主合同担保人2姓名与系统不一致'),
'保证人证件号2': ('保证人证件号2', 'se_common_compare', {}, '主合同担保人2证件号码与系统不一致'),
'主借人签字': ('主借人签字', 'se_common_compare', {}, '合同主借款人签字与系统不一致'),
'主借人日期': ('主借人日期', 'se_have_compare', {}, '合同主借款人签字日期无'),
'共借人签字': ('共借人签字', 'se_common_compare', {}, '合同共借人签字与系统不一致'),
'共借人日期': ('共借人日期', 'se_have_compare', {}, '合同共借人签字日期无'),
'保证人签字1': ('保证人签字1', 'se_common_compare', {}, '合同担保人1签字与系统不一致'),
'保证人日期1': ('保证人日期1', 'se_have_compare', {}, '合同担保人1签字日期无'),
'保证人签字2': ('保证人签字2', 'se_common_compare', {}, '合同担保人2签字与系统不一致'),
'保证人日期2': ('保证人日期2', 'se_have_compare', {}, '合同担保人2签字日期无'),
'见证人签字': ('见证人签字', 'se_have_compare', {}, '合同见证人无'),
'见证人日期': ('见证人日期', 'se_date_contain_compare', {}, '合同见证人签字日期不符合逻辑'),
'还款计划表': ('还款计划表', 'se_schedule_compare', {"value_idx": 1}, '合同还款计划表与系统不一致'),
'ASP项目详情-重要条款': ('ASP项目详情-重要条款', 'se_asp_compare', {}, '合同(重要条款)ASP名称或者金额与系统不一致'),
'ASP项目详情': ('ASP项目详情', 'se_asp_compare', {}, '合同ASP名称或者金额与系统不一致'),
'购置税校验': ('购置税校验', 'se_self_compare_gzs', {}, GZS_REASON_2),
'无ASP产品': ('无ASP产品', 'se_self_compare_other_asp', {}, GZS_REASON_1),
'文档': ('文档', 'se_common_compare', {}, '请确认是否已完成车辆抵押贷款合同签署'),
}
# 售后回租合同（E-Sign-SLB / OC）
HIL_CONTRACT_1_COMPARE_LOGIC = {
'合同编号-每页': ('合同编号-每页', 'se_list_compare', {}, '售后回租合同中合同编号系统不一致'),
'合同编号-正文': ('合同编号-正文', 'se_common_compare', {}, '售后回租合同正文中合同编号系统不一致'),
'车辆识别代码': ('车辆识别代码', 'se_common_compare', {}, '售后回租合同车辆识别代码与系统车架号不一致'),
'车辆卖方': ('车辆卖方', 'se_common_compare', {}, '售后回租合同车辆卖方与系统经销商不一致'),
'车辆原始销售价格': ('车辆原始销售价格', 'se_amount_str_compare', {}, '售后回租合同车辆原始销售价格与系统车辆价格不一致'),
'融资成本总额': ('融资成本总额', 'se_amount_str_compare', {}, '售后回租合同融资成本总额与系统不一致'),
'租期': ('租期', 'se_contain_compare', {}, '售后回租合同首页中贷款期限系统不一致'),
'还款计划表': ('还款计划表', 'se_schedule_compare', {"value_idx": 1}, '售后回租合同还款计划表与系统不一致'),
'ASP项目详情': ('ASP项目详情', 'se_asp_compare', {}, '售后回租合同ASP名称或者金额与系统不一致'),
'承租人法定代表人或授权代表': ('承租人法定代表人或授权代表', 'se_name_compare', {}, '售后回租合同承租人法定代表人或授权代表与系统不一致'),
'共同承租人法定代表人或授权代表': ('共同承租人法定代表人或授权代表', 'se_name_compare', {}, '售后回租合同共同承租人法定代表人或授权代表与系统不一致'),
'还款账号': ('还款账号', 'se_common_compare', {'remove_space': True}, '售后回租合同还款账号与系统不一致'),
'户名': ('户名', 'se_common_compare', {}, '售后回租合同户名与系统不一致'),
'开户行': ('开户行', 'se_both_contain_compare', {}, '售后回租合同开户行与系统不一致'),
'承租人姓名': ('承租人姓名', 'se_name_compare', {}, '售后回租合同承租人姓名与系统不一致'),
'承租人证件号': ('承租人证件号', 'se_common_compare', {}, '售后回租合同承租人证件号与系统不一致'),
'承租人签字': ('承租人签字', 'se_contain_compare', {}, '售后回租合同承租人签字与系统不一致'),
'共同承租人姓名': ('共同承租人姓名', 'se_name_compare', {}, '售后回租合同共同承租人姓名与系统不一致'),
'共同承租人&抵押人姓名': ('共同承租人&抵押人姓名', 'se_name_compare', {}, '售后回租合同共同承租人&抵押人姓名与系统不一致'),
'共同承租人证件号': ('共同承租人证件号', 'se_common_compare', {}, '售后回租合同共同承租人证件号与系统不一致'),
'共同承租人&抵押人证件号': ('共同承租人&抵押人证件号', 'se_common_compare', {}, '售后回租合同共同承租人&抵押人证件号与系统不一致'),
'共同承租人签字': ('共同承租人签字', 'se_contain_compare', {}, '售后回租合同共同承租人签字与系统不一致'),
'保证人姓名1': ('保证人姓名1', 'se_name_compare', {}, '售后回租合同保证人姓名1与系统不一致'),
'保证人证件号1': ('保证人证件号1', 'se_common_compare', {}, '售后回租合同保证人证件号1与系统不一致'),
'保证人签字1': ('保证人签字1', 'se_contain_compare', {}, '售后回租合同保证人签字1与系统不一致'),
'保证人姓名2': ('保证人姓名2', 'se_name_compare', {}, '售后回租合同保证人姓名2与系统不一致'),
'保证人证件号2': ('保证人证件号2', 'se_common_compare', {}, '售后回租合同保证人证件号2与系统不一致'),
'保证人签字2': ('保证人签字2', 'se_contain_compare', {}, '售后回租合同保证人签字2与系统不一致'),
'购置税校验': ('购置税校验', 'se_self_compare_gzs', {}, GZS_REASON_2),
'无ASP产品': ('无ASP产品', 'se_self_compare_other_asp', {}, GZS_REASON_1),
'文档': ('文档', 'se_common_compare', {}, '请确认是否已完成售后回租合同签署'),
}
# 车辆租赁抵押合同（E-Sign-SLB / OC）
HIL_CONTRACT_2_COMPARE_LOGIC = {
'合同编号': ('合同编号', 'se_common_compare', {}, '车辆租赁抵押合同合同编号与系统合同编号不一致'),
'合同编号-正文': ('合同编号-正文', 'se_common_compare', {}, '车辆租赁抵押合同正文合同编号与系统合同编号不一致'),
'车辆识别代码': ('车辆识别代码', 'se_common_compare', {}, '车辆租赁抵押合同车辆识别代码与系统车架号不一致'),
'租金总额': ('租金总额', 'se_amount_str_compare', {}, '车辆租赁抵押合同租金总额与系统租金总额不一致'),
'融资租赁期限': ('融资租赁期限', 'se_common_compare', {}, '车辆租赁抵押合同融资租赁期限与系统不一致'),
'抵押人': ('抵押人', 'se_name_compare', {}, '车辆租赁抵押合同抵押人姓名与系统不一致'),
'抵押人证件号码': ('抵押人证件号码', 'se_common_compare', {}, '车辆租赁抵押合同抵押人证件号码与系统不一致'),
'抵押人签字': ('抵押人签字', 'se_contain_compare', {}, '车辆租赁抵押合同抵押人签字与系统承租人姓名不一致'),
'抵押人配偶': ('抵押人配偶', 'se_name_compare', {}, '车辆租赁抵押合同抵押人配偶姓名与系统不一致'),
'抵押人配偶证件号码': ('抵押人配偶证件号码', 'se_common_compare', {}, '车辆租赁抵押合同抵押人配偶证件号码与系统不一致'),
'抵押人配偶签字': ('抵押人配偶签字', 'se_contain_compare', {}, '车辆租赁抵押合同抵押人签字与系统承租人配偶姓名不一致'),
'文档': ('文档', 'se_common_compare', {}, '请确认是否已完成车辆租赁抵押合同签署'),
}
# 抵押登记豁免函
HMH_COMPARE_LOGIC = {
'借款人/承租人姓名': ('借款/承租人姓名', 'se_name_compare', {}, '抵押登记豁免函借款人/承租人姓名与系统不符'),
'借款人/承租人证件号': ('证件号码', 'se_common_compare', {}, '抵押登记豁免函借款人/承租人证件号码与系统不符'),
'申请号': ('合同编号', 'se_common_compare', {}, '抵押登记豁免函申请号与系统不符'),
'渠道': ('渠道', 'se_channel_compare', {}, '抵押登记豁免函渠道与系统不符'),
'签字': ('借款人签字/盖章', 'se_have_compare', {}, '抵押登记豁免函签字需人工核查'),
'文档': ('文档', 'se_common_compare', {}, '请确认是否已完成抵押登记豁免函签署')
}
# 保单
BD_COMPARE_LOGIC = {
'被保险人姓名': ('被保险人姓名', 'super_list_compare', {'method': 'name'}, '保单被保险人姓名与系统不一致'),
'被保险人证件号码': ('被保险人证件号码', 'super_list_compare', {'method': 'common', 'is_bd_id': True}, '保单身份证号需人工核查'),
'车架号': ('车架号', 'se_common_compare', {}, '保单车架号与系统不一致'),
'机动车损失保险金额': ('机动车损失保险金额', 'se_amount_lte_compare', {}, '保单车损险异常'),
'第三者责任保险金额': ('机动车第三者责任保险金额', 'se_amount_lte_compare', {}, '保单三者险异常'),
'绝对免赔率': ('机动车损失保险绝对免赔率/绝对免赔额', 'se_one_compare', {}, '保单无绝对免赔项'),
'保险起始日期': ('保险起始日期', 'se_bd_date_compare', {'start': True}, '保单起始时间有误'),
'保险截止日期': ('保险截止日期', 'se_bd_date_compare', {}, '保单截止时间有误'),
'保单章': ('保单章', 'se_common_compare', {}, '保单无保单章'),
'第一受益人': ('特别约定第一受益人', 'se_common_compare', {}, '保单第一受益人需人工核查'),
'保险费合计': ('保险费合计', 'se_amount_lte_compare', {}, '保单保费疑似无法覆盖ASP保险融资'),
'文档': ('文档', 'se_common_compare', {}, '请确认是否提供保单')
}
SE_COMPARE_FIELD = {
ID_EN: (IC_OCR_FIELD, ID_COMPARE_LOGIC, True),
MVI_EN: (MVI_OCR_FIELD, MVI_COMPARE_LOGIC, False),
BC_EN: (BC_OCR_FIELD, BC_COMPARE_LOGIC, False),
HMH_EN: (HMH_OCR_FIELD, HMH_COMPARE_LOGIC, False),
AFC_CONTRACT_EN: (HT_FIELD, HT_COMPARE_LOGIC, False),
BD_EN: (BD_FIELD, BD_COMPARE_LOGIC, False),
HIL_CONTRACT_1_EN: (HIL_CONTRACT_1_FIELD, HIL_CONTRACT_1_COMPARE_LOGIC, False),
HIL_CONTRACT_2_EN: (HIL_CONTRACT_2_FIELD, HIL_CONTRACT_2_COMPARE_LOGIC, False)
}
OCR_COMPARE_COMMENT = {
ID_EN: {
'idExpiryDate': '身份证疑似过期',
'customerName': '身份证姓名与系统不一致',
'idNum': '身份证号码与系统不一致',
'other': '请确认是否提供身份证件'
},
MVI_EN: {
'firstSubmmisonDate': '请确认发票开票日期，若发票开票日期早于首次提交审批日期则无法受理放款申请',
'customerName': '发票购买方姓名与系统不一致',
'idNum': '发票购买方证件号码与系统不一致',
'vinNo': '发票车架号与系统不一致',
'vehicleTransactionAmount': '发票车辆价格与系统不一致',
'productGroupName': '请确认是否提供发票'
},
BC_EN: {
'bankName': '银行卡开户行与系统不一致',
'accountNo': '银行卡卡号与系统不一致',
'bankVerificationStatus': '请确认是否提供银行卡'
},
AFC_CONTRACT_EN: {
'applicationId & applicationVersion': '',
'service': '',
'amount': '',
'financedAmount': '',
'associatedServicePrincipal': '',
'vinNo': '',
'accountNo': '',
'accountHolderName': '',
'bankName': '',
'term': '',
'applicationEntity': ''
},
HIL_CONTRACT_1_EN: {
},
HIL_CONTRACT_2_EN: {
},
BD_EN: {
'': ''
}
}
src/prese/compare.py 0 → 100644
import json
import consts
from common.tools.comparison import cp
empty_str = ''
def get_pos_compare_info(pos_info):
compare_info = {}
application_entity = pos_info.get('applicationEntity', 'AFC')
application_id = pos_info.get('applicationId', '')
application_version = str(pos_info.get('applicationVersion', ''))
first_submission_date = pos_info.get('firstSubmmisonDate', '')
application_id_version = '{0}-{1}'.format(application_id, application_version)
custr_name = custr_id = ''
# 身份证
individual_cus_info_list = pos_info.get('individualCusInfo', [])
for individual_cus_info in individual_cus_info_list:
customer_name = individual_cus_info.get('customerName', '').strip()
id_num = individual_cus_info.get('idNum', '')
if individual_cus_info.get('applicantType', '') == 'CUSTR':
custr_name = customer_name
custr_id = id_num
if individual_cus_info.get('idType', '') == consts.ID_TYPE:
field_input = [(consts.ID_FIELDS[0], customer_name),
(consts.ID_FIELDS[1], id_num),
(consts.ID_FIELDS[2], individual_cus_info.get('idExpiryDate', ''))]
compare_info.setdefault(consts.ID_EN, []).append(field_input)
# 新车发票
vehicle_info = pos_info.get('vehicleInfo', {})
vin_no = vehicle_info.get('vinNo', '')
amount = str(vehicle_info.get('vehicleTransactionAmount', '0.0'))
field_input = [
(consts.MVI_FIELDS[0], vin_no),
(consts.MVI_FIELDS[1], amount),
(consts.MVI_FIELDS[2], custr_name),
(consts.MVI_FIELDS[3], custr_id),
(consts.MVI_FIELDS[4], first_submission_date),
(consts.MVI_FIELDS[5], consts.MVI_FPL_VALUE),
]
compare_info.setdefault(consts.MVI_EN, []).append(field_input)
# 银行卡
bank_info = pos_info.get('bankInfo', {})
bank_status = bank_info.get('bankVerificationStatus', '')
if bank_status != 'PASS':
account_no = bank_info.get('accountNo', '')
bank_name = bank_info.get('bankName', '')
field_input = [
(consts.BC_FIELDS[0], account_no),
(consts.BC_FIELDS[1], bank_name),
(consts.BC_FIELDS[2], consts.BC_TYPE_VALUE) # 是否为借记卡
]
compare_info.setdefault(consts.BC_EN, []).append(field_input)
# 抵押登记豁免函
quotationt_info = pos_info.get('quotationtInfo', {})
if quotationt_info.get('mortgageType', '') == 'MOTGF':
field_input = [
(consts.HMH_FIELDS[0], custr_name),
(consts.HMH_FIELDS[1], custr_id),
(consts.HMH_FIELDS[2], application_id_version),
# 宝马金融（中国）有限公司 先锋国际融资租赁有限公司
(consts.HMH_FIELDS[3], consts.HMH_CHANNEL_MAP.get(application_entity, '')),
(consts.HMH_FIELDS[4], consts.HAVE_CN),
]
compare_info.setdefault(consts.HMH_EN, []).append(field_input)
# 保单
insurance_info = pos_info.get('insuranceInfo', {})
if insurance_info.get('insuranceType') == 'ITCOM':
field_input = [
(consts.BD_FIELDS[0], [custr_name, ]),
(consts.BD_FIELDS[1], [custr_id, ]),
(consts.BD_FIELDS[2], vin_no)
]
compare_info.setdefault(consts.BD_EN, []).append(field_input)
# HIL合同
if application_entity in consts.HIL_SET:
pass
# AFC合同
else:
pass
return compare_info
def pre_compare_process(compare_info, ocr_res_dict, id_res_list):
compare_result = {}
for license_en, items_list in compare_info.items():
if license_en == consts.ID_EN:
for field_list in items_list:
result_list = pre_compare_license_id(license_en, id_res_list, field_list)
compare_result.setdefault(license_en, []).append(result_list)
elif license_en in [consts.HIL_CONTRACT_1_EN, consts.HIL_CONTRACT_2_EN,
consts.HIL_CONTRACT_3_EN, consts.AFC_CONTRACT_EN]:
pass
else:
for field_list in items_list:
result_list = pre_compare_license(license_en, ocr_res_dict, field_list)
compare_result.setdefault(license_en, []).append(result_list)
return compare_result
def pre_compare_license(license_en, ocr_res_dict, field_list):
ocr_field, compare_logic, no_find_comment = consts.PRE_COMPARE_LOGIC_MAP[license_en]
is_find = False
result_field_list = []
ocr_res_str = ocr_res_dict.get(ocr_field)
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
length = len(ocr_res_list)
for res_idx in range(length-1, -1, -1):
if is_find:
break
for idx, (name, value) in enumerate(field_list):
ocr_str = ocr_res_list[res_idx].get(compare_logic[name][0])
if isinstance(ocr_str, str):
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
else:
result = consts.RESULT_N
ocr_str = empty_str
if idx == 0 and result == consts.RESULT_N and length > 1:
break
is_find = True
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((value, ocr_str, result, compare_logic[name][3]))
if not is_find:
result_field_list.append((empty_str, empty_str, consts.RESULT_N, no_find_comment))
return result_field_list
def pre_compare_license_id(license_en, id_res_list, field_list):
ocr_field, compare_logic, no_find_comment = consts.PRE_COMPARE_LOGIC_MAP[license_en]
is_find = False
result_field_list = []
field_last_idx = len(field_list) - 1
for ca_or_se_idx, ocr_res_str in enumerate(id_res_list):
if is_find:
break
if ocr_res_str is not None:
ocr_res_list = json.loads(ocr_res_str)
length = len(ocr_res_list)
# 身份证、居住证 过期期限特殊处理
expiry_dates = set()
key = compare_logic.get('idExpiryDate')[0]
for ocr_res in ocr_res_list:
if key in ocr_res:
expiry_dates.add(ocr_res[key])
for res_idx in range(length-1, -1, -1):
if is_find:
break
result_field_list.clear()
for idx, (name, value) in enumerate(field_list):
ocr_str = ocr_res_list[res_idx].get(compare_logic[name][0])
if not isinstance(ocr_str, str):
result = consts.RESULT_N
ocr_str = empty_str
else:
result = getattr(cp, compare_logic[name][1])(value, ocr_str, **compare_logic[name][2])
if idx == 0 and result == consts.RESULT_N:
if ca_or_se_idx == 0:
break
elif length > 1:
break
# 过期期限特殊处理
if name == 'idExpiryDate' and result == consts.RESULT_N:
if len(expiry_dates) == 0:
ocr_str = empty_str
result = consts.RESULT_N
else:
for expiry_date in expiry_dates:
expiry_date_res = getattr(cp, compare_logic[name][1])(value, expiry_date,
**compare_logic[name][2])
if expiry_date_res == consts.RESULT_N:
ocr_str = expiry_date
break
else:
ocr_str = empty_str
result = consts.RESULT_Y
if ca_or_se_idx == 0 and result == consts.RESULT_N:
break
if ca_or_se_idx == 1:
is_find = True
elif idx == field_last_idx:
is_find = True
if isinstance(value, list):
value = json.dumps(value, ensure_ascii=False)
result_field_list.append((value, ocr_str, result, compare_logic[name][3]))
if not is_find:
result_field_list = [(empty_str, empty_str, consts.RESULT_N, no_find_comment)]
return result_field_list
def rebuild_result(compare_result):
# compare_result = {
# "is_pass": True,
# "particulars": [{
# "object_name": "PRC ID",
# "fields": [{
# "input": "张三",
# "ocr": "张三",
# "field_is_pass": True,
# "comments": "身份证姓名与系统不一致"
# }]
# }]
# }
is_pass = True
particulars_list = []
for license_en, items_list in compare_result.items():
for result_field_list in items_list:
field_list = []
for input_str, ocr_str, result, comments in result_field_list:
if result == consts.RESULT_N:
is_pass = False
field_list.append(
{
'input': input_str,
'ocr': ocr_str,
'field_is_pass': result == consts.RESULT_Y,
'comments': comments,
}
)
particulars_list.append(
{
'object_name': license_en,
'fields': field_list
}
)
rebuild_compare_result = {
'is_pass': is_pass,
'particulars': particulars_list
}
return rebuild_compare_result
def pre_compare(pos_info, ocr_res_dict, id_res_list):
compare_info = get_pos_compare_info(pos_info)
compare_result = pre_compare_process(compare_info, ocr_res_dict, id_res_list)
rebuild_compare_result = rebuild_result(compare_result)
return rebuild_compare_result
def get_empty_result():
empty_result = {
"is_pass": False,
"particulars": [{
"object_name": "",
"fields": [{
"input": "",
"ocr": "",
"field_is_pass": False,
"comments": "未查找到OCR识别结果"
}]
}]
}
return empty_result
src/prese/consts.py 0 → 100644
HIL_SET = {'HIL', 'HIl', 'HiL', 'Hil', 'hIL', 'hIl', 'hiL', 'hil', 'CO00002', 'SF5_CL'}
ID_TYPE = 'ITPRC'
ID_FIELDS = ['姓名', '身份证号码', '有效期限']
MVI_FIELDS = ['车架号', '价税合计(小写)', '购买方姓名', '购买方证件号码', '开票日期', '发票联']
MVI_FPL_VALUE = '发票联'
BC_FIELDS = ['卡号', '开户行名称', '卡片类型']
BC_TYPE_VALUE = '借记卡'
HMH_FIELDS = ['借款人/承租人姓名', '借款人/承租人证件号', '申请号', '渠道', '签字']
HMH_CHANNEL_MAP = {
'AFC': '宝马金融（中国）有限公司',
'HIL': '先锋国际融资租赁有限公司'
}
HAVE_CN = '有'
BD_FIELDS = ['被保险人姓名', '被保险人证件号码', '车架号']
ID_EN = 'idCard'
MVI_EN = 'newCar Invoice'
BC_EN = 'Bank Card'
HMH_EN = 'Mortgage Waiver Letter'
BD_EN = 'Insurance'
ID_OCR_FIELD = 'ic_ocr'
MVI_OCR_FIELD = 'mvi_ocr'
BC_OCR_FIELD = 'bc_ocr'
HMH_OCR_FIELD = 'hmh_ocr'
BD_FIELD = 'bd_ocr'
MVI_COMPARE_LOGIC = {
MVI_FIELDS[0]: ('车辆识别代码', 'se_common_compare', {}, '发票车架号与系统不一致'),
MVI_FIELDS[1]: ('价税合计小写', 'se_amount_compare', {}, '发票车辆价格与系统不一致'),
MVI_FIELDS[2]: ('购方名称', 'se_name_compare', {'is_passport': True, 'replace_kuohao': True}, '发票购买方姓名与系统不一致'),
MVI_FIELDS[3]: ('购买方身份证号或组织机构代码', 'se_common_compare', {}, '发票购买方证件号码与系统不一致'),
MVI_FIELDS[4]: ('开票日期', 'se_date_compare_2', {'three_month': True}, '请确认发票开票日期，若发票开票日期早于首次提交审批日期则无法受理放款申请'),
MVI_FIELDS[5]: ('发票类型', 'se_common_compare', {}, '发票疑似非发票联'),
}
BC_COMPARE_LOGIC = {
BC_FIELDS[0]: ('CardNum', 'se_common_compare', {'remove_space': True}, '银行卡卡号与系统不一致'),
BC_FIELDS[1]: ('BankName', 'se_both_contain_compare', {}, '银行卡开户行与系统不一致'),
BC_FIELDS[2]: ('CardType', 'se_common_compare', {}, '银行卡疑似非借记卡'),
}
HMH_COMPARE_LOGIC = {
HMH_FIELDS[0]: ('借款/承租人姓名', 'se_name_compare', {}, '抵押登记豁免函借款人/承租人姓名与系统不符'),
HMH_FIELDS[1]: ('证件号码', 'se_common_compare', {}, '抵押登记豁免函借款人/承租人证件号码与系统不符'),
HMH_FIELDS[2]: ('合同编号', 'se_common_compare', {}, '抵押登记豁免函申请号与系统不符'),
HMH_FIELDS[3]: ('渠道', 'se_channel_compare', {}, '抵押登记豁免函渠道与系统不符'),
HMH_FIELDS[4]: ('借款人签字/盖章', 'se_common_compare', {}, '抵押登记豁免函无签字'),
}
BD_COMPARE_LOGIC = {
BD_FIELDS[0]: ('被保险人姓名', 'super_list_compare', {'method': 'name'}, '保单被保险人姓名与系统不一致'),
BD_FIELDS[1]: ('被保险人证件号码', 'super_list_compare', {'method': 'common', 'is_bd_id': True}, '保单身份证号需人工核查'),
BD_FIELDS[2]: ('车架号', 'se_common_compare', {}, '保单车架号与系统不一致'),
}
PRE_COMPARE_LOGIC_MAP = {
ID_EN: (ID_OCR_FIELD, ID_COMPARE_LOGIC, '请确认是否提供身份证件'),
MVI_EN: (MVI_OCR_FIELD, MVI_COMPARE_LOGIC, '请确认是否提供发票'),
BC_EN: (BC_OCR_FIELD, BC_COMPARE_LOGIC, '请确认是否提供银行卡'),
HMH_EN: (HMH_OCR_FIELD, HMH_COMPARE_LOGIC, '请确认是否已完成抵押登记豁免函签署'),
BD_EN: (BD_FIELD, BD_COMPARE_LOGIC, '请确认是否提供保单'),
AFC_CONTRACT_EN: (HT_FIELD, HT_COMPARE_LOGIC, '请确认是否已完成车辆抵押贷款合同签署'),
HIL_CONTRACT_1_EN: (HIL_CONTRACT_1_FIELD, HIL_CONTRACT_1_COMPARE_LOGIC, '请确认是否已完成售后回租合同签署'),
HIL_CONTRACT_2_EN: (HIL_CONTRACT_2_FIELD, HIL_CONTRACT_2_COMPARE_LOGIC, '请确认是否已完成车辆租赁抵押合同签署'),
}
RESULT_Y = 'Y'
RESULT_N = 'N'
Styling with Markdown is supported
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!