"""消息通知 API""" from datetime import datetime from typing import Optional, List from fastapi import APIRouter, Depends, HTTPException from sqlalchemy import Column, String, DateTime, Integer, Text from sqlalchemy.orm import Session from pydantic import BaseModel, Field from app.core.database import get_db, Base from app.core.auth import get_current_user from app.core.response import success_response, error_response from app.core.utils import generate_id router = APIRouter(prefix="/api/web/admin/message", tags=["消息通知"]) class BizMessage(Base): """消息通知表(临时定义)""" __tablename__ = "biz_message" id = Column(String(32), primary_key=True, comment="主键ID") receiver_id = Column(String(32), nullable=False, comment="接收人ID") receiver_name = Column(String(100), nullable=True, comment="接收人姓名") message_type = Column(String(20), nullable=False, comment="消息类型") title = Column(String(200), nullable=False, comment="消息标题") content = Column(Text, nullable=True, comment="消息内容") is_read = Column(Integer, default=0, comment="是否已读: 0-未读, 1-已读") read_time = Column(DateTime, nullable=True, comment="阅读时间") is_del = Column(Integer, default=0, comment="是否删除") create_at = Column(DateTime, default=datetime.now, comment="创建时间") create_by = Column(String(32), nullable=True, comment="创建人ID") create_name = Column(String(100), nullable=True, comment="创建人姓名") update_at = Column(DateTime, default=datetime.now, onupdate=datetime.now, comment="修改时间") update_by = Column(String(32), nullable=True, comment="修改人ID") class MessageCRUD: """消息通知CRUD""" def __init__(self, model): self.model = model def create( self, db: Session, receiver_id: str, receiver_name: str, message_type: str, title: str, content: str = None, create_by: str = None, create_name: str = None ): data = { "id": generate_id(), "receiver_id": receiver_id, "receiver_name": receiver_name, "message_type": message_type, "title": title, "content": content, "is_read": 0, "create_by": create_by, "create_name": create_name } obj = self.model(**data) db.add(obj) db.commit() db.refresh(obj) return obj def get_unread_count(self, db: Session, receiver_id: str) -> int: """获取未读消息数量""" return db.query(self.model).filter( self.model.receiver_id == receiver_id, self.model.is_read == 0, self.model.is_del == 0 ).count() def get_list( self, db: Session, receiver_id: str = None, message_type: str = None, is_read: int = None, page: int = 1, page_size: int = 20 ): """获取消息列表""" query = db.query(self.model).filter(self.model.is_del == 0) if receiver_id: query = query.filter(self.model.receiver_id == receiver_id) if message_type: query = query.filter(self.model.message_type == message_type) if is_read is not None: query = query.filter(self.model.is_read == is_read) total = query.count() items = query.order_by(self.model.create_at.desc()).offset((page - 1) * page_size).limit(page_size).all() return items, total def mark_read(self, db: Session, message_id: str) -> bool: """标记消息已读""" message = db.query(self.model).filter( self.model.id == message_id, self.model.is_del == 0 ).first() if message: message.is_read = 1 message.read_time = datetime.now() db.commit() return True return False def mark_all_read(self, db: Session, receiver_id: str) -> int: """标记所有消息已读""" count = db.query(self.model).filter( self.model.receiver_id == receiver_id, self.model.is_read == 0, self.model.is_del == 0 ).update({ "is_read": 1, "read_time": datetime.now() }, synchronize_session=False) db.commit() return count def delete(self, db: Session, message_id: str) -> bool: """删除消息""" message = db.query(self.model).filter( self.model.id == message_id, self.model.is_del == 0 ).first() if message: message.is_del = 1 db.commit() return True return False message_crud = MessageCRUD(BizMessage) class MessageCreateRequest(BaseModel): """创建消息请求""" receiver_id: str = Field(..., description="接收人ID") message_type: str = Field(..., description="消息类型: system/order/delivery/payment") title: str = Field(..., description="消息标题") content: str = Field(None, description="消息内容") @router.post("/send", response_model=dict) def send_message( request: MessageCreateRequest, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """发送消息""" try: from app.crud.user import user_crud receiver = user_crud.get(db, request.receiver_id) receiver_name = receiver.real_name if receiver else None message = message_crud.create( db=db, receiver_id=request.receiver_id, receiver_name=receiver_name, message_type=request.message_type, title=request.title, content=request.content, create_by=current_user.get("user_id"), create_name=current_user.get("real_name") ) return success_response({ "id": message.id, "message": "消息发送成功" }) except Exception as e: return error_response(500, f"发送消息失败: {str(e)}") @router.get("/list", response_model=dict) def get_message_list( page: int = 1, page_size: int = 20, receiver_id: str = None, message_type: str = None, is_read: int = None, start_date: str = None, end_date: str = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """获取消息列表""" try: items, total = message_crud.get_list( db=db, receiver_id=receiver_id, message_type=message_type, is_read=is_read, page=page, page_size=page_size ) return success_response({ "list": [_to_response(item) for item in items], "total": total, "page": page, "page_size": page_size }) except Exception as e: return error_response(500, f"获取消息列表失败: {str(e)}") @router.get("/unread-count", response_model=dict) def get_unread_count( receiver_id: str = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """获取未读消息数量""" try: if not receiver_id: receiver_id = current_user.get("user_id") count = message_crud.get_unread_count(db, receiver_id) return success_response({"count": count}) except Exception as e: return error_response(500, f"获取未读数量失败: {str(e)}") @router.post("/mark-read/{message_id}", response_model=dict) def mark_read( message_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """标记消息已读""" try: success = message_crud.mark_read(db, message_id) if success: return success_response({"message": "标记成功"}) return error_response(404, "消息不存在") except Exception as e: return error_response(500, f"标记失败: {str(e)}") @router.post("/mark-all-read", response_model=dict) def mark_all_read( receiver_id: str = None, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """标记所有消息已读""" try: if not receiver_id: receiver_id = current_user.get("user_id") count = message_crud.mark_all_read(db, receiver_id) return success_response({"count": count, "message": "标记成功"}) except Exception as e: return error_response(500, f"标记失败: {str(e)}") @router.delete("/delete/{message_id}", response_model=dict) def delete_message( message_id: str, db: Session = Depends(get_db), current_user: dict = Depends(get_current_user) ): """删除消息""" try: success = message_crud.delete(db, message_id) if success: return success_response({"message": "删除成功"}) return error_response(404, "消息不存在") except Exception as e: return error_response(500, f"删除失败: {str(e)}") def _to_response(message: BizMessage) -> dict: """转换为响应格式""" return { "id": message.id, "receiver_id": message.receiver_id, "receiver_name": message.receiver_name, "message_type": message.message_type, "title": message.title, "content": message.content, "is_read": message.is_read, "read_time": str(message.read_time) if message.read_time else None, "create_at": str(message.create_at) if message.create_at else None }