import json as json_ import queue from datetime import datetime from core import context as runtime_context from core.event import Event import websocket import collections from core.logger import logger from core.database_manager import DbManager from core.cq import * class Echo: def __init__(self): self.echo_num = 0 self.echo_list = collections.deque(maxlen=20) def get(self): self.echo_num += 1 q = queue.Queue(maxsize=1) self.echo_list.append((self.echo_num, q)) return self.echo_num, q def match(self, context: dict): for obj in self.echo_list: if context["echo"] == obj[0]: obj[1].put(context) echo : Echo WS_APP : websocket.WebSocketApp class ApiWrapper: def __init__(self, raw_context: dict): self.ws = WS_APP self.context = Event(raw_context) self.dbmanager = DbManager() def call_api(self, action: str, params: dict) -> dict: echo_num, q = echo.get() data = json_.dumps({"action": action, "params": params, "echo": echo_num}) logger.info("发送调用 <- " + data) self.ws.send(data) try: # 阻塞至响应或者等待30s超时 return q.get(timeout=30) except queue.Empty: logger.error("API调用[{echo_num}] 超时......") return {"status": "failed"} def _build_title_prefix(self, user_id): try: equipped_titles = self.dbmanager.titles.equipped_all(user_id)[:3] if len(equipped_titles) == 0: return "" names = [] from plugins.title import get_title_def for tid in equipped_titles: data = get_title_def(tid) if data and data.get("name"): names.append(data["name"]) if len(names) == 0: return "" return "「{}」".format("·".join(names)) except Exception: return "" def _inject_titles_before_at(self, message): merged = [] for seg in message: if isinstance(seg, dict) and seg.get("type") == "at": qq = seg.get("data", {}).get("qq") if qq != "all": prefix = self._build_title_prefix(qq) if prefix: merged.append(text(prefix + " ")) merged.append(seg) return tuple(merged) def send_msg(self, *message) -> int: # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#send_msg-%E5%8F%91%E9%80%81%E6%B6%88%E6%81%AF message = self._inject_titles_before_at(message) if self.context.group_id != None: return self.send_group_msg(*message) elif self.context.user_id != None: return self.send_private_msg(*message) else: # 默认发送到群聊 return self.send_group_msg(*message) def send_private_msg(self, *message) -> int: # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#send_private_msg-%E5%8F%91%E9%80%81%E7%A7%81%E8%81%8A%E6%B6%88%E6%81%AF params = {"user_id": self.context.user_id, "message": message} ret = self.call_api("send_private_msg", params) return 0 if ret.get("status") != "ok" else ret.get("data", {}).get("message_id", 0) def send_group_msg(self, *message) -> int: # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#send_group_msg-%E5%8F%91%E9%80%81%E7%BE%A4%E6%B6%88%E6%81%AF group_id = self.context.group_id if not group_id: group_id = runtime_context.DEFAULT_GROUP_ID params = {"group_id": group_id, "message": message} ret = self.call_api("send_group_msg", params) msg_id = 0 if ret.get("status") != "ok" else ret.get("data", {}).get("message_id", 0) # 录制 bot 输出 if group_id and msg_id and runtime_context.is_group_recording(group_id): session = runtime_context.get_recording_session(group_id) if session: from core.base import BOT_QQ entry = { "type": "bot", "nickname": "小埃同学", "user_id": str(BOT_QQ), "message": list(message), "time": int(datetime.now().timestamp()), } session["messages"].append(entry) if str(BOT_QQ) not in session["participants"]: session["participants"][str(BOT_QQ)] = {"nickname": "小埃同学", "user_id": str(BOT_QQ)} return msg_id def get_group_member_info(self, user_id): #https://github.com/botuniverse/onebot-11/blob/master/api/public.md#get_group_member_info-%E8%8E%B7%E5%8F%96%E7%BE%A4%E6%88%90%E5%91%98%E4%BF%A1%E6%81%AF params = {"group_id": self.context.group_id, "user_id": user_id} ret = self.call_api("get_group_member_info", params) return ret.get("data", {}) def get_image(self, file): # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#get_image-%E8%8E%B7%E5%8F%96%E5%9B%BE%E7%89%87 params = {"file": file} ret = self.call_api("get_image", params) if ret.get("status") == "ok": return ret.get("data", {}).get("file", "") else : return "" def get_image_url(self, file): # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#get_image-%E8%8E%B7%E5%8F%96%E5%9B%BE%E7%89%87 params = {"file": file} ret = self.call_api("get_image", params) if ret.get("status") == "ok": return ret.get("data", {}).get("url", "") else : return "" def get_qq_avatar(self, user_id) -> str: """OneBot `get_qq_avatar`，成功返回头像 URL，失败返回空串。""" params = {"user_id": user_id} ret = self.call_api("get_qq_avatar", params) if not ret or ret.get("status") != "ok": return "" data = ret.get("data") if not isinstance(data, dict): return "" return data.get("url") or "" def set_friend_add_request(self, flag, approve = True): params = {"flag": flag, "approve" : approve, "remark": ""} self.call_api("set_friend_add_request", params) def send_forward_msg(self, message: list): if self.context.group_id != None: return self.send_group_forward_msg(message) else: return self.send_private_forward_msg(message) def send_group_forward_msg(self, message: list): group_id = self.context.group_id if not group_id: group_id = runtime_context.DEFAULT_GROUP_ID params = {"group_id": group_id, "messages": forward(message)} ret = self.call_api("send_group_forward_msg", params) return 0 if ret.get("status") != "ok" else 1 def send_private_forward_msg(self, message: list): params = {"user_id": self.context.user_id, "messages": forward(message)} ret = self.call_api("send_private_forward_msg", params) return 0 if ret.get("status") != "ok" else 1 def get_group_album_list(self, group_id): params = {"group_id": group_id} ret = self.call_api("get_group_album_list", params) return ret.get("data", {}) def set_group_special_title(self, group_id, user_id, title): params = {"group_id": group_id, "user_id": user_id, "special_title": title} ret = self.call_api("set_group_special_title", params) return 0 def get_msg(self, message_id): params = {"message_id": message_id} ret = self.call_api("get_msg", params) if ret and ret.get("status") == "ok": return ret.get("data", {}) return {} def delete_msg(self, message_id: int) -> bool: # https://github.com/botuniverse/onebot-11/blob/master/api/public.md#delete_msg-%E6%92%A4%E5%9B%9E%E6%B6%88%E6%81%AF ret = self.call_api("delete_msg", {"message_id": int(message_id)}) return bool(ret and ret.get("status") == "ok") def set_essence_msg(self, message_id: int) -> bool: # 扩展 API：设置群精华消息（如 NapCat / Lagrange.OneBot 等实现） ret = self.call_api("set_essence_msg", {"message_id": int(message_id)}) return bool(ret and ret.get("status") == "ok") def delete_essence_msg(self, message_id: int) -> bool: ret = self.call_api("delete_essence_msg", {"message_id": int(message_id)}) return bool(ret and ret.get("status") == "ok")