Spaces:
Runtime error
Runtime error
File size: 2,982 Bytes
24cdea4 | 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 | """
Logging utilities for Pipecat server.
Provides session-scoped logging with S3 upload capability.
File is named 'logging.py' but imported via utils package to avoid
conflicts with Python's built-in logging module.
"""
import os
import traceback
from typing import Callable, Optional
from loguru import logger
from utils.s3_client import get_s3_client
def get_log_filename(log_key: str) -> str:
"""
Get the temporary log file path for a session.
Args:
log_key: Session identifier (typically transcript_key with / replaced by _)
Returns:
Path to the log file in /tmp directory
"""
if not log_key or not isinstance(log_key, str):
log_key = "unknown"
safe_key = log_key.replace("/", "_")
return f"/tmp/{safe_key}.log"
def _log(level_func: Callable, log_key: str, msg: str) -> None:
"""
Helper to log a message at a given level to a temporary file sink.
Args:
level_func: Logger method (logger.info, logger.error, etc.)
log_key: Session identifier for file naming
msg: Message to log
"""
filename = get_log_filename(log_key)
sink_id = logger.add(filename)
level_func(f"[key={log_key}, msg={msg}]")
logger.remove(sink_id)
def log_info(log_key: Optional[str], msg: str) -> None:
"""Log info message for session."""
_log(logger.info, log_key or "system", msg)
def log_checkpoint(log_key: str, label: str) -> None:
"""Log checkpoint marker for debugging."""
_log(logger.info, log_key, f"Checkpoint {label}")
def log_debug(log_key: str, msg: str) -> None:
"""Log debug message for session."""
_log(logger.debug, log_key, msg)
def log_warning(log_key: str, msg: str) -> None:
"""Log warning message for session."""
_log(logger.warning, log_key, msg)
def log_error(log_key: Optional[str], msg: str) -> None:
"""Log error message with stack trace for session."""
_log(logger.error, log_key or "system", msg + "\n\n" + traceback.format_exc())
def save_log_to_s3(key_transcript: str) -> bool:
"""
Upload session log file to S3.
Args:
key_transcript: Transcript key used to derive log file name
Returns:
True if upload succeeded, False otherwise
"""
if not key_transcript or not isinstance(key_transcript, str):
logger.warning(f"Cannot save log to S3 - invalid key_transcript: {key_transcript}")
return False
try:
filename = get_log_filename(key_transcript)
if not os.path.isfile(filename):
log_info(
key_transcript,
f"Log file not uploaded to S3. File doesn't exist [log_file={filename}]",
)
return False
with open(filename, "r") as f:
logs = f.read()
s3_client = get_s3_client()
return s3_client.upload_log(key_transcript, logs)
except Exception as e:
logger.error(f"Failed to save log to S3: {e}")
return False
|