Spaces:
Sleeping
Sleeping
| """ | |
| Customer Service - Core business logic for customer management | |
| Handles customer CRUD operations with phone deduplication and authorization | |
| """ | |
| import logging | |
| from typing import Optional, Dict, Any, List, Tuple | |
| from datetime import datetime | |
| from sqlalchemy.orm import Session, joinedload | |
| from sqlalchemy import or_, and_, func | |
| from fastapi import HTTPException, status | |
| from uuid import UUID | |
| from app.models.customer import Customer | |
| from app.models.client import Client | |
| from app.models.project import ProjectRegion | |
| from app.models.user import User | |
| from app.models.enums import AppRole | |
| from app.schemas.customer import CustomerCreate, CustomerUpdate | |
| logger = logging.getLogger(__name__) | |
| class CustomerService: | |
| """Service for managing customers with authorization and deduplication""" | |
| # ============================================ | |
| # AUTHORIZATION HELPERS | |
| # ============================================ | |
| def can_user_create_customer(user: User, client_id: UUID) -> bool: | |
| """Check if user can create a customer for this client""" | |
| if user.role == AppRole.PLATFORM_ADMIN: | |
| return True | |
| if user.role == AppRole.CLIENT_ADMIN and user.client_id == client_id: | |
| return True | |
| return False | |
| def can_user_view_customer(user: User, customer: Customer) -> bool: | |
| """Check if user can view this customer""" | |
| if user.role == AppRole.PLATFORM_ADMIN: | |
| return True | |
| # Client admins can view their own customers | |
| if user.role == AppRole.CLIENT_ADMIN and user.client_id == customer.client_id: | |
| return True | |
| # Contractor users can view customers from projects they're working on | |
| if user.contractor_id: | |
| # Check if user's contractor has any projects with this client | |
| # (This would require checking project_team membership) | |
| # For now, allow if they share a client through projects | |
| return True | |
| return False | |
| def can_user_edit_customer(user: User, customer: Customer) -> bool: | |
| """Check if user can edit this customer""" | |
| if user.role == AppRole.PLATFORM_ADMIN: | |
| return True | |
| if user.role == AppRole.CLIENT_ADMIN and user.client_id == customer.client_id: | |
| return True | |
| return False | |
| # ============================================ | |
| # CUSTOMER CRUD | |
| # ============================================ | |
| def create_customer( | |
| db: Session, | |
| data: CustomerCreate, | |
| current_user: User | |
| ) -> Customer: | |
| """ | |
| Create a new customer with phone deduplication | |
| Business Rules: | |
| - Only client_admin and platform_admin can create customers | |
| - Client must exist and be active | |
| - phone_primary must be unique (deduplication) | |
| - If coordinates provided, both lat and lon required | |
| - If project_region_id provided, must exist | |
| Authorization: | |
| - platform_admin: Can create for any client | |
| - client_admin: Can only create for their own client | |
| """ | |
| # Authorization check | |
| if not CustomerService.can_user_create_customer(current_user, data.client_id): | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="You don't have permission to create customers for this client" | |
| ) | |
| # Validate client exists and is active | |
| client = db.query(Client).filter( | |
| Client.id == data.client_id, | |
| Client.deleted_at == None | |
| ).first() | |
| if not client: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Client with ID {data.client_id} not found" | |
| ) | |
| if not client.is_active: | |
| raise HTTPException( | |
| status_code=status.HTTP_400_BAD_REQUEST, | |
| detail="Client is not active" | |
| ) | |
| # Check for duplicate phone_primary (unique constraint) | |
| existing = db.query(Customer).filter( | |
| Customer.phone_primary == data.phone_primary, | |
| Customer.deleted_at == None | |
| ).first() | |
| if existing: | |
| raise HTTPException( | |
| status_code=status.HTTP_409_CONFLICT, | |
| detail=f"Customer with phone number {data.phone_primary} already exists (ID: {existing.id})" | |
| ) | |
| # Validate project_region if provided | |
| if data.project_region_id: | |
| region = db.query(ProjectRegion).filter( | |
| ProjectRegion.id == data.project_region_id, | |
| ProjectRegion.deleted_at == None | |
| ).first() | |
| if not region: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Project region with ID {data.project_region_id} not found" | |
| ) | |
| # Create customer | |
| customer = Customer(**data.model_dump()) | |
| db.add(customer) | |
| db.commit() | |
| db.refresh(customer) | |
| logger.info(f"Customer created: {customer.id} - {customer.customer_name} by user {current_user.id}") | |
| return customer | |
| def list_customers( | |
| db: Session, | |
| current_user: User, | |
| skip: int = 0, | |
| limit: int = 50, | |
| client_id: Optional[UUID] = None, | |
| project_region_id: Optional[UUID] = None, | |
| is_active: Optional[bool] = None, | |
| search: Optional[str] = None | |
| ) -> Tuple[List[Customer], int]: | |
| """ | |
| List customers with authorization filtering | |
| Filters: | |
| - client_id: Filter by specific client | |
| - project_region_id: Filter by region | |
| - is_active: Filter by active status | |
| - search: Search by name, phone, or email | |
| Authorization: | |
| - platform_admin: Can see all customers | |
| - client_admin: Can only see their own client's customers | |
| - Others: Can see customers from projects they're involved in | |
| """ | |
| query = db.query(Customer).filter(Customer.deleted_at == None) | |
| # Authorization filtering | |
| if current_user.role == AppRole.PLATFORM_ADMIN: | |
| # Platform admins see all | |
| pass | |
| elif current_user.role == AppRole.CLIENT_ADMIN: | |
| # Client admins only see their client's customers | |
| query = query.filter(Customer.client_id == current_user.client_id) | |
| else: | |
| # For other users (contractors), filter by projects they're involved in | |
| # This requires joining through projects | |
| if current_user.client_id: | |
| query = query.filter(Customer.client_id == current_user.client_id) | |
| else: | |
| # Contractor users - for now, allow all (can be refined later) | |
| pass | |
| # Apply filters | |
| if client_id: | |
| query = query.filter(Customer.client_id == client_id) | |
| if project_region_id: | |
| query = query.filter(Customer.project_region_id == project_region_id) | |
| if is_active is not None: | |
| query = query.filter(Customer.is_active == is_active) | |
| if search: | |
| search_filter = or_( | |
| Customer.customer_name.ilike(f"%{search}%"), | |
| Customer.phone_primary.ilike(f"%{search}%"), | |
| Customer.phone_alternative.ilike(f"%{search}%"), | |
| Customer.email.ilike(f"%{search}%"), | |
| Customer.id_number.ilike(f"%{search}%") | |
| ) | |
| query = query.filter(search_filter) | |
| # Get total count | |
| total = query.count() | |
| # Get paginated results with relationships | |
| customers = query.options( | |
| joinedload(Customer.client), | |
| joinedload(Customer.project_region) | |
| ).order_by(Customer.created_at.desc()).offset(skip).limit(limit).all() | |
| return customers, total | |
| def get_customer_by_id( | |
| db: Session, | |
| customer_id: UUID, | |
| current_user: User | |
| ) -> Customer: | |
| """Get a single customer by ID with authorization check""" | |
| customer = db.query(Customer).options( | |
| joinedload(Customer.client), | |
| joinedload(Customer.project_region) | |
| ).filter( | |
| Customer.id == customer_id, | |
| Customer.deleted_at == None | |
| ).first() | |
| if not customer: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Customer with ID {customer_id} not found" | |
| ) | |
| # Authorization check | |
| if not CustomerService.can_user_view_customer(current_user, customer): | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="You don't have permission to view this customer" | |
| ) | |
| return customer | |
| def get_customer_by_phone( | |
| db: Session, | |
| phone: str, | |
| current_user: User | |
| ) -> Optional[Customer]: | |
| """ | |
| Find customer by phone number (primary or alternative) | |
| Used for deduplication during order import | |
| """ | |
| customer = db.query(Customer).filter( | |
| or_( | |
| Customer.phone_primary == phone, | |
| Customer.phone_alternative == phone | |
| ), | |
| Customer.deleted_at == None | |
| ).first() | |
| if customer and not CustomerService.can_user_view_customer(current_user, customer): | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="You don't have permission to view this customer" | |
| ) | |
| return customer | |
| def update_customer( | |
| db: Session, | |
| customer_id: UUID, | |
| data: CustomerUpdate, | |
| current_user: User | |
| ) -> Customer: | |
| """ | |
| Update an existing customer | |
| Business Rules: | |
| - Cannot change phone_primary to one that already exists | |
| - If coordinates updated, both lat and lon required | |
| """ | |
| # Get existing customer | |
| customer = db.query(Customer).filter( | |
| Customer.id == customer_id, | |
| Customer.deleted_at == None | |
| ).first() | |
| if not customer: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Customer with ID {customer_id} not found" | |
| ) | |
| # Authorization check | |
| if not CustomerService.can_user_edit_customer(current_user, customer): | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="You don't have permission to edit this customer" | |
| ) | |
| # Check for duplicate phone if changing phone_primary | |
| if data.phone_primary and data.phone_primary != customer.phone_primary: | |
| existing = db.query(Customer).filter( | |
| Customer.phone_primary == data.phone_primary, | |
| Customer.deleted_at == None, | |
| Customer.id != customer_id | |
| ).first() | |
| if existing: | |
| raise HTTPException( | |
| status_code=status.HTTP_409_CONFLICT, | |
| detail=f"Customer with phone number {data.phone_primary} already exists" | |
| ) | |
| # Validate project_region if being updated | |
| if data.project_region_id: | |
| region = db.query(ProjectRegion).filter( | |
| ProjectRegion.id == data.project_region_id, | |
| ProjectRegion.deleted_at == None | |
| ).first() | |
| if not region: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Project region with ID {data.project_region_id} not found" | |
| ) | |
| # Update fields | |
| update_data = data.model_dump(exclude_unset=True) | |
| for field, value in update_data.items(): | |
| setattr(customer, field, value) | |
| customer.updated_at = datetime.utcnow() | |
| db.commit() | |
| db.refresh(customer) | |
| logger.info(f"Customer updated: {customer.id} by user {current_user.id}") | |
| return customer | |
| def delete_customer( | |
| db: Session, | |
| customer_id: UUID, | |
| current_user: User | |
| ) -> None: | |
| """ | |
| Soft delete a customer (platform_admin only) | |
| Business Rules: | |
| - Only platform_admin can delete customers | |
| - Performs soft delete (sets deleted_at) | |
| """ | |
| if current_user.role != AppRole.PLATFORM_ADMIN: | |
| raise HTTPException( | |
| status_code=status.HTTP_403_FORBIDDEN, | |
| detail="Only platform administrators can delete customers" | |
| ) | |
| customer = db.query(Customer).filter( | |
| Customer.id == customer_id, | |
| Customer.deleted_at == None | |
| ).first() | |
| if not customer: | |
| raise HTTPException( | |
| status_code=status.HTTP_404_NOT_FOUND, | |
| detail=f"Customer with ID {customer_id} not found" | |
| ) | |
| # Soft delete | |
| customer.deleted_at = datetime.utcnow() | |
| db.commit() | |
| logger.info(f"Customer deleted: {customer.id} by user {current_user.id}") | |