idcard_statistics.py
3.62 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
import re
import os
import ast
import datetime
from openpyxl import Workbook
from django.core.management import BaseCommand
from settings import conf
from common.mixins import LoggerMixin
from apps.doc.models import HILDoc, AFCDoc
from apps.doc import consts
class Command(BaseCommand, LoggerMixin):
def __init__(self):
super().__init__()
self.sheet_names = ('AFC', 'HIL')
self.header = ('申请号', '身份证号', '民族', '时间戳')
def add_arguments(self, parser):
parser.add_argument(
'--date',
default=datetime.date.today() - datetime.timedelta(days=1),
dest='date',
help='将要计算的日期，格式: 2018-01-01'
)
def handle(self, *args, **kwargs):
date = kwargs.get('date')
if isinstance(date, str):
if not re.match(r'\d{4}-\d{2}-\d{2}', date):
print('date format error')
return
date_str = date
else:
date_str = date.strftime('%Y-%m-%d')
excel_dir = os.path.join(conf.DATA_DIR, 'AFC', 'Logs')
if not os.path.exists(excel_dir):
print('excel dir not exists')
return
excel_path = os.path.join(excel_dir, 'idcard_{0}.xlsx'.format(date_str))
log_path = os.path.join(conf.LOG_DIR, 'idcard.log.{0}'.format(date_str))
if not os.path.exists(log_path):
print('log_path not exists')
return
wb = Workbook()
for name in self.sheet_names:
ws = wb.create_sheet(name)
ws.append(self.header)
wb.remove(wb.get_sheet_by_name('Sheet'))
with open(log_path, 'r', encoding='utf-8') as fp:
for line in fp:
search_obj = re.search(r'[(.*)] [task=(.*)] [idcard=(.*)]', line)
task_str = search_obj.group(1)
license_summary = ast.literal_eval(search_obj.group(2))
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
application_id = doc_class.objects.filter(id=doc_id).values_list('application_id', flat=True)
for classify, (_, name, field_order, side_diff, _, _) in consts.LICENSE_ORDER:
license_list = license_summary.get(classify)
if not license_list:
continue
ws = wb.get_sheet_by_name(name)
for license_dict in license_list:
if classify == consts.IC_CLASSIFY and license_dict.get('类别') == '1': # 居住证处理
license_summary.setdefault(consts.RP_CLASSIFY, []).append(license_dict)
continue
if side_diff:
key, field_order_yes, field_order_no = consts.FIELD_ORDER_MAP.get(classify)
field_order = field_order_yes if key in license_dict else field_order_no
all_value = []
for search_field, write_field in field_order:
if write_field is None:
continue
field_value = license_dict.get(search_field, '')
if isinstance(field_value, list):
all_value.append('\n'.join(field_value))
else:
all_value.append(field_value)
ws.append((application_id[0], *all_value))
wb.save(excel_path)