# manufacturer/service.py - оновлений з правильною обробкою помилок from datetime import datetime from typing import List, Optional from fastapi import HTTPException from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy import select, or_, asc, desc, func from src.error import _ERROR from src.manufacturer.models import Manufacturer, ManufacturerPhoto from src.manufacturer.schemas import ManufacturerCreateForm, ManufacturerOutSchema, ManufacturerPhotoResponse from src.product.models import Product from src.review.models import ManufacturerReview, ReviewStatusEnum from src.utils.image_manager import ASSETS_URL ALLOWED_SORT_FIELDS = {"name", "date_added", "date_modify", "rating"} class ManufacturerService: @staticmethod async def get_manufacturers_list( db: AsyncSession, needle: Optional[str] = None, sort_field: str = "date_added", sort_order: str = "ASC", page: int = 1, limit: int = 10, ) -> dict: if sort_field not in ALLOWED_SORT_FIELDS or not hasattr(Manufacturer, sort_field): _ERROR("ValidationError") order_func = asc if sort_order.upper() == "ASC" else desc # Получаем количество отзывов для каждого производителя manufacturer_reviews_query = await db.execute( select( ManufacturerReview.manufacturer_id, func.count(ManufacturerReview.manufacturer_review_id).label('reviews_count') ) .where(ManufacturerReview.status == ReviewStatusEnum.approved) .group_by(ManufacturerReview.manufacturer_id) ) manufacturer_reviews_counts = {row.manufacturer_id: row.reviews_count for row in manufacturer_reviews_query} base_query = ( select( Manufacturer, func.count(Product.product_id).label("product_count") ) .outerjoin(Product, Product.manufacturer_id == Manufacturer.manufacturer_id) .group_by(Manufacturer.manufacturer_id) ) count_query = select(func.count()).select_from(Manufacturer) if needle: like_term = f"%{needle.lower()}%" filters = or_( Manufacturer.name.ilike(like_term), Manufacturer.seo_keyword.ilike(like_term), Manufacturer.meta_title.ilike(like_term), Manufacturer.meta_description.ilike(like_term), Manufacturer.meta_keyword.ilike(like_term), ) base_query = base_query.where(filters) count_query = count_query.where(filters) base_query = base_query.order_by(order_func(getattr(Manufacturer, sort_field))) base_query = base_query.offset((page - 1) * limit).limit(limit) try: result = await db.execute(base_query) items = [] for m, product_count in result.all(): reviews_count = manufacturer_reviews_counts.get(m.manufacturer_id, 0) items.append(ManufacturerOutSchema( **m.__dict__, product_count=product_count, reviews_count=reviews_count # Добавляем количество отзывов )) total_result = await db.execute(count_query) total = total_result.scalar() return { "items": items, "total": total } except Exception: _ERROR("DatabaseError") @staticmethod async def get_manufacturer_by_id(db: AsyncSession, manufacturer_id: int) -> Manufacturer: stmt = select(Manufacturer).where(Manufacturer.manufacturer_id == manufacturer_id) result = await db.execute(stmt) manufacturer = result.scalar_one_or_none() if not manufacturer: _ERROR("ManufacturerNotFound") return manufacturer @staticmethod async def create_manufacturer( db: AsyncSession, data: ManufacturerCreateForm, logo_url: Optional[str] = None, ) -> Manufacturer: try: new_manufacturer = Manufacturer( name=data.name, address=data.address, description=data.description, phone=data.phone, email=data.email, status=data.status, seo_keyword=data.seo_keyword, meta_title=data.meta_title, meta_description=data.meta_description, meta_keyword=data.meta_keyword, logo=logo_url or f"{ASSETS_URL}placeholder.webp", date_added=datetime.utcnow(), date_modify=datetime.utcnow(), ) db.add(new_manufacturer) await db.commit() await db.refresh(new_manufacturer) return new_manufacturer except HTTPException: # Якщо це помилка користувача (400) — не перехоплюємо await db.rollback() raise except Exception as e: await db.rollback() raise HTTPException(status_code=500, detail=f"Database error: {str(e)}") @staticmethod async def update_manufacturer( db: AsyncSession, manufacturer_id: int, update_data: dict, logo_url: Optional[str] = None, ) -> Manufacturer: try: stmt = select(Manufacturer).where(Manufacturer.manufacturer_id == manufacturer_id) result = await db.execute(stmt) manufacturer = result.scalar_one_or_none() if not manufacturer: _ERROR("ManufacturerNotFound") for key, value in update_data.items(): setattr(manufacturer, key, value) if logo_url: manufacturer.logo = logo_url manufacturer.date_modify = datetime.utcnow() db.add(manufacturer) await db.commit() await db.refresh(manufacturer) return manufacturer except Exception: await db.rollback() _ERROR("DatabaseError") @staticmethod async def delete_manufacturer(db: AsyncSession, manufacturer_id: int): # Перевірка, чи існує виробник stmt = select(Manufacturer).where(Manufacturer.manufacturer_id == manufacturer_id) result = await db.execute(stmt) manufacturer = result.scalar_one_or_none() if not manufacturer: _ERROR("ManufacturerNotFound") # Перевірка на пов'язані продукти stmt_product = select(Product).where(Product.manufacturer_id == manufacturer_id) result = await db.execute(stmt_product) product = result.scalar_one_or_none() if product: _ERROR("ManufacturerNotEmpty") # Видалення виробника await db.delete(manufacturer) await db.commit() @staticmethod async def add_manufacturer_photo( db: AsyncSession, manufacturer_id: int, image_url: str ) -> ManufacturerPhoto: try: # Перевіряємо чи існує виробник manufacturer = await db.execute(select(Manufacturer).where(Manufacturer.manufacturer_id == manufacturer_id)) if not manufacturer.scalar_one_or_none(): _ERROR("ManufacturerNotFound") new_photo = ManufacturerPhoto( manufacturer_id=manufacturer_id, image=image_url ) db.add(new_photo) await db.commit() await db.refresh(new_photo) return new_photo except Exception: await db.rollback() _ERROR("DatabaseError") @staticmethod async def get_manufacturer_photo( manufacturer_id: int, db: AsyncSession, page: int, limit: int ) -> tuple[list[ManufacturerPhoto], int]: try: # Загальна кількість фото total = await db.scalar( select(func.count()).select_from( select(ManufacturerPhoto) .where(ManufacturerPhoto.manufacturer_id == manufacturer_id) .subquery() ) ) # Отримання фото з пагінацією result = await db.execute( select(ManufacturerPhoto) .where(ManufacturerPhoto.manufacturer_id == manufacturer_id) .order_by(desc(ManufacturerPhoto.manufacturer_photo_id)) .offset((page - 1) * limit) .limit(limit) ) photos = result.scalars().all() return photos, total except Exception: _ERROR("DatabaseError") @staticmethod async def delete_manufacturer_photo( db: AsyncSession, manufacturer_id: int, manufacturer_photo_id: int ): try: result = await db.execute( select(ManufacturerPhoto).where( ManufacturerPhoto.manufacturer_photo_id == manufacturer_photo_id, ManufacturerPhoto.manufacturer_id == manufacturer_id ) ) photo = result.scalar_one_or_none() if not photo: _ERROR("ManufacturerPhotoNotFound") await db.delete(photo) await db.commit() except Exception: await db.rollback() _ERROR("DatabaseError")