Spaces:
Running
Running
| """通知与公告的严格读取、白名单输出和锁内更新。""" | |
| 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, {}) | |