from datetime import datetime, timedelta from typing import Optional, List from fastapi.param_functions import Query from pydantic.networks import EmailStr from pydantic import Field, BaseModel, field_validator, model_validator, ConfigDict from src.apps.users.schemas.user_common import UserBase from src.core.utils.enums import PaymentMethodTypes, AccountsBulkActions from src.apps.payment_requests.enums import PaymentAuthorizationTypes from src.apps.base.utils import regexp from src.apps.base.schemas.common import AddressCreateRequestSchema from src.apps.files.schemas.file_requests import FileCreate from src.apps.customers.schemas.customer_contact import CustomerContactCreateRequestSchema from src.apps.customers.schemas.customer_common import CustomerBase from sqlalchemy import select, and_ from src.core.database import SessionLocal from src.apps.customers.models.customer import Customer from src.apps.merchants.models.merchant import Merchant class CustomerOwnerCreateRequestSchema(UserBase): model_config = ConfigDict( from_attributes=True, json_schema_extra={ "example": { "email": "johndoe@example.com", "password": "weakpassword", "phone": "+9176787678", "first_name": "John", "last_name": "Doe", } } ) email: EmailStr = Field(description="A valid email address") phone: str = Field(description="A valid phone number") is_active: Optional[bool] = Field( default=None, description="Activation status of user, false denotes that the user is inactive" ) is_verified: Optional[bool] = Field( default=None, description="Verification status of user, false denotes that the user is unverified" ) class CustomerCreateRequestSchema(CustomerBase): model_config = ConfigDict( from_attributes=True, json_schema_extra={ "example": { "first_name": "John", "middle_name": "Michael", "last_name": "Doe", "email": "john.doe@example.com", "phone": "+1234567890", "office_phone": "+1987654321", "business_legal_name": "Doe Industries LLC", "customer_type": "Business Customer", "is_vendor": False, "role_type": "Primary Contact", "timezone": "America/New_York", "is_active": True, "tags": ["premium", "b2b"], "merchant_id": 1, "user_account_id": 123, "account_literal": "ACC123456789012", "address": { "address_line_1": "123 Business Avenue", "address_line_2": "Suite 400", "zipcode": "10001", "city": "New York", "state": "NY", "country": "US", "address_type": "billing", "use_as_default": True }, "contact": { "first_name": "Jane", "last_name": "Smith", "email": "jane.smith@example.com", "account_phone": "+1555123456", "title": "Account Manager", "relation": "primary" } } } ) contact: Optional[CustomerContactCreateRequestSchema] = Field( default=None, description="A new contact for this customer that needs to be created" ) contacts: Optional[List[CustomerContactCreateRequestSchema]] = Field( default=None, description="Multiple contacts for this customer" ) address: AddressCreateRequestSchema = Field( description="Customer's billing address (required)" ) avatar: Optional[int] = Field( default=None, description="Uploaded image file id to be used as avatar" ) account_expires_on: datetime = Field(default_factory=lambda: datetime.now() + timedelta(days=(5 * 365))) email: Optional[EmailStr] = Field(default=None, description="A valid email address") phone: Optional[str] = Field(default=None, description="A valid phone number") first_name: Optional[str] = Field(default=None, description="First name of the customer") middle_name: Optional[str] = Field(default=None, description="Middle name of the customer") last_name: Optional[str] = Field(default=None, description="Last name of the customer") office_phone: Optional[str] = Field(default=None, description="Office Phone of the customer") business_legal_name: Optional[str] = Field(default=None, description="Legal name of the business") account_literal: Optional[str] = Field( default=None, description="User-provided account literal (15 characters). If not provided, system will auto-generate one. This is the public-facing account number.", min_length=15, max_length=15 ) role_type: Optional[str] = Field(default=None, description="Role type of the Customer") customer_type: str = Field( description="Customer type: any string value to categorize the customer" ) is_vendor: Optional[bool] = Field( default=False, description="Vendor type of the customer (auto-set based on account_type)" ) timezone: Optional[str] = Field(default=None, description="A valid timezone") tags: Optional[List[str]] = Field(default=None, description="Tags of customer") merchant_id: Optional[int] = Field(default=None, description="ID of the merchant") user_account_id: Optional[int] = Field(default=None, description="ID of the user account") is_active: Optional[bool] = Field( default=True, description="Activation status of customer, false denotes that the customer is inactive" ) @field_validator("account_literal") @classmethod def validate_account_literal_format(cls, v): """Validate account literal format if provided.""" if v is not None: if not v.isalnum(): raise ValueError("Account literal must contain only alphanumeric characters") if len(v) != 15: raise ValueError("Account literal must be exactly 15 characters long") return v @model_validator(mode='after') def validate_customer_data(self): """Validate customer data with database checks.""" with SessionLocal() as db: # Validate email uniqueness if provided if self.email: email_stmt = select(Customer).where( and_( Customer.email == self.email, Customer.deleted_at.is_(None) if hasattr(Customer, 'deleted_at') else True ) ) if db.execute(email_stmt).scalar_one_or_none(): raise ValueError("Customer with this email already exists") # Validate merchant exists if provided if self.merchant_id: merchant_stmt = select(Merchant).where(Merchant.id == self.merchant_id) if not db.execute(merchant_stmt).scalar_one_or_none(): raise ValueError(f"Merchant with ID {self.merchant_id} does not exist") # Validate user account exists if provided if self.user_account_id: # Lazy import to avoid circular dependency from src.apps.users.models.user import User user_stmt = select(User).where(User.id == self.user_account_id) if not db.execute(user_stmt).scalar_one_or_none(): raise ValueError(f"User account with ID {self.user_account_id} does not exist") # Validate account_literal uniqueness if provided if self.account_literal: literal_stmt = select(Customer).where(Customer.account_literal == self.account_literal) if db.execute(literal_stmt).scalar_one_or_none(): raise ValueError( f"Account literal '{self.account_literal}' already exists. " "Please choose a different account literal or use the /customers/account-suggestions endpoint for suggestions." ) return self class CustomerUpdateRequestSchema(CustomerBase): model_config = ConfigDict( from_attributes=True, json_schema_extra={ "example": { "first_name": "Updated John", "middle_name": "Updated Michael", "last_name": "Updated Doe", "email": "updated.john.doe@example.com", "phone": "+1234567891", "office_phone": "+1987654322", "business_legal_name": "Updated Doe Industries LLC", "customer_type": "Updated Business Type", "is_vendor": True, "role_type": "Updated Role", "timezone": "America/Los_Angeles", "is_active": False, "tags": ["updated", "premium", "enterprise"], "merchant_id": 2, "user_account_id": 456, "account_literal": "UPD123456789012", "address": { "address_line_1": "456 Updated Street", "address_line_2": "Floor 5", "zipcode": "90210", "city": "Los Angeles", "state": "CA", "country": "US", "address_type": "shipping", "use_as_default": False }, "contact": { "first_name": "Updated Jane", "last_name": "Updated Smith", "email": "updated.jane@example.com", "account_phone": "+1555654321", "title": "Senior Manager", "relation": "secondary" } } } ) first_name: Optional[str] = Field(default=None, description="First name of the customer") middle_name: Optional[str] = Field(default=None, description="Middle name of the customer") last_name: Optional[str] = Field(default=None, description="Last name of the customer") email: Optional[str] = Field(default=None, description="Email of the customer") phone: Optional[str] = Field(default=None, description="Phone of the customer") office_phone: Optional[str] = Field(default=None, description="Office Phone of the customer") business_legal_name: Optional[str] = Field(default=None, description="Legal name of the business") is_active: Optional[bool] = Field( default=None, description="Activation status of Customer, false denotes that the Customer is inactive" ) customer_type: Optional[str] = Field( default=None, description="Customer type: any string value to categorize the customer" ) is_vendor: Optional[bool] = Field( default=False, description="Boolean flag indicating if this entity is a vendor (independent of customer_type field)" ) address: Optional[AddressCreateRequestSchema] = Field( default=None, description="A new address for this customer that needs to be created" ) contact: Optional[CustomerContactCreateRequestSchema] = Field( default=None, description="A new contact for this customer that needs to be created" ) avatar: Optional[int] = Field( default=None, description="ID of Uploaded image file to be used as avatar" ) tags: Optional[List[str]] = Field(default=None, description="Tags of customer") # Additional fields from create schema that were missing in update role_type: Optional[str] = Field(default=None, description="Role type of the Customer") timezone: Optional[str] = Field(default=None, description="A valid timezone") merchant_id: Optional[int] = Field(default=None, description="ID of the merchant") user_account_id: Optional[int] = Field(default=None, description="ID of the user account") account_literal: Optional[str] = Field( default=None, description="User-provided account literal (15 characters). If not provided, system will auto-generate one. This is the public-facing account number.", min_length=15, max_length=15 ) @field_validator("account_literal") @classmethod def validate_account_literal_format(cls, v): """Validate account literal format if provided.""" if v is not None: if not v.isalnum(): raise ValueError("Account literal must contain only alphanumeric characters") if len(v) != 15: raise ValueError("Account literal must be exactly 15 characters long") return v class CustomerListFilterSchema(BaseModel): search: Optional[str] = Query( default=None, description="Search by name, email, phone, city, state, zip etc" ) address: Optional[str] = Query(default=None, description="Search by address") city: Optional[str] = Query(default=None, description="Search by city name") state: Optional[str] = Query( default=None, description="Search by comma separated list of states. Eg.: ?state=AR,AZ", ) zipcode: Optional[str] = Query(default=None, description="Search by zipcode") country: Optional[str] = Query( default=None, description="Search by comma separated list of countries. Eg.: ?country=US,CA", ) is_active: Optional[int] = Query(default=0, description="Active status of customer") is_vendor: Optional[bool] = Query(default=False) name: Optional[str] = Query(default="", description="name of the customer") phone: Optional[str] = Query(default="", description="phone of the customer") email: Optional[str] = Query(default="", description="email of the customer") class CustomerTransactionsFilterSchema(BaseModel): search: Optional[str] = Query( default=None, description="Search by customer name, email, transaction billing address, etc", ) date_from: Optional[str] = Query( default=None, pattern=regexp.ISO_DATE_FORMAT, description="Start date of listing to be done. Must be of format YYYY-MM-DD", ) date_to: Optional[str] = Query( default=None, pattern=regexp.ISO_DATE_FORMAT, description="End date of listing to be done. Must be of format YYYY-MM-DD", ) status: Optional[str] = Query( default=None, description="Comma separated list of status of transaction" ) class CustomerInvoicesFilterSchema(BaseModel): search: Optional[str] = Query( default=None, description="Search by customer name, email, transaction billing address, etc", ) currency: Optional[str] = Query( default=None, description="Search by comma separated list of currencies. Eg.: ?currency=usd,yuan", ) invoice_id: Optional[str] = Query( default=None, description="Search by the unique invoice id" ) payment_frequency: Optional[str] = Query( default=None, description="Search by comma separated list of frequencies. Eg.: ?payment_frequeny=split,one_time", ) authorization_type: Optional[PaymentAuthorizationTypes] = Query( default=None, description="Search by authorisation type" ) status: Optional[str] = Query( default=None, description="Search by comma separated list of status. Eg.: ?status=paid,full_amount_pending", ) class CustomerAccountBulkActions(BaseModel): action: AccountsBulkActions = Field( description="Type of action which needs to be performed." ) ids: List[str] = Field( description="An array of customer_ids of the items that you want to perform actions for" ) @field_validator("ids") @classmethod def validate_ids(cls, v): max_limit = 10 if v is None or len(v) == 0: raise ValueError(f"Please provide atleast one customer ids") if len(v) > max_limit: raise ValueError( f"You can provide atmost of {max_limit} customer ids at once" ) for item in v: if not regexp.test(item, regexp.CUSTOMER_UNIQUE_ID): raise ValueError(f"'{item}' is not a valid customer ids") return v class CustomerPaymentMethodsFilterSchema(BaseModel): search: Optional[str] = Query( default=None, description="Search by card number, card expiry, ach account number, etc", ) method: Optional[PaymentMethodTypes] = Query( default=None, description="Search by payment method type" ) class CustomerAccountCreateRequestSchema(UserBase): model_config = ConfigDict( from_attributes=True, json_schema_extra={ "example": { "email": "johndoe@example.com", "password": "weakpassword", "phone": "+9176787678", "first_name": "John", "last_name": "Doe", } } ) email: EmailStr = Field(description="A valid email address") phone: str = Field(description="A valid phone number") password: Optional[str] = Field( default=None, min_length=8, max_length=32, description="Make sure you choose a strong password.", ) confirm_password: Optional[str] = Field( default=None, min_length=8, max_length=32, description="Make sure the passwords are same." ) def compare_passwords(self) -> bool: return self.password == self.confirm_password class CustomerActionRequestSchema(BaseModel): """Schema for customer action requests (activate/deactivate).""" action: str = Field(..., description="Action to perform on customer", pattern="^(activate|deactivate)$") class CustomerBulkActionRequestSchema(BaseModel): """Schema for bulk customer actions. Limited to maximum 10 customers per request.""" model_config = ConfigDict( from_attributes=True, json_schema_extra={ "example": { "action": "activate", "ids": [ "ACCROS171217C9B", "ACCABI03758427F", "ACCXYZ123456789" ] } } ) action: str = Field( ..., description="Action to perform on customers", pattern="^(activate|deactivate|delete)$" ) ids: List[str] = Field( ..., description="List of customer IDs to perform action on (maximum 10 per request)", min_length=1, max_length=10 ) class CustomerPaginationSchema(BaseModel): """Schema for pagination parameters.""" model_config = ConfigDict(from_attributes=True) page: int = Field(1, ge=1, description="Page number for pagination", example=1) per_page: int = Field(20, ge=1, le=100, description="Number of items per page (max 100)", example=20) class CustomerFilterSchema(BaseModel): """Schema for consolidating customer list filters.""" model_config = ConfigDict(from_attributes=True) search: Optional[str] = Field(None, description="Search across customer name, email, customer_id, or business name", example="john") name: Optional[str] = Field(None, description="Filter by customer first or last name (partial match)", example="John") email: Optional[str] = Field(None, description="Filter by customer email address (partial match)", example="john@example.com") phone: Optional[str] = Field(None, description="Filter by customer phone number (partial match)", example="123-456-7890") address: Optional[str] = Field(None, description="Filter by customer address (searches address_line_1, city, state)", example="Main Street") zip_code: Optional[str] = Field(None, description="Filter by customer zip code (partial match)", example="12345") customer_type: Optional[str] = Field(None, description="Filter by customer type (any string value)", example="Business") account_type: Optional[str] = Field(None, description="Filter by account/business type", example="LLC") industry: Optional[str] = Field(None, description="Filter by industry (partial match)", example="finance") is_active: int = Field(0, description="Filter by active status: 0=show all, 1=active only, -1=inactive only", example=1) is_vendor: int = Field(0, description="Filter by vendor status: 0=show all, 1=vendors only, -1=customers only", example=-1) fields: Optional[str] = Field(None, description="Comma-separated list of related fields to include: addresses, contacts, merchant, attachments", example="addresses,contacts")