Dev/wikilog

core/parser/message.py

@@ -343,9 +343,11 @@ async def execute_submodule(msg: Bot.MessageSession, command_first_word, command
                                     func_params = inspect.signature(submodule.function).parameters
                                     if len(func_params) > 1 and msg.parsed_msg:
                                         parsed_msg_ = msg.parsed_msg.copy()
+                                        no_message_session = True
                                         for param_name, param_obj in func_params.items():
                                             if param_obj.annotation == Bot.MessageSession:
                                                 kwargs[param_name] = msg
+                                                no_message_session = False
                                             param_name_ = param_name
                                             if (param_name__ := f'<{param_name}>') in parsed_msg_:
                                                 param_name_ = param_name__
@@ -368,7 +370,9 @@ async def execute_submodule(msg: Bot.MessageSession, command_first_word, command
                                                         kwargs[param_name_] = param_obj.default
                                                     else:
                                                         kwargs[param_name_] = None
-
+                                        if no_message_session:
+                                            Logger.warning(f'{submodule.function.__name__} has no Bot.MessageSession parameter, did you forgot to add it?\n'
+                                                           'Remember: MessageSession IS NOT Bot.MessageSession')
                                     else:
                                         kwargs[func_params[list(func_params.keys())[0]].name] = msg
 

core/queue.py

@@ -94,5 +94,7 @@ async def check_job_queue():
                     Logger.error(traceback.format_exc())
                     return_val(tsk, {'send': False})
 
-        except Exception as e:
-            return_val(tsk, {'traceback': traceback.format_exc()}, status=False)
+        except Exception:
+            f = traceback.format_exc()
+            Logger.error(f)
+            return_val(tsk, {'traceback': f}, status=False)

modules/wiki/locales/zh_cn.json

@@ -146,7 +146,7 @@
     "wiki.message.user.username": "用户：",
     "wiki.message.user.users_groups": "用户组：",
     "wiki.message.user.wiki_unavailable": "${wikiurl} 不可用。",
-    "wiki.message.utils.redacted": "检测到外来信息介入，请前往滥用日志查看所有消息。",
+    "wiki.message.utils.redacted": "检测到特定文字被屏蔽，请前往日志查看所有消息。",
     "wiki.message.utils.wikilib.error.empty": "发生错误：API 未返回任何内容，请联系此站点管理员获取原因。",
     "wiki.message.utils.wikilib.error.text_too_long": "发生错误：请求的页面段落文本过长，无法显示。",
     "wiki.message.utils.wikilib.get_failed.api": "从 API 获取信息时出错：",

modules/wiki/utils/ab.py

@@ -1,7 +1,7 @@
 from core.builtins import Url, Bot
 from core.dirty_check import check
 from modules.wiki.utils.time import strptime2ts
-from modules.wiki.utils.wikilib import WikiLib
+from modules.wiki.utils.wikilib import WikiLib, WikiInfo
 
 
 async def ab(msg: Bot.MessageSession, wiki_url):
@@ -29,3 +29,47 @@ async def ab(msg: Bot.MessageSession, wiki_url):
             msg.locale.t("wiki.message.utils.redacted")}'
     else:
         return f'{str(Url(pageurl))}\n{y}\n{msg.locale.t("message.collapse", amount="5")}'
+
+
+async def convert_ab_to_detailed_format(abl: list, wiki_info: WikiInfo, msg: Bot.MessageSession):
+    ablist = []
+    userlist = []
+    titlelist = []
+    for x in abl:
+        if 'title' in x:
+            userlist.append(x['user'])
+            titlelist.append(x['title'])
+    checked_userlist = await check(*userlist)
+    user_checked_map = {}
+    for u in checked_userlist:
+        user_checked = u['content']
+        if user_checked.find("<吃掉了>") != -1 or user_checked.find("<全部吃掉了>") != -1:
+            user_checked = user_checked.replace("<吃掉了>", msg.locale.t(
+                "check.redacted") + '\n' + wiki_info.articlepath.replace('$1', "Special:AbuseLog"))
+            user_checked = user_checked.replace("<全部吃掉了>", msg.locale.t(
+                "check.redacted.all") + '\n' + wiki_info.articlepath.replace('$1', "Special:AbuseLog"))
+        user_checked_map[u['original']] = user_checked
+    checked_titlelist = await check(*titlelist)
+    title_checked_map = {}
+    for t in checked_titlelist:
+        title_checked = t['content']
+        if title_checked.find("<吃掉了>") != -1 or title_checked.find("<全部吃掉了>") != -1:
+            title_checked = title_checked.replace("<吃掉了>", msg.locale.t(
+                "check.redacted") + '\n' + wiki_info.articlepath.replace('$1', "Special:AbuseLog"))
+            title_checked = title_checked.replace("<全部吃掉了>", msg.locale.t(
+                "check.redacted.all") + '\n' + wiki_info.articlepath.replace('$1', "Special:AbuseLog"))
+        title_checked_map[t['original']] = title_checked
+    for x in abl:
+        if 'title' in x:
+            t = []
+            time = msg.ts2strftime(strptime2ts(x['timestamp']), iso=True)
+            t.append(time)
+            result = 'pass' if not x['result'] else x['result']
+            t.append(msg.locale.t("wiki.message.ab.qq.slice",
+                                  title=title_checked_map[x['title']],
+                                  user=user_checked_map[x['user']],
+                                  action=x['action'],
+                                  filter_name=x['filter'],
+                                  result=result))
+            ablist.append('\n'.join(t))
+    return ablist

modules/wiki/utils/ab_qq.py

@@ -4,6 +4,7 @@
 from core.logger import Logger
 from modules.wiki.utils.time import strptime2ts
 from modules.wiki.utils.wikilib import WikiLib
+from .ab import convert_ab_to_detailed_format
 
 
 async def ab_qq(msg: MessageSession, wiki_url):
@@ -21,42 +22,8 @@ async def ab_qq(msg: MessageSession, wiki_url):
                 {"type": "text", "data": {"text": pageurl}}]
         }
     }]
