-
Notifications
You must be signed in to change notification settings - Fork 0
/
Copy pathmain.py
170 lines (143 loc) · 5.93 KB
/
main.py
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
import eel
import pathlib
from random import randint
from collections import defaultdict
from objects.file import BaseFile, CsvFile, XlsFile
from objects.task import BaseTask, SumTask, DataMatchTask
import traceback
input_file_map = defaultdict(BaseFile)
prev_output_path = str(pathlib.Path().resolve() /
"结果{0}.xlsx".format(randint(1, 10000)))
prev_output_sheet_map = defaultdict(int)
output_task_map = defaultdict(lambda: defaultdict(BaseTask))
@eel.expose
def re_initialize():
global input_file_map, prev_output_path, prev_output_sheet_map
input_file_map.clear()
prev_output_path = str(pathlib.Path().resolve() /
"结果{0}.xlsx".format(randint(1, 10000)))
prev_output_sheet_map.clear()
@ eel.expose
def get_next_output_path():
global prev_output_path, prev_output_sheet_map
return [prev_output_path, "页{0}".format(prev_output_sheet_map[prev_output_path] + 1)]
@ eel.expose
def get_sheet_names(file_path):
global input_file_map
try:
return input_file_map[file_path].get_sheet_names()
except Exception:
traceback.print_exc()
return [False, "遇到一个异常，请查看日志！"]
@ eel.expose
def get_headers(file_path, sheet_name):
global input_file_map
try:
return input_file_map[file_path].get_headers(sheet_name)
except Exception:
traceback.print_exc()
return [False, "遇到一个异常，请查看日志！"]
@ eel.expose
def get_column_values(file_path, sheet_name, column, exclude_header=True):
global input_file_map
try:
cols = input_file_map[file_path].get_column_values(
sheet_name, (int)(column.split(' ')[0]), exclude_header)[1]
return [True, cols]
except Exception:
traceback.print_exc()
return [False, "遇到一个异常，请查看日志！"]
@ eel.expose
def get_all_files():
global input_file_map
return [k for k in input_file_map.keys()]
@ eel.expose
def load_file(file_path, file_name):
global input_file_map
try:
extension = file_name.split(".")[-1].lower()
if extension not in ('csv', 'xls', 'xlsx'):
return [False, "文件后缀名必须是以下之一：“.csv”, “.xls”, “.xlsx”"]
# build absolute path
absolute_file_path = file_path
if not file_path:
absolute_file_path = pathlib.Path().resolve()
absolute_file_path = pathlib.Path(
absolute_file_path) / file_name.split('\\')[-1]
# validation
if not absolute_file_path.exists():
return [False, str(absolute_file_path) + ' 不存在']
elif not absolute_file_path.is_file():
return [False, str(absolute_file_path) + ' 不是文件']
elif str(absolute_file_path) in input_file_map:
return [False, str(absolute_file_path) + ' 已加载']
# create file object
if extension == 'csv':
input_file_map[str(absolute_file_path)] = CsvFile(
str(absolute_file_path))
else:
input_file_map[str(absolute_file_path)] = XlsFile(
str(absolute_file_path))
return [True, str(absolute_file_path)]
except Exception:
traceback.print_exc()
return [False, "遇到一个异常，请查看日志！"]
@ eel.expose
def delete_file(file_path):
global input_file_map
delete_tasks = defaultdict(list)
if file_path in input_file_map:
for output_file_path, output_sheet_task in output_task_map.items():
for output_sheet, task in output_sheet_task.items():
for select in task.selects:
if select.file.path == file_path:
delete_tasks[output_file_path].append(output_sheet)
del input_file_map[file_path]
output_keys = list()
for output_file_path, output_sheet_list in delete_tasks.items():
for output_sheet in output_sheet_list:
output_keys.append(output_file_path + ":" + output_sheet)
del output_task_map[output_file_path][output_sheet]
return output_keys
@ eel.expose
def add_task(task_settings_json):
global output_task_map, input_file_map
output_file_path = task_settings_json['output-file-path']
output_sheet_name = task_settings_json['output-sheet-name']
if output_file_path in output_task_map and output_sheet_name in output_task_map[output_file_path]:
return [False, output_file_path + ':' + output_sheet_name + ' 已存在']
try:
if task_settings_json['task-type'] == 'sum':
output_task_map[output_file_path][output_sheet_name] = SumTask(
task_settings_json, input_file_map)
elif task_settings_json['task-type'] == 'data-match':
output_task_map[output_file_path][output_sheet_name] = DataMatchTask(
task_settings_json, input_file_map)
except Exception as e:
traceback.print_exc()
return [False, str(e)]
return [True, None]
@ eel.expose
def delete_task(output_file_path, output_sheet_name):
global output_task_map
if output_file_path in output_task_map and output_sheet_name in output_task_map[output_file_path]:
del output_task_map[output_file_path][output_sheet_name]
@ eel.expose
def run_all_tasks():
global output_task_map
if len(output_task_map) == 0:
return [False, "请先添加任务！"]
output_paths = set()
for output_path, output_sheet_task in output_task_map.items():
for _, task in output_sheet_task.items():
try:
task.run()
output_paths.add(output_path)
except Exception:
traceback.print_exc()
return [False, "遇到一个异常，请查看日志！"]
output_task_map.clear()
return [True, '任务完成，请在文件 [' + '、'.join(output_paths) + ' 中查看结果！']
if __name__ == "__main__":
eel.init('web')
eel.start('index.html', size=(1920, 720))