File size: 3,681 Bytes
3efe7a4 |
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 107 108 109 110 |
import logging
import os, json
from logging.handlers import TimedRotatingFileHandler
from datetime import datetime
class GetLogger:
def __init__(self, logging_level="INFO", log_to_console=True, log_dir="logs"):
"""
Advanced Logger
- Logs to both file (rotating) and console
- Default rotation: daily, keep last 7 logs
- Safe filename (no ':' in timestamp)
"""
self.logger = logging.getLogger(__name__)
self.logger.setLevel(logging_level.upper())
# Avoid duplicate handlers
if self.logger.hasHandlers():
self.logger.handlers.clear()
# Ensure log directory exists
os.makedirs(log_dir, exist_ok=True)
# File handler (rotates daily, keep 7 backups)
file_name = datetime.now().strftime("%Y-%m-%d_%H-%M-%S") + ".log"
log_path = os.path.join(log_dir, file_name)
file_handler = TimedRotatingFileHandler(
filename=log_path, when="D", interval=1, backupCount=3, encoding="utf-8"
)
formatter = logging.Formatter(
"%(asctime)s [%(levelname)s] [%(filename)s:%(lineno)d] %(funcName)s() - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S"
)
file_handler.setFormatter(formatter)
self.logger.addHandler(file_handler)
# Console handler (optional)
if log_to_console:
console_handler = logging.StreamHandler()
console_handler.setFormatter(formatter)
self.logger.addHandler(console_handler)
def get_logger(self):
return self.logger
def delete_logger(self):
"""Remove all handlers and delete logger."""
handlers = self.logger.handlers[:]
for handler in handlers:
self.logger.removeHandler(handler)
handler.close()
del self.logger
class MetricsLogger:
"""
Collects evaluation metrics and saves aggregated statistics.
"""
def __init__(self, save_path="logs/metrics_summary.json", logger=None):
self.save_path = save_path
self.metrics = [] # store per-query metrics
self.logger = logger or logging.getLogger(__name__)
def log_query_metrics(self, query, result_dict):
"""
Log metrics for a single query.
Example: result_dict = {"latency_sec": 0.5, "rougeL": 0.7, ...}
"""
record = {"query": query}
record.update(result_dict)
self.metrics.append(record)
self.logger.info(f"📊 Metrics logged for query: {query[:50]}...")
def summarize(self):
"""Aggregate metrics (mean values)."""
if not self.metrics:
return {}
summary = {}
keys = [k for k in self.metrics[0].keys() if k != "query"]
for key in keys:
values = [m[key] for m in self.metrics if key in m and isinstance(m[key], (int, float))]
if values:
summary[f"avg_{key}"] = float(sum(values) / len(values))
return summary
def save(self):
"""Save all metrics + summary to JSON."""
os.makedirs(os.path.dirname(self.save_path), exist_ok=True)
data = {
"per_query": self.metrics,
"summary": self.summarize()
}
with open(self.save_path, "w", encoding="utf-8") as f:
json.dump(data, f, indent=2)
self.logger.info(f"✅ Metrics saved to {self.save_path}")
return data
# Example
if __name__ == "__main__":
obj = GetLogger()
logger = obj.get_logger()
logger.info("✅ Logger initialized successfully")
logger.warning("⚠️ This is a warning")
logger.error("❌ This is an error")
|