Spaces:
Running
Running
Add logging for data export process and improve error handling in ChatAnalyzer
Browse files
src/analytics/chat_analyzer.py
CHANGED
|
@@ -1,11 +1,19 @@
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
| 1 |
from typing import List, Dict, Any, Tuple, Optional
|
| 2 |
-
from collections import Counter
|
| 3 |
-
from datetime import datetime
|
| 4 |
import re
|
| 5 |
-
import
|
| 6 |
-
|
| 7 |
from src.knowledge_base.dataset import DatasetManager
|
| 8 |
|
|
|
|
|
|
|
|
|
|
|
|
|
| 9 |
class ChatAnalyzer:
|
| 10 |
"""Chat history analyzer"""
|
| 11 |
|
|
@@ -186,8 +194,11 @@ class ChatAnalyzer:
|
|
| 186 |
qa_pairs = self.extract_question_answer_pairs()
|
| 187 |
|
| 188 |
if not qa_pairs:
|
|
|
|
| 189 |
return False, "Not enough data for export"
|
| 190 |
|
|
|
|
|
|
|
| 191 |
with open(output_file, "w", encoding="utf-8") as f:
|
| 192 |
for pair in qa_pairs:
|
| 193 |
training_example = {
|
|
@@ -198,8 +209,11 @@ class ChatAnalyzer:
|
|
| 198 |
}
|
| 199 |
f.write(json.dumps(training_example, ensure_ascii=False) + "\n")
|
| 200 |
|
|
|
|
| 201 |
return True, f"Training data successfully exported to {output_file}. Exported {len(qa_pairs)} examples."
|
|
|
|
| 202 |
except Exception as e:
|
|
|
|
| 203 |
return False, f"Error exporting training data: {str(e)}"
|
| 204 |
|
| 205 |
def get_chat_data(self) -> List[Dict[str, Any]]:
|
|
@@ -210,9 +224,8 @@ class ChatAnalyzer:
|
|
| 210 |
List of chat histories
|
| 211 |
"""
|
| 212 |
success, chat_data = self.dataset_manager.get_chat_history()
|
| 213 |
-
# Добавим логирование для отладки
|
| 214 |
if not success:
|
| 215 |
-
logger.error(f"Failed to get chat history: {chat_data}")
|
| 216 |
if not chat_data:
|
| 217 |
logger.warning("Chat data is empty")
|
| 218 |
return chat_data if success and chat_data else []
|
|
|
|
| 1 |
+
"""
|
| 2 |
+
Module for analyzing chat history and extracting useful data for training
|
| 3 |
+
"""
|
| 4 |
+
|
| 5 |
+
import json
|
| 6 |
+
import logging
|
| 7 |
from typing import List, Dict, Any, Tuple, Optional
|
| 8 |
+
from collections import Counter, defaultdict
|
|
|
|
| 9 |
import re
|
| 10 |
+
from datetime import datetime
|
|
|
|
| 11 |
from src.knowledge_base.dataset import DatasetManager
|
| 12 |
|
| 13 |
+
# Configure logging
|
| 14 |
+
logging.basicConfig(level=logging.INFO)
|
| 15 |
+
logger = logging.getLogger(__name__)
|
| 16 |
+
|
| 17 |
class ChatAnalyzer:
|
| 18 |
"""Chat history analyzer"""
|
| 19 |
|
|
|
|
| 194 |
qa_pairs = self.extract_question_answer_pairs()
|
| 195 |
|
| 196 |
if not qa_pairs:
|
| 197 |
+
logger.warning("Not enough data for export")
|
| 198 |
return False, "Not enough data for export"
|
| 199 |
|
| 200 |
+
logger.info(f"Found {len(qa_pairs)} question-answer pairs for export")
|
| 201 |
+
|
| 202 |
with open(output_file, "w", encoding="utf-8") as f:
|
| 203 |
for pair in qa_pairs:
|
| 204 |
training_example = {
|
|
|
|
| 209 |
}
|
| 210 |
f.write(json.dumps(training_example, ensure_ascii=False) + "\n")
|
| 211 |
|
| 212 |
+
logger.info(f"Data successfully exported to {output_file}")
|
| 213 |
return True, f"Training data successfully exported to {output_file}. Exported {len(qa_pairs)} examples."
|
| 214 |
+
|
| 215 |
except Exception as e:
|
| 216 |
+
logger.error(f"Error during data export: {str(e)}")
|
| 217 |
return False, f"Error exporting training data: {str(e)}"
|
| 218 |
|
| 219 |
def get_chat_data(self) -> List[Dict[str, Any]]:
|
|
|
|
| 224 |
List of chat histories
|
| 225 |
"""
|
| 226 |
success, chat_data = self.dataset_manager.get_chat_history()
|
|
|
|
| 227 |
if not success:
|
| 228 |
+
logger.error(f"Failed to get chat history: {chat_data}")
|
| 229 |
if not chat_data:
|
| 230 |
logger.warning("Chat data is empty")
|
| 231 |
return chat_data if success and chat_data else []
|