Spaces:
Sleeping
Sleeping
| import logging | |
| import json | |
| import os | |
| from logging.handlers import RotatingFileHandler | |
| from datetime import datetime | |
| # Ensure logs directory exists | |
| LOG_DIR = "logs" | |
| os.makedirs(LOG_DIR, exist_ok=True) | |
| LOG_FILE = os.path.join(LOG_DIR, "ora.log") | |
| class AuditLog: | |
| def __init__(self): | |
| self.logger = logging.getLogger("ORA_Audit") | |
| self.logger.setLevel(logging.INFO) | |
| # Prevent adding multiple handlers if re-initialized | |
| if not self.logger.handlers: | |
| # Rotate log after 10MB, keep 5 backups | |
| handler = RotatingFileHandler(LOG_FILE, maxBytes=10*1024*1024, backupCount=5) | |
| formatter = logging.Formatter('%(message)s') | |
| handler.setFormatter(formatter) | |
| self.logger.addHandler(handler) | |
| async def log_interaction(self, user_id: str, intent: str, guardrails_triggered: list[str]): | |
| """ | |
| Logs a user interaction and any triggered guardrails. | |
| """ | |
| entry = { | |
| "timestamp": datetime.now().isoformat(), | |
| "type": "INTERACTION", | |
| "user_id": user_id, | |
| "intent": intent, | |
| "guardrails": guardrails_triggered | |
| } | |
| self.logger.info(json.dumps(entry)) | |
| async def log_violation(self, user_id: str, violation_type: str, content: str): | |
| """ | |
| Logs a specific safety violation. | |
| """ | |
| entry = { | |
| "timestamp": datetime.now().isoformat(), | |
| "type": "VIOLATION", | |
| "user_id": user_id, | |
| "violation_type": violation_type, | |
| "content_snippet": content[:200] # Store first 200 chars for context | |
| } | |
| self.logger.info(json.dumps(entry)) | |
| audit_service = AuditLog() | |