import os
import time
import json
import shutil
import base64
import signal
import asyncio
import aiohttp
import difflib
import requests
import traceback
from collections import Counter
from datetime import datetime, date
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
from apps.doc.exceptions import EDMSException, OCR1Exception, OCR2Exception
from apps.doc.ocr.wb import BSWorkbook, Workbook
from apps.doc.models import DocStatus, HILDoc, AFCDoc, Keywords
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
# 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
# EDMS web_service_api
self.edms = EDMS()
# 优雅退出信号：15
signal.signal(signal.SIGTERM, self.signal_handler)
def signal_handler(self, sig, frame):
self.switch = False # 停止处理文件
@staticmethod
def 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_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.cronjob_log.info('{0} [get_doc_info] [queue empty]'.format(self.log_base))
return None, None, None
doc, business_type = self.get_doc_object(task_str)
if doc is None:
self.cronjob_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.cronjob_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()
self.cronjob_log.info('{0} [get_doc_info] [success] [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.cronjob_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.cronjob_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):
# 类别：'0'身份证， '1'居住证
license_data = ocr_data.get('data', [])
if not license_data:
res_list.append((pno, ino, part_idx, consts.RES_SUCCESS_EMPTY))
return
res_list.append((pno, ino, part_idx, consts.RES_SUCCESS))
license_summary.setdefault(classify, []).extend(license_data)
def license2_process(self, ocr_res_2, license_summary, pid, classify, res_list, pno, ino, part_idx):
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)
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 parse_img_path(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
@staticmethod
def get_most(value_list):
if value_list:
most_common = Counter(value_list).most_common(1)
return most_common[0][0] if most_common else None
@staticmethod
def date_format(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):
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]
merged_bs_summary[main_card]['classify'] = self.get_most(merged_bs_summary[main_card]['classify'])
merged_bs_summary[main_card]['role'] = self.get_most(merged_bs_summary[main_card]['role'])
del bs_summary
return merged_bs_summary
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']
# }
# }
# }
# 无卡号
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
# 多卡号
else:
merged_bs_summary = 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 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):
while self.switch:
# 1. 从队列获取文件信息
doc, business_type, task_str = self.get_doc_info()
# 队列为空时的处理
if doc is None:
time.sleep(self.sleep_time_doc_get)
continue
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)
pdf_path = os.path.join(doc_data_path, '{0}.pdf'.format(doc.id))
self.pdf_download(doc, pdf_path)
# 3.PDF文件提取图片
self.cronjob_log.info('{0} [pdf to img start] [task={1}]'.format(self.log_base, task_str))
start_time = time.time()
img_save_path = os.path.join(doc_data_path, 'img')
pdf_handler = PDFHandler(pdf_path, img_save_path)
pdf_handler.extract_image()
end_time = time.time()
speed_time = int(end_time - start_time)
self.cronjob_log.info('{0} [pdf to img end] [task={1}] [spend_time={2}]'.format(
self.log_base, task_str, speed_time))
with lock:
todo_count_dict[task_str] = len(pdf_handler.img_path_list)
for img_path in pdf_handler.img_path_list:
while img_queue.full():
self.cronjob_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:
doc.status = DocStatus.PROCESS_FAILED.value
doc.save()
self.cronjob_log.error('{0} [process failed (edms download)] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
except Exception as e:
doc.status = DocStatus.PROCESS_FAILED.value
doc.save()
self.cronjob_log.error('{0} [process failed (pdf to img)] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
def img_2_ocr_1(self, img_queue, todo_count_dict, res_dict, finish_queue, lock, url):
while True:
try:
img_path = img_queue.get(block=False)
except Exception as e:
# self.cronjob_log.info('{0} [img_2_ocr_1] [queue empty]'.format(self.log_base))
time.sleep(self.sleep_time_img_get)
continue
else:
self.cronjob_log.info('{0} [img_2_ocr_1] [get img] [img_path={1}]'.format(self.log_base, img_path))
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
}
for times in range(consts.RETRY_TIMES):
try:
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.cronjob_log.warn('{0} [ocr_1 failed] [times={1}] [img_path={2}] [error={3}]'.format(
self.log_base, times, 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.cronjob_log.info('{0} [ocr_1 success] [img={1}] [res={2}] [speed_time={3}]'.format(
self.log_base, img_path, ocr_1_res, speed_time))
break
else:
ocr_1_res = {}
self.cronjob_log.warn('{0} [ocr_1 failed] [img_path={1}]'.format(self.log_base, img_path))
# continue
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
def res_2_wb(self, res_dict, finish_queue, lock):
while True:
try:
task_str = finish_queue.get(block=False)
except Exception as e:
# self.cronjob_log.info('{0} [res_2_wb] [queue empty]'.format(self.log_base))
time.sleep(self.sleep_time_task_get)
continue
else:
self.cronjob_log.info('{0} [res_2_wb] [get task] [task={1}]'.format(self.log_base, task_str))
ocr_1_res = res_dict.get(task_str, {})
# self.cronjob_log.info('{0} [res_2_wb] [get task res] [task={1}]'.format(
# self.log_base, task_str))
try:
# 4.OCR结果并且构建excel文件
bs_summary = {}
license_summary = {}
unknown_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.cronjob_log.info('{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.cronjob_log.info('{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)
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.cronjob_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.cronjob_log.info(
'{0} [ocr_2 success] [img={1}] [res={2}] [speed_time={3}]'.format(
self.log_base, img_path, ocr_2_res, 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)
break
else:
res_list.append((pno, ino, part_idx, consts.RES_FAILED_2))
self.cronjob_log.warn(
'{0} [ocr_2 failed] [img_path={1}]'.format(self.log_base, img_path))
else: # 流水处理
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.cronjob_log.info('{0} [ocr_1 res error] [img={1}]'.format(self.log_base, img_path))
with lock:
del res_dict[task_str]
# self.cronjob_log.info('{0} [res_dict record] [res_dict={1}]'.format(
# self.log_base, res_dict))
self.cronjob_log.info('{0} [task={1}] [bs_summary={2}] [unknown_summary={3}] '
'[license_summary={4}]'.format(self.log_base, task_str, bs_summary,
unknown_summary, license_summary))
merged_bs_summary = self.rebuild_bs_summary(bs_summary, unknown_summary)
self.cronjob_log.info('{0} [task={1}] [merged_bs_summary={2}] [unknown_summary={3}] '
'[res_list={4}]'.format(self.log_base, task_str, merged_bs_summary,
unknown_summary, res_list))
del unknown_summary
# 4.2 重构Excel文件
doc, business_type = self.get_doc_object(task_str)
doc_data_path = os.path.join(self.data_dir, business_type, consts.TMP_DIR_NAME, str(doc.id))
excel_path = os.path.join(doc_data_path, '{0}.xlsx'.format(doc.id))
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:
with lock:
if task_str in res_dict:
del res_dict[task_str]
doc, business_type = self.get_doc_object(task_str)
doc.status = DocStatus.PROCESS_FAILED.value
doc.save()
self.cronjob_log.error('{0} [process failed (res to wb)] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
try:
doc_data_path = os.path.join(self.data_dir, business_type, consts.TMP_DIR_NAME, str(doc.id))
img_save_path = os.path.join(doc_data_path, 'img')
shutil.rmtree(img_save_path, ignore_errors=True)
pdf_path = os.path.join(doc_data_path, '{0}.pdf'.format(doc.id))
os.remove(pdf_path)
except Exception as e:
self.cronjob_log.error('{0} [file remove failed] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
else:
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)))
shutil.rmtree(img_save_path, ignore_errors=True)
# pdf_path = os.path.join(doc_data_path, '{0}.pdf'.format(doc.id))
# os.remove(pdf_path)
# os.remove(src_excel_path)
except Exception as e:
self.cronjob_log.error('{0} [file remove failed] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
try:
# 5.上传至EDMS
for times in range(consts.RETRY_TIMES):
try:
self.edms.upload(excel_path, doc, business_type)
except Exception as e:
self.cronjob_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:
doc.status = DocStatus.UPLOAD_FAILED.value
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()
self.cronjob_log.error('{0} [process failed (edms upload)] [task={1}] [error={2}]'.format(
self.log_base, task_str, traceback.format_exc()))
else:
doc.status = DocStatus.COMPLETE.value
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()
self.cronjob_log.info('{0} [process complete] [task={1}]'.format(self.log_base, task_str))
# os.remove(excel_path)
# TODO 细化文件状态，不同异常状态，归还队列，重试时采取不同的处理
# TODO 异常邮件通知
# 识别失败：普通异常，如PDF异常、构建过程异常
# EDMS异常：下载异常-->回队列-->邮件；上传异常-->重新上传队列-->邮件
# 算法异常：第一道异常-->识别失败-->邮件；第二道异常-->识别失败-->邮件
def handle(self, *args, **kwargs):
lock = Lock()
with Manager() as manager:
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))
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))
process_list.append(ocr_1_process)
wb_process = Process(target=self.res_2_wb, args=(res_dict, finish_queue, lock))
process_list.append(wb_process)
for p in process_list:
p.start()
for p in process_list:
p.join()
self.cronjob_log.info('{0} [stop safely]'.format(self.log_base))