-
Notifications
You must be signed in to change notification settings - Fork 0
/
ui_local_ct.py
140 lines (113 loc) · 3.98 KB
/
ui_local_ct.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
import gradio as gr
import time
from repolya._const import LOG_ROOT
from repolya._log import logger_jd
from repolya.autogen.wf_jd import (
generate_search_dict_for_event,
generate_event_context,
generate_event_context_textgen,
generate_event_plan,
)
_log_ans_ct_context = LOG_ROOT / '_ans_ct_context.txt'
_log_ans_ct_plan = LOG_ROOT / '_ans_ct_plan.txt'
_log_ref_ct = LOG_ROOT / '_ref_jd.txt'
ct_keyword = {
"基本情况_1": "发生发展时间线",
"基本情况_2": "冲突规模和强度",
"影响评估_1": "冲突对国际政治的影响",
"影响评估_2": "冲突对国际经济的影响",
}
ct_section = [
"基本情况",
"影响评估",
]
def write_log_ans(_log_ans, _txt, _status=None):
with open(_log_ans, 'w', encoding='utf-8') as wf:
if _status == "continue":
_txt += "\n\n计算中,预计需要10分钟,请稍候..."
# elif _status == "done":
# _txt += "\n\n[完成]"
wf.write(_txt)
def write_log_ref(_log_ref, _txt):
with open(_log_ref, 'w', encoding='utf-8') as wf:
wf.write(_txt)
def ct_read_logs():
with open(_log_ref_ct, "r") as f:
_ref_ct = f.read()
with open(_log_ans_ct_context, "r") as f:
_ans_ct_context = f.read()
with open(_log_ans_ct_plan, "r") as f:
_ans_ct_plan = f.read()
return [_ref_ct, _ans_ct_context, _ans_ct_plan]
def ct_clean_logs():
write_log_ref(_log_ref_ct,'')
write_log_ans(_log_ans_ct_context,'')
write_log_ans(_log_ans_ct_plan,'')
ct_clean_logs()
def ct_clean_all():
ct_clean_logs()
print('ct_clean_logs()')
return [
gr.Textbox(value=""), # ct_query
gr.Button(variant="secondary"), # ct_start_btn
gr.Button(variant="secondary"), # ct_plan_btn
gr.Textbox(value=""), # ct_log
gr.Textbox(value=""), # ct_context
gr.Textbox(value=""), # ct_plan
]
##### ct
def ct_sort_out_context(_event):
write_log_ans(_log_ans_ct_context, '')
start_time = time.time()
write_log_ans(_log_ans_ct_context, '', 'continue')
#####
_dict = generate_search_dict_for_event(_event, ct_keyword)
global _content_fp
_context, _content_fp = generate_event_context(_event, _dict, ct_keyword, ct_section)
write_log_ans(_log_ans_ct_context, _context, 'done')
#####
end_time = time.time()
execution_time = end_time - start_time
_time = f"'梳理脉络'耗时:{execution_time:.1f} seconds"
logger_jd.info(_time)
time.sleep(1)
return gr.Button(variant="primary")
def ct_sort_out_context_textgen(_event):
write_log_ans(_log_ans_ct_context, '')
start_time = time.time()
write_log_ans(_log_ans_ct_context, '', 'continue')
#####
_dict = generate_search_dict_for_event(_event, ct_keyword)
global _content_fp
_content_fp = ""
_textgen_url = "http://127.0.0.1:5552"
# _context, _content_fp = generate_event_context_textgen(_event, _dict, _textgen_url, ct_keyword, ct_section)
_context, _content_fp = generate_event_context(_event, _dict, ct_keyword, ct_section)
write_log_ans(_log_ans_ct_context, _context, 'done')
#####
end_time = time.time()
execution_time = end_time - start_time
_time = f"'梳理脉络'耗时:{execution_time:.1f} seconds"
logger_jd.info(_time)
time.sleep(1)
return gr.Button(variant="primary")
def ct_write_plan(_event, _context):
write_log_ans(_log_ans_ct_plan, '')
start_time = time.time()
write_log_ans(_log_ans_ct_plan, '', 'continue')
#####
_plan = generate_event_plan(_event, _context)
write_log_ans(_log_ans_ct_plan, _plan, 'done')
#####
end_time = time.time()
execution_time = end_time - start_time
_time = f"'总结报告'耗时:{execution_time:.1f} seconds"
logger_jd.info(_time)
def show_ct_context(text):
# print(f"text: {text}")
if text:
if _content_fp:
return gr.File(value=_content_fp, visible=True)
else:
if _content_fp:
return gr.File(value=_content_fp)