Spaces:
Sleeping
Sleeping
File size: 3,992 Bytes
7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 7c98bdb 2e93e81 | 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 | 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() |