Spaces:
Sleeping
Sleeping
File size: 1,726 Bytes
5e0532d |
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 |
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()
|