""" Test script to trigger burnout notifications This script creates test stamps to simulate excessive work patterns """ import os import sys import django from datetime import datetime, date, timedelta # Setup Django sys.path.append(os.path.dirname(os.path.abspath(__file__))) os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'worktimeapp.settings') django.setup() from stamps.models import Stamp from functions.models import Function from company.models import WorkingTimePolicy from userSettings.models import UserSettings from ml_service.models import BurnoutNotification from ml_service.tasks import analyze_burnout_daily, send_burnout_notification from django.utils import timezone # Configuration TEST_USER_ID = "575f8050-e027-4db9-9ff1-31e0b1b7dfad" # Your user ID COMPANY_ID = "your-company-id" # You may need to update this def create_test_stamps(): """Create test stamps to simulate excessive work""" print("Creating test stamps for burnout detection...") # Ensure user has burnout notifications enabled user_settings, created = UserSettings.objects.get_or_create( user_id=TEST_USER_ID, defaults={ 'lang': 'en', 'user_timezone': 'UTC', 'burnout_notifications_enabled': True, 'burnout_sensitivity': 'medium', } ) if not user_settings.burnout_notifications_enabled: user_settings.burnout_notifications_enabled = True user_settings.burnout_sensitivity = 'medium' user_settings.save() print(f"User settings configured: notifications={user_settings.burnout_notifications_enabled}, sensitivity={user_settings.burnout_sensitivity}") # Get or create a function for clock in/out clock_in_func, _ = Function.objects.get_or_create( function_ref_id='clock_in_001', defaults={ 'name': 'Clock In', 'out': False, 'break_flag': False, 'pre_function': False, 'question_type': 'clock_in', } ) clock_out_func, _ = Function.objects.get_or_create( function_ref_id='clock_out_001', defaults={ 'name': 'Clock Out', 'out': True, 'break_flag': False, 'pre_function': False, 'question_type': 'clock_out', } ) print(f"✓ Functions ready: {clock_in_func.name}, {clock_out_func.name}") # Create stamps for the last 7 days with excessive hours today = date.today() for day_offset in range(7): stamp_date = today - timedelta(days=day_offset) # Skip weekends if desired if stamp_date.weekday() >= 5: # Saturday or Sunday continue # Create excessive work hours (9-10 hours per day) clock_in_time = datetime.combine(stamp_date, datetime.strptime('07:00:00', '%H:%M:%S').time()) clock_out_time = datetime.combine(stamp_date, datetime.strptime('17:30:00', '%H:%M:%S').time()) # Get user from user.models import User try: user = User.objects.get(id=TEST_USER_ID) except User.DoesNotExist: print(f"❌ User {TEST_USER_ID} not found. Please update TEST_USER_ID in the script.") return # Clock In Stamp.objects.update_or_create( user=user, date=stamp_date, time=clock_in_time.time(), stamp_function=clock_in_func.id, defaults={ 'description': f'Test burnout detection - Day {day_offset + 1}', 'start_date': clock_in_time, } ) # Clock Out Stamp.objects.update_or_create( user=user, date=stamp_date, time=clock_out_time.time(), stamp_function=clock_out_func.id, defaults={ 'description': f'Test burnout detection - Day {day_offset + 1}', 'start_date': clock_out_time, } ) print(f"✓ Created stamps for {stamp_date}: 07:00 - 17:30 ({10.5} hours)") print(f"\n✓ Created test stamps for user {TEST_USER_ID}") print("✓ Average daily hours: 10.5 hours") print("✓ Consecutive days: 5") print("✓ This should trigger high burnout risk!") def trigger_burnout_analysis(): """Trigger burnout analysis for the test user""" print("\n" + "="*50) print("TRIGGERING BURNOUT ANALYSIS") print("="*50) try: # Import the detector from ml_service.burnout_detector import BurnoutDetector # Create detector instance detector = BurnoutDetector(TEST_USER_ID) # Analyze user print(f"\nAnalyzing user {TEST_USER_ID}...") analysis = detector.analyze_user(days=30) if analysis: print("\n" + "="*50) print("ANALYSIS RESULTS") print("="*50) print(f"Risk Score: {analysis['risk_score']}/100") print(f"Severity: {analysis['severity'].upper()}") print(f"\nRisk Factors:") for factor, score in analysis['risk_factors'].items(): print(f" - {factor}: {score} points") print(f"\nRecommendations:") for i, rec in enumerate(analysis['recommendations'], 1): print(f" {i}. {rec['balance_type']}: {rec['message']}") # Store analysis from ml_service.models import BurnoutAnalysis BurnoutAnalysis.objects.create( user_id=TEST_USER_ID, date=timezone.now().date(), risk_score=analysis['risk_score'], severity=analysis['severity'], risk_factors=analysis['risk_factors'], recommendations=analysis['recommendations'], ) print(f"\n✓ Analysis stored in database") # Send notification if analysis['severity'] in ['medium', 'high', 'critical']: print(f"\n" + "="*50) print("SENDING NOTIFICATION") print("="*50) send_burnout_notification.delay(TEST_USER_ID, analysis) print(f"✓ Notification sent to user!") # Also send immediately for testing print(f"\nSending immediate test notification...") from ml_service.push_service import PushNotificationService push_service = PushNotificationService() if analysis['severity'] == 'critical': title = "⚠️ High Burnout Risk Detected" body = f"Your work patterns show {analysis['risk_score']}/100 burnout risk. Take action immediately." elif analysis['severity'] == 'high': title = "⚠️ Consider Taking a Break" body = f"Your recent work patterns suggest high burnout risk. Consider using available balances." else: title = "ℹ️ Burnout Alert" body = f"Burnout risk detected: {analysis['risk_score']}/100. Monitor your work patterns." # Get recommendations if analysis['recommendations']: body += f"\n\n💡 Suggestion: {analysis['recommendations'][0]['action']}" # Create notification in database from ml_service.models import BurnoutNotification notification = BurnoutNotification.objects.create( user_id=TEST_USER_ID, notification_type='burnout_alert', title=title, message=body, risk_score=analysis['risk_score'], recommendation=analysis['recommendations'][0] if analysis['recommendations'] else {}, ) print(f"✓ Created notification record: {notification.id}") print(f"✓ Title: {title}") print(f"✓ Message: {body}") return analysis else: print("No analysis returned (user settings may be disabled)") return None except Exception as e: print(f"\n❌ Error during analysis: {e}") import traceback traceback.print_exc() return None def check_notifications(): """Check if notifications were created""" print("\n" + "="*50) print("CHECKING NOTIFICATIONS") print("="*50) notifications = BurnoutNotification.objects.filter(user_id=TEST_USER_ID).order_by('-sent_at') if notifications.exists(): print(f"\n✓ Found {notifications.count()} notifications") for notif in notifications[:3]: # Show latest 3 print(f"\nNotification {notif.id}:") print(f" Title: {notif.title}") print(f" Risk Score: {notif.risk_score}/100") print(f" Read: {'Yes' if notif.is_read else 'No'}") print(f" Sent: {notif.sent_at}") else: print("\n❌ No notifications found") def main(): """Main test function""" print("\n" + "="*50) print("BURNOUT NOTIFICATION TEST") print("="*50) print(f"\nUser ID: {TEST_USER_ID}") print(f"Date: {date.today()}") # Step 1: Create test stamps create_test_stamps() # Step 2: Wait a moment for balance calculation print("\n⏳ Waiting for balance calculation...") import time time.sleep(2) # Step 3: Trigger analysis analysis = trigger_burnout_analysis() # Step 4: Check notifications check_notifications() print("\n" + "="*50) print("TEST COMPLETE") print("="*50) print("\n📱 Check your app for the notification!") print("📊 You can also check the database for BurnoutNotification records") print("\nTo run again, execute: python test_burnout_notification.py") if __name__ == "__main__": main()