"""私信按真实参与者读取，所有消息修改使用锁内最新记录。""" import copy import logging import math import re import time import uuid from fastapi import HTTPException import 数据库连接 as db logger = logging.getLogger(__name__) HISTORY_TTL = 7 * 24 * 3600 MAX_MESSAGE_LENGTH = 1000 MESSAGE_FIELDS = frozenset({"id", "sender", "receiver", "content", "created_at", "is_read"}) 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(data): if not isinstance(data, dict): raise _unavailable() for key, messages in data.items(): if not isinstance(key, str) or not isinstance(messages, list): raise _unavailable() for message in messages: if not isinstance(message, dict): raise _unavailable() if any(not isinstance(message.get(field), str) or not message[field] for field in ("id", "sender", "receiver")): raise _unavailable() timestamp = message.get("created_at") if not isinstance(message.get("content"), str) or type(timestamp) not in (int, float): raise _unavailable() try: valid_timestamp = math.isfinite(timestamp) and timestamp >= 0 except (OverflowError, ValueError): valid_timestamp = False if not valid_timestamp or type(message.get("is_read", False)) is not bool: raise _unavailable() return data def _read(name): try: return db.load_data(name, default_data={}, skip_cache=True, strict=True) except Exception as error: logger.warning("聊天读取失败: %s", type(error).__name__) raise _unavailable() from None def _mutate(updater): try: return db.atomic_update("chats.json", lambda data: updater(_validate(data)), default_data={}) except HTTPException: raise except Exception as error: logger.warning("聊天写入失败: %s", type(error).__name__) raise _unavailable() from None def _pair(message, account, target): return ((message["sender"] == account and message["receiver"] == target) or (message["sender"] == target and message["receiver"] == account)) def _project(message): result = copy.deepcopy({key: value for key, value in message.items() if key in MESSAGE_FIELDS}) result.setdefault("is_read", False) return result def send_message(sender, receiver, content, client_request_id=None): if client_request_id is not None and (not isinstance(client_request_id, str) or re.fullmatch(r"[0-9a-f]{32}", client_request_id) is None): raise HTTPException(status_code=400, detail="发送请求标识不正确") if not isinstance(receiver, str) or not receiver or not isinstance(content, str) or not content.strip() or len(content) > MAX_MESSAGE_LENGTH: raise HTTPException(status_code=400, detail="私信内容或接收者不正确") users = _read("users.json") if not isinstance(users, dict): raise _unavailable() if receiver not in users: raise HTTPException(status_code=404, detail="接收者不存在") if not isinstance(users[receiver], dict) or users[receiver].get("account", receiver) != receiver: raise _unavailable() message = {"id": f"chat_{int(time.time())}_{uuid.uuid4().hex}", "sender": sender, "receiver": receiver, "content": content, "created_at": int(time.time()), "is_read": False} key = f"{min(sender, receiver)}_{max(sender, receiver)}" def append(data): if client_request_id is not None: for messages in data.values(): for existing in messages: if existing["sender"] == sender and existing.get("client_request_id") == client_request_id: if existing["receiver"] != receiver or existing["content"] != content: raise HTTPException(status_code=409, detail="发送请求标识已用于其它消息") return {"id": existing["id"], "client_request_id": client_request_id} message["client_request_id"] = client_request_id data.setdefault(key, []).append(message) if client_request_id is not None: return {"id": message["id"], "client_request_id": client_request_id} return _mutate(append) def chat_list(account): data = _validate(_read("chats.json")) users = _read("users.json") if not isinstance(users, dict): raise _unavailable() conversations = {} # 旧键的下划线编码可能相撞，账号子串也不能代表参与者。 for messages in data.values(): for message in messages: if message["sender"] == account: target = message["receiver"] elif message["receiver"] == account: target = message["sender"] else: continue conversations.setdefault(target, []).append(message) result = [] for target, messages in conversations.items(): user = users.get(target, {}) if not isinstance(user, dict): raise _unavailable() last = max(messages, key=_timestamp) result.append({"account": target, "name": user.get("name", target), "avatar": user.get("avatarDataUrl", ""), "last_message": last["content"], "last_time": last["created_at"], "unread_count": sum(message["receiver"] == account and not message.get("is_read", False) for message in messages)}) result.sort(key=lambda item: item["last_time"] / 1000 if item["last_time"] >= 1e12 else item["last_time"], reverse=True) return result def chat_history(account, target): now = time.time() data = _validate(_read("chats.json")) def expires(message): return message.get("is_read", False) and now - _timestamp(message) >= HISTORY_TTL def matching(message): return _pair(message, account, target) def needs_update(message): return matching(message) and (expires(message) or (message["receiver"] == account and not message.get("is_read", False))) def snapshot(data): return sorted((_project(message) for messages in data.values() for message in messages if matching(message)), key=_timestamp) if not any(needs_update(message) for messages in data.values() for message in messages): return snapshot(data) def update(data): for key, messages in data.items(): kept = [] for message in messages: if matching(message): if expires(message): continue if message["receiver"] == account: message["is_read"] = True kept.append(message) data[key] = kept return snapshot(data) return _mutate(update)