document-management/backend/dms/services/notification_service.py

"""站内通知服务：列表、未读数、全部已读与待办处理。"""

from __future__ import annotations

from datetime import UTC, datetime

from sqlalchemy import func, select, update

from dms.common.errors import ConflictError, InvalidArgumentError, ResourceNotFoundError
from dms.database.transaction import transaction
from dms.extensions import db
from dms.models.notification import Notification

# 列表返回条数上限：通知为待办提醒而非历史归档，固定取最近一批即可
LIST_LIMIT = 200
HANDLE_ACTIONS = ("AGREE", "REJECT")


def _format_utc(value: datetime | None) -> str | None:
    if value is None:
        return None
    return (
        value.replace(tzinfo=UTC)
        .isoformat(timespec="milliseconds")
        .replace("+00:00", "Z")
    )


def _serialize(item: Notification) -> dict[str, object]:
    return {
        "id": str(item.id),
        "title": item.title,
        "content": item.content,
        "notifyType": item.notify_type,
        "action": item.action,
        "status": item.status,
        "createdAt": _format_utc(item.created_at),
        "handledAt": _format_utc(item.handled_at),
    }


def list_notifications(user_id: int) -> dict[str, object]:
    """当前用户通知列表（id 倒序，最近 LIMIT 条）。"""
    items = db.session.scalars(
        select(Notification)
        .where(Notification.user_id == user_id)
        .order_by(Notification.id.desc())
        .limit(LIST_LIMIT)
    ).all()
    return {"items": [_serialize(item) for item in items]}


def unread_count(user_id: int) -> dict[str, object]:
    """未读数（红圈徽标数据源）。"""
    count = db.session.scalar(
        select(func.count())
        .select_from(Notification)
        .where(Notification.user_id == user_id, Notification.status == "UNREAD")
    )
    return {"count": int(count or 0)}


def mark_all_read(user_id: int) -> dict[str, object]:
    """全部标记已读（打开通知页后清红圈）。"""
    with transaction() as session:
        result = session.execute(
            update(Notification)
            .where(Notification.user_id == user_id, Notification.status == "UNREAD")
            .values(status="READ")
        )
    return {"updated": int(result.rowcount or 0)}


def handle_notification(user_id: int, notification_id: int, action: str) -> dict[str, object]:
    """处理待办通知：仅 PENDING 且归属本人的通知可同意/拒绝。"""
    if action not in HANDLE_ACTIONS:
        raise InvalidArgumentError("action必须是AGREE或REJECT")

    with transaction() as session:
        item = session.scalar(
            select(Notification).where(
                Notification.id == notification_id,
                Notification.user_id == user_id,
            )
        )
        if item is None:
            raise ResourceNotFoundError("通知不存在")
        if item.action != "PENDING":
            raise ConflictError("该通知已处理或无需处理")
        item.action = action
        item.status = "READ"
        item.handled_at = datetime.now(UTC).replace(tzinfo=None)
        handled = _serialize(item)

    return handled