""" Commands for revenue-related operations. """ from typing import Any, Dict, List, Optional from backend.core.models.revenues.revenues import Revenue from backend.core.repositories.revenues.revenues import RevenueRepository from backend.core.repositories.accounts.accounts import AccountRepository from backend.core.utils.validators import ( is_valid_uuid, is_valid_date, validate_required_fields, validate_model_exists ) from backend.core.utils.helpers import generate_uuid, parse_date from backend.core.commands.base import Command, CommandResult class CreateRevenueCommand(Command): """ Command to create a new revenue record. """ def __init__( self, revenue_repo: RevenueRepository, account_repo: AccountRepository, account_id: str, amount: str, start_date: str, end_date: Optional[str] = None ): """ Initialize the create revenue command. Args: revenue_repo: Repository for revenue operations. account_repo: Repository for account operations. account_id: ID of the account this revenue belongs to. amount: Amount of revenue. start_date: Start date of the revenue (YYYY-MM-DD). end_date: End date of the revenue (YYYY-MM-DD, optional). """ self.revenue_repo = revenue_repo self.account_repo = account_repo self.account_id = account_id self.amount = amount self.start_date = start_date self.end_date = end_date def validate(self) -> Dict[str, Any]: """ Validate the revenue creation data. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Check required fields required_fields = ['account_id', 'amount', 'start_date'] field_values = { 'account_id': self.account_id, 'amount': self.amount, 'start_date': self.start_date } missing_fields = validate_required_fields(field_values, required_fields) if missing_fields: errors.append(f"Required fields missing: {', '.join(missing_fields)}") # Validate account exists if not errors and self.account_id: account_validation = validate_model_exists( self.account_id, 'account', self.account_repo.get_by_id ) if not account_validation['valid']: errors.append(account_validation['error']) # Validate amount is a valid decimal if not errors and self.amount: try: amount = float(self.amount) if amount <= 0: errors.append("Amount must be greater than zero.") except ValueError: errors.append("Amount must be a valid number.") # Validate date formats if not errors and self.start_date and not is_valid_date(self.start_date): errors.append("Invalid start date format. Use YYYY-MM-DD.") if not errors and self.end_date and not is_valid_date(self.end_date): errors.append("Invalid end date format. Use YYYY-MM-DD.") # Validate start date is before end date if both provided if not errors and self.start_date and self.end_date: start = parse_date(self.start_date) end = parse_date(self.end_date) if start and end and start > end: errors.append("Start date must be before end date.") # Check if there's an active revenue record for this account if not errors: active_revenue = self.revenue_repo.get_active_by_account(self.account_id) if active_revenue: errors.append( f"Account already has an active revenue record. End the current record before creating a new one.") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[Revenue]: """ Execute the revenue creation command. Returns: CommandResult[Revenue]: Result of the command execution. """ # Validate command data validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Create revenue data revenue_id = generate_uuid() # Create revenue data dictionary revenue_data = { 'id': revenue_id, 'account_id': self.account_id, 'amount': self.amount, 'start_date': self.start_date, 'end_date': self.end_date } # Save to repository created_revenue = self.revenue_repo.create(revenue_data) return CommandResult.success_result( created_revenue, f"Revenue record created successfully for account {self.account_id}" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to create revenue record" ) class UpdateRevenueCommand(Command): """ Command to update an existing revenue record. """ def __init__( self, revenue_repo: RevenueRepository, account_repo: AccountRepository, revenue_id: str, amount: Optional[str] = None, start_date: Optional[str] = None, end_date: Optional[str] = None ): """ Initialize the update revenue command. Args: revenue_repo: Repository for revenue operations. account_repo: Repository for account operations. revenue_id: ID of the revenue to update. amount: New amount for the revenue. start_date: New start date for the revenue. end_date: New end date for the revenue. """ self.revenue_repo = revenue_repo self.account_repo = account_repo self.revenue_id = revenue_id self.amount = amount self.start_date = start_date self.end_date = end_date def validate(self) -> Dict[str, Any]: """ Validate the revenue update data. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Validate revenue exists if not is_valid_uuid(self.revenue_id): errors.append("Invalid revenue ID format") else: revenue = self.revenue_repo.get_by_id(self.revenue_id) if not revenue: errors.append(f"Revenue with ID {self.revenue_id} not found") # Validate amount is a valid decimal if provided if not errors and self.amount is not None: try: amount = float(self.amount) if amount <= 0: errors.append("Amount must be greater than zero.") except ValueError: errors.append("Amount must be a valid number.") # Validate date formats if provided if not errors and self.start_date is not None and not is_valid_date(self.start_date): errors.append("Invalid start date format. Use YYYY-MM-DD.") if not errors and self.end_date is not None and not is_valid_date(self.end_date): errors.append("Invalid end date format. Use YYYY-MM-DD.") # Validate start date is before end date if both provided if not errors and self.start_date and self.end_date: start = parse_date(self.start_date) end = parse_date(self.end_date) if start and end and start > end: errors.append("Start date must be before end date.") # If only updating end_date, validate it's after the existing start_date if not errors and self.end_date and not self.start_date: revenue = self.revenue_repo.get_by_id(self.revenue_id) if revenue: end = parse_date(self.end_date) start = parse_date(revenue.start_date) if end and start and start > end: errors.append("End date must be after the existing start date.") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[Revenue]: """ Execute the revenue update command. Returns: CommandResult[Revenue]: Result of the command execution. """ # Validate command data validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Create a dictionary of fields to update update_data = {} # Add fields to update_data if they were provided if self.amount is not None: update_data['amount'] = self.amount if self.start_date is not None: update_data['start_date'] = self.start_date if self.end_date is not None: update_data['end_date'] = self.end_date # Update the revenue with the data dictionary updated_revenue = self.revenue_repo.update(self.revenue_id, update_data) return CommandResult.success_result( updated_revenue, f"Revenue {self.revenue_id} updated successfully" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to update revenue" ) class DeleteRevenueCommand(Command): """ Command to delete a revenue record. """ def __init__( self, revenue_repo: RevenueRepository, revenue_id: str ): """ Initialize the delete revenue command. Args: revenue_repo: Repository for revenue operations. revenue_id: ID of the revenue to delete. """ self.revenue_repo = revenue_repo self.revenue_id = revenue_id def validate(self) -> Dict[str, Any]: """ Validate the revenue deletion request. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Validate revenue exists if not is_valid_uuid(self.revenue_id): errors.append("Invalid revenue ID format") else: revenue = self.revenue_repo.get_by_id(self.revenue_id) if not revenue: errors.append(f"Revenue with ID {self.revenue_id} not found") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[bool]: """ Execute the revenue deletion command. Returns: CommandResult[bool]: Result of the command execution. """ # Validate command data validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Delete the revenue success = self.revenue_repo.delete(self.revenue_id) if success: return CommandResult.success_result( True, f"Revenue {self.revenue_id} deleted successfully" ) else: return CommandResult.failure_result( "Failed to delete revenue", f"Revenue {self.revenue_id} could not be deleted" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to delete revenue" ) class EndRevenueCommand(Command): """ Command to end a revenue record by setting its end date. """ def __init__( self, revenue_repo: RevenueRepository, revenue_id: str, end_date: Optional[str] = None ): """ Initialize the end revenue command. Args: revenue_repo: Repository for revenue operations. revenue_id: ID of the revenue to end. end_date: End date for the revenue (defaults to today if not provided). """ self.revenue_repo = revenue_repo self.revenue_id = revenue_id self.end_date = end_date def validate(self) -> Dict[str, Any]: """ Validate the end revenue request. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Validate revenue exists if not is_valid_uuid(self.revenue_id): errors.append("Invalid revenue ID format") else: revenue = self.revenue_repo.get_by_id(self.revenue_id) if not revenue: errors.append(f"Revenue with ID {self.revenue_id} not found") elif revenue.end_date is not None: errors.append(f"Revenue is already ended") # Validate end date format if provided if not errors and self.end_date and not is_valid_date(self.end_date): errors.append("Invalid end date format. Use YYYY-MM-DD.") # Validate end date is after start date if provided if not errors and self.end_date: revenue = self.revenue_repo.get_by_id(self.revenue_id) if revenue: end = parse_date(self.end_date) start = parse_date(revenue.start_date) if end and start and start > end: errors.append("End date must be after the start date.") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[Revenue]: """Execute the end revenue command.""" validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # End the revenue record if self.end_date: # Use provided end date updated_revenue = self.revenue_repo.update(self.revenue_id, {'end_date': self.end_date}) else: # Use repository method that sets end date to today updated_revenue = self.revenue_repo.end_revenue(self.revenue_id) if updated_revenue: return CommandResult.success_result( updated_revenue, f"Revenue {self.revenue_id} ended successfully" ) else: return CommandResult.failure_result( "Failed to end revenue record", f"Revenue {self.revenue_id} could not be ended" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to end revenue record" ) class GetRevenueByDateRangeCommand(Command): """ Command to get revenue records within a date range. """ def __init__( self, revenue_repo: RevenueRepository, start_date: Optional[str] = None, end_date: Optional[str] = None, account_id: Optional[str] = None ): """ Initialize the get revenue by date range command. Args: revenue_repo: Repository for revenue operations. start_date: Start date for the range (YYYY-MM-DD). end_date: End date for the range (YYYY-MM-DD). account_id: Optional account ID to filter by. """ self.revenue_repo = revenue_repo self.start_date = start_date self.end_date = end_date self.account_id = account_id def validate(self) -> Dict[str, Any]: """ Validate the get revenue by date range request. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Validate date formats if provided if self.start_date and not is_valid_date(self.start_date): errors.append("Invalid start date format. Use YYYY-MM-DD.") if self.end_date and not is_valid_date(self.end_date): errors.append("Invalid end date format. Use YYYY-MM-DD.") # Validate start date is before end date if both provided if not errors and self.start_date and self.end_date: start = parse_date(self.start_date) end = parse_date(self.end_date) if start and end and start > end: errors.append("Start date must be before end date.") # Validate account ID if provided if not errors and self.account_id and not is_valid_uuid(self.account_id): errors.append("Invalid account ID format") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[List[Revenue]]: """ Execute the get revenue by date range command. Returns: CommandResult[List[Revenue]]: Result of the command execution with revenue data. """ # Validate command data validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Parse dates if provided start_date_obj = parse_date(self.start_date) if self.start_date else None end_date_obj = parse_date(self.end_date) if self.end_date else None # Get revenues by date range revenues = self.revenue_repo.get_by_date_range(start_date_obj, end_date_obj) # Filter by account if provided if self.account_id: revenues = revenues.filter(account_id=self.account_id) return CommandResult.success_result( list(revenues), f"Retrieved revenue data for specified date range" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to retrieve revenue data" ) class CalculateTotalRevenueCommand(Command): """ Command to calculate total revenue for an account or all accounts within a date range. """ def __init__( self, revenue_repo: RevenueRepository, start_date: Optional[str] = None, end_date: Optional[str] = None, account_id: Optional[str] = None ): self.revenue_repo = revenue_repo self.start_date = start_date self.end_date = end_date self.account_id = account_id def validate(self) -> Dict[str, Any]: """Validate the calculation request.""" errors = [] # Validate date formats if provided if self.start_date and not is_valid_date(self.start_date): errors.append("Invalid start date format. Use YYYY-MM-DD.") if self.end_date and not is_valid_date(self.end_date): errors.append("Invalid end date format. Use YYYY-MM-DD.") # Validate date range if both dates provided if self.start_date and self.end_date: start = parse_date(self.start_date) end = parse_date(self.end_date) if start and end and start > end: errors.append("Start date must be before end date.") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[float]: """Execute the revenue calculation command.""" validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Convert string dates to date objects if provided start_date_obj = parse_date(self.start_date) if self.start_date else None end_date_obj = parse_date(self.end_date) if self.end_date else None # Calculate total revenue using repository method total_revenue = self.revenue_repo.get_total_revenue( account_id=self.account_id, start_date=start_date_obj, end_date=end_date_obj ) # Create success message message = "Total revenue" if self.account_id: message += f" for account {self.account_id}" if self.start_date or self.end_date: message += " for period" if self.start_date: message += f" from {self.start_date}" if self.end_date: message += f" to {self.end_date}" message += f": ${total_revenue:.2f}" return CommandResult.success_result( total_revenue, message ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to calculate total revenue" ) class GetActiveRevenuesCommand(Command): """ Command to get all active revenue records. """ def __init__( self, revenue_repo: RevenueRepository, account_id: Optional[str] = None ): """ Initialize the get active revenues command. Args: revenue_repo: Repository for revenue operations. account_id: Optional account ID to filter by. """ self.revenue_repo = revenue_repo self.account_id = account_id def validate(self) -> Dict[str, Any]: """ Validate the get active revenues request. Returns: Dict[str, Any]: Validation result with 'is_valid' and optional 'errors'. """ errors = [] # Validate account ID if provided if self.account_id and not is_valid_uuid(self.account_id): errors.append("Invalid account ID format") return { 'is_valid': len(errors) == 0, 'errors': errors } def execute(self) -> CommandResult[List[Revenue]]: """ Execute the get active revenues command. Returns: CommandResult[List[Revenue]]: Result of the command execution with active revenues. """ # Validate command data validation = self.validate() if not validation['is_valid']: return CommandResult.failure_result(validation['errors']) try: # Get active revenues if self.account_id: active_revenue = self.revenue_repo.get_active_by_account(self.account_id) revenues = [active_revenue] if active_revenue else [] else: revenues = list(self.revenue_repo.get_active()) return CommandResult.success_result( revenues, f"Retrieved active revenue records" ) except Exception as e: return CommandResult.failure_result( str(e), "Failed to retrieve active revenue records" )