baoxiang/backend/app/services/announcement_service.py

"""
公告服务
"""
from sqlalchemy.orm import Session
from sqlalchemy import and_, desc, or_
from typing import List, Optional
from datetime import datetime
from ..models.announcement import Announcement, AnnouncementType
from ..schemas.announcement import (
    AnnouncementCreate,
    AnnouncementUpdate,
)


class AnnouncementService:
    """公告服务"""

    @staticmethod
    def create_announcement(db: Session, announcement_data: AnnouncementCreate) -> Announcement:
        """创建公告"""
        db_announcement = Announcement(**announcement_data.dict())
        db.add(db_announcement)
        db.commit()
        db.refresh(db_announcement)
        return db_announcement

    @staticmethod
    def get_announcement(db: Session, announcement_id: int) -> Optional[Announcement]:
        """获取公告"""
        return db.query(Announcement).filter(Announcement.id == announcement_id).first()

    @staticmethod
    def get_announcements(
        db: Session,
        skip: int = 0,
        limit: int = 20,
        include_expired: bool = False
    ) -> List[Announcement]:
        """获取公告列表"""
        query = db.query(Announcement)

        # 排除过期的公告（除非特别要求）
        if not include_expired:
            now = datetime.now()
            query = query.filter(
                or_(
                    Announcement.expires_at.is_(None),
                    Announcement.expires_at > now
                )
            )

        return query.order_by(desc(Announcement.priority), desc(Announcement.created_at)).offset(skip).limit(limit).all()

    @staticmethod
    def get_active_announcements(db: Session, limit: int = 5) -> List[Announcement]:
        """获取活跃公告（置顶优先）"""
        now = datetime.now()
        return db.query(Announcement).filter(
            or_(
                Announcement.starts_at.is_(None),
                Announcement.starts_at <= now
            ),
            or_(
                Announcement.expires_at.is_(None),
                Announcement.expires_at > now
            )
        ).order_by(desc(Announcement.is_pinned), desc(Announcement.priority), desc(Announcement.created_at)).limit(limit).all()

    @staticmethod
    def update_announcement(
        db: Session,
        announcement: Announcement,
        announcement_data: AnnouncementUpdate
    ) -> Announcement:
        """更新公告"""
        update_data = announcement_data.dict(exclude_unset=True)
        for field, value in update_data.items():
            setattr(announcement, field, value)

        db.commit()
        db.refresh(announcement)
        return announcement

    @staticmethod
    def delete_announcement(db: Session, announcement_id: int) -> bool:
        """删除公告"""
        announcement = db.query(Announcement).filter(Announcement.id == announcement_id).first()
        if not announcement:
            return False

        db.delete(announcement)
        db.commit()
        return True

    @staticmethod
    def pin_announcement(db: Session, announcement_id: int, pinned: bool = True) -> bool:
        """置顶/取消置顶公告"""
        announcement = db.query(Announcement).filter(Announcement.id == announcement_id).first()
        if not announcement:
            return False

        announcement.is_pinned = pinned
        if pinned:
            announcement.priority = 100  # 设置高优先级

        db.commit()
        return True

    @staticmethod
    def set_announcement_priority(db: Session, announcement_id: int, priority: int) -> bool:
        """设置公告优先级"""
        announcement = db.query(Announcement).filter(Announcement.id == announcement_id).first()
        if not announcement:
            return False

        announcement.priority = max(0, min(100, priority))
        db.commit()
        return True

    @staticmethod
    def get_pinned_announcements(db: Session) -> List[Announcement]:
        """获取置顶公告"""
        now = datetime.now()
        return db.query(Announcement).filter(
            Announcement.is_pinned == True,
            or_(
                Announcement.starts_at.is_(None),
                Announcement.starts_at <= now
            ),
            or_(
                Announcement.expires_at.is_(None),
                Announcement.expires_at > now
            )
        ).order_by(desc(Announcement.priority), desc(Announcement.created_at)).all()