""" Product services for business logic operations. """ import json from typing import Optional, List, Tuple, Any, Dict from sqlalchemy.orm import Session from sqlalchemy import select, update, delete, func from sqlalchemy.exc import IntegrityError from datetime import datetime import logging from fastapi import status, UploadFile from src.core.exceptions import APIException from src.core.utils.enums import FilterOperators from src.core.utils.constants import API_PREFIXES from src.apps.products.models.product import Product, product_attachments_map from src.apps.products.schemas.common import ProductResponseSchema from src.apps.products.schemas.common import ProductCreateSchema, ProductUpdateSchema,ProductListFilterSchema from src.helpers.pagination import QueryPaginator from src.apps.base.utils.functions import generate_unique_slug from src.core.config import settings from src.apps.files.models.file import File from src.apps.files.helper.io import remove_file_single, upload_file_multiple from src.apps.files.schemas.file_common import FileResponseSchema from src.apps.products.models.product import product_notes_map from src.apps.merchants.schemas.merchant_common import MerchantSchema from src.apps.users.schemas.user_common import UserSchema from src.core.utils.constants import ALLOWED_FILE_CONTENT_TYPES, MAX_UPLOAD_LIMIT from src.apps.products.schemas import requests as product_note_requests from src.apps.notes.models.note import Note from src.apps.notes.schemas.common import ( NoteCreateRequestSchema, NoteUpdateRequestSchema, NoteSchema, ) from src.apps.notes import services as note_services from src.apps.files.file_services import upload_multiple_files logger = logging.getLogger(__name__) async def get_products_list( db: Session, filters: ProductListFilterSchema, _from: Optional[str] = None, _to: Optional[str] = None, sort_by: str = "-created_at", fields: Optional[str] = None, skip: int = 0, limit: int = 20, paginate: bool = True, ) -> Dict[str, Any]: """ Get products list with filtering, sorting, and pagination. """ try: pagination_url: str = f"{settings.api_base_url()}{API_PREFIXES.PRODUCTS.value}" # Build base query stmt = select(Product) if hasattr(Product, 'deleted_at'): stmt = stmt.where(Product.deleted_at.is_(None)) if _from: from_date = datetime.strptime(_from, "%Y-%m-%d") stmt = stmt.where(Product.created_at >= from_date) if _to: to_date = datetime.strptime(_to, "%Y-%m-%d") to_date = to_date.replace(hour=23, minute=59, second=59, microsecond=999999) stmt = stmt.where(Product.created_at <= to_date) if filters.search: search_term = f"%{filters.search}%" stmt = stmt.where( (Product.name.ilike(search_term)) | (Product.description.ilike(search_term)) | (Product.sku.ilike(search_term)) ) # Individual field filters if filters.name: stmt = stmt.where(Product.name.ilike(f"%{filters.name}%")) if filters.sku: stmt = stmt.where(Product.sku.ilike(f"%{filters.sku}%")) if filters.code: stmt = stmt.where(Product.code.ilike(f"%{filters.code}%")) if filters.merchant_id is not None: stmt = stmt.where(Product.merchant_id == filters.merchant_id) if filters.item_type is not None: # Filter by item type string directly stmt = stmt.where(Product.item_type == filters.item_type.lower()) if filters.category_id is not None: stmt = stmt.where(Product.category_id == filters.category_id) # Convert integer filter to boolean for database comparison if filters.is_active != 0: # 0 means "all", so don't filter is_active_bool = filters.is_active == 1 # 1 = active (True), -1 = inactive (False) stmt = stmt.where(Product.is_active == is_active_bool) if filters.is_new != 0: # 0 means "all", so don't filter is_new_bool = filters.is_new == 1 # 1 = new (True), -1 = not new (False) stmt = stmt.where(Product.is_new == is_new_bool) if filters.is_purchase != 0: # 0 means "all", so don't filter is_purchase_bool = filters.is_purchase == 1 # 1 = purchase (True), -1 = not purchase (False) stmt = stmt.where(Product.is_purchase == is_purchase_bool) if filters.is_sell != 0: # 0 means "all", so don't filter is_sell_bool = filters.is_sell == 1 # 1 = sell (True), -1 = not sell (False) stmt = stmt.where(Product.is_sell == is_sell_bool) if filters.cost_price_range: cost_price_range_json = json.loads(filters.cost_price_range) if cost_price_range_json["operator"] == FilterOperators.EQUAL_TO: stmt = stmt.where( Product.unit_price == cost_price_range_json["start"] ) if cost_price_range_json["operator"] == FilterOperators.IN_BETWEEN: stmt = stmt.where( Product.unit_price.between( cost_price_range_json["start"], cost_price_range_json["end"] ) ) if cost_price_range_json["operator"] == FilterOperators.IS_GREATER_THAN: stmt = stmt.where( Product.unit_price > cost_price_range_json["start"] ) if cost_price_range_json["operator"] == FilterOperators.IS_LESS_THAN: stmt = stmt.where( Product.unit_price < cost_price_range_json["start"] ) # Apply sorting if sort_by.startswith("-"): field_name = sort_by[1:] if hasattr(Product, field_name): stmt = stmt.order_by(getattr(Product, field_name).desc()) else: if hasattr(Product, sort_by): stmt = stmt.order_by(getattr(Product, sort_by)) # Use QueryPaginator for pagination if paginate: paginator = QueryPaginator( query=stmt, schema=ProductResponseSchema, url=pagination_url, db=db, model=Product, offset=skip, limit=limit, use_orm=True, ) return paginator.to_dict() else: # Return raw results without pagination results = db.execute(stmt.offset(skip).limit(limit)).scalars().all() return [ProductResponseSchema.model_validate(product) for product in results] except Exception as e: logger.error(f"Error getting products list: {e}") return {"error": str(e)} async def create_product( db: Session, # merchant_id: int, payload: ProductCreateSchema ) -> ProductResponseSchema: """ Create a new product. """ # try: # Convert schema to dict and add merchant_id # Build dict from payload and ensure relationship fields are not passed as raw IDs product_dict = payload.model_dump() # product_dict["merchant_id"] = merchant_id # Extract attachments from the dict to avoid assigning List[int] to ORM relationship attachments = product_dict.pop("attachments", None) thumbnail = product_dict.pop("thumbnail", None) # Generate unique slug from product name if not product_dict.get("slug"): product_dict["slug"] = generate_unique_slug( db=db, instance=Product, slug_label=payload.name ) # Create new product new_product = Product(**product_dict) db.add(new_product) db.commit() db.refresh(new_product) if new_product and (attachments is not None) and (len(attachments) > 0): # Save the files try: for attachment_id in attachments: attachemnts_map_query = product_attachments_map.insert().values( product_id=new_product.id, file_id=attachment_id ) db.execute(attachemnts_map_query) db.commit() except IntegrityError: raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message="Attachment already exists", ) if new_product and thumbnail is not None: try: attachemnts_map_query = product_attachments_map.insert().values( product_id=new_product.id, file_id=thumbnail ) db.execute(attachemnts_map_query) db.commit() except IntegrityError: raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message="Thumbnail already exists", ) return ProductResponseSchema.model_validate(new_product) # except IntegrityError as e: # logger.error(f"IntegrityError creating product: {e}") # # Handle specific integrity errors # if "sku" in str(e.orig).lower(): # raise ValueError("Product with this SKU already exists") # elif "slug" in str(e.orig).lower(): # raise ValueError("Product with this slug already exists") # else: # raise ValueError("Product creation failed due to data conflict") # except Exception as e: # logger.error(f"Error creating product: {e}") # raise ValueError(f"Product creation failed: {str(e)}") async def get_product( db: Session, # merchant_id: int, slug: str ) -> Optional[ProductResponseSchema]: """ Get a product by ID. """ try: stmt = select(Product).where( Product.slug == slug, # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) result = db.execute(stmt) product = result.scalar_one_or_none() if product: return ProductResponseSchema.model_validate(product) return None except Exception as e: logger.error(f"Error getting product by ID: {e}") return None async def update_product( db: Session, # merchant_id: int, slug: str, payload: ProductUpdateSchema ) -> Optional[ProductResponseSchema]: """ Update a product by ID. """ try: # Check if product exists stmt = select(Product).where( Product.slug == slug, # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) result = db.execute(stmt) product = result.scalar_one_or_none() if not product: return None # Update product with new data update_data = payload.model_dump(exclude_unset=True) if update_data: # Add updated_at timestamp update_data["updated_at"] = datetime.utcnow() update_stmt = ( update(Product) .where( Product.id == product.id, # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) .values(**update_data) ) db.execute(update_stmt) db.commit() db.refresh(product) return ProductResponseSchema.model_validate(product) except IntegrityError as e: logger.error(f"IntegrityError updating product: {e}") if "sku" in str(e.orig).lower(): raise ValueError("Product with this SKU already exists") elif "slug" in str(e.orig).lower(): raise ValueError("Product with this slug already exists") else: raise ValueError("Product update failed due to data conflict") except Exception as e: logger.error(f"Error updating product: {e}") raise ValueError(f"Product update failed: {str(e)}") async def delete_product( db: Session, # merchant_id: int, slug: str ) -> bool: """ Delete a product by ID (soft delete). """ try: # Check if product exists stmt = select(Product).where( Product.slug == slug, # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) result = db.execute(stmt) product = result.scalar_one_or_none() if not product: return False # Soft delete by setting deleted_at timestamp update_stmt = ( update(Product) .where( Product.slug == slug, # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) .values(deleted_at=datetime.utcnow()) ) db.execute(update_stmt) db.commit() return True except Exception as e: logger.error(f"Error deleting product: {e}") raise ValueError(f"Product deletion failed: {str(e)}") async def bulk_delete_products( db: Session, # merchant_id: int, slugs: List[str] ) -> int: """ Delete multiple products by IDs (soft delete). """ try: # Update multiple products at once update_stmt = ( update(Product) .where( Product.slug.in_(slugs), # Product.merchant_id == merchant_id, Product.deleted_at.is_(None) ) .values(deleted_at=datetime.utcnow()) ) result = db.execute(update_stmt) db.commit() return result.rowcount except Exception as e: logger.error(f"Error bulk deleting products: {e}") raise ValueError(f"Bulk deletion failed: {str(e)}") async def upload_attachments_files( db: Session, files: List[UploadFile], file_type: str, current_user: UserSchema, merchant: MerchantSchema, slug: str, ) -> List[FileResponseSchema]: """ Create multiple files entry for a product and map attachments. """ if len(files) > MAX_UPLOAD_LIMIT: msg = f"A maximum of {MAX_UPLOAD_LIMIT} files can be uploaded at once" raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message=msg ) # Validate content types for f in files: extension = f.content_type.split("/")[1] if extension not in ALLOWED_FILE_CONTENT_TYPES: msg = ( f"Cannot upload {extension} files. Allowed file extensions are " f"{', '.join(ALLOWED_FILE_CONTENT_TYPES)}" ) raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message=msg ) uploaded_files: List[Dict] = upload_file_multiple( db, files, f"{merchant.merchant_id}/{slug}/" ) # Create database entries for uploaded files using file services new_files = await upload_multiple_files( db=db, files=uploaded_files, file_type=file_type, created_by=current_user, merchant=merchant ) msg = "Failed to upload files" product_data = await get_product(db=db, slug=slug) if not new_files: raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message=msg ) try: # Map newly created files to product for file_data in new_files: map_statement = product_attachments_map.insert().values( file_id=file_data.id, product_id=product_data.id ) db.execute(map_statement) db.commit() except Exception: logger.error( "Could not save product attachment mapping", extra={ "module": "product", "submodule": "product", "product_slug": slug, "merchant_id": merchant.merchant_id, "created_by_id": current_user.user_id, }, ) raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message="Could not save product attachment mapping", ) logger.info( "product attachment add", extra={ "module": "product", "submodule": "product", "merchant_id": merchant.merchant_id, "product_slug": product_data.slug, "created_by_id": current_user.user_id, }, ) return new_files async def delete_attachments_files( db: Session, slug: str, merchant: MerchantSchema, current_user: UserSchema, file_id: int, ) -> Dict: """ Delete a product's attachment file by product slug and file id. """ product_data = await get_product(db=db, slug=slug) # Fetch file using SQLAlchemy 2.0 style file_data = db.execute(select(File).where(File.id == file_id)).scalar_one_or_none() if file_data is None: raise APIException( module=__name__, error={}, status_code=status.HTTP_404_NOT_FOUND, message="The requested file was not found, ", ) try: # Remove mapping d = ( product_attachments_map.delete().where( product_attachments_map.c.file_id == file_id, product_attachments_map.c.product_id == product_data.id, ) ) db.execute(d) db.commit() # Fetch file via SQLAlchemy 2.0 style stmt = select(File).where(File.id == file_id) result = db.execute(stmt).scalar_one_or_none() if result is not None: # Remove physical file and delete record remove_file_single(path=result.upload_path) del_stmt = delete(File).where(File.id == file_id) db.execute(del_stmt) db.commit() except Exception as e: logger.error( "Could not delete product attachment", extra={ "module": "customer", "submodule": "product", "merchant_id": merchant.merchant_id, "product_slug": slug, "created_by_id": current_user.user_id, "error": str(e), }, ) raise APIException( module=__name__, error={}, status_code=status.HTTP_400_BAD_REQUEST, message="Could not delete attachment, " + str(e), ) logger.info( "product attachment file deleted", extra={ "module": "customer", "submodule": "product", "merchant_id": merchant.merchant_id, "product_slug": product_data.slug, "created_by_id": current_user.user_id, }, ) return {"deleted": True} async def create_note( db: Session, merchant: MerchantSchema, payload: NoteCreateRequestSchema, slug: str, created_by: UserSchema, ) -> NoteSchema: """Create Product Note""" product = await get_product(db=db, slug=slug) if not product: raise APIException( module=__name__, error={}, status_code=status.HTTP_412_PRECONDITION_FAILED, message="Product not found", ) # Create note using notes service note = await note_services.create_note( db=db, payload=payload, created_by_id=created_by.id ) map_statement = product_notes_map.insert().values( note_id=note.id, product_id=product.id ) db.execute(map_statement) db.commit() return note async def update_note( db: Session, note_id: int, payload: NoteUpdateRequestSchema, slug: str, ) -> NoteSchema: """Edit Customer Note""" product = await get_product(db=db, slug=slug) if not product: raise APIException( module=__name__, error={}, status_code=status.HTTP_412_PRECONDITION_FAILED, message="Product not found", ) # Update via notes service updated = await note_services.update_note_by_id( db=db, object_id=note_id, payload=payload ) return updated async def list_note( db: Session, merchant: MerchantSchema, slug: str ) -> List[NoteSchema]: """Get Note list""" product = await get_product(db=db, slug=slug) if not product: raise APIException( module=__name__, error={}, status_code=status.HTTP_412_PRECONDITION_FAILED, message="Product not found", ) stmt = ( select(Note) .join(product_notes_map, Note.id == product_notes_map.c.note_id) .where(product_notes_map.c.product_id == product.id) .order_by(Note.created_at.desc()) ) notes = db.execute(stmt).scalars().all() return [NoteSchema.model_validate(n) for n in notes] async def activities( db: Session, merchant: MerchantSchema, slug: str ) -> Optional[List[Dict]]: """Get activities list""" product = await get_product(db=db, slug=slug) if not product: raise APIException( module=__name__, error={}, status_code=status.HTTP_412_PRECONDITION_FAILED, message="Product not found", ) return []