buff3.0/backend-client/app/services/announcement_service.py

"""
公告服务
"""
from typing import Tuple, List, Dict, Any
from sqlalchemy.orm import Session
from app.models.announcement import Announcement
from datetime import datetime
from app.core.logger import logger


class AnnouncementService:
    """公告服务"""
    
    def __init__(self, db: Session):
        self.db = db
    
    def get_announcements(self, page: int = 1, page_size: int = 10, 
                          include_inactive: bool = False) -> Tuple[bool, str, dict]:
        """
        获取公告列表
        
        Args:
            page: 页码(从1开始)
            page_size: 每页数量
            include_inactive: 是否包含未启用的公告
            
        Returns:
            (success, message, data)
        """
        try:
            # 构建查询
            query = self.db.query(Announcement).filter(
                Announcement.is_deleted == 0
            )
            
            if not include_inactive:
                query = query.filter(Announcement.is_active == 1)
            
            # 统计总数
            total = query.count()
            
            # 分页查询：重要公告在前（priority 降序），同优先级按创建时间最新在前（created_at 降序）
            offset = (page - 1) * page_size
            announcements = query.order_by(
                Announcement.priority.desc(),
                Announcement.created_at.desc()
            ).offset(offset).limit(page_size).all()
            
            # 构建响应
            announcement_list = []
            for ann in announcements:
                announcement_list.append({
                    "id": ann.id,
                    "title": ann.title,
                    "content": ann.content,
                    "priority": ann.priority,
                    "publish_time": ann.publish_time.isoformat() if ann.publish_time else None,
                    "created_at": ann.created_at.isoformat() if ann.created_at else None
                })
            
            return True, "成功", {
                "announcements": announcement_list,
                "total": total,
                "page": page,
                "page_size": page_size
            }
            
        except Exception as e:
            logger.exception(f"获取公告列表失败: {e}")
            return False, f"获取公告失败: {str(e)}", {}
    
    def get_latest_announcement(self) -> Tuple[bool, str, dict]:
        """获取最新公告"""
        try:
            announcement = self.db.query(Announcement).filter(
                Announcement.is_deleted == 0,
                Announcement.is_active == 1
            ).order_by(
                Announcement.priority.desc(),
                Announcement.created_at.desc()
            ).first()
            
            if not announcement:
                return False, "暂无公告", {}
            
            return True, "成功", {
                "id": announcement.id,
                "title": announcement.title,
                "content": announcement.content,
                "priority": announcement.priority,
                "publish_time": announcement.publish_time.isoformat() if announcement.publish_time else None
            }
            
        except Exception as e:
            logger.exception(f"获取最新公告失败: {e}")
            return False, f"获取公告失败: {str(e)}", {}