-    ablist = []
-    userlist = []
-    titlelist = []
-    for x in query["query"]["abuselog"]:
-        if 'title' in x:
-            userlist.append(x['user'])
-            titlelist.append(x['title'])
-    checked_userlist = await check(*userlist)
-    user_checked_map = {}
-    for u in checked_userlist:
-        user_checked = u['content']
-        if user_checked.find("<吃掉了>") != -1 or user_checked.find("<全部吃掉了>") != -1:
-            user_checked = user_checked.replace("<吃掉了>", msg.locale.t("check.redacted"))
-            user_checked = user_checked.replace("<全部吃掉了>", msg.locale.t("check.redacted.all"))
-        user_checked_map[u['original']] = user_checked
-    checked_titlelist = await check(*titlelist)
-    title_checked_map = {}
-    for t in checked_titlelist:
-        title_checked = t['content']
-        if title_checked.find("<吃掉了>") != -1 or title_checked.find("<全部吃掉了>") != -1:
-            title_checked = title_checked.replace("<吃掉了>", msg.locale.t("check.redacted"))
-            title_checked = title_checked.replace("<全部吃掉了>", msg.locale.t("check.redacted.all"))
-        title_checked_map[t['original']] = title_checked
-    for x in query["query"]["abuselog"]:
-        if 'title' in x:
-            t = []
-            time = msg.ts2strftime(strptime2ts(x['timestamp']), iso=True)
-            t.append(time)
-            result = 'pass' if not x['result'] else x['result']
-            t.append(msg.locale.t("wiki.message.ab.qq.slice",
-                                  title=title_checked_map[x['title']],
-                                  user=user_checked_map[x['user']],
-                                  action=x['action'],
-                                  filter_name=x['filter'],
-                                  result=result))
-            ablist.append('\n'.join(t))
+
+    ablist = await convert_ab_to_detailed_format(query["query"]["abuselog"], wiki.wiki_info, msg)
     for x in ablist:
         nodelist.append(
             {

modules/wiki/utils/rc.py

@@ -1,7 +1,10 @@
+import urllib.parse
+
 from core.builtins import Url, Bot
 from core.dirty_check import check
+from core.logger import Logger
 from modules.wiki.utils.time import strptime2ts
-from modules.wiki.utils.wikilib import WikiLib
+from modules.wiki.utils.wikilib import WikiLib, WikiInfo
 
 
 async def rc(msg: Bot.MessageSession, wiki_url):
@@ -79,3 +82,117 @@ def compare_groups(old_groups, new_groups):
     added = "+" + ",".join(map(str, added_groups)) if added_groups else ""
     removed = "-" + ",".join(map(str, removed_groups)) if removed_groups else ""
     return f"{added} {removed}" if added and removed else f"{added}{removed}"
+
+
+async def convert_rc_to_detailed_format(rc: list, wiki_info: WikiInfo, msg: Bot.MessageSession):
+    rclist = []
+    userlist = []
+    titlelist = []
+    commentlist = []
+    for x in rc:
+        if 'title' in x:
+            userlist.append(x.get('user', ''))
+            titlelist.append(x.get('title'))
+            commentlist.append(x.get('comment', ''))
+    Logger.debug(userlist)
+    userlist = list(set(userlist))
+    titlelist = list(set(titlelist))
+    commentlist = list(set(commentlist))
+    checked_userlist = await check(*userlist)
+    user_checked_map = {}
+    for u in checked_userlist:
+        user_checked = u['content']
+        if user_checked.find("<吃掉了>") != -1 or user_checked.find("<全部吃掉了>") != -1:
+            user_checked = user_checked.replace("<吃掉了>", msg.locale.t(
+                "check.redacted") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+            user_checked = user_checked.replace("<全部吃掉了>", msg.locale.t(
+                "check.redacted.all") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+        user_checked_map[u['original']] = user_checked
+    checked_titlelist = await check(*titlelist)
+    title_checked_map = {}
+    for t in checked_titlelist:
+        title_checked = t['content']
+        if title_checked.find("<吃掉了>") != -1 or title_checked.find("<全部吃掉了>") != -1:
+            title_checked = title_checked.replace("<吃掉了>", msg.locale.t(
+                "check.redacted") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+            title_checked = title_checked.replace("<全部吃掉了>", msg.locale.t(
+                "check.redacted.all") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+        title_checked_map[t['original']] = title_checked
+    checked_commentlist = await check(*commentlist)
+    comment_checked_map = {}
+    for c in checked_commentlist:
+        comment_checked = c['content']
+        if comment_checked.find("<吃掉了>") != -1 or comment_checked.find("<全部吃掉了>") != -1:
+            comment_checked = comment_checked.replace("<吃掉了>", msg.locale.t(
+                "check.redacted") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+            comment_checked = comment_checked.replace("<全部吃掉了>", msg.locale.t(
+                "check.redacted.all") + '\n' + wiki_info.articlepath.replace('$1', "Special:RecentChanges"))
+        comment_checked_map[c['original']] = comment_checked
+    for x in rc:
+        if 'title' in x:
+            t = []
+            user = user_checked_map[x['user']]
+            title = title_checked_map[x['title']]
+            comment = comment_checked_map[x['comment']] if x.get('comment') else None
+
+            time = msg.ts2strftime(strptime2ts(x['timestamp']), iso=True)
+            t.append(time)
+            if x['type'] in ['edit', 'categorize']:
+                count = x['newlen'] - x['oldlen']
+                if count > 0:
+                    count = f'+{str(count)}'
+                else:
+                    count = str(count)
+                t.append(f"{title} .. ({count}) .. {user}")
+                if comment:
+                    t.append(comment)
+                t.append(
+                    wiki_info.articlepath.replace(
+                        '$1',
+                        f"{urllib.parse.quote(title)}?oldid={x['old_revid']}&diff={x['revid']}"))
+            if x['type'] == 'new':
+                r = msg.locale.t('message.brackets', msg=msg.locale.t(
+                    'wiki.message.rc.new_redirect')) if 'redirect' in x else ''
+                t.append(f"{title}{r} .. (+{x['newlen']}) .. {user}")
+                if comment:
+                    t.append(comment)
+            if x['type'] == 'log':
+                if x['logtype'] == x['logaction']:
+                    log = msg.locale.t(f"wiki.message.rc.action.{x['logtype']}", user=user, title=title)
+                else:
+                    log = msg.locale.t(
+                        f"wiki.message.rc.action.{
+                            x['logtype']}.{
+                            x['logaction']}",
+                        user=user,
+                        title=title)
+                if log.find("{") != -1 and log.find("}") != -1:
+                    if x['logtype'] == x['logaction']:
+                        log = f"{user} {x['logtype']} {title}"
+                    else:
+                        log = f"{user} {x['logaction']} {x['logtype']} {title}"
+                t.append(log)
+                params = x['logparams']
+                if 'suppressredirect' in params:
+                    t.append(msg.locale.t('wiki.message.rc.params.suppress_redirect'))
+                if 'oldgroups' and 'newgroups' in params:
+                    t.append(compare_groups(params['oldgroups'], params['newgroups']))
+                if 'oldmodel' and 'newmodel' in params:
+                    t.append(f"{params['oldmodel']} -> {params['newmodel']}")
+                if 'description' in params:
+                    t.append(params['description'])
+                if 'duration' in params:
+                    t.append(msg.locale.t('wiki.message.rc.params.duration') + params['duration'])
+                if 'flags' in params:
+                    t.append(', '.join(params['flags']))
+                if 'tag' in params:
+                    t.append(msg.locale.t('wiki.message.rc.params.tag') + params['tag'])
+                if 'target_title' in params:
+                    t.append(msg.locale.t('wiki.message.rc.params.target_title') + params['target_title'])
+                if comment:
+                    t.append(comment)
+                if x['revid'] != 0:
+                    t.append(wiki_info.articlepath.replace(
+                        "$1", f"{urllib.parse.quote(title_checked_map[x['title']])}"))
+            rclist.append('\n'.join(t))
+    return rclist