""" Customer Services for handling customer-related business logic. """ from typing import Dict, List, Optional, Tuple, Union, Any from datetime import datetime import logging from fastapi import HTTPException, status from sqlalchemy import select, and_, or_, func, desc, asc from sqlalchemy.orm import Session, selectinload from src.apps.customers.models.customer import Customer from src.apps.base.models.address import Address from src.apps.customers.schemas.customer_common import ( CustomerSchema, CustomerListSchema ) from src.apps.customers.schemas.customer_requests import ( CustomerCreateRequestSchema, CustomerUpdateRequestSchema, CustomerFilterSchema ) from src.helpers.pagination import QueryPaginator from src.core.functions import generate_id from src.core.exceptions import APIException from src.apps.merchants.models.merchant import Merchant from src.apps.base.models.address import Address from src.apps.customers.models.customer_contact import CustomerContact import uuid import re from pydantic import ValidationError logger = logging.getLogger(__name__) # Helper functions for query building def _apply_search_filter(stmt, search: str): """Apply search filter to customer query.""" search_term = f"%{search}%" return stmt.where( or_( Customer.first_name.ilike(search_term), Customer.last_name.ilike(search_term), Customer.email.ilike(search_term), Customer.uin.ilike(search_term), Customer.customer_id.ilike(search_term) ) ) def _apply_selective_loading(stmt, fields: str): """Apply selective loading based on fields parameter.""" requested_fields = [field.strip() for field in fields.split(',')] load_options = [] field_mapping = { 'addresses': Customer.addresses, 'contacts': Customer.contacts, 'merchant': Customer.merchant, 'attachments': Customer.attachments } for field in requested_fields: if field in field_mapping: load_options.append(selectinload(field_mapping[field])) if load_options: stmt = stmt.options(*load_options) return stmt async def get_customer(db: Session, customer_id: str, fields: str = None) -> CustomerSchema: """Get customer by customer_id with optional related fields.""" try: # Base query stmt = select(Customer).where(Customer.customer_id == customer_id) # Parse fields parameter and add selective loading if fields: stmt = _apply_selective_loading(stmt, fields) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() if not result: raise APIException( message="Customer not found", status_code=404, module="customers.services" ) # Use SQLAlchemy ORM with Pydantic model_validate for clean serialization return CustomerSchema.model_validate(result) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting customer: {e}") raise APIException( message=f"Failed to retrieve customer: {str(e)}", status_code=500, module="customers.services" ) async def create_customer( db: Session, customer_data: CustomerCreateRequestSchema ) -> Tuple[Union[CustomerSchema, Customer, None], str]: """Create a new customer with Pydantic validation.""" try: # Note: All validation is now handled by Pydantic validators in the schema # Always generate a new UIN (ignore any provided UIN to ensure uniqueness) # Generate UIN using the new helper function # Use customer name for more meaningful UINs if available customer_name = None if customer_data.first_name and customer_data.last_name: customer_name = f"{customer_data.first_name} {customer_data.last_name}" elif customer_data.first_name: customer_name = customer_data.first_name generated_uin = generate_id('uin', customer_name) logger.info(f"Generated UIN: {generated_uin} for customer: {customer_name}") # Ensure UIN uniqueness (should be unique with timestamp+UUID but double-check) max_attempts = 10 attempt = 0 while attempt < max_attempts: uin_check_stmt = select(Customer).where(Customer.uin == generated_uin) existing = db.execute(uin_check_stmt).scalar_one_or_none() if not existing: logger.info(f"UIN {generated_uin} is unique") break logger.warning(f"UIN {generated_uin} already exists, generating new one. Attempt: {attempt + 1}") generated_uin = generate_id('uin', customer_name) attempt += 1 if attempt >= max_attempts: logger.error(f"Failed to generate unique UIN after {max_attempts} attempts") return (None, "Failed to generate unique UIN after multiple attempts") # Update customer_data with generated UIN customer_data.uin = generated_uin logger.info(f"Final UIN assigned: {generated_uin}") # Generate internal customer_id (always auto-generated, not user-provided) customer_name = None if customer_data.first_name and customer_data.last_name: customer_name = f"{customer_data.first_name} {customer_data.last_name}" elif customer_data.first_name: customer_name = customer_data.first_name customer_id = generate_id('customer', customer_name) # Ensure customer_id uniqueness for internal use while await check_customer_id_exists(db, customer_id): customer_id = generate_id('customer', customer_name) # Handle account_literal generation or use user-provided account_literal account_literal = None # Check if user provided an account_literal if hasattr(customer_data, 'account_literal') and customer_data.account_literal: # Use user-provided account literal (already validated for uniqueness) account_literal = customer_data.account_literal else: # Auto-generate account literal using the helper function account_literal = generate_id('customer', customer_name) # Ensure uniqueness for auto-generated account literals while await check_account_literal_exists(db, account_literal): account_literal = generate_id('customer', customer_name) # Convert tags list to comma-separated string if provided tags_str = None if customer_data.tags and isinstance(customer_data.tags, list): tags_str = ",".join(customer_data.tags) # Get merchant_id (validation already done in early validation) merchant_id = customer_data.merchant_id if not merchant_id: # Get existing merchant or create a default one first_merchant_stmt = select(Merchant).limit(1) first_merchant = db.execute(first_merchant_stmt).scalar_one_or_none() if not first_merchant: # Create a default merchant for testing try: # Generate UIN for the default merchant merchant_uin = generate_id('merchant', "Default Merchant") default_merchant = Merchant( uin=merchant_uin, name="Default Merchant", email="default@merchant.com", phone="+1234567890", merchant_id=f"MERCH_{uuid.uuid4().hex[:8].upper()}", industry="general", is_active=True ) db.add(default_merchant) db.flush() merchant_id = default_merchant.id logger.info(f"Created default merchant with ID {merchant_id}") except Exception as e: logger.error(f"Error creating default merchant: {e}") return (None, f"Failed to create default merchant: {str(e)}") else: merchant_id = first_merchant.id # Convert schema to dict, excluding fields that don't exist in Customer model customer_dict = customer_data.model_dump(exclude={ 'contact', 'contacts', 'address', 'role_type', 'account_literal' }) # Handle user_account_id - use first available user if none provided user_account_id = customer_data.user_account_id # if not user_account_id: # # Get first available user as default # # Lazy import to avoid circular dependency # first_user_stmt = select(User).limit(1) # first_user = db.execute(first_user_stmt).scalar_one_or_none() # if first_user: # user_account_id = first_user.id # logger.info(f"Using default user_account_id: {user_account_id}") # Add computed/additional fields that aren't in the schema customer_dict["customer_id"] = customer_id customer_dict["account_literal"] = account_literal customer_dict["merchant_id"] = merchant_id customer_dict["user_account_id"] = user_account_id customer_dict["tags"] = tags_str # Handle optional fields with defaults customer_dict["is_active"] = getattr(customer_data, 'is_active', True) customer_dict["is_vendor"] = getattr(customer_data, 'is_vendor', False) # Create new customer customer_obj = Customer(**customer_dict) db.add(customer_obj) db.commit() db.refresh(customer_obj) logger.info(f"Customer inserted using ORM query: {customer_obj.email or customer_obj.uin}") logger.info(f"Customer committed to database with ID: {customer_obj.id}, customer_id: {customer_obj.customer_id}") # Create address record linked to customer if customer_data.address: try: # Convert address schema to dict, excluding None values address_dict = customer_data.address.model_dump(exclude_none=True) # Add customer relationship fields address_dict["customer_id"] = customer_obj.id address_dict["merchant_id"] = merchant_id # Set defaults for optional fields if not provided if "is_active" not in address_dict: address_dict["is_active"] = True if "use_as_default" not in address_dict: address_dict["use_as_default"] = True # Create address object with customer_id (one-to-many relationship) address_obj = Address(**address_dict) db.add(address_obj) db.commit() db.refresh(address_obj) logger.info(f"Address created and linked to customer {customer_obj.customer_id} with ID: {address_obj.id}") except Exception as e: logger.error(f"Error creating address for customer {customer_obj.customer_id}: {e}") # Don't fail customer creation if address creation fails # Just log the error and continue # Create contacts from customer data if provided contacts_to_create = [] # Handle single contact if customer_data.contact: contacts_to_create.append(customer_data.contact) # Handle multiple contacts if customer_data.contacts: contacts_to_create.extend(customer_data.contacts) # Create all contacts for contact_data in contacts_to_create: try: # Generate contact_id using project's standard helper function # Use contact name for more meaningful IDs if available contact_name = None if contact_data.first_name and contact_data.last_name: contact_name = f"{contact_data.first_name} {contact_data.last_name}" elif contact_data.first_name: contact_name = contact_data.first_name contact_id = generate_id('contact', contact_name) # Ensure contact_id uniqueness (should be unique with timestamp+UUID but double-check) max_attempts = 5 attempt = 0 while attempt < max_attempts: contact_check_stmt = select(CustomerContact).where(CustomerContact.contact_id == contact_id) existing_contact = db.execute(contact_check_stmt).scalar_one_or_none() if not existing_contact: break logger.warning(f"Contact ID {contact_id} already exists, generating new one. Attempt: {attempt + 1}") contact_id = generate_id('contact', contact_name) attempt += 1 if attempt >= max_attempts: logger.error(f"Failed to generate unique contact ID after {max_attempts} attempts") continue # Prepare contact data dict, excluding fields that don't exist in CustomerContact model contact_dict = contact_data.model_dump(exclude={'address', 'website'}) # Exclude legacy website field contact_dict["contact_id"] = contact_id contact_dict["customer_id"] = customer_obj.id contact_dict["user_account_id"] = None # Set to None since user_account_id is now optional contact_dict["is_active"] = contact_data.is_active if contact_data.is_active is not None else True # Create new contact contact_obj = CustomerContact(**contact_dict) db.add(contact_obj) db.commit() db.refresh(contact_obj) logger.info(f"Contact created for customer {customer_obj.customer_id} with ID: {contact_obj.contact_id}") except Exception as e: logger.error(f"Error creating contact for customer {customer_obj.customer_id}: {e}") # Don't fail customer creation if contact creation fails # Just log the error and continue # Re-query customer with relationships loaded for response customer_with_relations = await get_customer(db, customer_obj.customer_id) return (customer_with_relations, "Customer created successfully") except Exception as e: logger.error(f"Error creating customer: {e}") return (None, f"Failed to create customer: {str(e)}") async def get_customers_list( db: Session, filters: 'CustomerFilterSchema', limit: int = 10, offset: int = 0, paginate: bool = True ) -> Union[Dict, Tuple[List[CustomerSchema], int]]: """Get paginated list of customers using QueryPaginator.""" try: # Base query stmt = select(Customer) # Filter out deleted records if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) # Apply search filter if provided if filters.search: stmt = _apply_search_filter(stmt, filters.search) # Apply boolean filters with mapping boolean_filters = { filters.is_vendor: (Customer.is_vendor, "vendor"), filters.is_active: (Customer.is_active, "active") } for filter_value, (field, field_name) in boolean_filters.items(): if filter_value == -1: stmt = stmt.where(field == False) elif filter_value == 1: stmt = stmt.where(field == True) # filter_value == 0 or None means show all, so no filter needed # Apply customer_type filter if filters.customer_type: stmt = stmt.where(Customer.customer_type == filters.customer_type) # Apply account_type filter if filters.account_type: stmt = stmt.where(Customer.account_type == filters.account_type) # Apply industry filter if filters.industry: stmt = stmt.where(Customer.industry.ilike(f"%{filters.industry}%")) # Apply name filter if filters.name: name_term = f"%{filters.name}%" stmt = stmt.where( or_( Customer.first_name.ilike(name_term), Customer.last_name.ilike(name_term) ) ) # Apply email filter if filters.email: stmt = stmt.where(Customer.email.ilike(f"%{filters.email}%")) # Apply phone filter if filters.phone: stmt = stmt.where(Customer.phone.ilike(f"%{filters.phone}%")) # Apply address filter (searches in related addresses) if filters.address: address_term = f"%{filters.address}%" stmt = stmt.join(Customer.addresses).where( or_( Address.address_line_1.ilike(address_term), Address.address_line_2.ilike(address_term), Address.city.ilike(address_term), Address.state.ilike(address_term) ) ).distinct() # Apply zip_code filter (searches in related addresses) if filters.zip_code: zip_code_term = f"%{filters.zip_code}%" if filters.address: # If address filter is already applied, add to existing join stmt = stmt.where(Address.zipcode.ilike(zip_code_term)) else: # Create new join for zip_code only stmt = stmt.join(Customer.addresses).where( Address.zipcode.ilike(zip_code_term) ).distinct() # Apply selective loading based on fields parameter if filters.fields: stmt = _apply_selective_loading(stmt, filters.fields) # Apply default sorting (created_at descending) stmt = stmt.order_by(desc(Customer.created_at)) if paginate: # Use QueryPaginator for pagination paginator = QueryPaginator( query=stmt, schema=CustomerListSchema, db=db, model=Customer, url="/api/v1/customers", offset=offset, limit=limit, use_orm=True, ) pagination_result = paginator.paginate() return { "data": pagination_result.result, "total": pagination_result.total, "page": pagination_result.page, "per_page": pagination_result.per_page, "next": pagination_result.next, "previous": pagination_result.previous, "first": pagination_result.first, "last": pagination_result.last, } else: # Return raw results for non-paginated queries results = db.execute(stmt).scalars().all() customers_list = [CustomerSchema.model_validate(result) for result in results] return customers_list, len(customers_list) except Exception as e: logger.error(f"Error getting customers list: {e}") if paginate: return { "data": [], "total": 0, "page": 1, "per_page": limit, "next": None, "previous": None, "first": None, "last": None, } else: return [], 0 async def get_customer_by_email(db: Session, email: str) -> CustomerSchema: """Get customer by email using proper ORM.""" try: stmt = select(Customer).where(Customer.email == email) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() if not result: raise APIException( message="Customer not found", status_code=404, module="customers.services" ) return CustomerSchema.model_validate(result) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting customer by email: {e}") raise APIException( message=f"Failed to retrieve customer by email: {str(e)}", status_code=500, module="customers.services" ) async def activate_customer(db: Session, customer_id: str) -> bool: """Activate a customer by customer_id (alphanumeric unique identifier). Can restore soft-deleted customers.""" try: # Query by customer_id field (alphanumeric) - include soft-deleted customers for activation stmt = select(Customer).where(Customer.customer_id == customer_id) customer = db.execute(stmt).scalar_one_or_none() if not customer: return False customer.is_active = True customer.updated_at = datetime.utcnow() # If customer was soft-deleted, restore them by clearing deleted_at if hasattr(customer, 'deleted_at') and customer.deleted_at is not None: customer.deleted_at = None db.commit() return True except Exception as e: logger.error(f"Error activating customer: {e}") return False async def deactivate_customer(db: Session, customer_id: str) -> bool: """Deactivate a customer by customer_id (alphanumeric unique identifier).""" try: # Query by customer_id field (alphanumeric) instead of id field (integer) stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(stmt).scalar_one_or_none() if not customer: return False customer.is_active = False customer.updated_at = datetime.utcnow() db.commit() return True except Exception as e: logger.error(f"Error deactivating customer: {e}") return False async def update_customer(db: Session, customer_id: str, customer_data: CustomerUpdateRequestSchema) -> CustomerSchema: """Update customer by ID using SQLAlchemy ORM.""" try: # Query by customer_id field (alphanumeric) instead of id field (integer) stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.services" ) # Update fields using model_dump to get only set fields (per PR feedback) update_data = customer_data.model_dump(exclude_unset=True) # Handle tags conversion from list to comma-separated string if 'tags' in update_data and update_data['tags'] is not None: if isinstance(update_data['tags'], list): update_data['tags'] = ",".join(update_data['tags']) for field, value in update_data.items(): if hasattr(customer, field): setattr(customer, field, value) customer.updated_at = datetime.utcnow() db.commit() db.refresh(customer) # Return CustomerSchema using model_validate as per PR feedback return CustomerSchema.model_validate(customer) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error updating customer: {e}") raise APIException( message=f"Failed to update customer: {str(e)}", status_code=500, module="customers.services" ) async def delete_customer(db: Session, customer_id: str) -> bool: """Soft delete customer by customer_id (alphanumeric unique identifier).""" try: # Query by customer_id field (alphanumeric) instead of id field (integer) stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(stmt).scalar_one_or_none() if not customer: return False customer.deleted_at = datetime.utcnow() customer.updated_at = datetime.utcnow() db.commit() return True except Exception as e: logger.error(f"Error deleting customer: {e}") return False async def get_customer_by_merchant(db: Session, customer_id: str, merchant_id: int) -> Optional[CustomerSchema]: """ Get customer by customer_id and merchant_id. Args: db: Database session customer_id: Customer account literal identifier merchant_id: Merchant ID to filter by Returns: CustomerSchema: Customer data if found, None otherwise Raises: APIException: If database error occurs """ try: stmt = select(Customer).where( and_( Customer.customer_id == customer_id, Customer.merchant_id == merchant_id ) ) # Filter out soft-deleted records if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() if result: return CustomerSchema.model_validate(result) return None except Exception as e: logger.error(f"Error getting customer by account_literal: {e}") raise APIException( message=f"Failed to retrieve customer by account_literal: {str(e)}", status_code=500, module="customers.services" ) async def check_customer_id_exists(db: Session, customer_id: str) -> bool: """Check if a customer ID already exists in the database.""" try: stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() return result is not None except Exception as e: logger.error(f"Error checking customer ID existence: {e}") return True # Return True to be safe and avoid conflicts async def get_customer_contact( db: Session, contact_id:str, customer_id:Optional[int]=None ) -> Optional[CustomerContact]: """Get customer contact by contact_id and optional customer_id.""" try: stmt = select(CustomerContact).where(CustomerContact.contact_id == contact_id) if customer_id: stmt = stmt.where(CustomerContact.customer_id == customer_id) if hasattr(CustomerContact, 'deleted_at'): stmt = stmt.where(CustomerContact.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() return result except Exception as e: logger.error(f"Error getting customer contact: {e}") return None async def check_account_literal_exists(db: Session, account_literal: str) -> bool: """Check if an account literal already exists in the database.""" try: stmt = select(Customer).where(Customer.account_literal == account_literal) if hasattr(Customer, 'deleted_at'): stmt = stmt.where(Customer.deleted_at.is_(None)) result = db.execute(stmt).scalar_one_or_none() return result is not None except Exception as e: logger.error(f"Error checking account literal existence: {e}") return True # Return True to be safe and avoid conflicts async def get_customer_activities(db: Session, customer_id: str) -> Dict: """Get customer activities - currently returns static data.""" # Verify customer exists first await get_customer(db, customer_id) # Return static activity data return { "data": [ { "level": "info", "event": "EVENTS.CUSTOMER.CONTACT.CREATE", "message": "Contact Created", "module": "customer", "submodule": "contact", "data": { "name": "Garry Porter", "designation": "CEO", "relation": None, "email": "garryporter375@yopmail.com", "phone": "+12565537655", "office_phone": None, "tilled_id": None, "is_active": True, "dob": "1992-01-10T18:30:00", "website": "https://www.dreamztech.com", "timezone": "America/Denver", "id": 217, "contact_id": "con_HyGutVzqME42nXos", "created_at": "2025-04-04T12:33:05.705000", "updated_at": None, "deleted_at": None, "avatar": None, "addresses": [], }, "created_by": None, "created_by_id": None, "merchant_id": "mer_DkhsfjDM28Kf7yPu", "customer_id": customer_id, "contact_id": None, "receipt_id": None, "invoice_id": None, "payment_request_id": None, "product_slug": None, "authorization_type": None, "authorization_id": None, "ip_address": None, "phone": None, "contract_id": None, "checkout_id": None, "created_at": "2025-04-04T12:33:05.843000" }, { "level": "info", "event": "EVENTS.CUSTOMER.CREATE", "message": "New customer created", "module": "customer", "submodule": "account", "data": { "uin": "GBTF8E3339", "customer_type": "False", "account_type": None, "industry": None, "office_phone": None, "account_expires_on": "2030-04-03T08:12:30.265000", "timezone": "America/Denver", "website": "https://www.dreamztech.com", "account_tax_id": None, "account_literal": "592970", "id": 162, "customer_id": customer_id, "tilled_id": None, "is_active": True, "created_at": "2025-04-04T12:33:05.480000", "updated_at": None, "deleted_at": None, "merchant": None, "avatar": None, "addresses": None, "tags_list": [], "contacts": None, "is_vendor": None, "is_registered": False, "user_account": None, "default_address": None, "billing_address": None, "shipping_address": None, "attachments": None, "first_name": "GBT Corp.", "middle_name": None, "last_name": None, "email": "garryporter375@yopmail.com", "phone": "+12565537655", "business_legal_name": "GBT Corporation", "last_active_date": None }, "created_by": None, "created_by_id": None, "merchant_id": "mer_DkhsfjDM28Kf7yPu", "customer_id": customer_id, "contact_id": None, "receipt_id": None, "invoice_id": None, "payment_request_id": None, "product_slug": None, "authorization_type": None, "authorization_id": None, "ip_address": None, "phone": None, "contract_id": None, "checkout_id": None, "created_at": "2025-04-04T12:33:05.596000" } ], "status_code": 200, "success": True, "message": "Activities fetched successfully" } async def get_customer_summary(db: Session, is_vendor: bool = False) -> Dict: """Get customer summary - currently returns static data.""" # Return static summary data for now return { "data": { "top_customers": [ { "id": 201, "first_name": "IBM", "last_name": None, "total_amount": 1859900.0 }, { "id": 165, "first_name": "Mason", "last_name": None, "total_amount": 264328.0 }, { "id": 119, "first_name": "RITZ Solution", "last_name": None, "total_amount": 112500.0 } ], "lowest_customers": [ { "id": 168, "first_name": "Holigan", "last_name": None, "total_amount": 8700.0 }, { "id": 119, "first_name": "RITZ Solution", "last_name": None, "total_amount": 112500.0 }, { "id": 165, "first_name": "Mason", "last_name": None, "total_amount": 264328.0 } ], "most_frequent_customers": [ { "id": 201, "first_name": "IBM", "last_name": None, "total_amount": 1859900.0 }, { "id": 165, "first_name": "Mason", "last_name": None, "total_amount": 264328.0 }, { "id": 119, "first_name": "RITZ Solution", "last_name": None, "total_amount": 112500.0 } ] }, "status_code": 200, "success": True, "message": "Customer Summary details" } async def bulk_customer_actions(db: Session, action: str, customer_ids: List[str]) -> Dict[str, Any]: """Perform bulk actions on multiple customers. Limited to 10 customers per request.""" try: # Enforce upper limit of 10 customers per bulk request if len(customer_ids) > 10: raise APIException( message="Bulk action limited to maximum 10 customers per request", status_code=400, module="customers.services" ) results = { "success": [], "failed": [], "total_processed": len(customer_ids), "success_count": 0, "failed_count": 0 } for customer_id in customer_ids: try: # Check if customer exists (including soft-deleted customers) stmt = select(Customer).where(Customer.customer_id == customer_id) customer = db.execute(stmt).scalar_one_or_none() if not customer: results["failed"].append({ "customer_id": customer_id, "error": "Customer not found" }) results["failed_count"] += 1 continue # Check if customer is soft-deleted and handle based on action if hasattr(customer, 'deleted_at') and customer.deleted_at is not None: if action == "activate": # Allow activation of soft-deleted customers (restore them) pass elif action == "deactivate": # Skip deactivation of already deleted customers results["failed"].append({ "customer_id": customer_id, "error": "Customer is already deleted" }) results["failed_count"] += 1 continue elif action == "delete": # Allow re-deletion (will update the timestamp) pass # Perform the action if action == "activate": success = await activate_customer(db, customer_id) action_verb = "activated" elif action == "deactivate": success = await deactivate_customer(db, customer_id) action_verb = "deactivated" elif action == "delete": success = await delete_customer(db, customer_id) action_verb = "deleted" else: results["failed"].append({ "customer_id": customer_id, "error": f"Invalid action: {action}" }) results["failed_count"] += 1 continue if success: results["success"].append({ "customer_id": customer_id, "message": f"Customer {action_verb} successfully" }) results["success_count"] += 1 else: results["failed"].append({ "customer_id": customer_id, "error": f"Failed to {action} customer" }) results["failed_count"] += 1 except Exception as e: logger.error(f"Error processing customer {customer_id}: {e}") results["failed"].append({ "customer_id": customer_id, "error": str(e) }) results["failed_count"] += 1 return results except Exception as e: logger.error(f"Error in bulk customer actions: {e}") raise APIException( message=f"Failed to perform bulk actions: {str(e)}", status_code=500, module="customers.services" )