Spaces:
Runtime error
Runtime error
| """ | |
| 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 | |