""" Batch Processing and Scheduling Service for Report Generation Provides background processing, progress tracking, scheduling, and email delivery """ from django.db import models from django.contrib.auth.models import User from django.core.mail import EmailMessage from django.conf import settings from django.utils import timezone from django.core.cache import cache from celery import shared_task from celery.result import AsyncResult import json import logging from datetime import datetime, timedelta from typing import Dict, List, Any, Optional, Union import uuid import os from io import BytesIO import zipfile from .enhanced_pdf_service import EnhancedPDFService from .export_functions import EnhancedExcelExportService from .advanced_filtering_service import AdvancedFilteringService logger = logging.getLogger(__name__) class BatchJob(models.Model): """Model to track batch processing jobs""" STATUS_CHOICES = [ ('pending', 'Pending'), ('processing', 'Processing'), ('completed', 'Completed'), ('failed', 'Failed'), ('cancelled', 'Cancelled') ] JOB_TYPE_CHOICES = [ ('single_report', 'Single Report'), ('batch_export', 'Batch Export'), ('scheduled_report', 'Scheduled Report'), ('bulk_processing', 'Bulk Processing') ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) job_type = models.CharField(max_length=20, choices=JOB_TYPE_CHOICES) status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='pending') # Job configuration report_types = models.JSONField() # List of report types to process filters = models.JSONField(default=dict) export_formats = models.JSONField(default=list) # ['pdf', 'excel'] # Progress tracking total_items = models.IntegerField(default=0) processed_items = models.IntegerField(default=0) progress_percentage = models.FloatField(default=0.0) # Results output_files = models.JSONField(default=list) # List of generated file paths error_messages = models.JSONField(default=list) # Timing created_at = models.DateTimeField(auto_now_add=True) started_at = models.DateTimeField(null=True, blank=True) completed_at = models.DateTimeField(null=True, blank=True) estimated_completion = models.DateTimeField(null=True, blank=True) # Email delivery email_recipients = models.JSONField(default=list) email_sent = models.BooleanField(default=False) # Celery task ID celery_task_id = models.CharField(max_length=255, null=True, blank=True) class Meta: ordering = ['-created_at'] def __str__(self): return f"{self.job_type} - {self.status} ({self.user.username})" @property def duration(self): """Calculate job duration""" if self.started_at: end_time = self.completed_at or timezone.now() return end_time - self.started_at return None @property def estimated_time_remaining(self): """Estimate remaining time based on progress""" if self.progress_percentage > 0 and self.started_at: elapsed = timezone.now() - self.started_at total_estimated = elapsed / (self.progress_percentage / 100) return total_estimated - elapsed return None class ScheduledReport(models.Model): """Model for scheduled report generation""" FREQUENCY_CHOICES = [ ('daily', 'Daily'), ('weekly', 'Weekly'), ('monthly', 'Monthly'), ('quarterly', 'Quarterly'), ('yearly', 'Yearly') ] WEEKDAY_CHOICES = [ (0, 'Monday'), (1, 'Tuesday'), (2, 'Wednesday'), (3, 'Thursday'), (4, 'Friday'), (5, 'Saturday'), (6, 'Sunday') ] id = models.UUIDField(primary_key=True, default=uuid.uuid4, editable=False) user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=200) description = models.TextField(blank=True) # Schedule configuration frequency = models.CharField(max_length=20, choices=FREQUENCY_CHOICES) time_of_day = models.TimeField() day_of_week = models.IntegerField(choices=WEEKDAY_CHOICES, null=True, blank=True) day_of_month = models.IntegerField(null=True, blank=True) # 1-31 # Report configuration report_types = models.JSONField() filters = models.JSONField(default=dict) export_formats = models.JSONField(default=list) # Email configuration email_recipients = models.JSONField(default=list) email_subject_template = models.CharField(max_length=200, default="Scheduled Report: {report_name}") email_body_template = models.TextField(default="Please find attached your scheduled report.") # Status is_active = models.BooleanField(default=True) last_run = models.DateTimeField(null=True, blank=True) next_run = models.DateTimeField(null=True, blank=True) run_count = models.IntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: ordering = ['next_run'] def __str__(self): return f"{self.name} ({self.frequency})" def calculate_next_run(self): """Calculate the next run time based on frequency""" now = timezone.now() if self.frequency == 'daily': next_run = now.replace(hour=self.time_of_day.hour, minute=self.time_of_day.minute, second=0, microsecond=0) if next_run <= now: next_run += timedelta(days=1) elif self.frequency == 'weekly': days_ahead = self.day_of_week - now.weekday() if days_ahead <= 0: # Target day already happened this week days_ahead += 7 next_run = now + timedelta(days=days_ahead) next_run = next_run.replace(hour=self.time_of_day.hour, minute=self.time_of_day.minute, second=0, microsecond=0) elif self.frequency == 'monthly': next_run = now.replace(day=self.day_of_month, hour=self.time_of_day.hour, minute=self.time_of_day.minute, second=0, microsecond=0) if next_run <= now: # Move to next month if next_run.month == 12: next_run = next_run.replace(year=next_run.year + 1, month=1) else: next_run = next_run.replace(month=next_run.month + 1) elif self.frequency == 'quarterly': # Find next quarter current_quarter = (now.month - 1) // 3 + 1 next_quarter_month = current_quarter * 3 + 1 if next_quarter_month > 12: next_quarter_month = 1 year = now.year + 1 else: year = now.year next_run = now.replace(year=year, month=next_quarter_month, day=1, hour=self.time_of_day.hour, minute=self.time_of_day.minute, second=0, microsecond=0) elif self.frequency == 'yearly': next_run = now.replace(month=1, day=1, hour=self.time_of_day.hour, minute=self.time_of_day.minute, second=0, microsecond=0) if next_run <= now: next_run = next_run.replace(year=next_run.year + 1) self.next_run = next_run self.save(update_fields=['next_run']) return next_run class BatchProcessingService: """ Service class for batch processing and scheduling """ def __init__(self): """Initialize the batch processing service""" self.pdf_service = EnhancedPDFService() self.excel_service = EnhancedExcelExportService() self.filtering_service = AdvancedFilteringService() def create_batch_job(self, user, job_type: str, report_types: List[str], filters: Dict[str, Any] = None, export_formats: List[str] = None, email_recipients: List[str] = None) -> BatchJob: """Create a new batch processing job""" try: job = BatchJob.objects.create( user=user, job_type=job_type, report_types=report_types, filters=filters or {}, export_formats=export_formats or ['excel'], email_recipients=email_recipients or [], total_items=len(report_types) * len(export_formats or ['excel']) ) # Start the batch job asynchronously task = process_batch_job.delay(str(job.id)) job.celery_task_id = task.id job.save(update_fields=['celery_task_id']) return job except Exception as e: logger.error(f"Error creating batch job: {str(e)}") raise def get_job_status(self, job_id: str) -> Dict[str, Any]: """Get the status of a batch job""" try: job = BatchJob.objects.get(id=job_id) status_data = { 'id': str(job.id), 'status': job.status, 'progress_percentage': job.progress_percentage, 'processed_items': job.processed_items, 'total_items': job.total_items, 'created_at': job.created_at.isoformat(), 'started_at': job.started_at.isoformat() if job.started_at else None, 'completed_at': job.completed_at.isoformat() if job.completed_at else None, 'duration': str(job.duration) if job.duration else None, 'estimated_time_remaining': str(job.estimated_time_remaining) if job.estimated_time_remaining else None, 'output_files': job.output_files, 'error_messages': job.error_messages, 'email_sent': job.email_sent } return status_data except BatchJob.DoesNotExist: return {'error': 'Job not found'} except Exception as e: logger.error(f"Error getting job status: {str(e)}") return {'error': str(e)} def cancel_job(self, job_id: str, user) -> bool: """Cancel a batch job""" try: job = BatchJob.objects.get(id=job_id, user=user) if job.status in ['pending', 'processing']: # Cancel Celery task if job.celery_task_id: AsyncResult(job.celery_task_id).revoke(terminate=True) job.status = 'cancelled' job.completed_at = timezone.now() job.save(update_fields=['status', 'completed_at']) return True return False except BatchJob.DoesNotExist: return False except Exception as e: logger.error(f"Error cancelling job: {str(e)}") return False def create_scheduled_report(self, user, name: str, description: str, frequency: str, time_of_day, report_types: List[str], filters: Dict[str, Any] = None, export_formats: List[str] = None, email_recipients: List[str] = None, **schedule_params) -> ScheduledReport: """Create a new scheduled report""" try: scheduled_report = ScheduledReport.objects.create( user=user, name=name, description=description, frequency=frequency, time_of_day=time_of_day, day_of_week=schedule_params.get('day_of_week'), day_of_month=schedule_params.get('day_of_month'), report_types=report_types, filters=filters or {}, export_formats=export_formats or ['excel'], email_recipients=email_recipients or [] ) # Calculate next run time scheduled_report.calculate_next_run() return scheduled_report except Exception as e: logger.error(f"Error creating scheduled report: {str(e)}") raise def get_user_jobs(self, user, status: str = None, limit: int = 50) -> List[BatchJob]: """Get user's batch jobs""" try: queryset = BatchJob.objects.filter(user=user) if status: queryset = queryset.filter(status=status) return list(queryset[:limit]) except Exception as e: logger.error(f"Error getting user jobs: {str(e)}") return [] def get_user_scheduled_reports(self, user, active_only: bool = True) -> List[ScheduledReport]: """Get user's scheduled reports""" try: queryset = ScheduledReport.objects.filter(user=user) if active_only: queryset = queryset.filter(is_active=True) return list(queryset) except Exception as e: logger.error(f"Error getting user scheduled reports: {str(e)}") return [] def get_due_scheduled_reports(self) -> List[ScheduledReport]: """Get scheduled reports that are due to run""" try: now = timezone.now() return list(ScheduledReport.objects.filter( is_active=True, next_run__lte=now )) except Exception as e: logger.error(f"Error getting due scheduled reports: {str(e)}") return [] def execute_scheduled_report(self, scheduled_report: ScheduledReport) -> BatchJob: """Execute a scheduled report""" try: # Create batch job for the scheduled report job = self.create_batch_job( user=scheduled_report.user, job_type='scheduled_report', report_types=scheduled_report.report_types, filters=scheduled_report.filters, export_formats=scheduled_report.export_formats, email_recipients=scheduled_report.email_recipients ) # Update scheduled report scheduled_report.last_run = timezone.now() scheduled_report.run_count += 1 scheduled_report.calculate_next_run() return job except Exception as e: logger.error(f"Error executing scheduled report: {str(e)}") raise def cleanup_old_jobs(self, days_old: int = 30): """Clean up old completed jobs""" try: cutoff_date = timezone.now() - timedelta(days=days_old) old_jobs = BatchJob.objects.filter( status__in=['completed', 'failed', 'cancelled'], completed_at__lt=cutoff_date ) # Delete associated files for job in old_jobs: for file_path in job.output_files: try: if os.path.exists(file_path): os.remove(file_path) except Exception as e: logger.warning(f"Could not delete file {file_path}: {str(e)}") # Delete job records deleted_count = old_jobs.count() old_jobs.delete() logger.info(f"Cleaned up {deleted_count} old batch jobs") return deleted_count except Exception as e: logger.error(f"Error cleaning up old jobs: {str(e)}") return 0 def create_job_template(self, user, name: str, description: str, job_type: str, report_types: List[str], filters: Dict[str, Any] = None, export_formats: List[str] = None, email_recipients: List[str] = None) -> 'BatchJobTemplate': """Create a reusable batch job template""" try: template = BatchJobTemplate.objects.create( user=user, name=name, description=description, job_type=job_type, report_types=report_types, filters=filters or {}, export_formats=export_formats or ['excel'], email_recipients=email_recipients or [] ) return template except Exception as e: logger.error(f"Error creating job template: {str(e)}") raise def create_job_from_template(self, template: 'BatchJobTemplate', override_params: Dict[str, Any] = None) -> BatchJob: """Create a batch job from a template""" try: params = { 'user': template.user, 'job_type': template.job_type, 'report_types': template.report_types, 'filters': template.filters, 'export_formats': template.export_formats, 'email_recipients': template.email_recipients } # Apply overrides if override_params: params.update(override_params) job = self.create_batch_job(**params) # Update template usage template.usage_count += 1 template.last_used = timezone.now() template.save(update_fields=['usage_count', 'last_used']) return job except Exception as e: logger.error(f"Error creating job from template: {str(e)}") raise def get_user_templates(self, user) -> List['BatchJobTemplate']: """Get user's job templates""" try: return list(BatchJobTemplate.objects.filter(user=user)) except Exception as e: logger.error(f"Error getting user templates: {str(e)}") return [] def get_queue_status(self) -> Dict[str, Any]: """Get current queue status""" try: pending_jobs = BatchJob.objects.filter(status='pending').count() processing_jobs = BatchJob.objects.filter(status='processing').count() # Get queue information queue_items = BatchJobQueue.objects.filter( job__status__in=['pending', 'processing'] ).order_by('-priority', 'queue_position') queue_info = [] for item in queue_items[:10]: # Top 10 in queue queue_info.append({ 'job_id': str(item.job.id), 'priority': item.priority, 'position': item.queue_position, 'estimated_time': str(item.estimated_processing_time) if item.estimated_processing_time else None, 'user': item.job.user.username, 'job_type': item.job.job_type }) return { 'pending_jobs': pending_jobs, 'processing_jobs': processing_jobs, 'queue_length': len(queue_info), 'queue_items': queue_info } except Exception as e: logger.error(f"Error getting queue status: {str(e)}") return {'error': str(e)} def set_job_priority(self, job_id: str, priority: str, user) -> bool: """Set job priority""" try: job = BatchJob.objects.get(id=job_id, user=user) if job.status == 'pending': queue_item, created = BatchJobQueue.objects.get_or_create( job=job, defaults={'priority': priority} ) if not created: queue_item.priority = priority queue_item.save(update_fields=['priority']) return True return False except BatchJob.DoesNotExist: return False except Exception as e: logger.error(f"Error setting job priority: {str(e)}") return False def get_processing_metrics(self, date_range: Dict[str, Any] = None) -> Dict[str, Any]: """Get batch processing metrics""" try: if date_range: start_date = datetime.strptime(date_range['start'], '%Y-%m-%d').date() end_date = datetime.strptime(date_range['end'], '%Y-%m-%d').date() else: end_date = timezone.now().date() start_date = end_date - timedelta(days=30) metrics = BatchProcessingMetrics.objects.filter( date__range=[start_date, end_date] ) # Aggregate metrics total_jobs = sum(m.total_jobs_created for m in metrics) completed_jobs = sum(m.total_jobs_completed for m in metrics) failed_jobs = sum(m.total_jobs_failed for m in metrics) total_files = sum(m.total_files_generated for m in metrics) success_rate = (completed_jobs / total_jobs * 100) if total_jobs > 0 else 0 return { 'period': { 'start': start_date.isoformat(), 'end': end_date.isoformat() }, 'totals': { 'jobs_created': total_jobs, 'jobs_completed': completed_jobs, 'jobs_failed': failed_jobs, 'files_generated': total_files, 'success_rate': round(success_rate, 2) }, 'daily_metrics': [ { 'date': m.date.isoformat(), 'jobs_created': m.total_jobs_created, 'jobs_completed': m.total_jobs_completed, 'jobs_failed': m.total_jobs_failed, 'files_generated': m.total_files_generated } for m in metrics ] } except Exception as e: logger.error(f"Error getting processing metrics: {str(e)}") return {'error': str(e)} def estimate_job_completion_time(self, job: BatchJob) -> Optional[datetime]: """Estimate job completion time based on historical data""" try: # Get similar completed jobs similar_jobs = BatchJob.objects.filter( job_type=job.job_type, status='completed', total_items__range=[job.total_items * 0.8, job.total_items * 1.2] ).order_by('-completed_at')[:10] if not similar_jobs: return None # Calculate average processing time total_duration = timedelta() for similar_job in similar_jobs: if similar_job.duration: total_duration += similar_job.duration if similar_jobs: avg_duration = total_duration / len(similar_jobs) estimated_completion = timezone.now() + avg_duration # Update job with estimate job.estimated_completion = estimated_completion job.save(update_fields=['estimated_completion']) return estimated_completion return None except Exception as e: logger.error(f"Error estimating job completion time: {str(e)}") return None def create_bulk_job_from_filters(self, user, base_filters: Dict[str, Any], filter_variations: List[Dict[str, Any]], report_types: List[str], export_formats: List[str] = None) -> BatchJob: """Create bulk job with multiple filter variations""" try: # Create individual jobs for each filter variation sub_jobs = [] for variation in filter_variations: # Merge base filters with variation combined_filters = {**base_filters, **variation} sub_job = self.create_batch_job( user=user, job_type='bulk_processing', report_types=report_types, filters=combined_filters, export_formats=export_formats or ['excel'] ) sub_jobs.append(sub_job) # Create parent bulk job bulk_job = BatchJob.objects.create( user=user, job_type='bulk_processing', report_types=report_types, filters={'sub_job_ids': [str(job.id) for job in sub_jobs]}, export_formats=export_formats or ['excel'], total_items=sum(job.total_items for job in sub_jobs) ) return bulk_job except Exception as e: logger.error(f"Error creating bulk job from filters: {str(e)}") raise def pause_job(self, job_id: str, user) -> bool: """Pause a running job""" try: job = BatchJob.objects.get(id=job_id, user=user) if job.status == 'processing' and job.celery_task_id: # This would require custom Celery task implementation # For now, we'll just mark it as paused in the database cache.set(f"job_paused_{job_id}", True, timeout=3600) return True return False except BatchJob.DoesNotExist: return False except Exception as e: logger.error(f"Error pausing job: {str(e)}") return False def resume_job(self, job_id: str, user) -> bool: """Resume a paused job""" try: job = BatchJob.objects.get(id=job_id, user=user) if cache.get(f"job_paused_{job_id}"): cache.delete(f"job_paused_{job_id}") return True return False except BatchJob.DoesNotExist: return False except Exception as e: logger.error(f"Error resuming job: {str(e)}") return False # Celery tasks for background processing @shared_task(bind=True) def process_batch_job(self, job_id: str): """Process a batch job in the background""" try: job = BatchJob.objects.get(id=job_id) job.status = 'processing' job.started_at = timezone.now() job.save(update_fields=['status', 'started_at']) service = BatchProcessingService() pdf_service = EnhancedPDFService() excel_service = EnhancedExcelExportService() filtering_service = AdvancedFilteringService() output_files = [] errors = [] processed = 0 # Process each report type for report_type in job.report_types: try: # Get data for this report type report_data = get_report_data(report_type, job.filters, job.user) # Generate each requested format for export_format in job.export_formats: try: if export_format == 'pdf': # Generate PDF pdf_content = pdf_service.generate_analytics_pdf( report_type=report_type, data=report_data, date_range=job.filters.get('date_range', {}), charts_config={}, template='standard', enable_signature=False ) # Save PDF file filename = f"{report_type}_{job_id}_{processed}.pdf" file_path = os.path.join(settings.MEDIA_ROOT, 'batch_reports', filename) os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, 'wb') as f: f.write(pdf_content) output_files.append(file_path) elif export_format == 'excel': # Generate Excel excel_response = excel_service.create_multi_worksheet_export( report_data, report_type, job.filters ) # Save Excel file filename = f"{report_type}_{job_id}_{processed}.xlsx" file_path = os.path.join(settings.MEDIA_ROOT, 'batch_reports', filename) os.makedirs(os.path.dirname(file_path), exist_ok=True) with open(file_path, 'wb') as f: f.write(excel_response.content) output_files.append(file_path) processed += 1 # Update progress progress = (processed / job.total_items) * 100 job.processed_items = processed job.progress_percentage = progress job.save(update_fields=['processed_items', 'progress_percentage']) # Update task progress self.update_state( state='PROGRESS', meta={'current': processed, 'total': job.total_items, 'progress': progress} ) except Exception as e: error_msg = f"Error generating {export_format} for {report_type}: {str(e)}" errors.append(error_msg) logger.error(error_msg) processed += 1 except Exception as e: error_msg = f"Error processing report type {report_type}: {str(e)}" errors.append(error_msg) logger.error(error_msg) # Update job with results job.output_files = output_files job.error_messages = errors job.status = 'completed' if not errors else 'failed' job.completed_at = timezone.now() job.progress_percentage = 100.0 job.save(update_fields=['output_files', 'error_messages', 'status', 'completed_at', 'progress_percentage']) # Send email if recipients specified if job.email_recipients and output_files: send_batch_job_email.delay(str(job.id)) return { 'status': job.status, 'output_files': len(output_files), 'errors': len(errors) } except Exception as e: logger.error(f"Error processing batch job {job_id}: {str(e)}") # Update job status to failed try: job = BatchJob.objects.get(id=job_id) job.status = 'failed' job.completed_at = timezone.now() job.error_messages = [str(e)] job.save(update_fields=['status', 'completed_at', 'error_messages']) except: pass raise @shared_task def send_batch_job_email(job_id: str): """Send email with batch job results""" try: job = BatchJob.objects.get(id=job_id) if not job.email_recipients or not job.output_files: return # Create email subject = f"Batch Report Generation Complete - {job.job_type}" body = f""" Your batch report generation has been completed. Job Details: - Job Type: {job.job_type} - Status: {job.status} - Reports Generated: {len(job.output_files)} - Processing Time: {job.duration} - Completed At: {job.completed_at} """ if job.error_messages: body += f"\nErrors encountered:\n" for error in job.error_messages: body += f"- {error}\n" email = EmailMessage( subject=subject, body=body, from_email=settings.DEFAULT_FROM_EMAIL, to=job.email_recipients ) # Attach files (or create zip if many files) if len(job.output_files) <= 5: # Attach individual files for file_path in job.output_files: if os.path.exists(file_path): email.attach_file(file_path) else: # Create zip file zip_filename = f"batch_reports_{job_id}.zip" zip_path = os.path.join(settings.MEDIA_ROOT, 'batch_reports', zip_filename) with zipfile.ZipFile(zip_path, 'w') as zip_file: for file_path in job.output_files: if os.path.exists(file_path): zip_file.write(file_path, os.path.basename(file_path)) email.attach_file(zip_path) # Send email email.send() # Update job job.email_sent = True job.save(update_fields=['email_sent']) logger.info(f"Batch job email sent for job {job_id}") except Exception as e: logger.error(f"Error sending batch job email: {str(e)}") @shared_task def process_scheduled_reports(): """Process due scheduled reports""" try: service = BatchProcessingService() due_reports = service.get_due_scheduled_reports() processed_count = 0 for scheduled_report in due_reports: try: job = service.execute_scheduled_report(scheduled_report) logger.info(f"Started scheduled report job {job.id} for {scheduled_report.name}") processed_count += 1 except Exception as e: logger.error(f"Error processing scheduled report {scheduled_report.id}: {str(e)}") logger.info(f"Processed {processed_count} scheduled reports") return processed_count except Exception as e: logger.error(f"Error processing scheduled reports: {str(e)}") return 0 @shared_task def cleanup_old_batch_jobs(): """Clean up old batch jobs""" try: service = BatchProcessingService() deleted_count = service.cleanup_old_jobs(days_old=30) logger.info(f"Cleaned up {deleted_count} old batch jobs") return deleted_count except Exception as e: logger.error(f"Error cleaning up old batch jobs: {str(e)}") return 0 def get_report_data(report_type: str, filters: Dict[str, Any], user) -> Dict[str, Any]: """Get report data for a specific report type""" # This would be implemented based on your specific report data sources # For now, returning sample data structure sample_data = { 'summary': { 'total_loans': 100, 'total_amount': 5000000, 'average_amount': 50000 }, 'loans': [ # Sample loan data would go here ] } return sample_data class BatchJobTemplate(models.Model): """Model for reusable batch job templates""" user = models.ForeignKey(User, on_delete=models.CASCADE) name = models.CharField(max_length=200) description = models.TextField(blank=True) # Template configuration job_type = models.CharField(max_length=20, choices=BatchJob.JOB_TYPE_CHOICES) report_types = models.JSONField() filters = models.JSONField(default=dict) export_formats = models.JSONField(default=list) email_recipients = models.JSONField(default=list) # Usage tracking usage_count = models.IntegerField(default=0) last_used = models.DateTimeField(null=True, blank=True) created_at = models.DateTimeField(auto_now_add=True) updated_at = models.DateTimeField(auto_now=True) class Meta: unique_together = ['user', 'name'] ordering = ['-last_used', '-created_at'] def __str__(self): return f"{self.name} ({self.user.username})" class BatchJobQueue(models.Model): """Model for managing batch job queues and priorities""" PRIORITY_CHOICES = [ ('low', 'Low'), ('normal', 'Normal'), ('high', 'High'), ('urgent', 'Urgent') ] job = models.OneToOneField(BatchJob, on_delete=models.CASCADE) priority = models.CharField(max_length=10, choices=PRIORITY_CHOICES, default='normal') queue_position = models.IntegerField(default=0) estimated_processing_time = models.DurationField(null=True, blank=True) dependencies = models.ManyToManyField('self', blank=True, symmetrical=False) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ['-priority', 'queue_position', 'created_at'] def __str__(self): return f"Queue: {self.job.id} ({self.priority})" class BatchProcessingMetrics(models.Model): """Model for tracking batch processing metrics""" date = models.DateField(unique=True) # Job statistics total_jobs_created = models.IntegerField(default=0) total_jobs_completed = models.IntegerField(default=0) total_jobs_failed = models.IntegerField(default=0) # Processing statistics average_processing_time = models.DurationField(null=True, blank=True) total_files_generated = models.IntegerField(default=0) total_data_processed_mb = models.FloatField(default=0.0) # Queue statistics peak_queue_length = models.IntegerField(default=0) average_queue_wait_time = models.DurationField(null=True, blank=True) # Email statistics emails_sent = models.IntegerField(default=0) email_delivery_failures = models.IntegerField(default=0) created_at = models.DateTimeField(auto_now_add=True) class Meta: ordering = ['-date'] def __str__(self): return f"Metrics for {self.date}"