Spaces:
Sleeping
Sleeping
| 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 | |
| from lead_scoring_agent import score_lead | |
| class AnalyticsAgent: | |
| """Analytics Agent with integrated lead scoring metrics for visibility in analytics and UI.""" | |
| 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), including lead scoring data.""" | |
| try: | |
| metrics = fetch_metrics(spec or self.config["live_metrics_spec"]) | |
| # Include lead scoring metrics in analytics | |
| metrics["lead_scoring"] = self._get_lead_scoring_metrics() | |
| 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) with scoring breakdowns.""" | |
| try: | |
| report = fetch_metrics(self.config["end_of_day_spec"]) | |
| # Add lead scoring summary to daily report | |
| report["lead_scoring_summary"] = self._get_lead_scoring_metrics() | |
| 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) including score distribution.""" | |
| try: | |
| exported = export_metrics(period_days) | |
| # Include lead scoring trends in export | |
| exported["lead_scoring_trends"] = self._get_lead_scoring_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_lead_scoring_metrics(self, period_days: Optional[int] = None) -> Dict[str, Any]: | |
| """Retrieve lead scoring metrics for analytics visibility.""" | |
| # Query database/cache for lead scoring data | |
| # This would integrate with your data layer to fetch scored leads | |
| return { | |
| "total_scored_leads": 0, # Populate from database | |
| "score_distribution": { | |
| "high": 0, | |
| "medium": 0, | |
| "low": 0 | |
| }, | |
| "average_score": 0.0, | |
| "scoring_enabled": True | |
| } | |
| 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 with lead scoring data.""" | |
| 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() # Now includes lead_scoring metrics | |
| # eod = agent.run_end_of_day_report() # Now includes lead_scoring_summary | |
| # sevenday_export = agent.batch_export(7) # Now includes lead_scoring_trends | |
| # agent.expose_endpoint() | |
| # errors = agent.get_errors() |