""" Contact Services for customer contact management. """ from typing import Dict, List, Optional, Tuple, Union from datetime import datetime, timezone 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_contact import CustomerContact from src.apps.customers.models.customer import Customer from src.apps.customers.schemas.customer_contact import ( CustomerContactSchema, CustomerContactCreateRequestSchema, CustomerContactUpdateRequestSchema ) from src.helpers.pagination import QueryPaginator from src.core.exceptions import APIException from src.core.functions import generate_id logger = logging.getLogger(__name__) async def create_customer_contact( db: Session, customer_id: str, contact_data: CustomerContactCreateRequestSchema ) -> CustomerContactSchema: """Create a new contact for a customer.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.contact_services" ) # Generate contact_id using project's standard helper function 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 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") raise APIException( message="Failed to generate unique contact ID", status_code=500, module="customers.contact_services" ) # Convert schema to dict, excluding fields that don't exist in CustomerContact model contact_dict = contact_data.model_dump(exclude={'address', 'website'}) # Add customer relationship fields contact_dict["contact_id"] = contact_id contact_dict["customer_id"] = customer.id # user_account_id is now optional - set to None if not provided contact_dict["user_account_id"] = None # Set defaults for optional fields if not provided if "is_active" not in contact_dict or contact_dict["is_active"] is None: contact_dict["is_active"] = True # Create contact object contact_obj = CustomerContact(**contact_dict) db.add(contact_obj) db.commit() db.refresh(contact_obj) logger.info(f"Contact created for customer {customer_id} with ID: {contact_obj.contact_id}") return CustomerContactSchema.model_validate(contact_obj) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error creating contact for customer {customer_id}: {e}") db.rollback() raise APIException( message=f"Failed to create contact: {str(e)}", status_code=500, module="customers.contact_services" ) async def get_customer_contacts( db: Session, customer_id: str, limit: int = 10, offset: int = 0, is_active: Optional[bool] = None, search: Optional[str] = None, title: Optional[str] = None, relation: Optional[str] = None ) -> Dict: """Get paginated list of customer contacts.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.contact_services" ) # Base query stmt = select(CustomerContact).where(CustomerContact.customer_id == customer.id) # Filter out deleted records if hasattr(CustomerContact, 'deleted_at'): stmt = stmt.where(CustomerContact.deleted_at.is_(None)) # Apply filters if is_active is not None: stmt = stmt.where(CustomerContact.is_active == is_active) if title: stmt = stmt.where(CustomerContact.title.ilike(f"%{title}%")) if relation: stmt = stmt.where(CustomerContact.relation.ilike(f"%{relation}%")) # Apply search filter if search: search_term = f"%{search}%" stmt = stmt.where( or_( CustomerContact.first_name.ilike(search_term), CustomerContact.last_name.ilike(search_term), CustomerContact.email.ilike(search_term), CustomerContact.title.ilike(search_term), CustomerContact.phone.ilike(search_term), CustomerContact.account_phone.ilike(search_term) ) ) # Apply default sorting (created_at descending) stmt = stmt.order_by(desc(CustomerContact.created_at)) # Use QueryPaginator for pagination paginator = QueryPaginator( query=stmt, schema=CustomerContactSchema, db=db, model=CustomerContact, url=f"/api/v1/customers/{customer_id}/contacts", 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, } except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting contacts for customer {customer_id}: {e}") raise APIException( message=f"Failed to get customer contacts: {str(e)}", status_code=500, module="customers.contact_services" ) async def get_customer_contact( db: Session, customer_id: str, contact_id: str ) -> CustomerContactSchema: """Get a specific contact by contact_id for a customer.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.contact_services" ) # Query for specific contact stmt = select(CustomerContact).where( and_( CustomerContact.contact_id == contact_id, CustomerContact.customer_id == customer.id ) ) if hasattr(CustomerContact, 'deleted_at'): stmt = stmt.where(CustomerContact.deleted_at.is_(None)) contact = db.execute(stmt).scalar_one_or_none() if not contact: raise APIException( message="Contact not found", status_code=404, module="customers.contact_services" ) return CustomerContactSchema.model_validate(contact) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error getting contact {contact_id} for customer {customer_id}: {e}") raise APIException( message=f"Failed to get contact: {str(e)}", status_code=500, module="customers.contact_services" ) async def update_customer_contact( db: Session, customer_id: str, contact_id: str, contact_data: CustomerContactUpdateRequestSchema ) -> CustomerContactSchema: """Update a customer contact.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.contact_services" ) # Query for specific contact stmt = select(CustomerContact).where( and_( CustomerContact.contact_id == contact_id, CustomerContact.customer_id == customer.id ) ) if hasattr(CustomerContact, 'deleted_at'): stmt = stmt.where(CustomerContact.deleted_at.is_(None)) contact = db.execute(stmt).scalar_one_or_none() if not contact: raise APIException( message="Contact not found", status_code=404, module="customers.contact_services" ) # Update contact fields using model_dump to get only set fields update_data = contact_data.model_dump(exclude_unset=True, exclude={'address', 'website'}) for field, value in update_data.items(): if hasattr(contact, field): setattr(contact, field, value) contact.updated_at = datetime.now(timezone.utc) db.commit() db.refresh(contact) logger.info(f"Contact {contact_id} updated for customer {customer_id}") return CustomerContactSchema.model_validate(contact) except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error updating contact {contact_id} for customer {customer_id}: {e}") db.rollback() raise APIException( message=f"Failed to update contact: {str(e)}", status_code=500, module="customers.contact_services" ) async def delete_customer_contact( db: Session, customer_id: str, contact_id: str ) -> bool: """Soft delete a customer contact.""" try: # Validate customer exists customer_stmt = select(Customer).where(Customer.customer_id == customer_id) if hasattr(Customer, 'deleted_at'): customer_stmt = customer_stmt.where(Customer.deleted_at.is_(None)) customer = db.execute(customer_stmt).scalar_one_or_none() if not customer: raise APIException( message="Customer not found", status_code=404, module="customers.contact_services" ) # Query for specific contact stmt = select(CustomerContact).where( and_( CustomerContact.contact_id == contact_id, CustomerContact.customer_id == customer.id ) ) if hasattr(CustomerContact, 'deleted_at'): stmt = stmt.where(CustomerContact.deleted_at.is_(None)) contact = db.execute(stmt).scalar_one_or_none() if not contact: raise APIException( message="Contact not found", status_code=404, module="customers.contact_services" ) # Soft delete the contact if hasattr(contact, 'deleted_at'): contact.deleted_at = datetime.now(timezone.utc) else: # If no soft delete support, mark as inactive contact.is_active = False contact.updated_at = datetime.now(timezone.utc) db.commit() logger.info(f"Contact {contact_id} deleted for customer {customer_id}") return True except APIException: raise # Re-raise APIException as-is except Exception as e: logger.error(f"Error deleting contact {contact_id} for customer {customer_id}: {e}") db.rollback() raise APIException( message=f"Failed to delete contact: {str(e)}", status_code=500, module="customers.contact_services" )