Support Group Admin Change Event and Member Restrict Event

efb_qq_plugin_go_cqhttp/GoCQHttp.py

@@ -26,9 +26,10 @@ from ehforwarderbot.exceptions import (
     EFBMessageError,
     EFBOperationNotSupported,
 )
+from ehforwarderbot.chat import GroupChat
 from ehforwarderbot.message import MessageCommand, MessageCommands
 from ehforwarderbot.status import MessageRemoval
-from ehforwarderbot.types import ChatID
+from ehforwarderbot.types import ChatID, MessageID
 from ehforwarderbot.utils import extra
 from PIL import Image
 from pkg_resources import resource_filename
@@ -49,8 +50,11 @@ from .Utils import (
     download_user_avatar,
     process_quote_text,
     qq_emoji_list,
+    strf_time,
 )
 
+all_group_list = []
+
 
 class GoCQHttp(BaseClient):
     client_name: str = "GoCQHttp Client"
@@ -204,12 +208,23 @@ class GoCQHttp(BaseClient):
             chat: Chat
             author: ChatMember
 
-            user = self.get_user_info(qq_uid)
+            try:
+                user = self.get_user_info(qq_uid)
+            except:
+                return
             if context["message_type"] == "private":
                 context["alias"] = user["remark"]
                 chat: PrivateChat = self.chat_manager.build_efb_chat_as_private(context)
             else:
-                chat = self.chat_manager.build_efb_chat_as_group(context)
+                try:
+                    chat = self.chat_manager.build_efb_chat_as_group(context)
+                except:
+                    return
+                is_discuss = False if context['message_type'] == 'group' else True
+                chat_uid = context['discuss_id'] if is_discuss else context['group_id']
+                if len(all_group_list) > 0 and chat_uid not in all_group_list:
+                    print(f"Filter 1 msg from {chat.uid} {chat.name}.")
+                    return
 
             if "anonymous" not in context or context["anonymous"] is None:
                 if context["message_type"] == "group":
@@ -236,13 +251,13 @@ class GoCQHttp(BaseClient):
 
             if main_text != "":
                 messages.append(self.msg_decorator.qq_text_simple_wrapper(main_text, at_dict))
-            uid: str = str(uuid.uuid4())
             coolq_msg_id = context["message_id"]
             for i in range(len(messages)):
                 if not isinstance(messages[i], Message):
                     continue
                 efb_msg: Message = messages[i]
-                efb_msg.uid = uid + "_" + str(coolq_msg_id) + "_" + str(i)
+                efb_msg.uid = f"{chat.uid.split('_')[-1]}_{coolq_msg_id}_{i}" if i > 0 else \
+                    f"{chat.uid.split('_')[-1]}_{coolq_msg_id}"
                 efb_msg.chat = chat
                 efb_msg.author = author
                 # if qq_uid != '80000000':
@@ -260,6 +275,9 @@ class GoCQHttp(BaseClient):
 
         @self.coolq_bot.on_notice("group_increase")
         def handle_group_increase_msg(context):
+            if len(all_group_list) > 0 and context["group_id"] not in all_group_list:
+                print(f"Filter 1 group_increase from {context['group_id']}.")
+                return
             context["event_description"] = self._("\u2139 Group Member Increase Event")
             if (context["sub_type"]) == "invite":
                 text = self._("{nickname}({context[user_id]}) joined the group({group_name}) via invitation")
@@ -281,6 +299,9 @@ class GoCQHttp(BaseClient):
 
         @self.coolq_bot.on_notice("group_decrease")
         def handle_group_decrease_msg(context):
+            if len(all_group_list) > 0 and context["group_id"] not in all_group_list:
+                print(f"Filter 1 group_decrease from {context['group_id']}.")
+                return
             context["event_description"] = self._("\u2139 Group Member Decrease Event")
             original_group = self.get_group_info(context["group_id"], False)
             group_name = context["group_id"]
@@ -302,6 +323,64 @@ class GoCQHttp(BaseClient):
             context["message"] = text
             self.send_efb_group_notice(context)
 
