idcard_daily.py
3.5 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
87
88
89
90
91
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_name = '身份证'
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')
afc_excel_dir = os.path.join(conf.DATA_DIR, 'AFC', 'IdCard')
hil_excel_dir = os.path.join(conf.DATA_DIR, 'HIL', 'IdCard')
if not os.path.exists(afc_excel_dir) or not os.path.exists(hil_excel_dir):
print('excel_dir not exist')
return
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_afc = Workbook()
ws_afc = wb_afc.create_sheet(self.sheet_name)
ws_afc.append(self.header)
wb_afc.remove(wb_afc.get_sheet_by_name('Sheet'))
wb_hil = Workbook()
ws_hil = wb_hil.create_sheet(self.sheet_name)
ws_hil.append(self.header)
wb_hil.remove(wb_hil.get_sheet_by_name('Sheet'))
with open(log_path, 'r', encoding='utf-8') as fp:
for line in fp:
search_obj = re.match(r'\[(.*)] \[task=(.*)] \[idcard=(.*)]', line)
idcard_str = search_obj.group(3)
idcard_list = ast.literal_eval(idcard_str)
content_list = []
for idcard_dict in idcard_list:
nation = idcard_dict.get('民族')
if nation is None:
continue
if idcard_dict.get('类别') == '1':
continue
content_list.append((idcard_dict.get('公民身份号码'), nation))
if len(content_list) == 0:
continue
time_str = search_obj.group(1)
task_str = search_obj.group(2)
business_type, doc_id_str = task_str.split(consts.SPLIT_STR)
doc_class = HILDoc if business_type == consts.HIL_PREFIX else AFCDoc
application_id = doc_class.objects.filter(id=int(doc_id_str)).values_list('application_id', flat=True)
if business_type == consts.HIL_PREFIX:
for id_num, nation in content_list:
ws_hil.append((application_id[0], id_num, nation, time_str))
else:
for id_num, nation in content_list:
ws_afc.append((application_id[0], id_num, nation, time_str))
afc_excel_path = os.path.join(afc_excel_dir, 'idcard_{0}.xlsx'.format(date_str))
hil_excel_path = os.path.join(hil_excel_dir, 'idcard_{0}.xlsx'.format(date_str))
wb_afc.save(afc_excel_path)
wb_hil.save(hil_excel_path)