"""APS Scheduler service for the Lin application.""" import logging from datetime import datetime, timedelta from apscheduler.schedulers.background import BackgroundScheduler from apscheduler.triggers.cron import CronTrigger from apscheduler.jobstores.memory import MemoryJobStore from apscheduler.executors.pool import ThreadPoolExecutor from backend.services.content_service import ContentService from backend.services.linkedin_service import LinkedInService from backend.utils.database import init_supabase from backend.utils.image_utils import ensure_bytes_format from backend.config import Config from backend.utils.timezone_utils import ( parse_timezone_schedule, get_server_timezone, convert_time_to_timezone, validate_timezone ) # Configure logging logger = logging.getLogger(__name__) class APSchedulerService: """Service for managing APScheduler tasks.""" def __init__(self, app=None): self.app = app self.scheduler = None self.supabase_client = None # Initialize scheduler if app is provided if app is not None: self.init_app(app) def init_app(self, app): """Initialize the scheduler with the Flask app.""" try: self.app = app logger.info("🚀 APScheduler starting...") # Initialize Supabase client self.supabase_client = init_supabase( app.config['SUPABASE_URL'], app.config['SUPABASE_KEY'] ) # Configure job stores and executors jobstores = { 'default': MemoryJobStore() } executors = { 'default': ThreadPoolExecutor(20), } job_defaults = { 'coalesce': False, 'max_instances': 3 } # Create scheduler self.scheduler = BackgroundScheduler( jobstores=jobstores, executors=executors, job_defaults=job_defaults, timezone='UTC' ) # Add the scheduler to the app app.scheduler = self # Start the scheduler self.scheduler.start() logger.info("✅ APScheduler started successfully") # Add the periodic job to load schedules from database self.scheduler.add_job( func=self.load_schedules, trigger=CronTrigger(minute='*/5'), # Every 5 minutes id='load_schedules', name='Load schedules from database', replace_existing=True ) # Load schedules immediately when the app starts self.load_schedules() except Exception as e: logger.error(f"❌ APScheduler initialization failed: {str(e)}") import traceback logger.error(traceback.format_exc()) def load_schedules(self): """Load schedules from the database and create jobs.""" try: # Run within application context with self.app.app_context(): if not self.supabase_client: logger.error("❌ Supabase client not initialized") return # Fetch all schedules from Supabase response = ( self.supabase_client .table("Scheduling") .select("*, Social_network(id_utilisateur, token, sub)") .execute() ) schedules = response.data if response.data else [] logger.info(f"📋 Found {len(schedules)} schedules in database") # Remove existing scheduled jobs (except the loader job) jobs_to_remove = [] for job in self.scheduler.get_jobs(): if job.id != 'load_schedules': jobs_to_remove.append(job.id) for job_id in jobs_to_remove: try: self.scheduler.remove_job(job_id) except Exception as e: logger.warning(f"Failed to remove job {job_id}: {str(e)}") # Create jobs for each schedule for schedule in schedules: try: schedule_id = schedule.get('id') schedule_time = schedule.get('schedule_time') adjusted_time = schedule.get('adjusted_time') if not schedule_time or not adjusted_time: logger.warning(f"⚠️ Invalid schedule format for schedule {schedule_id}") continue # Parse timezone information server_timezone = get_server_timezone() schedule_time_part, schedule_timezone = parse_timezone_schedule(schedule_time) adjusted_time_part, adjusted_timezone = parse_timezone_schedule(adjusted_time) # Convert to server timezone for APScheduler if schedule_timezone and validate_timezone(schedule_timezone): server_schedule_time = convert_time_to_timezone(schedule_time_part, schedule_timezone, server_timezone) server_adjusted_time = convert_time_to_timezone(adjusted_time_part, adjusted_timezone or schedule_timezone, server_timezone) else: # Use original time if no valid timezone server_schedule_time = schedule_time_part server_adjusted_time = adjusted_time_part # Parse schedule times for server timezone content_gen_cron = self._parse_schedule_time(server_adjusted_time) publish_cron = self._parse_schedule_time(server_schedule_time) # Create content generation job (5 minutes before publishing) gen_job_id = f"gen_{schedule_id}" self.scheduler.add_job( func=self.generate_content_task, trigger=CronTrigger( minute=content_gen_cron['minute'], hour=content_gen_cron['hour'], day_of_week=content_gen_cron['day_of_week'] ), id=gen_job_id, name=f"Content generation for schedule {schedule_id}", args=[schedule.get('Social_network', {}).get('id_utilisateur'), schedule_id], replace_existing=True ) # Create publishing job pub_job_id = f"pub_{schedule_id}" self.scheduler.add_job( func=self.publish_post_task, trigger=CronTrigger( minute=publish_cron['minute'], hour=publish_cron['hour'], day_of_week=publish_cron['day_of_week'] ), id=pub_job_id, name=f"Post publishing for schedule {schedule_id}", args=[schedule_id], replace_existing=True ) logger.info(f"📅 Created schedule jobs for {schedule_id}") except Exception as e: logger.error(f"❌ Error creating jobs for schedule {schedule.get('id')}: {str(e)}") except Exception as e: logger.error(f"❌ Error loading schedules: {str(e)}") def _parse_schedule_time(self, schedule_time): """ Parse schedule time string into cron format. Args: schedule_time (str): Schedule time in format "Day HH:MM" Returns: dict: Cron parameters """ try: day_name, time_str = schedule_time.split() hour, minute = map(int, time_str.split(':')) # Map day names to cron format day_map = { 'Monday': 0, 'Tuesday': 1, 'Wednesday': 2, 'Thursday': 3, 'Friday': 4, 'Saturday': 5, 'Sunday': 6 } day_of_week = day_map.get(day_name, '*') return { 'minute': minute, 'hour': hour, 'day_of_week': day_of_week } except Exception as e: logger.error(f"Error parsing schedule time {schedule_time}: {str(e)}") # Default to every minute for error cases return { 'minute': '*', 'hour': '*', 'day_of_week': '*' } def generate_content_task(self, user_id: str, schedule_id: str): """ APScheduler task to generate content for a scheduled post. Args: user_id (str): User ID schedule_id (str): Schedule ID """ try: logger.info(f"🎨 Generating content for schedule {schedule_id}") # Run within application context with self.app.app_context(): # Initialize content service content_service = ContentService() # Generate content using content service generated_result = content_service.generate_post_content(user_id) # Ensure proper extraction of text content and image data from tuple # ContentService.generate_post_content() always returns a tuple: (text_content, image_data) if isinstance(generated_result, (tuple, list)) and len(generated_result) >= 1: # Extract text content (first element) and ensure it's a string text_content = generated_result[0] if generated_result[0] is not None else "Generated content will appear here..." # Extract image data (second element) if it exists image_data = generated_result[1] if len(generated_result) >= 2 and generated_result[1] is not None else None # Additional safeguard: ensure text_content is always a string, never a list/tuple if not isinstance(text_content, str): text_content = str(text_content) if text_content is not None else "Generated content will appear here..." else: # Fallback for unexpected return types text_content = str(generated_result) if generated_result is not None else "Generated content will appear here..." image_data = None # Final validation to ensure text_content is never stored as a list/tuple if isinstance(text_content, (list, tuple)): # Convert list/tuple to string representation text_content = str(text_content) # Process image data for proper storage processed_image_data = None if image_data is not None: try: # Use the shared utility function to ensure proper bytes format processed_image_data = ensure_bytes_format(image_data) logger.info(f"✅ Image data processed for schedule {schedule_id}") except Exception as e: logger.error(f"❌ Error processing image data for schedule {schedule_id}: {str(e)}") # Continue with text content even if image processing fails processed_image_data = None # Store generated content in database # We need to get the social account ID from the schedule schedule_response = ( self.supabase_client .table("Scheduling") .select("id_social") .eq("id", schedule_id) .execute() ) if not schedule_response.data: raise Exception(f"Schedule {schedule_id} not found") social_account_id = schedule_response.data[0]['id_social'] # Prepare post data post_data = { "id_social": social_account_id, "Text_content": text_content, "is_published": False, "sched": schedule_id } # Add processed image data if present if processed_image_data is not None: post_data["image_content_url"] = processed_image_data # Store the generated content response = ( self.supabase_client .table("Post_content") .insert(post_data) .execute() ) if response.data: logger.info(f"✅ Content generated and stored for schedule {schedule_id}") else: logger.error(f"❌ Failed to store generated content for schedule {schedule_id}") except Exception as e: logger.error(f"❌ Error in content generation task for schedule {schedule_id}: {str(e)}") def publish_post_task(self, schedule_id: str): """ APScheduler task to publish a scheduled post. Args: schedule_id (str): Schedule ID """ try: logger.info(f"🚀 Publishing post for schedule {schedule_id}") # Run within application context with self.app.app_context(): # Fetch the post to publish response = ( self.supabase_client .table("Post_content") .select("*") .eq("sched", schedule_id) .eq("is_published", False) .order("created_at", desc=True) .limit(1) .execute() ) if not response.data: logger.info(f"📭 No unpublished posts found for schedule {schedule_id}") return post = response.data[0] post_id = post.get('id') text_content = post.get('Text_content') image_url = post.get('image_content_url') # Get social network credentials schedule_response = ( self.supabase_client .table("Scheduling") .select("Social_network(token, sub)") .eq("id", schedule_id) .execute() ) if not schedule_response.data: raise Exception(f"Schedule {schedule_id} not found") social_network = schedule_response.data[0].get('Social_network', {}) access_token = social_network.get('token') user_sub = social_network.get('sub') if not access_token or not user_sub: logger.error(f"❌ Missing social network credentials for schedule {schedule_id}") return # Publish to LinkedIn linkedin_service = LinkedInService() publish_response = linkedin_service.publish_post( access_token, user_sub, text_content, image_url ) # Update post status in database update_response = ( self.supabase_client .table("Post_content") .update({"is_published": True}) .eq("id", post_id) .execute() ) logger.info(f"✅ Post published successfully for schedule {schedule_id}") except Exception as e: logger.error(f"❌ Error in publishing task for schedule {schedule_id}: {str(e)}") def trigger_immediate_update(self): """Trigger immediate schedule update.""" try: logger.info("🔄 Triggering immediate schedule update...") self.load_schedules() return True except Exception as e: logger.error(f"❌ Error triggering immediate schedule update: {str(e)}") return False def shutdown(self): """Shutdown the scheduler.""" if self.scheduler: self.scheduler.shutdown() logger.info("🛑 APS Scheduler shutdown")