notification_store.py

"""通知与公告的严格读取、白名单输出和锁内更新。"""
import copy
import logging
import math
import time
import uuid

from fastapi import HTTPException
import 数据库连接 as db

logger = logging.getLogger(__name__)
TTL = 7 * 24 * 3600
FIELDS = frozenset({"id", "type", "from_user", "from_name", "from_avatar", "target_item_id",
                    "target_item_title", "content", "is_read", "created_at", "receiver"})


def _unavailable():
    return HTTPException(status_code=503, detail="通知数据暂不可用")


def _timestamp(message):
    value = message["created_at"]
    return value / 1000 if value >= 1e12 else value


def _validate(message):
    if not isinstance(message, dict):
        raise _unavailable()
    if any(not isinstance(message.get(key), str) or not message[key] for key in ("id", "type")):
        raise _unavailable()
    timestamp = message.get("created_at")
    try:
        valid_timestamp = type(timestamp) in (int, float) and math.isfinite(timestamp) and timestamp >= 0
    except (OverflowError, ValueError):
        valid_timestamp = False
    if not valid_timestamp:
        raise _unavailable()
    if not isinstance(message.get("content", ""), str) or type(message.get("is_read", False)) is not bool:
        raise _unavailable()
    return message


def _messages(data, account):
    if not isinstance(data, dict) or not isinstance(data.get(account, []), list):
        raise _unavailable()
    messages = data.get(account, [])
    for message in messages:
        _validate(message)
        if "receiver" in message and message["receiver"] != account:
            raise _unavailable()
    return messages


def _announcements(data):
    if not isinstance(data, list):
        raise _unavailable()
    for message in data:
        if _validate(message)["type"] != "system":
            raise _unavailable()
    return data


def _read(name, default):
    try:
        return db.load_data(name, default_data=default, skip_cache=True, strict=True)
    except Exception as error:
        logger.warning("通知读取失败: %s", type(error).__name__)
        raise _unavailable() from None


def _mutate(name, updater, default):
    try:
        return db.atomic_update(name, updater, default_data=default)
    except HTTPException:
        raise
    except Exception as error:
        logger.warning("通知写入失败: %s", type(error).__name__)
        raise _unavailable() from None


def _profile(account):
    users = _read("users.json", {})
    if not isinstance(users, dict) or not isinstance(users.get(account, {}), dict):
        raise _unavailable()
    return users.get(account, {})


def publish_announcement(account, content):
    if not isinstance(content, str) or not content.strip():
        raise HTTPException(status_code=400, detail="公告内容不能为空")
    user = _profile(account)
    announcement = {"id": f"sys_{int(time.time())}_{uuid.uuid4().hex}", "type": "system", "from_user": account,
                    "from_name": user.get("name", account), "from_avatar": user.get("avatarDataUrl", ""),
                    "content": content, "created_at": int(time.time())}
    def update(data):
        _announcements(data).append(announcement)
    _mutate("announcements.json", update, [])


def append_notification(account, payload):
    source = payload.get("from_user")
    if not account or source == account:
        return
    user = _profile(source) if source and source != "system" else {}
    message = {"id": f"msg_{int(time.time())}_{uuid.uuid4().hex}", "type": payload.get("type"),
               "from_user": source, "from_name": user.get("name", source) if source and source != "system" else "系统通知",
               "from_avatar": user.get("avatarDataUrl", ""), "target_item_id": payload.get("target_item_id", ""),
               "target_item_title": payload.get("target_item_title", ""), "content": payload.get("content", ""),
               "is_read": False, "created_at": int(time.time())}
    _validate(message)
    def update(data):
        messages = _messages(data, account)
        data[account] = [message, *messages]
    _mutate("messages.json", update, {})


def get_messages(account, count_only=False):
    now = time.time()
    announcements = _announcements(_read("announcements.json", []))
    # 过期公告不重复注入已被清理的旧通知，也不在角标中重新出现。
    active = [message for message in announcements if now - _timestamp(message) < TTL]
    messages = _messages(_read("messages.json", {}), account)

    def combine(messages):
        result = copy.deepcopy(messages)
        ids = {message["id"] for message in result}
        for message in active:
            if message["id"] not in ids:
                result.append({**copy.deepcopy(message), "receiver": account, "is_read": False})
                ids.add(message["id"])
        return result

    def unread(messages):
        return sum(not message.get("is_read", False) and message["type"] not in ("private", "plugin_update")
                   and now - _timestamp(message) < TTL for message in messages)

    def project(messages):
        return [copy.deepcopy({key: value for key, value in message.items() if key in FIELDS}) for message in messages]

    combined = combine(messages)
    if count_only:
        return {"status": "success", "unread_count": unread(combined)}
    if len(combined) == len(messages) and all(message.get("is_read", False) and now - _timestamp(message) < TTL for message in messages):
        return {"status": "success", "data": project(messages), "unread_count": 0}

    def update(data):
        latest = combine(_messages(data, account))
        count = unread(latest)
        kept = [message for message in latest if not message.get("is_read", False) or now - _timestamp(message) < TTL]
        for message in kept:
            message["is_read"] = True
        kept.sort(key=_timestamp, reverse=True)
        data[account] = kept
        return {"status": "success", "data": project(kept), "unread_count": count}
    return _mutate("messages.json", update, {})


def mark_messages_read(account):
    messages = _messages(_read("messages.json", {}), account)
    if all(message.get("is_read", False) for message in messages):
        return
    def update(data):
        for message in _messages(data, account):
            message["is_read"] = True
    _mutate("messages.json", update, {})