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 apps.doc import consts from apps.doc.ocr.edms import EDMS, 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, HILOCRReport, AFCOCRReport, DDARecords, IDBCRecords from celery_compare.tasks import compare class Command(BaseCommand, LoggerMixin): def __init__(self): super().__init__() self.log_base = '[doc 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_3 = conf.BC_URL self.ocr_url_4 = conf.IC_URL # EDMS web_service_api self.edms = EDMS() # 优雅退出信号: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 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) 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 = 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 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 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, task_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 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 classify == consts.DDA_CLASSIFY: # DDA处理 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 = {key: value.get('words', '') for key, value in license_data.get('result', {}).items()} dda_ocr_result[consts.DDA_IMG_PATH] = img_path dda_ocr_result[consts.DDA_PRO] = pro license_summary.setdefault(classify, []).append(dda_ocr_result) elif classify == consts.MVC_CLASSIFY: # 车辆登记证 3/4页结果整合 for mvc_dict in license_data: 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: # 身份证真伪 for id_card_dict in license_data: try: base64_img = id_card_dict.pop('base64_img') except Exception as e: continue else: 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) finally: 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[consts.IC_KEY_FIELD[0]].strip() ic_id = id_card_dict[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, []).extend(license_data) else: 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): 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, '') 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: # 营业执照等 for result_dict in ocr_res_2.get('ResultList', []): res_dict = {} for field_dict in result_dict.get('FieldList', []): res_dict[field_dict.get('chn_key', '')] = field_dict.get('value', '') license_summary.setdefault(classify, []).append(res_dict) 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): while self.switch: try: # 1. 从队列获取文件信息 doc, business_type, task_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: try: # 2. 从EDMS获取PDF文件 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) 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.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() 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') img_count = len(pdf_handler.img_path_list) if 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') else: with lock: todo_count_dict[task_str] = img_count for img_path in 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) img_queue.put(img_path) # 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 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: img_path = 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 } 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 = {} 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) wb = BSWorkbook(interest_keyword, salary_keyword, loan_keyword, wechat_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 } for times in range(consts.RETRY_TIMES): try: start_time = time.time() ocr_2_response = requests.post(self.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: name = '有' json_data_3 = { "file": file_data, 'card_res': 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 = '无' ocr_2_res['Name'] = name self.license2_process(ocr_2_res, license_summary, pid, classify, res_list, pno, ino, part_idx, img_path, do_dda, dda_id_bc_mapping) 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)) 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}] ' '[res_list={4}]'.format(self.log_base, task_str, merged_bs_summary, license_summary, 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) 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 识别结果存一张表,方便跑报表 # CA比对 if doc.document_scheme == consts.DOC_SCHEME_LIST[0]: try: # 更新OCR累计识别结果表 result_class = HILOCRResult if business_type == consts.HIL_PREFIX else AFCOCRResult 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(): 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) if not hasattr(res_obj, field): continue 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), 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)) 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))