+        @self.coolq_bot.on_notice("group_admin")
+        def handle_group_admin_msg(context):
+            if len(all_group_list) > 0 and context["group_id"] not in all_group_list:
+                print(f"Filter 1 group_admin from {context['group_id']}.")
+                return
+            context["event_description"] = self._("\u2139 Group Admin Change Event")
+            if (context["sub_type"]) == "set":
+                text = self._("{nickname}({context[user_id]}) "
+                              "has been appointed as the group({group_name}) administrator")
+            else:
+                text = self._("{nickname}({context[user_id]}) "
+                              "has been de-appointed as the group({group_name}) administrator")
+
+            original_group = self.get_group_info(context["group_id"], False)
+            group_name = context["group_id"]
+            if original_group is not None and "group_name" in original_group:
+                group_name = original_group["group_name"]
+            text = text.format(
+                nickname=self.get_stranger_info(context["user_id"])["nickname"],
+                context=context,
+                group_name=group_name,
+            )
+
+            context["message"] = text
+            self.send_efb_group_notice(context)
+
+        @self.coolq_bot.on_notice("group_ban")
+        def handle_group_ban_msg(context):
+            if len(all_group_list) > 0 and context["group_id"] not in all_group_list:
+                print(f"Filter 1 group_ban from {context['group_id']}.")
+                return
+            context["event_description"] = self._("\u2139 Group Member Restrict Event")
+            if (context["sub_type"]) == "ban":
+                text = self._("{nickname}({context[user_id]}) "
+                              "is restricted for speaking for `{time}` at the group({group_name}) by "
+                              "{nickname_}({context[operator_id]})")
+                time_text = strf_time(context["duration"])
+            else:
+                text = self._("{nickname}({context[user_id]}) "
+                              "is lifted from restrictions at the group({group_name}) by "
+                              "{nickname_}({context[operator_id]}){time}")
+                time_text = ""
+
+            original_group = self.get_group_info(context["group_id"], False)
+            group_name = context["group_id"]
+            if original_group is not None and "group_name" in original_group:
+                group_name = original_group["group_name"]
+            text = text.format(
+                nickname=self.get_stranger_info(context["user_id"])["nickname"],
+                context=context,
+                time=time_text,
+                group_name=group_name,
+                nickname_=self.get_stranger_info(context["operator_id"])["nickname"],
+            )
+
+            context["message"] = text
+            self.send_efb_group_notice(context)
+
         @self.coolq_bot.on_notice("offline_file")
         def handle_offline_file_upload_msg(context):
             context["event_description"] = self._("\u2139 Offline File Upload Event")
@@ -320,6 +399,9 @@ class GoCQHttp(BaseClient):
 
         @self.coolq_bot.on_notice("group_upload")
         def handle_group_file_upload_msg(context):
+            if len(all_group_list) > 0 and context["group_id"] not in all_group_list:
+                print(f"Filter 1 file from {context['group_id']}.")
+                return
             context["event_description"] = self._("\u2139 Group File Upload Event")
             context["uid_prefix"] = "group_upload"
             original_group = self.get_group_info(context["group_id"], False)
@@ -358,6 +440,37 @@ class GoCQHttp(BaseClient):
             context["message"] = text
             self.send_msg_to_master(context)
 
+        @self.coolq_bot.on_notice("group_recall")
+        def handle_group_recall_msg(context):
+            coolq_msg_id = context["message_id"]
+            chat = GroupChat(channel=self.channel, uid=f"group_{context['group_id']}")
+            if len(all_group_list) > 0 and int(chat.uid.split('_')[-1]) not in all_group_list:
+                print(f"Filter 1 group recall from {chat.uid} {chat.name}.")
+                return
+            efb_msg = Message(
+                chat=chat,
+                uid=MessageID([[f"{chat.uid.split('_')[-1]}_{coolq_msg_id}"]])
+            )
+            coordinator.send_status(MessageRemoval(source_channel=self.channel,
+                                                   destination_channel=coordinator.master,
+                                                   message=efb_msg))
+
+        @self.coolq_bot.on_notice("friend_recall")
+        def handle_friend_recall_msg(context):
+            coolq_msg_id = context["message_id"]
+
+            try:
+                chat: PrivateChat = self.chat_manager.build_efb_chat_as_private(context)
+            except:
+                return
+            efb_msg = Message(
+                chat=chat,
+                uid=MessageID([[f"{chat.uid.split('_')[-1]}_{coolq_msg_id}"]])
+            )
+            coordinator.send_status(MessageRemoval(source_channel=self.channel,
+                                                   destination_channel=coordinator.master,
+                                                   message=efb_msg))
+
         @self.coolq_bot.on_request("friend")  # Add friend request
         def handle_add_friend_request(context):
             self.logger.debug(repr(context))
