275 lines
12 KiB
Python
275 lines
12 KiB
Python
"""
|
|
Simple debug logger for AI prompts and responses.
|
|
Writes files chronologically to the configured log directory with sequential numbering.
|
|
"""
|
|
import os
|
|
from datetime import datetime, UTC
|
|
from typing import List, Optional, Any
|
|
from modules.shared.configuration import APP_CONFIG
|
|
|
|
|
|
def _resolveLogDir() -> str:
|
|
"""Resolve the absolute log directory from configuration."""
|
|
logDir = APP_CONFIG.get("APP_LOGGING_LOG_DIR", "./")
|
|
if not os.path.isabs(logDir):
|
|
# If relative path, make it relative to the gateway directory
|
|
gatewayDir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
logDir = os.path.join(gatewayDir, logDir)
|
|
return logDir
|
|
|
|
def _ensureDir(path: str) -> None:
|
|
"""Create directory if it does not exist."""
|
|
os.makedirs(path, exist_ok=True)
|
|
|
|
def _isDebugEnabled() -> bool:
|
|
"""Check if debug workflow logging is enabled."""
|
|
return APP_CONFIG.get("APP_DEBUG_CHAT_WORKFLOW_ENABLED", False)
|
|
|
|
def _getBaseDebugDir() -> str:
|
|
"""Get the base debug directory path from configuration."""
|
|
# Check if custom debug directory is configured
|
|
customDebugDir = APP_CONFIG.get("APP_DEBUG_CHAT_WORKFLOW_DIR", None)
|
|
if customDebugDir:
|
|
# Use custom debug directory if configured
|
|
if not os.path.isabs(customDebugDir):
|
|
# If relative path, make it relative to the gateway directory
|
|
gatewayDir = os.path.dirname(os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
|
|
customDebugDir = os.path.join(gatewayDir, customDebugDir)
|
|
return customDebugDir
|
|
|
|
# Default: Get log directory from config (same as used by main logging system)
|
|
logDir = _resolveLogDir()
|
|
|
|
# Create debug subdirectory within the log directory
|
|
return os.path.join(logDir, 'debug')
|
|
|
|
def _getDebugDir() -> str:
|
|
"""Get the debug prompts directory path from configuration."""
|
|
baseDebugDir = _getBaseDebugDir()
|
|
return os.path.join(baseDebugDir, 'prompts')
|
|
|
|
def _getNextSequenceNumber() -> int:
|
|
"""Get the next sequence number by counting existing files."""
|
|
debugDir = _getDebugDir()
|
|
if not os.path.exists(debugDir):
|
|
return 1
|
|
|
|
# Count existing numbered files
|
|
files = [f for f in os.listdir(debugDir) if f.startswith(('0', '1', '2', '3', '4', '5', '6', '7', '8', '9'))]
|
|
return len(files) + 1
|
|
|
|
|
|
def writeDebugFile(content: str, fileType: str, documents: Optional[List] = None) -> None:
|
|
"""
|
|
Write debug content to a file with sequential numbering.
|
|
Writes the content as-is since it's already the final integrated prompt.
|
|
Includes document list labels for tracing enhancement.
|
|
Only writes if debug logging is enabled via _isDebugEnabled() function.
|
|
|
|
Args:
|
|
content: The main content to write (already integrated)
|
|
fileType: Type of file (e.g., 'prompt_final', 'response')
|
|
documents: Optional list of documents for tracing
|
|
"""
|
|
try:
|
|
# Check if debug logging is enabled
|
|
if not _isDebugEnabled():
|
|
return
|
|
|
|
debugDir = _getDebugDir()
|
|
_ensureDir(debugDir)
|
|
|
|
seqNum = _getNextSequenceNumber()
|
|
ts = datetime.now(UTC).strftime('%Y%m%d-%H%M%S')
|
|
# Add 3-digit sequence number for uniqueness
|
|
tsWithSeq = f"{ts}-{seqNum:03d}"
|
|
|
|
# Allow callers to pass an extension; if none, default to .txt
|
|
if "." in (fileType or ""):
|
|
filename = f"{tsWithSeq}-{fileType}"
|
|
else:
|
|
filename = f"{tsWithSeq}-{fileType}.txt"
|
|
filepath = os.path.join(debugDir, filename)
|
|
|
|
# Build content with document tracing
|
|
debug_content = content
|
|
|
|
# Add document list labels for tracing enhancement
|
|
if documents:
|
|
debug_content += "\n\n=== DOCUMENT LIST FOR TRACING ===\n"
|
|
for i, doc in enumerate(documents):
|
|
if hasattr(doc, 'fileName'):
|
|
debug_content += f"Document {i+1}: {doc.fileName} ({doc.mimeType})\n"
|
|
elif hasattr(doc, 'fileId'):
|
|
debug_content += f"Document {i+1}: {doc.fileId} ({getattr(doc, 'mimeType', 'unknown')})\n"
|
|
else:
|
|
debug_content += f"Document {i+1}: {str(doc)[:100]}...\n"
|
|
|
|
# Write the content with document tracing
|
|
with open(filepath, 'w', encoding='utf-8') as f:
|
|
f.write(debug_content)
|
|
except Exception as e:
|
|
# Don't log debug errors to avoid recursion
|
|
pass
|
|
|
|
def debugLogToFile(message: str, context: str = "DEBUG") -> None:
|
|
"""
|
|
Log debug message to file if debug logging is enabled.
|
|
|
|
Args:
|
|
message: Debug message to log
|
|
context: Context identifier for the debug message
|
|
"""
|
|
try:
|
|
# Check if debug logging is enabled
|
|
if not _isDebugEnabled():
|
|
return
|
|
|
|
# Get debug directory (use base debug dir, not prompts subdirectory)
|
|
debug_dir = _getBaseDebugDir()
|
|
_ensureDir(debug_dir)
|
|
|
|
# Create debug file path
|
|
debug_file = os.path.join(debug_dir, "debug_workflow.log")
|
|
|
|
# Format the debug entry
|
|
from modules.shared.timeUtils import getUtcTimestamp
|
|
timestamp = getUtcTimestamp()
|
|
debug_entry = f"[{timestamp}] [{context}] {message}\n"
|
|
|
|
# Write to debug file
|
|
with open(debug_file, "a", encoding="utf-8") as f:
|
|
f.write(debug_entry)
|
|
|
|
except Exception as e:
|
|
# Don't log debug errors to avoid recursion
|
|
pass
|
|
|
|
def storeDebugMessageAndDocuments(message, currentUser) -> None:
|
|
"""
|
|
Store message and documents (metadata and file bytes) for debugging purposes.
|
|
Structure: {log_dir}/debug/messages/m_round_task_action_timestamp/documentlist_label/
|
|
- message.json, message_text.txt
|
|
- document_###_metadata.json
|
|
- document_###_<original_filename> (actual file bytes)
|
|
|
|
Args:
|
|
message: ChatMessage object to store
|
|
currentUser: Current user for component interface access
|
|
"""
|
|
try:
|
|
import json
|
|
|
|
# Create base debug directory (use base debug dir, not prompts subdirectory)
|
|
baseDebugDir = _getBaseDebugDir()
|
|
debug_root = os.path.join(baseDebugDir, 'messages')
|
|
_ensureDir(debug_root)
|
|
|
|
# Generate timestamp
|
|
timestamp = datetime.now(UTC).strftime('%Y%m%d-%H%M%S-%f')[:-3]
|
|
|
|
# Create message folder name: m_round_task_action_timestamp
|
|
# Use actual values from message, not defaults
|
|
round_str = str(message.roundNumber) if message.roundNumber is not None else "0"
|
|
task_str = str(message.taskNumber) if message.taskNumber is not None else "0"
|
|
action_str = str(message.actionNumber) if message.actionNumber is not None else "0"
|
|
message_folder = f"{timestamp}_m_{round_str}_{task_str}_{action_str}"
|
|
|
|
message_path = os.path.join(debug_root, message_folder)
|
|
os.makedirs(message_path, exist_ok=True)
|
|
|
|
# Store message data - use dict() instead of model_dump() for compatibility
|
|
message_file = os.path.join(message_path, "message.json")
|
|
with open(message_file, "w", encoding="utf-8") as f:
|
|
# Convert message to dict manually to avoid model_dump() issues
|
|
message_dict = {
|
|
"id": message.id,
|
|
"workflowId": message.workflowId,
|
|
"parentMessageId": message.parentMessageId,
|
|
"message": message.message,
|
|
"role": message.role,
|
|
"status": message.status,
|
|
"sequenceNr": message.sequenceNr,
|
|
"publishedAt": message.publishedAt,
|
|
"roundNumber": message.roundNumber,
|
|
"taskNumber": message.taskNumber,
|
|
"actionNumber": message.actionNumber,
|
|
"documentsLabel": message.documentsLabel,
|
|
"actionId": message.actionId,
|
|
"actionMethod": message.actionMethod,
|
|
"actionName": message.actionName,
|
|
"success": message.success,
|
|
"documents": []
|
|
}
|
|
json.dump(message_dict, f, indent=2, ensure_ascii=False, default=str)
|
|
|
|
# Store message content as text
|
|
if message.message:
|
|
message_text_file = os.path.join(message_path, "message_text.txt")
|
|
with open(message_text_file, "w", encoding="utf-8") as f:
|
|
f.write(str(message.message))
|
|
|
|
# Store documents if provided
|
|
if message.documents and len(message.documents) > 0:
|
|
# Group documents by documentsLabel
|
|
documents_by_label = {}
|
|
for doc in message.documents:
|
|
label = message.documentsLabel or 'default'
|
|
if label not in documents_by_label:
|
|
documents_by_label[label] = []
|
|
documents_by_label[label].append(doc)
|
|
|
|
# Create subfolder for each document label
|
|
for label, docs in documents_by_label.items():
|
|
# Sanitize label for filesystem
|
|
safe_label = "".join(c for c in str(label) if c.isalnum() or c in (' ', '-', '_')).rstrip()
|
|
safe_label = safe_label.replace(' ', '_')
|
|
if not safe_label:
|
|
safe_label = "default"
|
|
|
|
label_folder = os.path.join(message_path, safe_label)
|
|
_ensureDir(label_folder)
|
|
|
|
# Store each document
|
|
for i, doc in enumerate(docs):
|
|
# Create document metadata file
|
|
doc_meta = {
|
|
"id": doc.id,
|
|
"messageId": doc.messageId,
|
|
"fileId": doc.fileId,
|
|
"fileName": doc.fileName,
|
|
"fileSize": doc.fileSize,
|
|
"mimeType": doc.mimeType,
|
|
"roundNumber": doc.roundNumber,
|
|
"taskNumber": doc.taskNumber,
|
|
"actionNumber": doc.actionNumber,
|
|
"actionId": doc.actionId
|
|
}
|
|
|
|
doc_meta_file = os.path.join(label_folder, f"document_{i+1:03d}_metadata.json")
|
|
with open(doc_meta_file, "w", encoding="utf-8") as f:
|
|
json.dump(doc_meta, f, indent=2, ensure_ascii=False, default=str)
|
|
|
|
# Also store the actual file bytes next to metadata for debugging
|
|
try:
|
|
# Lazy import to avoid circular deps at module load
|
|
from modules.interfaces import interfaceDbComponentObjects as comp
|
|
componentInterface = comp.getInterface(currentUser)
|
|
file_bytes = componentInterface.getFileData(doc.fileId)
|
|
if file_bytes:
|
|
# Build a safe filename preserving original name
|
|
safe_name = doc.fileName or f"document_{i+1:03d}"
|
|
# Avoid path traversal
|
|
safe_name = os.path.basename(safe_name)
|
|
doc_file_path = os.path.join(label_folder, f"document_{i+1:03d}_" + safe_name)
|
|
with open(doc_file_path, "wb") as df:
|
|
df.write(file_bytes)
|
|
else:
|
|
pass
|
|
except Exception as e:
|
|
pass
|
|
|
|
except Exception as e:
|
|
# Silent fail - don't break main flow
|
|
pass
|
|
|