from datetime import datetime from fastapi import HTTPException from sqlalchemy import select, and_, asc, desc, func, or_, text from sqlalchemy.ext.asyncio import AsyncSession from typing import Optional, List, Tuple, Dict, Any from sqlalchemy.orm import selectinload from src.customer.models import Customer from src.manufacturer.models import Manufacturer from src.product.models import Product from src.review.models import StoreReview, ReviewStatusEnum, ManufacturerReview, ProductReview, ServiceReview from src.review.schemas import SortField, SortOrder from src.store.models import Store class ReviewService: @staticmethod async def get_reviews( db: AsyncSession, page: int = 1, limit: int = 10, sort_field: SortField = SortField.date_added, sort_order: SortOrder = SortOrder.desc, search: Optional[str] = None, status: Optional[int] = None, store_id: Optional[int] = None, date_range: Optional[List[str]] = None, ) -> Tuple[List[StoreReview], int, dict]: stmt = select(StoreReview).options( selectinload(StoreReview.customer), selectinload(StoreReview.store), ) filters = [] if search: search_pattern = f"%{search.lower()}%" filters.append( or_( func.lower(StoreReview.text).like(search_pattern), func.lower(Customer.name).like(search_pattern), func.lower(Customer.phone).like(search_pattern), func.lower(Store.name).like(search_pattern), ) ) if status is not None: filters.append(StoreReview.status == ReviewStatusEnum(status)) if store_id is not None: filters.append(StoreReview.store_id == store_id) if date_range and len(date_range) == 2: try: start_date = datetime.strptime(date_range[0], "%Y-%m-%d") end_date = datetime.strptime(date_range[1], "%Y-%m-%d") filters.append( and_( StoreReview.date_added >= start_date, StoreReview.date_added <= end_date ) ) except ValueError: pass if filters: stmt = stmt.join(StoreReview.customer).join(StoreReview.store).where(*filters) sort_column = { SortField.date_added: StoreReview.date_added, SortField.rating: StoreReview.rating, "store_review_id": StoreReview.store_review_id }.get(sort_field, StoreReview.date_added) stmt = stmt.order_by(asc(sort_column) if sort_order == SortOrder.asc else desc(sort_column)) stmt = stmt.offset((page - 1) * limit).limit(limit) # --- Total --- count_stmt = select(func.count()).select_from(StoreReview) if filters: count_stmt = count_stmt.join(StoreReview.customer).join(StoreReview.store).where(*filters) total_result = await db.execute(count_stmt) total = total_result.scalar_one() # --- Dashboard --- dashboard_stmt = select( func.count(StoreReview.store_review_id), func.avg(StoreReview.rating) ).select_from(StoreReview) if filters: dashboard_stmt = dashboard_stmt.join(StoreReview.customer).join(StoreReview.store).where(*filters) dashboard_result = await db.execute(dashboard_stmt) total_quantity, medium_rating = dashboard_result.first() dashboard = { "total_quantity": total_quantity or 0, "medium_rating": float(medium_rating) if medium_rating is not None else 0 } result = await db.execute(stmt) reviews = result.scalars().unique().all() return reviews, total, dashboard @staticmethod async def update_store_review_status( db: AsyncSession, store_review_id: int, new_status: ReviewStatusEnum ) -> StoreReview: query = select(StoreReview).where(StoreReview.store_review_id == store_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"StoreReview with id={store_review_id} not found") old_status = review.status review.status = new_status review.date_modify = datetime.utcnow() db.add(review) await db.commit() await db.refresh(review) # Перераховуємо рейтинг, якщо статус змінився на approved або з approved if old_status != new_status and (new_status == ReviewStatusEnum.approved or old_status == ReviewStatusEnum.approved): await ReviewService.recalculate_ratings(db, 'store', review.store_id) return review @staticmethod async def delete_store_review(db: AsyncSession, store_review_id: int) -> None: query = select(StoreReview).where(StoreReview.store_review_id == store_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"StoreReview with id={store_review_id} not found") await db.delete(review) await db.commit() @staticmethod async def get_manufacturer_reviews( db: AsyncSession, page: int = 1, limit: int = 10, sort_field: str = "date_added", sort_order: str = "desc", search: Optional[str] = None, status: Optional[int] = None, manufacturer_id: Optional[int] = None, date_range: Optional[List[str]] = None, ) -> Tuple[List[ManufacturerReview], int, dict]: stmt = select(ManufacturerReview).options( selectinload(ManufacturerReview.customer), selectinload(ManufacturerReview.manufacturer), ) conditions = [] need_join_customer = False need_join_manufacturer = False if search: pattern = f"%{search.lower()}%" conditions.append( or_( func.lower(ManufacturerReview.text).like(pattern), func.lower(Customer.name).like(pattern), func.lower(Customer.phone).like(pattern), func.lower(Manufacturer.name).like(pattern), ) ) need_join_customer = True need_join_manufacturer = True if status is not None: conditions.append(ManufacturerReview.status == ReviewStatusEnum(status)) if manufacturer_id is not None: conditions.append(ManufacturerReview.manufacturer_id == manufacturer_id) need_join_manufacturer = True # Діапазон дат if date_range and len(date_range) == 2: try: start_date = datetime.strptime(date_range[0], "%Y-%m-%d") end_date = datetime.strptime(date_range[1], "%Y-%m-%d") conditions.append( and_( ManufacturerReview.date_added >= datetime.combine(start_date, datetime.min.time()), ManufacturerReview.date_added <= datetime.combine(end_date, datetime.max.time()) ) ) except ValueError: pass if need_join_customer: stmt = stmt.outerjoin(ManufacturerReview.customer) if need_join_manufacturer: stmt = stmt.outerjoin(ManufacturerReview.manufacturer) if conditions: stmt = stmt.where(and_(*conditions)) sort_column = { "date_added": ManufacturerReview.date_added, "rating": ManufacturerReview.rating, "manufacturer_review_id": ManufacturerReview.manufacturer_review_id }.get(sort_field, ManufacturerReview.date_added) stmt = stmt.order_by(asc(sort_column) if sort_order == "asc" else desc(sort_column)) stmt = stmt.offset((page - 1) * limit).limit(limit) count_stmt = select(func.count()).select_from(ManufacturerReview) if need_join_customer: count_stmt = count_stmt.outerjoin(ManufacturerReview.customer) if need_join_manufacturer: count_stmt = count_stmt.outerjoin(ManufacturerReview.manufacturer) if conditions: count_stmt = count_stmt.where(and_(*conditions)) total_result = await db.execute(count_stmt) total = total_result.scalar_one() stats_stmt = select( func.count(ManufacturerReview.manufacturer_review_id), func.avg(ManufacturerReview.rating) ).select_from(ManufacturerReview) if need_join_customer: stats_stmt = stats_stmt.outerjoin(ManufacturerReview.customer) if need_join_manufacturer: stats_stmt = stats_stmt.outerjoin(ManufacturerReview.manufacturer) if conditions: stats_stmt = stats_stmt.where(and_(*conditions)) stats_result = await db.execute(stats_stmt) total_quantity, medium_rating = stats_result.first() medium_rating = round(float(medium_rating or 0), 1) result = await db.execute(stmt) reviews = result.scalars().unique().all() dashboard = { "total_quantity": int(total_quantity or 0), "medium_rating": medium_rating } return reviews, total, dashboard @staticmethod async def update_manufacturer_review_status( db: AsyncSession, manufacturer_review_id: int, new_status: ReviewStatusEnum ) -> ManufacturerReview: query = select(ManufacturerReview).where(ManufacturerReview.manufacturer_review_id == manufacturer_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ManufacturerReview with id={manufacturer_review_id} not found") old_status = review.status review.status = new_status review.date_modify = datetime.utcnow() db.add(review) await db.commit() await db.refresh(review) # Перераховуємо рейтинг, якщо статус змінився на approved або з approved if old_status != new_status and (new_status == ReviewStatusEnum.approved or old_status == ReviewStatusEnum.approved): await ReviewService.recalculate_ratings(db, 'manufacturer', review.manufacturer_id) return review @staticmethod async def delete_manufacturer_review(db: AsyncSession, manufacturer_review_id: int): query = select(ManufacturerReview).where(ManufacturerReview.manufacturer_review_id == manufacturer_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ManufacturerReview id={manufacturer_review_id} not found") await db.delete(review) await db.commit() @staticmethod async def get_product_reviews( db: AsyncSession, page: int = 1, limit: int = 10, sort_field: str = "date_added", sort_order: str = "desc", search: Optional[str] = None, status: Optional[int] = None, manufacturer_id: Optional[int] = None, date_range: Optional[List[str]] = None, ) -> Tuple[List[ProductReview], int, dict]: stmt = select(ProductReview).options( selectinload(ProductReview.customer), selectinload(ProductReview.product), # підвантажуємо продукт selectinload(ProductReview.manufacturer), # підвантажуємо виробника ) conditions = [] need_join_customer = False need_join_manufacturer = False # Визначаємо, чи потрібні join-и if search or manufacturer_id is not None: need_join_customer = True if manufacturer_id is not None: need_join_manufacturer = True # Додаємо join-и до основного запиту if need_join_customer: stmt = stmt.outerjoin(ProductReview.customer) if need_join_manufacturer: stmt = stmt.outerjoin(ProductReview.manufacturer) # Умови пошуку if search: search_pattern = f"%{search.lower()}%" search_conditions = [ func.lower(ProductReview.text).like(search_pattern), ] # тільки якщо реально join потрібен if need_join_customer: search_conditions.append(func.lower(Customer.name).like(search_pattern)) search_conditions.append(func.lower(Customer.phone).like(search_pattern)) if need_join_manufacturer: search_conditions.append(func.lower(Product.name).like(search_pattern)) # об'єднуємо всередині одного or_ для цього joined ряду conditions.append(or_(*search_conditions)) # Фільтр по статусу if status is not None: conditions.append(ProductReview.status == ReviewStatusEnum(status)) # Фільтр по виробнику if manufacturer_id is not None: conditions.append(ProductReview.manufacturer_id == manufacturer_id) # Фільтр по датах if date_range and len(date_range) == 2: try: start_date_obj = datetime.strptime(date_range[0], "%Y-%m-%d").date() end_date_obj = datetime.strptime(date_range[1], "%Y-%m-%d").date() start_date = datetime.combine(start_date_obj, datetime.min.time()) end_date = datetime.combine(end_date_obj, datetime.max.time()) conditions.append( and_( ProductReview.date_added >= start_date, ProductReview.date_added <= end_date ) ) except Exception: pass # Додаємо умови до основного запиту if conditions: stmt = stmt.where(and_(*conditions)) # Сортування sort_column = { "date_added": ProductReview.date_added, "rating": ProductReview.rating, "product_review_id": ProductReview.product_review_id }.get(sort_field, ProductReview.date_added) stmt = stmt.order_by(asc(sort_column) if sort_order == "asc" else desc(sort_column)) # Пагінація stmt = stmt.offset((page - 1) * limit).limit(limit) # Виконуємо основний запит для отримання списку відгуків result = await db.execute(stmt) reviews = result.scalars().unique().all() # Тепер виконуємо один агрегатний запит, щоб отримати total_quantity і medium_rating stats_stmt = select( func.count(func.distinct(ProductReview.product_review_id)), func.avg(ProductReview.rating) ).select_from(ProductReview) if need_join_customer: stats_stmt = stats_stmt.outerjoin(ProductReview.customer) if need_join_manufacturer: stats_stmt = stats_stmt.outerjoin(ProductReview.manufacturer) if conditions: stats_stmt = stats_stmt.where(and_(*conditions)) stats_result = await db.execute(stats_stmt) total_quantity, medium_rating = stats_result.first() medium_rating = round(float(medium_rating or 0), 1) total_quantity = int(total_quantity or 0) return reviews, total_quantity, { "total_quantity": total_quantity, "medium_rating": medium_rating } @staticmethod async def update_product_review_status( db: AsyncSession, product_review_id: int, new_status: ReviewStatusEnum ) -> ProductReview: query = select(ProductReview).where(ProductReview.product_review_id == product_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ProductReview with id={product_review_id} not found") old_status = review.status review.status = new_status review.date_modify = datetime.utcnow() db.add(review) await db.commit() await db.refresh(review) # Перераховуємо рейтинг, якщо статус змінився на approved або з approved if old_status != new_status and (new_status == ReviewStatusEnum.approved or old_status == ReviewStatusEnum.approved): await ReviewService.recalculate_ratings(db, 'product', review.product_id) return review @staticmethod async def delete_product_review( db: AsyncSession, product_review_id: int ): query = select(ProductReview).where(ProductReview.product_review_id == product_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ProductReview with id={product_review_id} not found") await db.delete(review) await db.commit() @staticmethod async def get_service_reviews( db: AsyncSession, page: int = 1, limit: int = 10, sort_field: str = "date_added", sort_order: str = "desc", search: Optional[str] = None, status: Optional[int] = None, date_range: Optional[List[str]] = None, ) -> Tuple[List[ServiceReview], int, Dict[str, Any]]: # Базовий SELECT stmt = select(ServiceReview).options(selectinload(ServiceReview.customer)) conditions = [] # Пошук if search: pattern = f"%{search.lower()}%" stmt = stmt.join(ServiceReview.customer) conditions.append( or_( func.lower(ServiceReview.text).like(pattern), func.lower(Customer.name).like(pattern), func.lower(Customer.phone).like(pattern), ) ) # Статус if status is not None: conditions.append(ServiceReview.status == ReviewStatusEnum(status)) # Дата if date_range and len(date_range) == 2: start_date = datetime.strptime(date_range[0], "%Y-%m-%d") end_date = datetime.strptime(date_range[1], "%Y-%m-%d") conditions.append(and_( ServiceReview.date_added >= datetime.combine(start_date, datetime.min.time()), ServiceReview.date_added <= datetime.combine(end_date, datetime.max.time()) )) # Застосовуємо умови if conditions: stmt = stmt.where(and_(*conditions)) # Сортування sort_column = { "date_added": ServiceReview.date_added, "rating": ServiceReview.rating, "service_review_id": ServiceReview.service_review_id }.get(sort_field, ServiceReview.date_added) stmt = stmt.order_by(asc(sort_column) if sort_order == "asc" else desc(sort_column)) # Пагінація stmt = stmt.offset((page - 1) * limit).limit(limit) # Виконання основного запиту result = await db.execute(stmt) reviews = result.scalars().unique().all() # Підрахунок total count_stmt = select(func.count()).select_from(ServiceReview) if conditions: count_stmt = count_stmt.join(ServiceReview.customer).where(and_(*conditions)) total_result = await db.execute(count_stmt) total = total_result.scalar_one() # Підрахунок dashboard dashboard_stmt = select( func.count(ServiceReview.service_review_id), func.avg(ServiceReview.rating) ).select_from(ServiceReview) if conditions: dashboard_stmt = dashboard_stmt.join(ServiceReview.customer).where(and_(*conditions)) dashboard_result = await db.execute(dashboard_stmt) total_quantity, medium_rating = dashboard_result.one() dashboard = { "total_quantity": total_quantity or 0, "medium_rating": float(medium_rating or 0) } return reviews, total, dashboard @staticmethod async def update_service_review_service( db: AsyncSession, service_review_id: int, new_status: ReviewStatusEnum ) -> ServiceReview: query = select(ServiceReview).where(ServiceReview.service_review_id == service_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ServiceReview with id={service_review_id} not found") old_status = review.status review.status = new_status review.date_modify = datetime.utcnow() db.add(review) await db.commit() await db.refresh(review) # Для сервісних відгуків рейтинг не рахуємо, оскільки немає окремої таблиці для сервісу # але логіку можна розширити в майбутньому return review @staticmethod async def delete_service_review( db: AsyncSession, service_review_id: int ): query = select(ServiceReview).where(ServiceReview.service_review_id == service_review_id) result = await db.execute(query) review = result.scalar_one_or_none() if not review: raise HTTPException(status_code=404, detail=f"ServiceReview with id={service_review_id} not found") await db.delete(review) await db.commit() @staticmethod async def recalculate_ratings(db: AsyncSession, review_type: str, entity_id: int) -> None: """ Перераховує рейтинг для сутності на основі затверджених відгуків Args: db: Сесія бази даних review_type: Тип відгуку ('store', 'manufacturer', 'product') entity_id: ID сутності для якої перераховуємо рейтинг """ if review_type == 'store': # Отримуємо середній рейтинг затверджених відгуків магазину avg_rating_stmt = select(func.avg(StoreReview.rating)).where( and_( StoreReview.store_id == entity_id, StoreReview.status == ReviewStatusEnum.approved ) ) result = await db.execute(avg_rating_stmt) avg_rating = result.scalar() # Оновлюємо рейтинг магазину store_stmt = select(Store).where(Store.store_id == entity_id) store_result = await db.execute(store_stmt) store = store_result.scalar_one_or_none() if store: store.rating = int(avg_rating) if avg_rating else 0 db.add(store) elif review_type == 'manufacturer': # Отримуємо середній рейтинг затверджених відгуків виробника avg_rating_stmt = select(func.avg(ManufacturerReview.rating)).where( and_( ManufacturerReview.manufacturer_id == entity_id, ManufacturerReview.status == ReviewStatusEnum.approved ) ) result = await db.execute(avg_rating_stmt) avg_rating = result.scalar() # Оновлюємо рейтинг виробника manufacturer_stmt = select(Manufacturer).where(Manufacturer.manufacturer_id == entity_id) manufacturer_result = await db.execute(manufacturer_stmt) manufacturer = manufacturer_result.scalar_one_or_none() if manufacturer: manufacturer.rating = float(avg_rating) if avg_rating else 0.0 db.add(manufacturer) elif review_type == 'product': # Отримуємо середній рейтинг затверджених відгуків продукту avg_rating_stmt = select(func.avg(ProductReview.rating)).where( and_( ProductReview.product_id == entity_id, ProductReview.status == ReviewStatusEnum.approved ) ) result = await db.execute(avg_rating_stmt) avg_rating = result.scalar() # Оновлюємо рейтинг продукту product_stmt = select(Product).where(Product.product_id == entity_id) product_result = await db.execute(product_stmt) product = product_result.scalar_one_or_none() if product: product.rating = float(avg_rating) if avg_rating else 0.0 db.add(product) await db.commit()