mtsd/backend/app/crud/product_review.py

"""CRUD - 售后评价"""
from typing import List, Optional
from sqlalchemy.orm import Session
from sqlalchemy import and_, func

from app.crud.base import CRUDBase, HistoryCRUD
from app.models.product_review import BizProductReview, BizProductReviewHis
from app.schemas.product_review import ProductReviewCreate, ProductReviewReplyRequest


class ProductReviewCRUD(CRUDBase[BizProductReview]):
    """售后评价CRUD操作"""

    def get_by_product(
        self,
        db: Session,
        product_id: str,
        skip: int = 0,
        limit: int = 10,
        status: str = "approved",
        has_media: int = None
    ) -> tuple:
        """获取商品的评价列表"""
        query = db.query(self.model).filter(
            self.model.product_id == product_id,
            self.model.is_del == 0
        )
        
        if status:
            query = query.filter(self.model.status == status)
        
        if has_media == 1:
            from app.models.product_review_media import BizProductReviewMedia
            review_ids_with_media = db.query(BizProductReviewMedia.review_id).filter(
                BizProductReviewMedia.is_del == 0
            ).distinct().subquery()
            query = query.filter(self.model.id.in_(review_ids_with_media))
        
        total = query.count()
        
        items = query.order_by(
            self.model.is_top.desc(),
            self.model.create_at.desc()
        ).offset(skip).limit(limit).all()
        
        return items, total

    def get_admin_list(
        self,
        db: Session,
        skip: int = 0,
        limit: int = 10,
        product_id: str = None,
        product_name: str = None,
        user_name: str = None,
        status: str = None,
        rating: int = None,
        start_date: str = None,
        end_date: str = None,
    ) -> tuple:
        """管理员获取评价列表（支持多条件筛选）"""
        query = db.query(self.model).filter(self.model.is_del == 0)
        
        if product_id:
            query = query.filter(self.model.product_id == product_id)
        
        if user_name:
            query = query.filter(self.model.user_name.like(f"%{user_name}%"))
        
        if status:
            query = query.filter(self.model.status == status)
        
        if rating:
            query = query.filter(self.model.rating == rating)
        
        if start_date:
            query = query.filter(self.model.create_at >= start_date)
        
        if end_date:
            query = query.filter(self.model.create_at <= end_date)
        
        total = query.count()
        
        items = query.order_by(
            self.model.is_top.desc(),
            self.model.create_at.desc()
        ).offset(skip).limit(limit).all()
        
        return items, total

    def get_statistics(self, db: Session, product_id: str = None) -> dict:
        """获取评价统计"""
        query = db.query(self.model).filter(
            self.model.is_del == 0,
            self.model.status == "approved"
        )
        
        if product_id:
            query = query.filter(self.model.product_id == product_id)
        
        total_count = query.count()
        
        if total_count > 0:
            avg_rating = db.query(func.avg(self.model.rating)).filter(
                self.model.is_del == 0,
                self.model.status == "approved",
                self.model.product_id == product_id if product_id else True
            ).scalar() or 0
        else:
            avg_rating = 0
        
        rating_dist = {}
        for r in range(1, 6):
            count = db.query(self.model).filter(
                self.model.is_del == 0,
                self.model.status == "approved",
                self.model.rating == r,
                self.model.product_id == product_id if product_id else True
            ).count()
            rating_dist[str(r)] = count
        
        pending_count = db.query(self.model).filter(
            self.model.is_del == 0,
            self.model.status == "pending",
            self.model.product_id == product_id if product_id else True
        ).count()
        
        rejected_count = db.query(self.model).filter(
            self.model.is_del == 0,
            self.model.status == "rejected",
            self.model.product_id == product_id if product_id else True
        ).count()
        
        return {
            "total_count": total_count,
            "pending_count": pending_count,
            "approved_count": total_count,
            "rejected_count": rejected_count,
            "avg_rating": round(float(avg_rating), 1),
            "rating_distribution": rating_dist,
        }

    def reply(
        self,
        db: Session,
        review_id: str,
        reply_content: str,
        updater: dict = None
    ) -> Optional[BizProductReview]:
        """回复评价"""
        review = self.get(db, review_id)
        if review:
            review.reply_content = reply_content
            review.reply_time = func.now()
            if updater:
                review.reply_by = updater.get("user_id")
                review.reply_by_name = updater.get("real_name")
            db.commit()
            db.refresh(review)
        return review

    def reject(
        self,
        db: Session,
        review_id: str,
        reject_reason: str,
        updater: dict = None
    ) -> Optional[BizProductReview]:
        """拒绝评价"""
        review = self.get(db, review_id)
        if review:
            review.status = "rejected"
            review.reject_reason = reject_reason
            db.commit()
            db.refresh(review)
        return review

    def approve(
        self,
        db: Session,
        review_id: str
    ) -> Optional[BizProductReview]:
        """审核通过评价"""
        review = self.get(db, review_id)
        if review:
            review.status = "approved"
            db.commit()
            db.refresh(review)
        return review

    def toggle_top(
        self,
        db: Session,
        review_id: str,
        is_top: int
    ) -> Optional[BizProductReview]:
        """切换置顶状态"""
        review = self.get(db, review_id)
        if review:
            review.is_top = is_top
            db.commit()
            db.refresh(review)
        return review

    def like(
        self,
        db: Session,
        review_id: str
    ) -> Optional[BizProductReview]:
        """点赞评价"""
        review = self.get(db, review_id)
        if review:
            review.like_count = (review.like_count or 0) + 1
            db.commit()
            db.refresh(review)
        return review

    def user_can_review(
        self,
        db: Session,
        user_id: str,
        product_id: str,
        order_id: str = None
    ) -> bool:
        """检查用户是否已评价过该商品"""
        query = db.query(self.model).filter(
            self.model.user_id == user_id,
            self.model.product_id == product_id,
            self.model.is_del == 0
        )
        if order_id:
            query = query.filter(self.model.order_id == order_id)
        return query.first() is not None


class ProductReviewHisCRUD(HistoryCRUD[BizProductReviewHis]):
    """售后评价历史CRUD操作"""
    pass


product_review = ProductReviewCRUD(BizProductReview)
product_review_his = ProductReviewHisCRUD(BizProductReviewHis)