""" CODH Core Celery Tasks Background task processing """ import logging from datetime import datetime, timedelta from celery import shared_task from django.conf import settings from django.utils import timezone from .models import BackgroundTask, SystemHealth, SystemConfiguration logger = logging.getLogger('codh') # Task mapping for dynamic retries TASK_MAPPING = {} def register_task(task_type, task_func): """Register a task for dynamic retries""" TASK_MAPPING[task_type] = task_func @shared_task(bind=True, max_retries=3) def test_celery_task(self): """Test Celery is working""" logger.info("Celery test task executed successfully") return {'status': 'success', 'message': 'Celery is working'} @shared_task(bind=True, max_retries=3) def run_health_check(self): """Run system health check""" try: import redis import psutil from django.db import connection health = SystemHealth() # Check database try: with connection.cursor() as cursor: cursor.execute("SELECT 1") health.database_status = 'ok' except Exception as e: health.database_status = 'error' logger.error(f"Database health check failed: {e}") # Check Redis try: r = redis.from_url(settings.CACHES['default']['LOCATION']) r.ping() health.redis_status = 'ok' except Exception as e: health.redis_status = 'error' logger.error(f"Redis health check failed: {e}") # Check Celery try: from celery import current_app inspector = current_app.control.inspect() ping = inspector.ping() health.celery_status = 'ok' if ping else 'warning' except Exception as e: health.celery_status = 'error' logger.error(f"Celery health check failed: {e}") # System metrics try: disk = psutil.disk_usage('/') health.disk_usage_percent = disk.percent health.disk_status = 'ok' if disk.percent < 80 else 'warning' if disk.percent < 90 else 'error' mem = psutil.virtual_memory() health.memory_usage_percent = mem.percent health.memory_status = 'ok' if mem.percent < 80 else 'warning' if mem.percent < 90 else 'error' health.cpu_usage_percent = psutil.cpu_percent(interval=1) except Exception as e: logger.error(f"System metrics check failed: {e}") # Queue metrics try: health.pending_tasks = BackgroundTask.objects.filter(status='pending').count() health.failed_tasks = BackgroundTask.objects.filter(status='failed').count() except: pass health.save() return {'status': 'success', 'health_id': str(health.id)} except Exception as e: logger.error(f"Health check task failed: {e}") raise self.retry(exc=e, countdown=60) @shared_task(bind=True, max_retries=3) def cleanup_old_logs(self): """Clean up old audit logs and tasks""" try: from .models import AuditLog # Delete logs older than 90 days cutoff_date = timezone.now() - timedelta(days=90) deleted_logs = AuditLog.objects.filter(created_at__lt=cutoff_date).delete() # Delete completed tasks older than 30 days task_cutoff = timezone.now() - timedelta(days=30) deleted_tasks = BackgroundTask.objects.filter( status__in=['completed', 'cancelled'], completed_at__lt=task_cutoff ).delete() logger.info(f"Cleanup completed: {deleted_logs[0]} logs, {deleted_tasks[0]} tasks deleted") return { 'status': 'success', 'logs_deleted': deleted_logs[0], 'tasks_deleted': deleted_tasks[0] } except Exception as e: logger.error(f"Cleanup task failed: {e}") raise self.retry(exc=e, countdown=300) @shared_task(bind=True, max_retries=5) def send_notification(self, user_id, notification_type, data): """Send notification to user""" try: from apps.notifications.models import Notification from apps.notifications.utils import send_email, send_push, send_sms notification = Notification.objects.create( user_id=user_id, type=notification_type, data=data ) # Send via appropriate channels config = SystemConfiguration.get_config() if config.enable_email_notifications: send_email.delay(notification.id) if config.enable_push_notifications: send_push.delay(notification.id) if config.enable_sms_notifications and notification_type == 'sms': send_sms.delay(notification.id) return {'status': 'success', 'notification_id': str(notification.id)} except Exception as e: logger.error(f"Notification task failed: {e}") raise self.retry(exc=e, countdown=60) @shared_task(bind=True, max_retries=3) def generate_daily_report(self): """Generate daily system report""" try: from apps.analytics.utils import generate_report yesterday = timezone.now().date() - timedelta(days=1) report = generate_report(start_date=yesterday, end_date=yesterday) # Send to admin config = SystemConfiguration.get_config() if config.admin_notification_email: send_notification.delay( None, 'daily_report', { 'email': config.admin_notification_email, 'report': report } ) return {'status': 'success', 'report_date': str(yesterday)} except Exception as e: logger.error(f"Daily report task failed: {e}") raise self.retry(exc=e, countdown=3600) @shared_task(bind=True, max_retries=3) def process_expired_subscriptions(self): """Process expired subscriptions""" try: from apps.subscriptions.models import Subscription expired = Subscription.objects.filter( status='active', end_date__lt=timezone.now() ) suspended_count = 0 for subscription in expired: subscription.status = 'expired' subscription.save() # Suspend hosting if configured config = SystemConfiguration.get_config() if config.auto_suspend_on_expiry: from apps.hosting.tasks import suspend_hosting_account suspend_hosting_account.delay(subscription.hosting_account_id) # Notify user send_notification.delay( subscription.user_id, 'subscription_expired', {'subscription_id': str(subscription.id)} ) suspended_count += 1 logger.info(f"Processed {suspended_count} expired subscriptions") return {'status': 'success', 'suspended_count': suspended_count} except Exception as e: logger.error(f"Subscription expiry task failed: {e}") raise self.retry(exc=e, countdown=3600) @shared_task(bind=True, max_retries=3) def backup_database(self): """Create database backup""" try: import subprocess from django.conf import settings db_settings = settings.DATABASES['default'] timestamp = datetime.now().strftime('%Y%m%d_%H%M%S') backup_file = f"/backups/codh_backup_{timestamp}.sql" # Create backup command cmd = [ 'mysqldump', '-h', db_settings['HOST'], '-u', db_settings['USER'], '-p' + db_settings['PASSWORD'], db_settings['NAME'] ] with open(backup_file, 'w') as f: subprocess.run(cmd, stdout=f, check=True) logger.info(f"Database backup created: {backup_file}") return {'status': 'success', 'backup_file': backup_file} except Exception as e: logger.error(f"Database backup failed: {e}") raise self.retry(exc=e, countdown=3600)