⚡ 启用联合黑名单

liteyuki/plugins/liteyuki_uniblacklist/__init__.py

@@ -1,4 +1,5 @@
 from nonebot.plugin import PluginMetadata
+from .api import *
 
 __author__ = "snowykami"
 __plugin_meta__ = PluginMetadata(

liteyuki/plugins/liteyuki_uniblacklist/api.py

@@ -0,0 +1,59 @@
+import datetime
+
+import httpx
+import nonebot
+from nonebot import require
+from nonebot.exception import IgnoredException
+from nonebot.message import event_preprocessor
+from nonebot_plugin_alconna.typings import Event
+
+require("nonebot_plugin_apscheduler")
+
+from nonebot_plugin_apscheduler import scheduler
+
+blacklist_data: dict[str, set[str]] = {}
+blacklist: set[str] = set()
+
+
+@scheduler.scheduled_job("interval", minutes=10, next_run_time=datetime.datetime.now())
+async def update_blacklist():
+    await request_for_blacklist()
+
+
+async def request_for_blacklist():
+    global blacklist
+    urls = [
+            "https://cdn.liteyuki.icu/static/ubl/"
+    ]
+
+    platforms = [
+            "qq"
+    ]
+
+    for plat in platforms:
+        for url in urls:
+            url += f"{plat}.txt"
+            async with httpx.AsyncClient() as client:
+                resp = await client.get(url)
+                blacklist_data[plat] = set(resp.text.splitlines())
+    blacklist = get_uni_set()
+    nonebot.logger.info("blacklists updated")
+
+
+def get_uni_set() -> set:
+    s = set()
+    for new_set in blacklist_data.values():
+        s.update(new_set)
+    return s
+
+
+@event_preprocessor
+async def pre_handle(event: Event):
+    try:
+        user_id = str(event.get_user_id())
+
+    except:
+        return
+
+    if user_id in get_uni_set():
+        raise IgnoredException("UserId in blacklist")

main.py

@@ -5,13 +5,12 @@ from liteyuki.utils.base.config import load_from_yaml
 from liteyuki.utils.base.data_manager import StoredConfig, common_db
 from liteyuki.utils.base.ly_api import liteyuki_api
 
-init()
-store_config: dict = common_db.first(StoredConfig(), default=StoredConfig()).config
-static_config = load_from_yaml("config.yml")
-store_config.update(static_config)
-nonebot.init(**store_config)
-
 if __name__ == "__mp_main__":
+    init()
+    store_config: dict = common_db.first(StoredConfig(), default=StoredConfig()).config
+    static_config = load_from_yaml("config.yml")
+    store_config.update(static_config)
+    nonebot.init(**store_config)
     adapters = [v11.Adapter, v12.Adapter]
     driver = nonebot.get_driver()