Create analytics_agent.py

analytics_agent.py implementing live metrics on demand, end-of-day reports (scheduled), batch export (7/30/90 days), API/UI hooks, error logging/reporting, and config-driven specs. See functions: get_live_metrics, run_end_of_day_report, batch_export, expose_endpoint, get_errors. Needs analytics_utils.py. Update README for usage/testing.

analytics_agent.py

@@ -0,0 +1,72 @@
+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()