@@ -397,13 +510,13 @@ class GoCQHttp(BaseClient):
             context["group_id"] = str(context["group_id"]) + "_notification"
             context["message_type"] = "group"
             context["event_description"] = "\u2139 New Group Join Request"
-            original_group = self.get_group_info(context["group_id"], False)
+            original_group = self.get_group_info(context["group_id_orig"], False)
             group_name = context["group_id"]
             if original_group is not None and "group_name" in original_group:
                 group_name = original_group["group_name"]
             msg = Message()
             msg.uid = "group" + "_" + str(context["group_id"])
-            msg.author = self.chat_manager.build_efb_chat_as_system_user(context)
+            msg.author = (self.chat_manager.build_efb_chat_as_system_user(context)).other
             msg.chat = self.chat_manager.build_efb_chat_as_group(context)
             msg.deliver_to = coordinator.master
             msg.type = MsgType.Text
@@ -736,7 +849,7 @@ class GoCQHttp(BaseClient):
     def coolq_send_message(self, msg_type, uid, message):
         keyword = msg_type if msg_type != "private" else "user"
         res = self.coolq_api_query("send_msg", message_type=msg_type, **{keyword + "_id": uid}, message=message)
-        return str(uuid.uuid4()) + "_" + str(res["message_id"])
+        return str(uid) + "_" + str(res["message_id"])
 
     def _coolq_api_wrapper(self, func_name, **kwargs):
         try:
@@ -895,12 +1008,13 @@ class GoCQHttp(BaseClient):
             author = chat.get_member(SystemChatMember.SYSTEM_ID)
         except KeyError:
             author = chat.add_system_member()
+        event_description = context.get("event_description", "")
         msg = Message(
             uid="__group_notice__.%s" % int(time.time()),
             type=MsgType.Text,
             chat=chat,
             author=author,
-            text=context["message"],
+            text=event_description + "\n\n" + context["message"] if event_description else context["message"],
             deliver_to=coordinator.master,
         )
         coordinator.send_message(msg)

efb_qq_plugin_go_cqhttp/MsgDecorator.py

@@ -28,6 +28,13 @@ class QQMsgProcessor:
             efb_msg.text = self._("[Image Source missing]")
             return [efb_msg]
 
+        # flash
+        if 'type' in data:
+            if 'flash' == data['type']:
+                data['url'] = f'https://gchat.qpic.cn/gchatpic_new/1/1-1-' \
+                              f'{data["file"].replace(".image", "").upper()}/0?term=3%27'
+                efb_msg.text = self._('Send a flash picture.')
+
         efb_msg.file = cq_get_image(data["url"])
         if efb_msg.file is None:
             efb_msg.type = MsgType.Text

efb_qq_plugin_go_cqhttp/Utils.py

@@ -783,3 +783,15 @@ def download_voice(voice_url: str):
     else:
         audio_file = origin_file
     return audio_file
+
+
+def strf_time(seconds: int) -> str:
+    minutes, seconds = divmod(seconds, 60)
+    hours, minutes = divmod(minutes, 60)
+    days, hours = divmod(hours, 24)
+    text = ""
+    text += f"{days}d" if days else ""
+    text += f"{hours}h" if hours else ""
+    text += f"{minutes}m" if minutes else ""
+    text += f"{seconds}s" if seconds else ""
+    return text