import os from celery import Celery from celery.schedules import crontab from datetime import datetime, timedelta import logging from services.database_service import DatabaseService from services.notification_service import NotificationService from utils.image_processor import ImageProcessor # Initialize Celery redis_url = os.getenv("REDIS_URL", "redis://localhost:6379") celery_app = Celery("glowai", broker=redis_url, backend=redis_url) # Configure Celery celery_app.conf.update( task_serializer="json", accept_content=["json"], result_serializer="json", timezone="UTC", enable_utc=True, task_track_started=True, task_time_limit=30 * 60, # 30 minutes task_soft_time_limit=25 * 60, # 25 minutes worker_max_tasks_per_child=200, worker_prefetch_multiplier=4, ) # Schedule periodic tasks celery_app.conf.beat_schedule = { "delete-expired-images": { "task": "worker.delete_expired_images", "schedule": crontab(hour="*/6"), # Every 6 hours }, "send-streak-reminders": { "task": "worker.send_streak_reminders", "schedule": crontab(hour="20", minute="0"), # 8 PM daily }, "update-leaderboard": { "task": "worker.update_leaderboard", "schedule": crontab(hour="0", minute="0"), # Midnight daily }, "cleanup-privacy-logs": { "task": "worker.cleanup_privacy_logs", "schedule": crontab(day_of_month="1"), # First day of month }, "backup-database": { "task": "worker.backup_database", "schedule": crontab(hour="2", minute="0"), # 2 AM daily }, "sync-product-catalog": { "task": "worker.sync_product_catalog", "schedule": crontab(hour="3", minute="0"), # 3 AM daily }, } logger = logging.getLogger(__name__) db_service = DatabaseService() notification_service = NotificationService() image_processor = ImageProcessor() @celery_app.task def delete_expired_images(): """ Delete images older than retention period (GDPR compliance) """ try: logger.info("Starting expired image deletion task") cutoff_time = datetime.utcnow() - timedelta(hours=24) # Get expired images from database expired_images = db_service.get_expired_images(cutoff_time) deleted_count = 0 for image in expired_images: try: # Delete from storage image_processor.delete_image(image["url"]) # Update database db_service.mark_image_deleted(image["id"]) deleted_count += 1 except Exception as e: logger.error(f"Failed to delete image {image['id']}: {e}") logger.info(f"Deleted {deleted_count} expired images") return {"deleted_count": deleted_count} except Exception as e: logger.error(f"Image deletion task failed: {e}") raise @celery_app.task def send_streak_reminders(): """ Send push notifications to users who haven't checked in """ try: logger.info("Starting streak reminder task") # Get users who haven't checked in today users_to_remind = db_service.get_users_for_reminder() sent_count = 0 for user in users_to_remind: try: streak = db_service.get_streak(user["id"]) notification_service.send_push_notification( user_id=user["id"], title="Don't Break Your Streak! 🔥", body=f"You're on a {streak['current_streak']} day streak. Check in now!", data={"type": "streak_reminder"} ) sent_count += 1 except Exception as e: logger.error(f"Failed to send reminder to user {user['id']}: {e}") logger.info(f"Sent {sent_count} streak reminders") return {"sent_count": sent_count} except Exception as e: logger.error(f"Streak reminder task failed: {e}") raise @celery_app.task def update_leaderboard(): """ Update leaderboard rankings """ try: logger.info("Starting leaderboard update task") # Calculate weekly leaderboard weekly = db_service.calculate_leaderboard("weekly") # Calculate monthly leaderboard monthly = db_service.calculate_leaderboard("monthly") # Calculate all-time leaderboard all_time = db_service.calculate_leaderboard("all_time") # Store in cache db_service.update_leaderboard_cache({ "weekly": weekly, "monthly": monthly, "all_time": all_time }) logger.info("Leaderboard updated successfully") return { "weekly_count": len(weekly), "monthly_count": len(monthly), "all_time_count": len(all_time) } except Exception as e: logger.error(f"Leaderboard update failed: {e}") raise @celery_app.task def cleanup_privacy_logs(): """ Clean up old privacy logs (GDPR compliance) """ try: logger.info("Starting privacy logs cleanup") # Keep logs for 90 days cutoff_date = datetime.utcnow() - timedelta(days=90) deleted_count = db_service.delete_old_privacy_logs(cutoff_date) logger.info(f"Deleted {deleted_count} old privacy logs") return {"deleted_count": deleted_count} except Exception as e: logger.error(f"Privacy logs cleanup failed: {e}") raise @celery_app.task def backup_database(): """ Create database backup """ try: logger.info("Starting database backup") backup_file = f"backup_{datetime.utcnow().strftime('%Y%m%d_%H%M%S')}.sql" db_service.create_backup(backup_file) # Upload to cloud storage cloudinary_service.upload_backup(backup_file) logger.info(f"Database backup created: {backup_file}") return {"backup_file": backup_file} except Exception as e: logger.error(f"Database backup failed: {e}") raise @celery_app.task def sync_product_catalog(): """ Sync product catalog with suppliers """ try: logger.info("Starting product catalog sync") # Fetch latest product data from suppliers updated_products = db_service.sync_products_with_suppliers() # Update prices and availability db_service.update_product_prices(updated_products) logger.info(f"Synced {len(updated_products)} products") return {"synced_count": len(updated_products)} except Exception as e: logger.error(f"Product catalog sync failed: {e}") raise @celery_app.task def process_analysis_batch(image_batch): """ Process batch of images for analysis (for bulk operations) """ try: logger.info(f"Processing batch of {len(image_batch)} images") from models.skin_analysis_model import SkinAnalysisModel model = SkinAnalysisModel() results = [] for image_data in image_batch: try: result = model.analyze_skin(image_data["bytes"]) results.append({ "user_id": image_data["user_id"], "result": result }) except Exception as e: logger.error(f"Failed to process image: {e}") results.append({ "user_id": image_data["user_id"], "error": str(e) }) return {"processed": len(results), "results": results} except Exception as e: logger.error(f"Batch processing failed: {e}") raise @celery_app.task def generate_recommendations_for_user(user_id): """ Generate product recommendations for user """ try: logger.info(f"Generating recommendations for user {user_id}") # Get user's latest analysis analysis = db_service.get_latest_analysis(user_id) if not analysis: return {"error": "No analysis found"} # Generate recommendations from models.cosmetics_matcher import CosmeticMatcher matcher = CosmeticMatcher() recommendations = matcher.find_matches( skin_tone=analysis["skin_tone"], conditions=analysis["conditions"] ) # Store recommendations db_service.save_recommendations(user_id, recommendations) return {"user_id": user_id, "recommendations_count": len(recommendations)} except Exception as e: logger.error(f"Recommendation generation failed: {e}") raise @celery_app.task def send_onboarding_emails(): """ Send onboarding emails to new users """ try: logger.info("Sending onboarding emails") # Get users who registered in last 24h and haven't received onboarding new_users = db_service.get_new_users_for_onboarding() sent_count = 0 for user in new_users: try: notification_service.send_email( to=user["email"], subject="Welcome to Glow AI! ✨", template="onboarding", context={ "name": user["name"], "get_started_link": "https://glowai.app/get-started" } ) db_service.mark_onboarding_email_sent(user["id"]) sent_count += 1 except Exception as e: logger.error(f"Failed to send onboarding email to {user['email']}: {e}") logger.info(f"Sent {sent_count} onboarding emails") return {"sent_count": sent_count} except Exception as e: logger.error(f"Onboarding email task failed: {e}") raise