Spaces:
Paused
Paused
File size: 9,304 Bytes
20adca1 1f8ac0c 20adca1 |
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 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 |
from datetime import datetime
from typing import Dict, Any, List, Optional
from ..services.notifications import create_and_broadcast_notification
from ..db.models import ActivityType
from ..utils.logger import logger
class PerformanceNotificationService:
# Performance thresholds for notifications
THRESHOLDS = {
"high_efficiency": 9.0, # Notify for efficiency scores above 9
"low_efficiency": 5.0, # Warn for efficiency scores below 5
"sales_milestone": 10000, # Notify for sales milestones (every 10k)
"high_void_rate": 0.1, # Warn when void rate exceeds 10%
"customer_service": 20, # Notify for high customer service interactions
}
@staticmethod
async def check_and_notify_performance(
user_id: int,
branch_id: int,
metrics: Dict[str, Any]
):
"""Check performance metrics and send notifications if thresholds are met"""
try:
notifications = []
# Check efficiency score
if metrics["efficiency_score"] >= PerformanceNotificationService.THRESHOLDS["high_efficiency"]:
notifications.append({
"title": "Outstanding Performance!",
"message": "Your efficiency score is exceptional. Keep up the great work!",
"type": "performance_achievement",
"data": {
"metric": "efficiency_score",
"value": metrics["efficiency_score"]
}
})
elif metrics["efficiency_score"] <= PerformanceNotificationService.THRESHOLDS["low_efficiency"]:
notifications.append({
"title": "Performance Alert",
"message": "Your efficiency score needs improvement. Contact your supervisor for support.",
"type": "performance_alert",
"data": {
"metric": "efficiency_score",
"value": metrics["efficiency_score"]
}
})
# Check sales milestones
milestone = PerformanceNotificationService.THRESHOLDS["sales_milestone"]
if metrics["total_sales"] > 0 and metrics["total_sales"] % milestone < 100:
notifications.append({
"title": "Sales Milestone Achieved!",
"message": f"Congratulations! You've reached {int(metrics['total_sales'] / 1000)}k in sales.",
"type": "sales_milestone",
"data": {
"metric": "total_sales",
"value": metrics["total_sales"]
}
})
# Check void rate
if metrics["transaction_count"] > 0:
void_rate = metrics["void_count"] / metrics["transaction_count"]
if void_rate > PerformanceNotificationService.THRESHOLDS["high_void_rate"]:
notifications.append({
"title": "High Void Rate Alert",
"message": "Your void transaction rate is above average. Please review procedures.",
"type": "void_rate_alert",
"data": {
"metric": "void_rate",
"value": void_rate
}
})
# Check customer service interactions
if metrics["customer_interaction_count"] >= PerformanceNotificationService.THRESHOLDS["customer_service"]:
notifications.append({
"title": "Customer Service Achievement",
"message": "Great job handling customer interactions today!",
"type": "customer_service_achievement",
"data": {
"metric": "customer_interactions",
"value": metrics["customer_interaction_count"]
}
})
# Send all notifications
for notif in notifications:
await create_and_broadcast_notification(
user_id=str(user_id),
title=notif["title"],
message=notif["message"],
notification_type=notif["type"],
data=notif["data"]
)
# If it's an alert, also notify supervisor
if "alert" in notif["type"]:
await create_and_broadcast_notification(
user_id="supervisor",
title=f"Staff Alert: {notif['title']}",
message=f"Staff member {user_id} in branch {branch_id}: {notif['message']}",
notification_type=f"supervisor_{notif['type']}",
data={
**notif["data"],
"staff_id": user_id,
"branch_id": branch_id
}
)
except Exception as e:
logger.error(f"Error in performance notifications: {str(e)}")
@staticmethod
async def notify_performance_insights(
branch_id: int,
insights: List[Dict[str, Any]]
):
"""Send notifications for performance insights"""
try:
for insight in insights:
if insight["type"] in ["warning", "positive"]:
await create_and_broadcast_notification(
user_id="supervisor",
title=f"Branch Performance Insight",
message=insight["message"],
notification_type=f"performance_insight_{insight['type']}",
data={
"branch_id": branch_id,
"category": insight["category"],
"insight_type": insight["type"]
}
)
except Exception as e:
logger.error(f"Error in insight notifications: {str(e)}")
@staticmethod
async def notify_realtime_alerts(
branch_id: int,
current_metrics: Dict[str, Any],
previous_metrics: Optional[Dict[str, Any]] = None
):
"""Send notifications for significant real-time changes"""
try:
if not previous_metrics:
return
alerts = []
# Check for significant drops in performance
if (previous_metrics["efficiency_score"] - current_metrics["efficiency_score"]) > 2:
alerts.append({
"title": "Sudden Performance Drop",
"message": "Notable decrease in staff performance detected.",
"type": "realtime_performance_alert"
})
# Check for unusual void patterns
current_void_rate = (
current_metrics["void_count"] / current_metrics["transaction_count"]
if current_metrics["transaction_count"] > 0 else 0
)
previous_void_rate = (
previous_metrics["void_count"] / previous_metrics["transaction_count"]
if previous_metrics["transaction_count"] > 0 else 0
)
if current_void_rate > previous_void_rate * 2:
alerts.append({
"title": "Unusual Void Pattern",
"message": "Significant increase in void transactions detected.",
"type": "realtime_void_alert"
})
# Send alerts
for alert in alerts:
await create_and_broadcast_notification(
user_id="supervisor",
title=alert["title"],
message=alert["message"],
notification_type=alert["type"],
data={
"branch_id": branch_id,
"current_metrics": current_metrics,
"previous_metrics": previous_metrics
}
)
except Exception as e:
logger.error(f"Error in realtime alert notifications: {str(e)}")
@staticmethod
async def process_activity_notifications(
user_id: int,
branch_id: int,
activity: Any,
prev_metrics: Optional[Dict[str, Any]],
new_metrics: Dict[str, Any]
):
"""Process notifications for a new staff activity"""
await PerformanceNotificationService.check_and_notify_performance(
user_id=user_id,
branch_id=branch_id,
metrics=new_metrics
)
if prev_metrics:
await PerformanceNotificationService.notify_realtime_alerts(
branch_id=branch_id,
current_metrics=new_metrics,
previous_metrics=prev_metrics
)
performance_notifications = PerformanceNotificationService() |