import os import time import json import shutil import base64 import signal import difflib import requests import traceback from collections import Counter from datetime import datetime, date from django import db from django.utils import timezone from django.core.management import BaseCommand from multiprocessing import Process, Queue, Manager, Lock from settings import conf from common.mixins import LoggerMixin from common.tools.file_tools import write_zip_file from common.tools.pdf_to_img import PDFHandler from common.electronic_afc_contract.afc_contract_ocr import predict as afc_predict from common.electronic_hil_contract.hil_contract_ocr import predict as hil_predict from apps.doc import consts # from apps.doc.ocr.edms import EDMS, rh from apps.doc.ocr.ecm import ECM, rh from apps.doc.named_enum import KeywordsType, FailureReason, WorkflowName, ProcessName, RequestTeam, RequestTrigger from apps.doc.exceptions import EDMSException, OCR1Exception, OCR2Exception, OCR4Exception from apps.doc.ocr.wb import BSWorkbook from apps.doc.models import ( DocStatus, HILDoc, AFCDoc, Keywords, HILOCRResult, AFCOCRResult, AFCSEOCRResult, HILOCRReport, HILSEOCRResult, AFCOCRReport, DDARecords, IDBCRecords, Configs, ) from celery_compare.tasks import compare class Command(BaseCommand, LoggerMixin): def __init__(self): super().__init__() self.log_base = '[doc ocr process]' self.e_log_base = '[e-contract ocr process]' # 处理文件开关 self.switch = True # 睡眠时间 self.sleep_time_doc_get = float(conf.SLEEP_SECOND_DOC_GET) self.sleep_time_img_put = float(conf.SLEEP_SECOND_IMG_PUT) self.sleep_time_img_get = float(conf.SLEEP_SECOND_IMG_GET) self.sleep_time_task_get = float(conf.SLEEP_SECOND_TASK_GET) # 队列长度 self.img_queue_size = int(conf.IMG_QUEUE_SIZE) # 数据目录 self.data_dir = conf.DATA_DIR # DDA目录 self.dda_dir = os.path.join(self.data_dir, 'HIL', 'SF5-CL-S-1', 'DDA') self.dda_complete_dir = os.path.join(self.dda_dir, 'complete') self.dda_wanting_dir = os.path.join(self.dda_dir, 'wanting') # ocr相关 self.ocr_1_urls = conf.get_namespace('OCR_URL_1_') self.ocr_url_2 = conf.OCR_URL_2 self.ocr_url_2_bc = conf.OCR_URL_2_BC self.ocr_url_3 = conf.BC_URL self.ocr_url_4 = conf.IC_URL # EDMS web_service_api # self.edms = EDMS() self.edms = ECM() # 优雅退出信号:15 signal.signal(signal.SIGTERM, self.signal_handler) def signal_handler(self, sig, frame): self.switch = False # 停止处理文件 # def get_doc_object(self, task_str): # business_type, doc_id_str = task_str.split(consts.SPLIT_STR) # doc_id = int(doc_id_str) # doc_class = HILDoc if business_type == consts.HIL_PREFIX else AFCDoc # # doc_info = doc_class.objects.filter(id=doc_id, status=DocStatus.INIT.value).values( # # 'id', 'metadata_version_id', 'application_id', 'document_name', 'document_scheme').first() # doc = doc_class.objects.filter(id=doc_id).first() # return doc, business_type def get_doc_info(self): task_str, is_priority = rh.dequeue() if task_str is None: self.online_log.info('{0} [get_doc_info] [queue empty]'.format(self.log_base)) return None, None, None, None, None self.online_log.info('{0} [get_doc_info] [task={1}] [is_priority={2}]'.format( self.log_base, task_str, is_priority)) try: # doc, business_type = self.get_doc_object(task_str) info_tuple = task_str.split(consts.SPLIT_STR) if len(info_tuple) == 2: business_type, doc_id_str = info_tuple classify_1_str = classify_2_str = '0' rebuild_task_str = task_str else: business_type, doc_id_str, classify_1_str, classify_2_str = info_tuple rebuild_task_str = '{0}{1}{2}'.format(business_type, consts.SPLIT_STR, doc_id_str) doc_id = int(doc_id_str) doc_class = HILDoc if business_type == consts.HIL_PREFIX else AFCDoc doc = doc_class.objects.filter(id=doc_id).first() if doc is None: self.online_log.warn('{0} [get_doc_info] [doc not exist] [task_str={1}] [is_priority={2}]'.format( self.log_base, task_str, is_priority)) return None, None, None, None, None elif doc.status != DocStatus.INIT.value: self.online_log.warn('{0} [get_doc_info] [doc status error] [task_str={1}] [is_priority={2}] ' '[doc_status={3}]'.format(self.log_base, task_str, is_priority, doc.status)) return None, None, None, None, None doc.status = DocStatus.PROCESSING.value doc.start_time = timezone.now() doc.save() except Exception as e: rh.enqueue([task_str], is_priority) self.online_log.error('{0} [process error (get doc info in)] [error={1}]'.format( self.log_base, traceback.format_exc())) raise e else: self.online_log.info('{0} [get_doc_info] [db save end] [task_str={1}] [is_priority={2}]'.format( self.log_base, task_str, is_priority)) return doc, business_type, rebuild_task_str, classify_1_str, classify_2_str # def pdf_download(self, doc, pdf_path): # if not doc.application_id.startswith(consts.FIXED_APPLICATION_ID_PREFIX): # for times in range(consts.RETRY_TIMES): # try: # self.edms.download(pdf_path, doc.metadata_version_id) # except Exception as e: # self.online_log.warn('{0} [edms download failed] [times={1}] [pdf_path={2}] ' # '[error={3}]'.format(self.log_base, times, pdf_path, traceback.format_exc())) # edms_exc = str(e) # else: # break # else: # raise EDMSException(edms_exc) # self.online_log.info('{0} [edms download success] [pdf_path={1}]'.format(self.log_base, pdf_path)) def bs_process(self, wb, ocr_data, bs_summary, unknown_summary, classify, res_list, pno, ino, part_idx): sheets = ocr_data.get('data', []) if not sheets: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY)) return # confidence = ocr_data.get('confidence', 1) img_name = 'page_{0}_img_{1}_{2}'.format(pno, ino, part_idx) cells_exists = False for i, sheet in enumerate(sheets): cells = sheet.get('cells') if not cells: continue cells_exists = True sheet_name = '{0}_{1}'.format(img_name, i) ws = wb.create_sheet(sheet_name) for cell in cells: c1 = cell.get('start_column') r1 = cell.get('start_row') words = cell.get('words') ws.cell(row=r1 + 1, column=c1 + 1, value=words) # ['户名', '卡号', '页码', '回单验证码', '打印时间', '起始时间', '终止时间'] summary = sheet.get('summary') card = summary[1] if card is None: classify_dict = unknown_summary.setdefault(classify, {}) role = consts.UNKNOWN_ROLE if summary[0] is None else summary[0] role_dict = classify_dict.setdefault(role, {}) role_dict['classify'] = classify role_dict['role'] = role role_dict.setdefault('sheet', []).append(sheet_name) # role_dict.setdefault('confidence', []).append(confidence) code_list = role_dict.setdefault('code', []) pt_list = role_dict.setdefault('print_time', []) sd_list = role_dict.setdefault('start_date', []) ed_list = role_dict.setdefault('end_date', []) if summary[3] is not None: code_list.append((summary[2], summary[3])) if summary[4] is not None: pt_list.append(summary[4]) if summary[5] is not None: sd_list.append(summary[5]) if summary[6] is not None: ed_list.append(summary[6]) else: card_dict = bs_summary.setdefault(card, {}) card_dict['count'] = card_dict.get('count', 0) + 1 card_dict.setdefault('classify', []).append(classify) # card_dict.setdefault('confidence', []).append(confidence) card_dict.setdefault('sheet', []).append(sheet_name) role_list = card_dict.setdefault('role', []) role_set = card_dict.setdefault('role_set', set()) code_list = card_dict.setdefault('code', []) pt_list = card_dict.setdefault('print_time', []) sd_list = card_dict.setdefault('start_date', []) ed_list = card_dict.setdefault('end_date', []) if summary[0] is not None: role_list.append(summary[0]) role_set.add(summary[0]) if summary[3] is not None: code_list.append((summary[2], summary[3])) if summary[4] is not None: pt_list.append(summary[4]) if summary[5] is not None: sd_list.append(summary[5]) if summary[6] is not None: ed_list.append(summary[6]) if cells_exists: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS)) else: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY)) def contract_process(self, classify, ocr_data, contract_result, res_list, pno, ino, part_idx, img_path): contract_dict = ocr_data.get('data') if not contract_dict or contract_dict.get('page_num') is None or contract_dict.get('page_info') is None: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY)) return res_list.append((pno, ino, part_idx, consts.RES_SUCCESS)) page_num = contract_dict.get('page_num') if page_num.startswith('page_'): page_num_only = page_num.split('_')[-1] else: page_num_only = page_num rebuild_page_info = [] text_key = 'words' for key, value in contract_dict.get('page_info', {}).items(): if value is None: rebuild_page_info.append((key, )) elif text_key in value: if value[text_key] is None: rebuild_page_info.append((key,)) elif isinstance(value[text_key], str): rebuild_page_info.append((key, value[text_key])) elif isinstance(value[text_key], list): rebuild_page_info.append((key,)) for row_list in value[text_key]: rebuild_page_info.append(row_list) else: rebuild_page_info.append((key,)) for sub_key, sub_value in value.items(): if sub_value is None: rebuild_page_info.append((sub_key,)) elif text_key in sub_value: if sub_value[text_key] is None: rebuild_page_info.append((sub_key,)) elif isinstance(sub_value[text_key], str): rebuild_page_info.append((sub_key, sub_value[text_key])) elif isinstance(sub_value[text_key], list): rebuild_page_info.append((sub_key,)) for row_list in sub_value[text_key]: rebuild_page_info.append(row_list) # contract_result.setdefault(page_num_only, []).append(rebuild_page_info) contract_result.setdefault(classify, dict()).setdefault(page_num_only, []).append(rebuild_page_info) # def rebuild_result(self, ocr_data, classify, img_path): # license_data = ocr_data.get('data') # if not license_data: # return # if classify == consts.IC_CLASSIFY: # rebuild_data_dict = {} # card_type = license_data.get('type', '') # is_ic = card_type.startswith('身份证') # is_info_side = card_type.endswith('信息面') # rebuild_data_dict['类别'] = '0' if is_ic else '1' # if is_ic: # field_map = consts.IC_MAP_0 if is_info_side else consts.IC_MAP_1 # else: # field_map = consts.RP_MAP_0 if is_info_side else consts.RP_MAP_1 # for write_field, search_field in field_map: # rebuild_data_dict[write_field] = license_data.get('words_result', {}).get(search_field, {}).get('words', '') # if not is_info_side: # start_time = license_data.get('words_result', {}).get('签发日期', {}).get('words', '') # end_time = license_data.get('words_result', {}).get('失效日期', {}).get('words', '') # rebuild_data_dict['有效期限'] = '{0}-{1}'.format(start_time, end_time) # return [rebuild_data_dict] # elif classify == consts.MVC_CLASSIFY: # # license_data[consts.IMG_PATH_KEY] = img_path # rebuild_data_dict = {} # mvc_page = license_data.pop('page', 'VehicleRCI') # mvc_res = license_data.pop('results', {}) # if mvc_page == 'VehicleRegArea': # rebuild_data_dict['机动车登记证书编号'] = mvc_res.get('机动车登记证书编号', {}).get('words', '') # for register_info in mvc_res.get('登记信息', []): # register_info.pop('register_type', None) # register_info.pop('register_type_name', None) # for cn_key, detail_dict in register_info.items(): # rebuild_data_dict.setdefault(cn_key, []).append( # detail_dict.get('words', '')) # else: # for cn_key, detail_dict in mvc_res.items(): # rebuild_data_dict[cn_key] = detail_dict.get('words', '') # del mvc_res # return [rebuild_data_dict] # elif classify == consts.MVI_CLASSIFY: # rebuild_data_dict = {} # mvi_res = license_data.pop('result', {}) # for en_key, detail_dict in mvi_res.items(): # rebuild_data_dict[detail_dict.get('chinese_key', '')] = detail_dict.get('words', '') # return [rebuild_data_dict] # elif classify == consts.UCI_CLASSIFY: # rebuild_data_dict = {} # mvi_res = license_data.pop('result', {}) # for en_key, detail_dict in mvi_res.items(): # rebuild_data_dict[detail_dict.get('chinese_key', '')] = detail_dict.get('words', '') # return [rebuild_data_dict] def license1_process(self, ocr_data, license_summary, classify, res_list, pno, ino, part_idx, img_path, do_dda, dda_id_bc_mapping): # 类别:'0'身份证, '1'居住证 license_data = ocr_data.get('data') if not license_data: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY)) return if isinstance(license_data, dict): pre, suf = os.path.splitext(img_path) base64_img = license_data.pop('base64_img', '') is_save = True if len(base64_img) > 0 else False section_img_path = '{0}_{1}{2}'.format(pre, part_idx, suf) if is_save else img_path if is_save: try: with open(section_img_path, "wb") as fh: fh.write(base64.b64decode(base64_img.encode())) except Exception as e: self.online_log.warn( '{0} [section img save failed] [img_path={1}]' ' [part_idx={2}]'.format(self.log_base, img_path, part_idx)) else: is_save = False section_img_path = img_path # 保单 if classify == consts.INSURANCE_CLASSIFY: product_result = ['', '', ''] for product in license_data.get('result', {}).get('productList', []): name = product.get('name', {}).get('words', '') if name.find('机动车损失') != -1: product_result[0] = product.get('coverage', {}).get('words', '') product_result[2] = product.get('deductible_franchise', {}).get('words', '') elif name.find('第三者责任') != -1: product_result[1] = product.get('coverage', {}).get('words', '') special_str = license_data.get('result', {}).get('1stBeneficiary', {}).get('words', '') special = '无' if special_str.find('宝马') != -1 or special_str.find('先锋国际融资租赁有限公司') != -1: special = '有' insurance_ocr_result = { '被保险人姓名': license_data.get('result', {}).get('insured', {}).get('name', {}).get('words', ''), '被保险人证件号码': license_data.get('result', {}).get('insured', {}).get('certiCode', {}).get('words', ''), '车架号': license_data.get('result', {}).get('vehicle', {}).get('VIN', {}).get('words', ''), '机动车损失保险金额': product_result[0], '机动车第三者责任保险金额': product_result[1], '机动车损失保险绝对免赔率/绝对免赔额': product_result[2], '保险费合计': license_data.get('result', {}).get('premiumSum', {}).get('words', ''), '保险起始日期': license_data.get('result', {}).get('startDate', {}).get('words', ''), '保险截止日期': license_data.get('result', {}).get('endDate', {}).get('words', ''), '保单章': license_data.get('result', {}).get('seal', {}).get('words', ''), '特别约定第一受益人': special, consts.IMG_PATH_KEY: img_path, consts.SECTION_IMG_PATH_KEY: section_img_path, } # position_dict = { # '': {consts.FIELD_POSITION_KEY: {}} # } # insurance_ocr_result[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(insurance_ocr_result) # DDA elif classify == consts.DDA_CLASSIFY: pro = ocr_data.get('confidence', 0) if pro < consts.DDA_PRO_MIN: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY)) return dda_ocr_result = {} position_dict = {} for key, value in license_data.get('result', {}).items(): dda_ocr_result[key] = value.get('words', '') position_dict[key] = { consts.FIELD_POSITION_KEY: value.get('position', {}) } dda_ocr_result[consts.DDA_IMG_PATH] = img_path dda_ocr_result[consts.DDA_PRO] = pro dda_ocr_result[consts.IMG_PATH_KEY] = img_path dda_ocr_result[consts.SECTION_IMG_PATH_KEY] = section_img_path dda_ocr_result[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(dda_ocr_result) # 抵押登记豁免函 elif classify == consts.HMH_CLASSIFY: hmh_ocr_result = {} position_dict = {} for key, value in license_data.get('words_result', {}).items(): hmh_ocr_result[key] = value.get('words', '') location_list = value.get('location', [-1, -1, -1, -1]) if len(location_list) == 4: position_dict[key] = { consts.FIELD_POSITION_KEY: { 'top': location_list[1], 'left': location_list[0], 'height': location_list[-1] - location_list[1], 'width': location_list[2] - location_list[0] } } hmh_ocr_result[consts.IMG_PATH_KEY] = img_path hmh_ocr_result[consts.SECTION_IMG_PATH_KEY] = section_img_path hmh_ocr_result[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(hmh_ocr_result) # 二手车交易凭证 elif classify == consts.JYPZ_CLASSIFY: jypz_ocr_result = {} position_dict = {} for key, value in license_data.get('result', {}).items(): jypz_ocr_result[key] = value.get('words', '') position_dict[key] = { consts.FIELD_POSITION_KEY: value.get('position', {}) } jypz_ocr_result[consts.IMG_PATH_KEY] = img_path jypz_ocr_result[consts.SECTION_IMG_PATH_KEY] = section_img_path jypz_ocr_result[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(jypz_ocr_result) # 车辆登记证 3/4页结果整合 elif classify == consts.MVC_CLASSIFY: rebuild_data_dict = {} position_dict = {} mvc_page = license_data.pop('page', 'VehicleRCI') mvc_res = license_data.pop('results', {}) if mvc_page == 'VehicleRegArea': rebuild_data_dict['机动车登记证书编号'] = mvc_res.get('机动车登记证书编号', {}).get('words', '') code_position_list = mvc_res.get('机动车登记证书编号', {}).get('position', [0, 0, 0, 0]) if len(code_position_list) == 4: position_dict['机动车登记证书编号'] = { consts.FIELD_POSITION_KEY: { 'top': code_position_list[1], 'left': code_position_list[0], 'height': code_position_list[-1], 'width': code_position_list[2], } } for register_info in mvc_res.get('登记信息', []): register_info.pop('register_type', None) register_info.pop('register_type_name', None) for cn_key, detail_dict in register_info.items(): rebuild_data_dict.setdefault(cn_key, []).append( detail_dict.get('words', '')) tmp_position_list = detail_dict.get('position', [0, 0, 0, 0]) if len(tmp_position_list) == 4: position_dict.setdefault(cn_key, []).append( { consts.FIELD_POSITION_KEY: { 'top': tmp_position_list[1], 'left': tmp_position_list[0], 'height': tmp_position_list[-1], 'width': tmp_position_list[2], } } ) rebuild_data_dict[consts.ALL_POSITION_KEY_2] = position_dict rebuild_data_dict[consts.IMG_PATH_KEY_2] = img_path rebuild_data_dict[consts.SECTION_IMG_PATH_KEY_2] = section_img_path else: for cn_key, detail_dict in mvc_res.items(): rebuild_data_dict[cn_key] = detail_dict.get('words', '') position_list = detail_dict.get('position', [0, 0, 0, 0]) if len(position_list) == 4: position_dict[cn_key] = { consts.FIELD_POSITION_KEY: { 'top': position_list[1], 'left': position_list[0], 'height': position_list[-1], 'width': position_list[2], } } rebuild_data_dict[consts.ALL_POSITION_KEY] = position_dict rebuild_data_dict[consts.IMG_PATH_KEY] = img_path rebuild_data_dict[consts.SECTION_IMG_PATH_KEY] = section_img_path del mvc_res license_summary.setdefault(classify, []).append(rebuild_data_dict) # for mvc_dict in license_data: # mvc_dict[consts.IMG_PATH_KEY] = img_path # try: # mvc_page = mvc_dict.pop('page') # except Exception as e: # pass # else: # if mvc_page == 'VehicleRegArea': # mvc_res = mvc_dict.pop('results', {}) # mvc_dict['机动车登记证书编号'] = mvc_res.get('register_no', {}).get('words', '') # for register_info in mvc_res.get('register_info', []): # for detail_dict in register_info.get('details', {}).values(): # mvc_dict.setdefault(detail_dict.get('chinese_key', '未知'), []).append( # detail_dict.get('words', '')) # del mvc_res # license_summary.setdefault(classify, []).extend(license_data) # 身份证真伪 elif classify == consts.IC_CLASSIFY: id_card_dict = {} position_dict = {} card_type = license_data.get('type', '') is_ic = card_type.startswith('身份证') is_info_side = card_type.endswith('信息面') id_card_dict['类别'] = '0' if is_ic else '1' if is_ic: field_map = consts.IC_MAP_0 if is_info_side else consts.IC_MAP_1 else: field_map = consts.RP_MAP_0 if is_info_side else consts.RP_MAP_1 for write_field, search_field in field_map: id_card_dict[write_field] = license_data.get('words_result', {}).get(search_field, {}).get('words', '') location_list = license_data.get('words_result', {}).get(search_field, {}).get( 'location', [-1, -1, -1, -1]) if len(location_list) == 4: position_dict[write_field] = { consts.FIELD_POSITION_KEY: { 'top': location_list[1], 'left': location_list[0], 'height': location_list[-1] - location_list[1], 'width': location_list[2] - location_list[0] } } if not is_info_side: start_time = license_data.get('words_result', {}).get('签发日期', {}).get('words', '') end_time = license_data.get('words_result', {}).get('失效日期', {}).get('words', '') id_card_dict['有效期限'] = '{0}-{1}'.format(start_time, end_time) end_time_location_list = license_data.get('words_result', {}).get('失效日期', {}).get( 'location', [-1, -1, -1, -1]) if len(end_time_location_list) == 4: position_dict['有效期限'] = { consts.FIELD_POSITION_KEY: { 'top': end_time_location_list[1], 'left': end_time_location_list[0], 'height': end_time_location_list[-1] - end_time_location_list[1], 'width': end_time_location_list[2] - end_time_location_list[0] } } if not is_info_side: id_card_dict[consts.IMG_PATH_KEY_2] = img_path id_card_dict[consts.ALL_POSITION_KEY_2] = position_dict id_card_dict[consts.SECTION_IMG_PATH_KEY_2] = section_img_path else: id_card_dict[consts.ALL_POSITION_KEY] = position_dict id_card_dict[consts.IMG_PATH_KEY] = img_path id_card_dict[consts.SECTION_IMG_PATH_KEY] = section_img_path if is_ic and is_save: card_type = -1 json_data_4 = { 'mode': 1, 'user_info': { 'image_content': base64_img, }, 'options': { 'distinguish_type': 1, 'auto_rotate': True, }, } for times in range(consts.RETRY_TIMES): try: start_time = time.time() ocr_4_response = requests.post(self.ocr_url_4, json=json_data_4) if ocr_4_response.status_code != 200: raise OCR4Exception('ocr_4 status code: {0}'.format(ocr_4_response.status_code)) except Exception as e: self.online_log.warn( '{0} [ocr_4 failed] [times={1}] [img_path={2}] [error={3}]'.format( self.log_base, times, img_path, traceback.format_exc())) else: ocr_4_res = ocr_4_response.json() end_time = time.time() speed_time = int(end_time - start_time) if ocr_4_res.get('code') == 0 and ocr_4_res.get('result', {}).get('rtn') == 0: card_type = ocr_4_res.get('result', {}).get( 'idcard_distinguish_result', {}).get('result', -1) self.online_log.info( '{0} [ocr_4 success] [img_path={1}] [speed_time={2}]'.format( self.log_base, img_path, speed_time)) break else: self.online_log.warn( '{0} [ocr_4 failed] [img_path={1}]'.format(self.log_base, img_path)) id_card_dict[consts.IC_TURE_OR_FALSE] = consts.IC_RES_MAPPING.get(card_type) if do_dda and isinstance(id_card_dict.get(consts.IC_KEY_FIELD[0]), str) and \ isinstance(id_card_dict.get(consts.IC_KEY_FIELD[1]), str): ic_name = id_card_dict.get(consts.IC_KEY_FIELD[0], '').strip() ic_id = id_card_dict.get(consts.IC_KEY_FIELD[1], '').strip() if len(ic_name) > 0 and len(ic_id) > 0: dda_id_bc_mapping.setdefault(consts.IC_FIELD, []).append((ic_name, ic_id, img_path)) license_summary.setdefault(classify, []).append(id_card_dict) # 购车发票 & 二手车发票 elif classify == consts.MVI_CLASSIFY or classify == consts.UCI_CLASSIFY: rebuild_data_dict = {} position_dict = {} mvi_res = license_data.pop('result', {}) for en_key, detail_dict in mvi_res.items(): rebuild_data_dict[detail_dict.get('chinese_key', '')] = detail_dict.get('words', '') position_dict[detail_dict.get('chinese_key', '')] = { consts.FIELD_POSITION_KEY: detail_dict.get('position', {}) } rebuild_data_dict[consts.IMG_PATH_KEY] = img_path rebuild_data_dict[consts.SECTION_IMG_PATH_KEY] = section_img_path rebuild_data_dict[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(rebuild_data_dict) # 其他 else: for res_dict in license_data: res_dict[consts.IMG_PATH_KEY] = img_path res_dict[consts.SECTION_IMG_PATH_KEY] = section_img_path license_summary.setdefault(classify, []).extend(license_data) res_list.append((pno, ino, part_idx, consts.RES_SUCCESS)) def license2_process(self, ocr_res_2, license_summary, pid, classify, res_list, pno, ino, part_idx, img_path, do_dda, dda_id_bc_mapping, file_data): if ocr_res_2.get('ErrorCode') in consts.SUCCESS_CODE_SET: res_list.append((pno, ino, part_idx, consts.RES_SUCCESS)) if pid == consts.BC_PID: # 银行卡 # res_dict = {} # for en_key, chn_key in consts.BC_FIELD: # res_dict[chn_key] = ocr_res_2.get(en_key, '') ocr_res_2[consts.IMG_PATH_KEY] = img_path license_summary.setdefault(classify, []).append(ocr_res_2) if do_dda and isinstance(ocr_res_2.get(consts.BC_KEY_FIELD), str): bc_no = ocr_res_2[consts.BC_KEY_FIELD].strip() if len(bc_no) > 0: dda_id_bc_mapping.setdefault(consts.BC_FIELD, []).append((bc_no, img_path)) else: # 营业执照等 pre, suf = os.path.splitext(img_path) src_section_img_path = img_path if file_data is None else '{0}_{1}{2}'.format(pre, part_idx, suf) is_save = False for res_idx, result_dict in enumerate(ocr_res_2.get('ResultList', [])): image_data = result_dict.get('image_data', '') if len(image_data) > 0: position = {} angle = 0 section_img_path = '{0}_{1}_{2}{3}'.format(pre, part_idx, res_idx, suf) try: with open(section_img_path, "wb") as fh: fh.write(base64.b64decode(image_data.encode())) except Exception as e: self.online_log.warn( '{0} [section img save failed] [img_path={1}]' ' [part_idx={2}] [res_idx={3}]'.format(self.log_base, img_path, part_idx, res_idx)) else: is_save = True section_img_path = src_section_img_path position = result_dict.get('position', {}) angle = result_dict.get('angle', 0) res_dict = {} position_dict = {} for field_dict in result_dict.get('FieldList', []): res_dict[field_dict.get('chn_key', '')] = field_dict.get('value', '') position_dict[field_dict.get('chn_key', '')] = { consts.FIELD_POSITION_KEY: field_dict.get('position', {}), consts.FIELD_QUAD_KEY: field_dict.get('quad', []), } position_dict[consts.POSITION_KEY] = position position_dict[consts.ANGLE_KEY] = angle res_dict[consts.IMG_PATH_KEY] = img_path res_dict[consts.SECTION_IMG_PATH_KEY] = section_img_path res_dict[consts.ALL_POSITION_KEY] = position_dict license_summary.setdefault(classify, []).append(res_dict) if is_save and file_data is not None: try: with open(src_section_img_path, "wb") as fh: fh.write(base64.b64decode(file_data.encode())) except Exception as e: self.online_log.warn( '{0} [section img save failed] [img_path={1}]' ' [part_idx={2}]'.format(self.log_base, img_path, part_idx)) else: res_list.append((pno, ino, part_idx, consts.RES_FAILED_2)) @staticmethod def license_rebuild(license_summary): ic_merge = False rp_merge = False for classify in (consts.IC_CLASSIFY, consts.MVI_CLASSIFY, consts.MVC_CLASSIFY): license_list = license_summary.get(classify) if not license_list: continue if classify == consts.IC_CLASSIFY: # 身份证、居住证分开,先正面,后反面 key, _, _ = consts.FIELD_ORDER_MAP.get(classify) ic_side1_list = [] ic_side2_list = [] rp_side1_list = [] rp_side2_list = [] for license_dict in license_list: is_rp = license_dict.pop('类别', '0') if key in license_dict: if is_rp == '1': rp_side2_list.append(license_dict) else: ic_side2_list.append(license_dict) elif is_rp == '1': rp_side1_list.append(license_dict) else: ic_side1_list.append(license_dict) ic_merge = len(ic_side1_list) == len(ic_side2_list) == 1 rp_merge = len(rp_side1_list) == len(rp_side2_list) == 1 ic_side1_list.extend(ic_side2_list) rp_side1_list.extend(rp_side2_list) if ic_side1_list: # license_list = ic_side1_list license_summary[classify] = ic_side1_list else: license_summary.pop(classify, None) if rp_side1_list: license_summary[consts.RP_CLASSIFY] = rp_side1_list ic_side1_list = ic_side2_list = rp_side1_list = rp_side2_list = None if classify == consts.MVI_CLASSIFY: # 机动车销售统一发票, 增加不含税价(逻辑计算) for license_dict in license_list: price = '' rate_str = license_dict.get('增值税税率') price_total_str = license_dict.get('价税合计小写') if rate_str is not None and price_total_str is not None: try: rate = int(rate_str.rstrip('%')) price_total = float(price_total_str) except Exception as e: pass else: price = round(price_total * 100 / (rate + 100), 2) license_dict['不含税价(逻辑计算)'] = price if classify == consts.MVC_CLASSIFY: # 机动车登记证先1/2页,后3/4页 key, _, _ = consts.FIELD_ORDER_MAP.get(classify) page_1_2 = [] page_3_4 = [] for license_dict in license_list: if key in license_dict: page_3_4.append(license_dict) else: page_1_2.append(license_dict) page_1_2.extend(page_3_4) license_summary[classify] = page_1_2 page_1_2 = page_3_4 = None return ic_merge, rp_merge def parse_img_path(self, img_path): img_name, _ = os.path.splitext(os.path.basename(img_path)) part_list = img_name.split('_') # page_7_img_11_0 return int(part_list[1])+1, int(part_list[3])+1 def get_most(self, value_list): if value_list: most_common = Counter(value_list).most_common(1) return most_common[0][0] if most_common else None def date_format(self, date_str, format_str): try: date_res = datetime.strptime(date_str, format_str).date() except Exception as e: return else: return date_res def get_validate_date(self, date_list): for date_str in date_list: for format_str in consts.DATE_FORMAT: date_res = self.date_format(date_str, format_str) if isinstance(date_res, date): return date_res def merge_card(self, bs_summary): classify_info = {} merged_bs_summary = {} sorted_card = sorted(bs_summary.keys(), key=lambda x: bs_summary[x]['count'], reverse=True) for main_card in sorted_card: if bs_summary.get(main_card) is None: continue merged_bs_summary[main_card] = bs_summary.pop(main_card) del merged_bs_summary[main_card]['count'] merge_cards = [] for card in bs_summary.keys(): if difflib.SequenceMatcher(None, main_card, card).quick_ratio() > consts.CARD_RATIO: merged_bs_summary[main_card]['classify'].extend(bs_summary[card]['classify']) # merged_bs_summary[main_card]['confidence'].extend(bs_summary[card]['confidence']) merged_bs_summary[main_card]['sheet'].extend(bs_summary[card]['sheet']) merged_bs_summary[main_card]['role'].extend(bs_summary[card]['role']) merged_bs_summary[main_card]['role_set'].update(bs_summary[card]['role_set']) merged_bs_summary[main_card]['code'].extend(bs_summary[card]['code']) merged_bs_summary[main_card]['print_time'].extend(bs_summary[card]['print_time']) merged_bs_summary[main_card]['start_date'].extend(bs_summary[card]['start_date']) merged_bs_summary[main_card]['end_date'].extend(bs_summary[card]['end_date']) merge_cards.append(card) for card in merge_cards: del bs_summary[card] most_classify = self.get_most(merged_bs_summary[main_card]['classify']) classify_count = classify_info.get(most_classify, 0) classify_info[most_classify] = classify_count + 1 merged_bs_summary[main_card]['classify'] = most_classify merged_bs_summary[main_card]['role'] = self.get_most(merged_bs_summary[main_card]['role']) del bs_summary return merged_bs_summary, classify_info def prune_bs_summary(self, bs_summary): for summary in bs_summary.values(): del summary['count'] summary['classify'] = self.get_most(summary['classify']) summary['role'] = self.get_most(summary['role']) return bs_summary def rebuild_bs_summary(self, bs_summary, unknown_summary): # bs_summary = { # '卡号': { # 'count': 100, # 'classify': [], # 'confidence': [], # 'role': [], # 'code': [('page', 'code')], # 'print_time': [], # 'start_date': [], # 'end_date': [], # 'sheet': ['sheet_name'] # } # } # # unknown_summary = { # 0: { # '户名': { # 'classify': 0, # 'confidence': [], # 'role': '户名', # 'code': [('page', 'code')], # 'print_time': [], # 'start_date': [], # 'end_date': [], # 'sheet': ['sheet_name'] # } # } # } # 归为同一份流水的逻辑 # 所有图片均无卡号:同一分类同一户名归为同一份流水(如果同一分类下只有一个已知户名,则此分类下其他未知户名归为此户名) # 所有图片只已知1卡号:其他未知卡号流水归为此卡号 # 所有图片已知多卡号: 1.根据相似度和图片数目合并相似已知卡号,并整理多数分类和户名集合 # 2.遍历所有未知卡号,进行过滤:当未知卡号分类与某已知卡号一致,且此未知卡号户名在此已知卡号户名集合中时,将未知卡号归为已知卡号。剩余未知卡号同一分类同一户名归为同一流水 # 无卡号 if len(bs_summary) == 0: del bs_summary merged_bs_summary = {} card_num = 1 for role_dict in unknown_summary.values(): if len(role_dict) == 2 and consts.UNKNOWN_ROLE in role_dict: summary_dict = role_dict.pop(consts.UNKNOWN_ROLE, {}) for summary in role_dict.values(): # summary_dict['confidence'].extend(summary['confidence']) summary_dict['role'] = summary['role'] summary_dict['code'].extend(summary['code']) summary_dict['print_time'].extend(summary['print_time']) summary_dict['start_date'].extend(summary['start_date']) summary_dict['end_date'].extend(summary['end_date']) summary_dict['sheet'].extend(summary['sheet']) card = '{0}_{1}'.format(consts.UNKNOWN_CARD, card_num) merged_bs_summary[card] = summary_dict else: for summary in role_dict.values(): card = '{0}_{1}'.format(consts.UNKNOWN_CARD, card_num) card_num += 1 merged_bs_summary[card] = summary else: # 1卡号 one_card = False if len(bs_summary) == 1: merged_bs_summary = self.prune_bs_summary(bs_summary) one_card = True classify_info = {} # 多卡号 else: merged_bs_summary, classify_info = self.merge_card(bs_summary) for card_summary in merged_bs_summary.values(): merge_role = [] classify_summary = unknown_summary.get(card_summary['classify'], {}) for role, summary in classify_summary.items(): if one_card or classify_info.get(card_summary['classify'], 0) == 1 or role in card_summary['role_set']: merge_role.append(role) # card_summary['confidence'].extend(summary['confidence']) card_summary['sheet'].extend(summary['sheet']) card_summary['code'].extend(summary['code']) card_summary['print_time'].extend(summary['print_time']) card_summary['start_date'].extend(summary['start_date']) card_summary['end_date'].extend(summary['end_date']) for role in merge_role: del classify_summary[role] card_num = 1 for role_dict in unknown_summary.values(): for summary in role_dict.values(): card = '{0}_{1}'.format(consts.UNKNOWN_CARD, card_num) card_num += 1 merged_bs_summary[card] = summary del unknown_summary for summary in merged_bs_summary.values(): if summary.get('role_set') is not None: del summary['role_set'] summary['print_time'] = self.get_validate_date(summary['print_time']) summary['start_date'] = self.get_validate_date(summary['start_date']) summary['end_date'] = self.get_validate_date(summary['end_date']) # summary['confidence'] = max(summary['confidence']) return merged_bs_summary def pdf_2_img_2_queue(self, img_queue, todo_count_dict, lock, error_list, res_dict, finish_queue): while self.switch: try: # 1. 从队列获取文件信息 doc, business_type, task_str, classify_1_str, classify_2_str = self.get_doc_info() # 队列为空时的处理 if doc is None: time.sleep(self.sleep_time_doc_get) continue except Exception as e: self.online_log.error('{0} [process error (get doc info out)] [error={1}]'.format( self.log_base, traceback.format_exc())) error_list.append(1) return else: doc_data_path = os.path.join(self.data_dir, business_type, consts.TMP_DIR_NAME, str(doc.id)) os.makedirs(doc_data_path, exist_ok=True) img_save_path = os.path.join(doc_data_path, 'img') pdf_path = os.path.join(doc_data_path, '{0}.pdf'.format(doc.id)) pdf_handler = PDFHandler(pdf_path, img_save_path, doc.document_name) if classify_1_str == '0': try: # 2. 从EDMS获取PDF文件 max_count_obj = Configs.objects.filter(id=2).first() try: max_img_count = int(max_count_obj.value) except Exception as e: max_img_count = 500 for times in range(consts.RETRY_TIMES): try: if not doc.application_id.startswith(consts.FIXED_APPLICATION_ID_PREFIX): # self.edms.download(pdf_path, doc.metadata_version_id) self.edms.download(pdf_path, doc.metadata_version_id, doc.document_scheme, business_type) self.online_log.info('{0} [edms download success] [task={1}] [times={2}] ' '[pdf_path={3}]'.format(self.log_base, task_str, times, pdf_path)) # 3.PDF文件提取图片 self.online_log.info('{0} [pdf to img start] [task={1}] [times={2}]'.format( self.log_base, task_str, times)) start_time = time.time() pdf_handler.extract_image(max_img_count) end_time = time.time() speed_time = int(end_time - start_time) self.online_log.info('{0} [pdf to img end] [task={1}] [times={2}] [spend_time={3}]'.format( self.log_base, task_str, times, speed_time)) except Exception as e: self.online_log.warn('{0} [download or pdf to img failed] [task={1}] [times={2}] ' '[error={3}]'.format(self.log_base, task_str, times, traceback.format_exc())) else: break else: raise Exception('download or pdf to img failed') if pdf_handler.img_count == 0: self.online_log.warn('{0} [pdf to img failed (pdf img empty)] [task={1}]'.format( self.log_base, task_str)) raise Exception('pdf img empty') elif pdf_handler.img_count >= max_img_count: self.online_log.info('{0} [too many pdf image] [task={1}] [img_count={2}]'.format( self.log_base, task_str, pdf_handler.img_count)) try: report_table = HILOCRReport if business_type == consts.HIL_PREFIX else AFCOCRReport report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.get_value(doc.data_source, 0), input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=doc.start_time, successful_at_this_level=False, failure_reason=FailureReason.IMG_LIMIT.value, process_name=ProcessName.ALL.value, notes='pdf page count: {0}'.format(str(pdf_handler.img_count)) ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) try: doc.status = DocStatus.PROCESS_FAILED.value doc.save() except Exception as e: self.online_log.error('{0} [process error (db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) else: with lock: todo_count_dict[task_str] = pdf_handler.img_count for img_idx, img_path in enumerate(pdf_handler.img_path_list): while img_queue.full(): self.online_log.info('{0} [pdf_2_img_2_queue] [img queue full]'.format(self.log_base)) time.sleep(self.sleep_time_img_put) if pdf_handler.is_ebank: try: text_list = pdf_handler.page_text_list[img_idx].pop('rebuild_text') except Exception as e: text_list = [] else: text_list = [] img_queue.put((business_type, img_path, text_list)) # except EDMSException as e: # try: # doc.status = DocStatus.PROCESS_FAILED.value # doc.save() # self.online_log.warn('{0} [process failed (edms download)] [task={1}] [error={2}]'.format( # self.log_base, task_str, traceback.format_exc())) # except Exception as e: # self.online_log.error('{0} [process error (db save 1)] [error={1}]'.format( # self.log_base, traceback.format_exc())) # error_list.append(1) # return except Exception as e: try: end_time = timezone.now() report_table = HILOCRReport if business_type == consts.HIL_PREFIX else AFCOCRReport report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.get_value(doc.data_source, 0), input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, successful_at_this_level=False, failure_reason=FailureReason.PDF.value, process_name=ProcessName.ALL.value, ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) try: doc.status = DocStatus.PROCESS_FAILED.value doc.save() self.online_log.warn('{0} [process failed (pdf_2_img_2_queue)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) except Exception as e: self.online_log.error('{0} [process error (db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) error_list.append(1) return else: # e-contract try: # pdf下载 处理 图片存储 识别 for times in range(consts.RETRY_TIMES): try: self.edms.download(pdf_path, doc.metadata_version_id, doc.document_scheme, business_type) self.online_log.info('{0} [edms download success] [task={1}] [times={2}] ' '[pdf_path={3}]'.format(self.e_log_base, task_str, times, pdf_path)) self.online_log.info('{0} [pdf to img start] [task={1}] [times={2}]'.format( self.e_log_base, task_str, times)) pdf_handler.e_contract_process() self.online_log.info( '{0} [pdf to img end] [task={1}] [times={2}]'.format(self.e_log_base, task_str, times)) except Exception as e: self.online_log.warn('{0} [download or pdf to img failed] [task={1}] [times={2}] ' '[error={3}]'.format(self.e_log_base, task_str, times, traceback.format_exc())) else: break else: raise Exception('download or pdf to img failed') if classify_1_str == str(consts.CONTRACT_CLASSIFY): ocr_result = afc_predict(pdf_handler.pdf_info) page_res = {} for page_num, page_info in ocr_result.get('page_info', {}).items(): if isinstance(page_num, str) and page_num.startswith('page_'): page_res[page_num] = { 'classify': int(classify_1_str), 'page_num': page_num, 'page_info': page_info } else: file_type_1 = consts.HIL_CONTRACT_TYPE_MAP.get(classify_1_str) file_type_2 = consts.HIL_CONTRACT_TYPE_MAP.get(classify_2_str) ocr_result_1 = hil_predict(pdf_handler.pdf_info, file_type_1) rebuild_res_1 = {} page_res = {} for field_name, field_info in ocr_result_1.items(): page_num = field_info.pop('page', 'page_1') rebuild_res_1.setdefault(page_num, dict())[field_name] = field_info for page_num, page_info in rebuild_res_1.items(): if isinstance(page_num, str) and page_num.startswith('page_'): page_res[page_num] = { 'classify': int(classify_1_str), 'page_num': page_num, 'page_info': page_info } if isinstance(file_type_2, int): rebuild_res_2 = {} ocr_result_2 = hil_predict(pdf_handler.pdf_info, file_type_2) for field_name, field_info in ocr_result_2.items(): page_num = field_info.pop('page', 'page_1') rebuild_res_2.setdefault(page_num, dict())[field_name] = field_info for page_num, page_info in ocr_result_2.items(): if isinstance(page_num, str) and page_num.startswith('page_'): page_res[page_num] = { 'classify': int(classify_2_str), 'page_num': page_num, 'page_info': page_info } contract_res = {} for img_path_tmp, page_key in pdf_handler.img_path_pno_list: if page_key in page_res: img_contract_res = { 'code': 1, 'data': [ { 'classify': page_res[page_key].pop('classify', consts.OTHER_CLASSIFY), 'data': page_res[page_key] } ] } else: img_contract_res = { 'code': 1, 'data': [ { 'classify': int(classify_1_str), } ] } contract_res[img_path_tmp] = img_contract_res with lock: res_dict[task_str] = contract_res finish_queue.put(task_str) except Exception as e: try: doc.status = DocStatus.PROCESS_FAILED.value doc.save() self.online_log.warn('{0} [process failed (e-contract)] [task={1}] ' '[error={2}]'.format(self.e_log_base, task_str, traceback.format_exc())) except Exception as e: self.online_log.error('{0} [process error (db save)] [error={1}]'.format( self.e_log_base, traceback.format_exc())) error_list.append(1) return def img_2_ocr_1(self, img_queue, todo_count_dict, res_dict, finish_queue, lock, url, error_list): while len(error_list) == 0 or not img_queue.empty(): try: channel, img_path, text_list = img_queue.get(block=False) except Exception as e: # self.online_log.info('{0} [img_2_ocr_1] [queue empty]'.format(self.log_base)) time.sleep(self.sleep_time_img_get) continue else: try: self.online_log.info('{0} [img_2_ocr_1] [get img] [img_path={1}]'.format(self.log_base, img_path)) for times in range(consts.RETRY_TIMES): try: with open(img_path, 'rb') as f: base64_data = base64.b64encode(f.read()) # 获取解码后的base64值 file_data = base64_data.decode() json_data_1 = { "file": file_data, "channel": channel, } if len(text_list) > 0: json_data_1['text_list'] = text_list start_time = time.time() ocr_1_response = requests.post(url, json=json_data_1) if ocr_1_response.status_code != 200: raise OCR1Exception('ocr_1 status code: {0}'.format(ocr_1_response.status_code)) except Exception as e: self.online_log.warn('{0} [ocr_1 failed] [times={1}] [url={2}] [img_path={3}] ' '[error={4}]'.format(self.log_base, times, url, img_path, traceback.format_exc())) else: ocr_1_res = ocr_1_response.json() end_time = time.time() speed_time = int(end_time - start_time) self.online_log.info('{0} [ocr_1 success] [img={1}] [url={2}] [speed_time={3}]'.format( self.log_base, img_path, url, speed_time)) break else: ocr_1_res = {} self.online_log.warn('{0} [ocr_1 failed] [img_path={1}] [url={2}]'.format( self.log_base, img_path, url)) # continue except Exception as e: self.online_log.error('{0} [process error (ocr fetch)] [img_path={1}] [error={2}]'.format( self.log_base, img_path, traceback.format_exc())) else: try: del json_data_1 # /data/bmw-ocr-data/AFC/tmp/6/img/page_0_img_0.jpeg # AFC_2 path_split = img_path.split('/') task_str = consts.SPLIT_STR.join((path_split[-5], path_split[-3])) with lock: doc_res_dict = res_dict.setdefault(task_str, {}) doc_res_dict[img_path] = ocr_1_res res_dict[task_str] = doc_res_dict todo_count = todo_count_dict.get(task_str) if todo_count == 1: finish_queue.put(task_str) del todo_count_dict[task_str] else: todo_count_dict[task_str] = todo_count - 1 except Exception as e: self.online_log.error('{0} [process error (store ocr res)] [img_path={1}] [error={2}]'.format( self.log_base, img_path, traceback.format_exc())) def res_2_wb(self, res_dict, img_queue, finish_queue, lock, error_list): while len(error_list) == 0 or not img_queue.empty() or not finish_queue.empty(): try: task_str = finish_queue.get(block=False) except Exception as e: # self.online_log.info('{0} [res_2_wb] [queue empty]'.format(self.log_base)) time.sleep(self.sleep_time_task_get) continue else: self.online_log.info('{0} [res_2_wb] [get task] [task={1}]'.format(self.log_base, task_str)) ocr_1_res = res_dict.pop(task_str, {}) business_type, doc_id_str = task_str.split(consts.SPLIT_STR) doc_id = int(doc_id_str) doc_class = HILDoc if business_type == consts.HIL_PREFIX else AFCDoc is_hil = True if business_type == consts.HIL_PREFIX else False dda_id_bc_mapping = dict() doc_data_path = os.path.join(self.data_dir, business_type, consts.TMP_DIR_NAME, doc_id_str) excel_path = os.path.join(doc_data_path, '{0}.xlsx'.format(doc_id_str)) try: doc = doc_class.objects.filter(id=doc_id).first() # report_dict = { # 'process': None or pdf or excel or edms # 'idcard': True or False, # 'bs': None or normal or mobile, # } report_list = [None, False, None, None] do_dda = is_hil and doc.document_scheme == consts.DOC_SCHEME_LIST[1] except Exception as e: self.online_log.error('{0} [process error (db filter)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: try: # 4.OCR结果并且构建excel文件 bs_classify_set = set() bs_summary = {} unknown_summary = {} license_summary = {} contract_result = {} res_list = [] interest_keyword = Keywords.objects.filter( type=KeywordsType.INTEREST.value, on_off=True).values_list('keyword', flat=True) salary_keyword = Keywords.objects.filter( type=KeywordsType.SALARY.value, on_off=True).values_list('keyword', flat=True) loan_keyword = Keywords.objects.filter( type=KeywordsType.LOAN.value, on_off=True).values_list('keyword', flat=True) wechat_keyword = Keywords.objects.filter( type=KeywordsType.ALI_WECHART.value, on_off=True).values_list('keyword', flat=True) repayments_keyword = Keywords.objects.filter( type=KeywordsType.REPAYMENTS.value, on_off=True).values_list('keyword', flat=True) wb = BSWorkbook(interest_keyword, salary_keyword, loan_keyword, wechat_keyword, repayments_keyword) for img_path, res in ocr_1_res.items(): pno, ino = self.parse_img_path(img_path) part_idx = 1 if res.get('code') == 1: ocr_data_list = res.get('data', []) if not isinstance(ocr_data_list, list): res_list.append((pno, ino, part_idx, consts.RES_FAILED_3)) self.online_log.warn('{0} [ocr_1 res error] [img={1}]'.format(self.log_base, img_path)) else: for part_idx, ocr_data in enumerate(ocr_data_list): part_idx = part_idx + 1 classify = ocr_data.get('classify') if classify is None: res_list.append((pno, ino, part_idx, consts.RES_FAILED_3)) self.online_log.warn('{0} [ocr_1 res error] [img={1}]'.format( self.log_base, img_path)) continue elif classify in consts.OTHER_CLASSIFY_SET: # 其他类 res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_OTHER)) continue elif classify in consts.LICENSE_CLASSIFY_SET_1: # 证件1 self.license1_process(ocr_data, license_summary, classify, res_list, pno, ino, part_idx, img_path, do_dda, dda_id_bc_mapping) elif classify in consts.LICENSE_CLASSIFY_SET_2: # 证件2 pid, _, _, _, _, _ = consts.LICENSE_CLASSIFY_MAPPING.get(classify) file_data = ocr_data.get('section_img') if file_data is None: with open(img_path, 'rb') as f: base64_data = base64.b64encode(f.read()) # 获取解码后的base64值 file_data = base64_data.decode() json_data_2 = { "pid": str(pid), "filedata": file_data } ocr_url_2 = self.ocr_url_2_bc if classify == consts.BC_CLASSIFY else self.ocr_url_2 for times in range(consts.RETRY_TIMES): try: start_time = time.time() ocr_2_response = requests.post(ocr_url_2, data=json_data_2) if ocr_2_response.status_code != 200: raise OCR2Exception('ocr_2 status code: {0}'.format(ocr_2_response.status_code)) except Exception as e: self.online_log.warn( '{0} [ocr_2 failed] [times={1}] [img_path={2}] [error={3}]'.format( self.log_base, times, img_path, traceback.format_exc())) else: ocr_2_res = json.loads(ocr_2_response.text) end_time = time.time() speed_time = int(end_time - start_time) self.online_log.info( '{0} [ocr_2 success] [img={1}] [speed_time={2}]'.format( self.log_base, img_path, speed_time)) if classify == consts.BC_CLASSIFY: new_ocr_2_res = {} position_dict = {} get_info = False bc_image_data = '' position = {} angle = 0 for page_info in ocr_2_res.get('PageInfo', []): if get_info: break if page_info.get('ErrorCode', 1) in consts.SUCCESS_CODE_SET: for bc_res_all in page_info.get('Result', []): if get_info: break if bc_res_all.get('ErrorCode', 1) in consts.SUCCESS_CODE_SET: for bc_res in bc_res_all.get('ResultList', []): if get_info: break get_info = True bc_image_data = bc_res.get('image_data', '') position = bc_res.get('position', {}) angle = bc_res.get('angle', 0) for field_info in bc_res.get('FieldList', []): new_ocr_2_res[field_info.get('key', '')] = field_info.get('value', '') position_dict[field_info.get('key', '')] = { consts.FIELD_POSITION_KEY: field_info.get('position', {}), consts.FIELD_QUAD_KEY: field_info.get('quad', []), } if get_info: pre, suf = os.path.splitext(img_path) if len(bc_image_data) > 0: position = {} angle = 0 section_img_path = '{0}_{1}_0{2}'.format(pre, part_idx, suf) try: with open(section_img_path, "wb") as fh: fh.write(base64.b64decode(bc_image_data.encode())) except Exception as e: self.online_log.warn( '{0} [bc section img save failed] [img_path={1}]' ' [part_idx={2}]]'.format(self.log_base, img_path, part_idx)) else: section_img_path = img_path if ocr_data.get('section_img') is None else '{0}_{1}{2}'.format(pre, part_idx, suf) if ocr_data.get('section_img') is not None: try: with open(section_img_path, "wb") as fh: fh.write(base64.b64decode(ocr_data.get('section_img').encode())) except Exception as e: self.online_log.warn( '{0} [bc section img save failed] [img_path={1}]' ' [part_idx={2}]'.format(self.log_base, img_path, part_idx)) new_ocr_2_res['ErrorCode'] = 0 position_dict[consts.POSITION_KEY] = position position_dict[consts.ANGLE_KEY] = angle new_ocr_2_res[consts.SECTION_IMG_PATH_KEY] = section_img_path new_ocr_2_res[consts.ALL_POSITION_KEY] = position_dict name = '有' json_data_3 = { "file": file_data, 'card_res': new_ocr_2_res } card_name_response = requests.post(self.ocr_url_3, json_data_3) if card_name_response.status_code == 200: card_name_res = card_name_response.json() if isinstance(card_name_res, dict) and \ card_name_res.get('data', {}).get('is_exists_name') == 0: name = '无' new_ocr_2_res['Name'] = name else: new_ocr_2_res = ocr_2_res self.license2_process(new_ocr_2_res, license_summary, pid, classify, res_list, pno, ino, part_idx, img_path, do_dda, dda_id_bc_mapping, file_data=ocr_data.get('section_img')) break else: res_list.append((pno, ino, part_idx, consts.RES_FAILED_2)) self.online_log.warn( '{0} [ocr_2 failed] [img_path={1}]'.format(self.log_base, img_path)) elif classify in consts.CONTRACT_SET: self.contract_process(classify, ocr_data, contract_result, res_list, pno, ino, part_idx, img_path) else: # 流水处理 bs_classify_set.add(classify) self.bs_process(wb, ocr_data, bs_summary, unknown_summary, classify, res_list, pno, ino, part_idx) else: res_list.append((pno, ino, part_idx, consts.RES_FAILED_1)) self.online_log.info('{0} [ocr_1 res error] [img={1}]'.format(self.log_base, img_path)) self.online_log.info('{0} [task={1}] [bs_summary={2}] [unknown_summary={3}]'.format( self.log_base, task_str, bs_summary, unknown_summary)) # self.license_log.info('[task={0}] [license_summary={1}]'.format(task_str, license_summary)) idcard_list = license_summary.get(consts.IC_CLASSIFY) if idcard_list: report_list[1] = True self.idcard_log.info('[task={0}] [idcard={1}]'.format(task_str, idcard_list)) if len(bs_classify_set) > 0: if consts.ALI_WECHART_CLASSIFY & bs_classify_set: report_list[2] = WorkflowName.MOBILE.value else: report_list[2] = WorkflowName.NORMAL.value merged_bs_summary = self.rebuild_bs_summary(bs_summary, unknown_summary) del unknown_summary ic_merge, rp_merge = self.license_rebuild(license_summary) # self.bs_log.info('[task={0}] [bs_summary={1}]'.format(task_str, merged_bs_summary)) self.online_log.info('{0} [task={1}] [merged_bs_summary={2}] [license_summary={3}] [contract={4}] ' '[res_list={5}]'.format(self.log_base, task_str, merged_bs_summary, license_summary, contract_result, res_list)) except Exception as e: report_list[0] = FailureReason.EXCEL.value self.online_log.warn('{0} [process failed (res conformity)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) try: doc.status = DocStatus.PROCESS_FAILED.value doc.save() except Exception as e: self.online_log.error('{0} [process error (db save)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: try: # 重构Excel文件 # src_excel_path = os.path.join(doc_data_path, 'src.xlsx') # wb.save(src_excel_path) count_list = wb.rebuild(merged_bs_summary, license_summary, res_list, doc.document_scheme, contract_result) wb.save(excel_path) except Exception as e: report_list[0] = FailureReason.EXCEL.value self.online_log.warn('{0} [process failed (wb rebuild)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) try: doc.status = DocStatus.PROCESS_FAILED.value doc.save() except Exception as e: self.online_log.error('{0} [process error (db save)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: try: # 上传至EDMS for times in range(consts.RETRY_TIMES): try: self.edms.upload(excel_path, doc, business_type) except Exception as e: self.online_log.warn( '{0} [edms upload failed] [times={1}] [task={2}] [error={3}]'.format( self.log_base, times, task_str, traceback.format_exc())) edms_exc = str(e) else: break else: raise EDMSException(edms_exc) except Exception as e: report_list[0] = FailureReason.EDMS.value doc.status = DocStatus.UPLOAD_FAILED.value self.online_log.warn('{0} [process failed (edms upload)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: doc.status = DocStatus.COMPLETE.value self.online_log.info('{0} [edms upload success] [task={1}]'.format(self.log_base, task_str)) finally: try: doc.end_time = timezone.now() doc.duration = min((doc.end_time - doc.start_time).seconds, 32760) for field, count in count_list: if hasattr(doc, field): setattr(doc, field, count) doc.save() except Exception as e: self.online_log.error('{0} [process error (db save)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: self.online_log.info('{0} [process complete] [task={1}]'.format(self.log_base, task_str)) os.remove(excel_path) finally: # TODO 识别结果存一张表,方便跑报表 # 比对 if len(license_summary) > 0 and doc.document_scheme != consts.DOC_SCHEME_LIST[2]: try: is_ca = True if doc.document_scheme == consts.DOC_SCHEME_LIST[0] else False # 更新OCR累计识别结果表 if business_type == consts.HIL_PREFIX: result_class = HILOCRResult if is_ca else HILSEOCRResult else: result_class = AFCOCRResult if is_ca else AFCSEOCRResult res_obj = result_class.objects.filter(application_id=doc.application_id).first() if res_obj is None: res_obj = result_class() res_obj.application_id = doc.application_id for classify, field in consts.RESULT_MAPPING.items(): if not hasattr(res_obj, field): continue license_list = license_summary.get(classify) if not license_list: continue if classify == consts.IC_CLASSIFY and ic_merge: license_list[0].update(license_list[1]) license_list.pop(1) elif classify == consts.RP_CLASSIFY and rp_merge: license_list[0].update(license_list[1]) license_list.pop(1) old_res_str = getattr(res_obj, field) if old_res_str is None: last_res_str = json.dumps(license_list) else: old_res_list = json.loads(old_res_str) old_res_list.extend(license_list) last_res_str = json.dumps(old_res_list) setattr(res_obj, field, last_res_str) res_obj.save() except Exception as e: self.online_log.error( '{0} [process error (ocr result save)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: self.online_log.info('{0} [ocr result save success] [task={1}] [res_id={2}]'.format( self.log_base, task_str, res_obj.id)) # 触发比对 try: # pass compare.apply_async((doc.application_id, business_type, None, res_obj.id, is_ca, True), queue='queue_compare') except Exception as e: self.online_log.error( '{0} [process error (comparison info send)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) else: self.online_log.info('{0} [comparison info send success] [task={1}] ' '[res_id={2}]'.format(self.log_base, task_str, res_obj.id)) # DDA处理 if do_dda: # 入库 try: dda_record = DDARecords.objects.filter( application_id=doc.application_id).first() if dda_record is None: dda_record = DDARecords(application_id=doc.application_id) except Exception as e: report_list[3] = False self.online_log.error('{0} [process error (dda db get)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) else: try: if not dda_record.all_found: found_time = timezone.now() move_img_path_dict = dict() ic_res_list = dda_id_bc_mapping.get(consts.IC_FIELD, []) bc_res_list = dda_id_bc_mapping.get(consts.BC_FIELD, []) self.online_log.info('{0} [dda process] [task={1}] [ic={2}] ' '[bc={3}]'.format(self.log_base, task_str,ic_res_list, bc_res_list)) if not dda_record.is_dda_found: try: # DDA过滤,获取有效DDA best_dda_res = None dda_res_list = license_summary.get(consts.DDA_CLASSIFY, []) if len(dda_res_list) > 0: dda_res_list.sort(key=lambda x: x.get(consts.DDA_PRO, 0), reverse=True) best_dda_res = dda_res_list[0] # tmp_best_dda_res = dda_res_list[0] # if tmp_best_dda_res.get(consts.DDA_PRO, 0) >= consts.DDA_PRO_MIN: # best_dda_res = tmp_best_dda_res self.online_log.info( '{0} [dda process] [task={1}] [dda={2}]'.format( self.log_base, task_str, dda_res_list)) except Exception as e: best_dda_res = None dda_record.is_dda_found = False if best_dda_res is None else True if dda_record.is_dda_found: dda_path = best_dda_res.get(consts.DDA_IMG_PATH, '') customer_name = best_dda_res.get(consts.DDA_IC_NAME, '') customer_id = best_dda_res.get(consts.DDA_IC_ID, '') account_id = best_dda_res.get(consts.DDA_BC_ID, '') dda_record.dda_path = dda_path dda_record.dda_found_time = found_time dda_record.customer_name = customer_name dda_record.customer_id = customer_id dda_record.account_id = account_id # move move_img_path_dict.setdefault( consts.DDA_FIELD, set()).add(dda_path) if dda_record.is_dda_found: try: if not dda_record.is_id_found: for ic_name, ic_id, ic_img_path in ic_res_list: if ic_id == dda_record.customer_id \ or ic_name == dda_record.customer_name: dda_record.is_id_found = True dda_record.id_path = ic_img_path dda_record.id_found_time = found_time move_img_path_dict.setdefault( consts.IC_FIELD, set()).add(ic_img_path) break else: id_record = IDBCRecords.objects.filter( application_id=doc.application_id, target_id=dda_record.customer_id, is_id=True).first() if id_record is None: id_record = IDBCRecords.objects.filter( application_id=doc.application_id, target_name=dda_record.customer_name, is_id=True).first() if id_record is not None: dda_record.is_id_found = True dda_record.id_path = id_record.file_path dda_record.id_found_time = id_record.create_time move_img_path_dict.setdefault( consts.IC_FIELD, set()).add(id_record.file_path) except Exception as e: report_list[3] = False self.online_log.error( '{0} [process error (dda id process)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) try: if not dda_record.is_bc_found: for bc_no, bc_img_path in bc_res_list: if bc_no == dda_record.account_id: dda_record.is_bc_found = True dda_record.bc_path = bc_img_path dda_record.bc_found_time = found_time move_img_path_dict.setdefault( consts.BC_FIELD, set()).add(bc_img_path) break else: bc_record = IDBCRecords.objects.filter( application_id=doc.application_id, target_id=dda_record.account_id, is_id=False).first() if bc_record is not None: dda_record.is_bc_found = True dda_record.bc_path = bc_record.file_path dda_record.bc_found_time = bc_record.create_time move_img_path_dict.setdefault( consts.BC_FIELD, set()).add(bc_record.file_path) except Exception as e: report_list[3] = False self.online_log.error( '{0} [process error (dda bc process)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) if dda_record.is_dda_found and dda_record.is_id_found and dda_record.is_bc_found: dda_record.all_found = True dda_record.save() # 图片移动 try: if len(move_img_path_dict) > 0: self.online_log.info( '{0} [dda process] [task={1}] [move_img_path={2}]'.format( self.log_base, task_str, move_img_path_dict)) wanting_dir = os.path.join(self.dda_wanting_dir, doc.application_id) wanting_dir_exists = os.path.isdir(wanting_dir) if dda_record.all_found: target_dir = os.path.join(self.dda_complete_dir, doc.application_id) if wanting_dir_exists: shutil.move(wanting_dir, target_dir) else: os.makedirs(target_dir, exist_ok=True) else: target_dir = wanting_dir if not wanting_dir_exists: os.makedirs(target_dir, exist_ok=True) for prefix, path_set in move_img_path_dict.items(): for idx, path in enumerate(path_set): if os.path.isfile(path): file_name = '{0}_{1}{2}'.format( prefix, idx, os.path.splitext(path)[-1]) target_path = os.path.join(target_dir, file_name) shutil.copyfile(path, target_path) else: self.online_log.warn( '{0} [dda process] [img path empty] [task={1}] ' '[path={2}]'.format(self.log_base, task_str, path)) except Exception as e: report_list[3] = False self.online_log.error( '{0} [process error (dda img move)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) # id & bc 入库 try: if not dda_record.is_dda_found and not dda_record.is_id_found: ic_set = set() for ic_name, ic_id, ic_img_path in ic_res_list: query_str = '{0}{1}'.format(ic_name, ic_id) if query_str in ic_set: continue ic_set.add(query_str) IDBCRecords.objects.create( application_id=doc.application_id, target_name=ic_name, target_id=ic_id, is_id=True, file_path=ic_img_path) if not dda_record.is_dda_found and not dda_record.is_bc_found: bc_set = set() for bc_no, bc_img_path in bc_res_list: if bc_no in bc_set: continue bc_set.add(bc_no) IDBCRecords.objects.create( application_id=doc.application_id, target_id=bc_no, is_id=False, file_path=bc_img_path) except Exception as e: report_list[3] = False self.online_log.error( '{0} [process error (dda id&bc db save)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) except Exception as e: report_list[3] = False self.online_log.error( '{0} [process error (dda process)] [task={1}] ' '[error={2}]'.format(self.log_base, task_str, traceback.format_exc())) else: if report_list[3] is None: report_list[3] = True finally: # report_dict = { # 'process': None or pdf or excel or edms # 'idcard': True or False, # 'bs': None or normal or mobile, # } end_time = timezone.now() report_table = HILOCRReport if business_type == consts.HIL_PREFIX else AFCOCRReport try: if report_list[0] is None: report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.get_value(doc.data_source, 0), input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, process_name=ProcessName.ALL.value, ) else: report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.get_value(doc.data_source, 0), input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, successful_at_this_level=False, failure_reason=report_list[0], process_name=ProcessName.ALL.value, ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) try: if report_list[1]: report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.CONTROLLING.value, request_trigger=RequestTrigger.DOCUPLOAD.value, input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, process_name=ProcessName.IDCARD.value, ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) try: if report_list[2] is not None: report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.DOCUPLOAD.value, input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, process_name=ProcessName.BS.value, workflow_name=report_list[2], ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) try: if report_list[3] is not None: report_table.objects.create( case_number=doc.application_id, request_team=RequestTeam.get_value(doc.document_scheme, 0), request_trigger=RequestTrigger.DOCUPLOAD.value, input_file=doc.document_name, transaction_start=doc.start_time, transaction_end=end_time, successful_at_this_level=report_list[3], process_name=ProcessName.DDA.value, ) except Exception as e: self.online_log.error('{0} [process error (report db save)] [error={1}]'.format( self.log_base, traceback.format_exc())) finally: try: # img_save_path = os.path.join(doc_data_path, 'img') # write_zip_file(img_save_path, os.path.join(doc_data_path, '{0}_img.zip'.format(doc_id_str))) # shutil.rmtree(img_save_path, ignore_errors=True) pdf_path = os.path.join(doc_data_path, '{0}.pdf'.format(doc_id_str)) os.remove(pdf_path) self.online_log.info('{0} [pdf & img removed] [task={1}]'.format(self.log_base, task_str)) except Exception as e: self.online_log.error('{0} [process error (pdf & img remove)] [task={1}] [error={2}]'.format( self.log_base, task_str, traceback.format_exc())) def handle(self, *args, **kwargs): db.close_old_connections() lock = Lock() with Manager() as manager: error_list = manager.list() todo_count_dict = manager.dict() res_dict = manager.dict() img_queue = Queue(self.img_queue_size) finish_queue = Queue() process_list = [] pdf_process = Process(target=self.pdf_2_img_2_queue, args=(img_queue, todo_count_dict, lock, error_list, res_dict, finish_queue)) process_list.append(pdf_process) for url in self.ocr_1_urls.values(): ocr_1_process = Process(target=self.img_2_ocr_1, args=( img_queue, todo_count_dict, res_dict, finish_queue, lock, url, error_list)) process_list.append(ocr_1_process) wb_process = Process(target=self.res_2_wb, args=(res_dict, img_queue, finish_queue, lock, error_list)) process_list.append(wb_process) for p in process_list: p.start() for p in process_list: p.join() self.online_log.info('{0} [stop safely]'.format(self.log_base))