273 lines
7.4 KiB
Python
273 lines
7.4 KiB
Python
"""
|
|
Logging configuration for Code of Conquest.
|
|
|
|
Sets up structured logging using structlog with JSON output
|
|
and context-aware logging throughout the application.
|
|
"""
|
|
|
|
import logging
|
|
import sys
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
|
|
import structlog
|
|
from structlog.stdlib import LoggerFactory
|
|
|
|
|
|
def setup_logging(
|
|
log_level: str = "INFO",
|
|
log_format: str = "json",
|
|
log_file: Optional[str] = None
|
|
) -> None:
|
|
"""
|
|
Configure structured logging for the application.
|
|
|
|
Args:
|
|
log_level: Logging level (DEBUG, INFO, WARNING, ERROR, CRITICAL)
|
|
log_format: Output format ('json' or 'console')
|
|
log_file: Optional path to log file
|
|
|
|
Example:
|
|
>>> from app.utils.logging import setup_logging
|
|
>>> setup_logging(log_level="DEBUG", log_format="json")
|
|
>>> logger = structlog.get_logger(__name__)
|
|
>>> logger.info("Application started", version="0.1.0")
|
|
"""
|
|
# Convert log level string to logging constant
|
|
numeric_level = getattr(logging, log_level.upper(), logging.INFO)
|
|
|
|
# Configure standard library logging
|
|
logging.basicConfig(
|
|
format="%(message)s",
|
|
stream=sys.stdout,
|
|
level=numeric_level,
|
|
)
|
|
|
|
# Create logs directory if logging to file
|
|
if log_file:
|
|
log_path = Path(log_file)
|
|
log_path.parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
# Add file handler
|
|
file_handler = logging.FileHandler(log_file)
|
|
file_handler.setLevel(numeric_level)
|
|
logging.root.addHandler(file_handler)
|
|
|
|
# Configure structlog processors
|
|
processors = [
|
|
# Add log level to event dict
|
|
structlog.stdlib.add_log_level,
|
|
# Add logger name to event dict
|
|
structlog.stdlib.add_logger_name,
|
|
# Add timestamp
|
|
structlog.processors.TimeStamper(fmt="iso"),
|
|
# Add stack info for exceptions
|
|
structlog.processors.StackInfoRenderer(),
|
|
# Format exceptions
|
|
structlog.processors.format_exc_info,
|
|
# Clean up _record and _from_structlog keys
|
|
structlog.processors.UnicodeDecoder(),
|
|
]
|
|
|
|
# Add format-specific processor
|
|
if log_format == "json":
|
|
# JSON output for production
|
|
processors.append(structlog.processors.JSONRenderer())
|
|
else:
|
|
# Console-friendly output for development
|
|
processors.append(
|
|
structlog.dev.ConsoleRenderer(
|
|
colors=True,
|
|
exception_formatter=structlog.dev.plain_traceback
|
|
)
|
|
)
|
|
|
|
# Configure structlog
|
|
structlog.configure(
|
|
processors=processors,
|
|
context_class=dict,
|
|
logger_factory=LoggerFactory(),
|
|
wrapper_class=structlog.stdlib.BoundLogger,
|
|
cache_logger_on_first_use=True,
|
|
)
|
|
|
|
|
|
def get_logger(name: str) -> structlog.stdlib.BoundLogger:
|
|
"""
|
|
Get a configured logger instance.
|
|
|
|
Args:
|
|
name: Logger name (typically __name__)
|
|
|
|
Returns:
|
|
BoundLogger: Configured structlog logger
|
|
|
|
Example:
|
|
>>> logger = get_logger(__name__)
|
|
>>> logger.info("User logged in", user_id="123", email="user@example.com")
|
|
"""
|
|
return structlog.get_logger(name)
|
|
|
|
|
|
class LoggerMixin:
|
|
"""
|
|
Mixin class to add logging capabilities to any class.
|
|
|
|
Provides a `self.logger` attribute with context automatically
|
|
bound to the class name.
|
|
|
|
Example:
|
|
>>> class MyService(LoggerMixin):
|
|
... def do_something(self, user_id):
|
|
... self.logger.info("Doing something", user_id=user_id)
|
|
"""
|
|
|
|
@property
|
|
def logger(self) -> structlog.stdlib.BoundLogger:
|
|
"""Get logger for this class."""
|
|
if not hasattr(self, '_logger'):
|
|
self._logger = get_logger(self.__class__.__name__)
|
|
return self._logger
|
|
|
|
|
|
# Common logging utilities
|
|
|
|
def log_function_call(logger: structlog.stdlib.BoundLogger):
|
|
"""
|
|
Decorator to log function calls with arguments and return values.
|
|
|
|
Args:
|
|
logger: Logger instance to use
|
|
|
|
Example:
|
|
>>> logger = get_logger(__name__)
|
|
>>> @log_function_call(logger)
|
|
... def process_data(data_id):
|
|
... return {"status": "processed"}
|
|
"""
|
|
def decorator(func):
|
|
def wrapper(*args, **kwargs):
|
|
logger.debug(
|
|
f"Calling {func.__name__}",
|
|
function=func.__name__,
|
|
args=args,
|
|
kwargs=kwargs
|
|
)
|
|
try:
|
|
result = func(*args, **kwargs)
|
|
logger.debug(
|
|
f"{func.__name__} completed",
|
|
function=func.__name__,
|
|
result=result
|
|
)
|
|
return result
|
|
except Exception as e:
|
|
logger.error(
|
|
f"{func.__name__} failed",
|
|
function=func.__name__,
|
|
error=str(e),
|
|
exc_info=True
|
|
)
|
|
raise
|
|
return wrapper
|
|
return decorator
|
|
|
|
|
|
def log_ai_call(
|
|
logger: structlog.stdlib.BoundLogger,
|
|
user_id: str,
|
|
model: str,
|
|
tier: str,
|
|
tokens_used: int,
|
|
cost_estimate: float,
|
|
context_type: str
|
|
) -> None:
|
|
"""
|
|
Log AI API call for cost tracking and analytics.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
user_id: User making the request
|
|
model: AI model used
|
|
tier: Model tier (free, standard, premium)
|
|
tokens_used: Number of tokens consumed
|
|
cost_estimate: Estimated cost in USD
|
|
context_type: Type of context (narrative, combat, etc.)
|
|
"""
|
|
logger.info(
|
|
"AI call completed",
|
|
event_type="ai_call",
|
|
user_id=user_id,
|
|
model=model,
|
|
tier=tier,
|
|
tokens_used=tokens_used,
|
|
cost_estimate=cost_estimate,
|
|
context_type=context_type
|
|
)
|
|
|
|
|
|
def log_combat_action(
|
|
logger: structlog.stdlib.BoundLogger,
|
|
session_id: str,
|
|
character_id: str,
|
|
action_type: str,
|
|
target_id: Optional[str] = None,
|
|
damage: Optional[int] = None,
|
|
effects: Optional[list] = None
|
|
) -> None:
|
|
"""
|
|
Log combat action for analytics and debugging.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
session_id: Game session ID
|
|
character_id: Acting character ID
|
|
action_type: Type of action (attack, cast, item, defend)
|
|
target_id: Target of action (if applicable)
|
|
damage: Damage dealt (if applicable)
|
|
effects: Effects applied (if applicable)
|
|
"""
|
|
logger.info(
|
|
"Combat action executed",
|
|
event_type="combat_action",
|
|
session_id=session_id,
|
|
character_id=character_id,
|
|
action_type=action_type,
|
|
target_id=target_id,
|
|
damage=damage,
|
|
effects=effects
|
|
)
|
|
|
|
|
|
def log_marketplace_transaction(
|
|
logger: structlog.stdlib.BoundLogger,
|
|
transaction_id: str,
|
|
buyer_id: str,
|
|
seller_id: str,
|
|
item_id: str,
|
|
price: int,
|
|
transaction_type: str
|
|
) -> None:
|
|
"""
|
|
Log marketplace transaction for analytics and auditing.
|
|
|
|
Args:
|
|
logger: Logger instance
|
|
transaction_id: Transaction ID
|
|
buyer_id: Buyer user ID
|
|
seller_id: Seller user ID
|
|
item_id: Item ID
|
|
price: Transaction price
|
|
transaction_type: Type of transaction
|
|
"""
|
|
logger.info(
|
|
"Marketplace transaction",
|
|
event_type="marketplace_transaction",
|
|
transaction_id=transaction_id,
|
|
buyer_id=buyer_id,
|
|
seller_id=seller_id,
|
|
item_id=item_id,
|
|
price=price,
|
|
transaction_type=transaction_type
|
|
)
|