ouyangyouzhang/QueryRewrite
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

优化对话到工单的处理逻辑，新增进度条显示，调整日期参数的默认值，并修复日志目录创建的冗余代码。同时，更新DifyExporter类以支持按日期范围过滤消息，重构查询日志加载逻辑，新增备注提取功能，提升代码可读性和可维护性。

Browse Source
This commit is contained in:
ouyangyouzhang
2025-07-14 17:21:21 +08:00
parent b06a84c059
commit af1e1a9d9b
6 changed files with 88 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rag2_0/dify/export_new_dify.py
+49 -25
View File
@@ -37,9 +37,9 @@ class DifyExporter:
self.query_log_dir = os.path.join(os.getcwd(), "data", "query_logs")
self.query_log_file = query_log_file or os.path.join(self.query_log_dir, "answer_type_logs.json")
# 设置日期过滤
self.start_date = start_date
self.end_date = end_date
# 设置日期过滤，转换为datetime对象
self.start_date = datetime.datetime.strptime(start_date, "%Y-%m-%d %H") if start_date else None
self.end_date = datetime.datetime.strptime(end_date, "%Y-%m-%d %H") if end_date else None
# 初始化工具类
self.dify_pgsql = PgSql()
@@ -49,22 +49,22 @@ class DifyExporter:
self.message_info_list = []
self.query_logs = {}
def load_query_logs(self):
def load_query_logs(self,path):
"""
从文件加载查询日志
"""
try:
with open(self.query_log_file, 'r', encoding='utf-8') as f:
with open(path, 'r', encoding='utf-8') as f:
query_logs_list = json.load(f)
# 创建字典来存储每个查询的最新记录
# 创建字典来存储每个查询的最新记录workflow_run_id
for record in query_logs_list:
query = record['query']
workflow_run_id = record['workflow_run_id']
timestamp = record.get('timestamp')
# 如果查询不在字典中或者当前记录的时间戳更新，则更新字典
if query not in self.query_logs or (timestamp and self.query_logs.get(query, {}).get('timestamp') and
if workflow_run_id not in self.query_logs or (timestamp and self.query_logs.get(workflow_run_id, {}).get('timestamp') and
datetime.datetime.fromisoformat(timestamp) >
datetime.datetime.fromisoformat(self.query_logs[query]['timestamp'])):
self.query_logs[query] = record
datetime.datetime.fromisoformat(self.query_logs[workflow_run_id]['timestamp'])):
self.query_logs[workflow_run_id] = record
return True
except Exception as e:
print(f"加载查询日志失败: {e}")
@@ -103,6 +103,27 @@ class DifyExporter:
message_chain_new.append(msg)
return message_chain_new
def get_remark(self, msg_debug_info):
"""
获取备注
"""
intent_node_execution_info = [node_execution_info for node_execution_info in msg_debug_info['workflow_node_executions_info']
if node_execution_info["title"] == "意图识别结果解析"]
if len(intent_node_execution_info) == 0:
return ""
if intent_node_execution_info[0]["outputs"] is None:
return ""
intent_result = json.loads(intent_node_execution_info[0]["outputs"])
vertical_classification = intent_result.get("vertical_classification", "")
sub_classification = intent_result.get("sub_classification", "")
if vertical_classification == "固定话术类":
return "使用固定话术"
if sub_classification == "软件锁类":
return "固定引导至博微软件助手中操作"
return ""
def extract_message_info(self, message):
"""
从消息中提取信息
@@ -121,7 +142,7 @@ class DifyExporter:
user_name = msg_inputs.get("user_name", "")
msg_query = message["query"]
msg_answer = message["answer"]
msg_answer = msg_answer.split("----------------------------------------")[0]
# 将UTC+0时间转换为UTC+8时间
created_at_utc = message['created_at']
created_at_utc8 = created_at_utc + datetime.timedelta(hours=8)
@@ -143,13 +164,17 @@ class DifyExporter:
document_name = knowledge['metadata']['document_name']
wiki_list.append(document_name.split("/")[-1])
# 获取备注
remark = self.get_remark(msg_debug_info)
wiki_list = list(set(wiki_list))
wiki_list_str = "\n".join(wiki_list)
if wiki_list_str == "":
wiki_list_str = "无"
rating = self.dify_pgsql.get_message_rating(msg_id)
# 直接通过字典键获取query_type
query_type = self.query_logs.get(msg_query, {}).get('query_type', "")
workflow_run_id = message['workflow_run_id']
query_type = self.query_logs.get(workflow_run_id, {}).get('query_type', "")
return {
"msg_id": msg_id,
@@ -159,7 +184,8 @@ class DifyExporter:
"提问时间": created_at,
"评价": rating,
"问题分类": query_type,
"检索到的词条": wiki_list_str
"检索到的词条": wiki_list_str,
"备注": remark
}
def process_conversations(self):
@@ -184,13 +210,10 @@ class DifyExporter:
created_at_utc = message['created_at']
created_at_utc8 = created_at_utc + datetime.timedelta(hours=8)
# 提取消息的创建日期时间，精确到小时
created_at_hour = created_at_utc8.strftime("%Y-%m-%d %H")
# 应用日期时间过滤
if self.start_date and created_at_hour < self.start_date:
if self.start_date and created_at_utc8 < self.start_date:
continue
if self.end_date and created_at_hour > self.end_date:
if self.end_date and created_at_utc8 > self.end_date:
continue
message_info = self.extract_message_info(message)
@@ -281,7 +304,8 @@ class DifyExporter:
数据库中的时间是UTC+0时区，会自动转换为UTC+8时区进行过滤和显示
"""
# 加载查询日志
self.load_query_logs()
self.load_query_logs(self.query_log_file)
self.load_query_logs("data/query_logs/answer_type_logs_071409.json")
# 处理会话数据
self.process_conversations()
@@ -294,12 +318,12 @@ class DifyExporter:
# 如果指定了日期范围，则在文件名中体现
date_suffix = ""
if self.start_date:
# 将空格替换为下划线，使文件名更规范
formatted_start = self.start_date.replace(" ", "_")
# 格式化日期对象为字符串
formatted_start = self.start_date.strftime("%Y-%m-%d_%H")
date_suffix += f"_from_{formatted_start}"
if self.end_date:
# 将空格替换为下划线，使文件名更规范
formatted_end = self.end_date.replace(" ", "_")
# 格式化日期对象为字符串
formatted_end = self.end_date.strftime("%Y-%m-%d_%H")
date_suffix += f"_to_{formatted_end}"
output_file = os.path.join(os.getcwd(), "data", "excel", f"dify_export{date_suffix}_{timestamp}.xlsx")
@@ -321,9 +345,9 @@ if __name__ == "__main__":
help='Dify应用ID')
parser.add_argument('--query_log_file', '-q', type=str, default="data/query_logs/answer_type_logs.json",
help='查询日志文件路径')
parser.add_argument('--start_date', '-s', type=str, default="2025-07-09 13",
parser.add_argument('--start_date', '-s', type=str, default="2025-07-14 00",
help='开始日期时间，格式为YYYY-MM-DD HH，例如2025-07-08 14表示2025年7月8日14时（UTC+8时区）')
parser.add_argument('--end_date', '-e', type=str, default=None,
parser.add_argument('--end_date', '-e', type=str, default="2025-07-14 15",
help='结束日期时间，格式为YYYY-MM-DD HH，例如2025-07-08 18表示2025年7月8日18时（UTC+8时区）')
args = parser.parse_args()