Spaces:
Sleeping
Sleeping
File size: 2,615 Bytes
7c98bdb | 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 | import datetime
import json
import threading
from typing import Dict, Any, List, Optional
from .analytics_utils import fetch_metrics, export_metrics, log_error, save_report
class AnalyticsAgent:
def __init__(self, config_path: str = "configs/analytics_config.json"):
with open(config_path, "r") as f:
self.config = json.load(f)
self.metrics_cache = {}
self.error_log = []
self.lock = threading.Lock()
def get_live_metrics(self, spec: Optional[Dict[str, Any]] = None) -> Dict[str, Any]:
"""Aggregate real-time metrics on demand (API/button/query)."""
try:
metrics = fetch_metrics(spec or self.config["live_metrics_spec"])
with self.lock:
self.metrics_cache["live"] = metrics
return metrics
except Exception as e:
log_error(self.error_log, e)
return {"error": str(e)}
def run_end_of_day_report(self) -> Dict[str, Any]:
"""Aggregate metrics, save, notify/report at end of day (scheduled)."""
try:
report = fetch_metrics(self.config["end_of_day_spec"])
save_report(report, period="day")
with self.lock:
self.metrics_cache["end_of_day"] = report
return report
except Exception as e:
log_error(self.error_log, e)
return {"error": str(e)}
def batch_export(self, period_days: int) -> Dict[str, Any]:
"""Export summary for past period (7/30/90 days)."""
try:
exported = export_metrics(period_days)
with self.lock:
self.metrics_cache[f"export_{period_days}"] = exported
return exported
except Exception as e:
log_error(self.error_log, e)
return {"error": str(e)}
def get_errors(self) -> List[str]:
"""Return error logs or missing data events."""
return self.error_log
def expose_endpoint(self, endpoint_type: str = "live") -> Dict[str, Any]:
"""Hook for UI/API to query/display current metrics."""
return self.metrics_cache.get(endpoint_type, {})
def scheduled_tasks(self):
"""Set up periodic triggers for end-of-day rollup/reports."""
# Implementation for actual scheduler integration needed.
pass
def clear_error_log(self):
self.error_log.clear()
# Usage Example:
# agent = AnalyticsAgent()
# live = agent.get_live_metrics()
# eod = agent.run_end_of_day_report()
# sevenday_export = agent.batch_export(7)
# agent.expose_endpoint()
# errors = agent.get_errors()
|