feat: matrix 推送支持

app/setting_interface.py

@@ -619,7 +619,7 @@ def __initCard(self):
             # "lark": FIF.MAIL,
             # "custom": FIF.MAIL
         }
-        self.notifySupportImage = ["telegram", "smtp", "wechatworkapp", "onebot", "gocqhttp", "lark", "custom"]
+        self.notifySupportImage = ["telegram", "matrix", "smtp", "wechatworkapp", "onebot", "gocqhttp", "lark", "custom"]
 
         for key, _ in cfg.config.items():
             if key.startswith("notify_") and key.endswith("_enable"):

assets/config/config.example.yaml

@@ -282,6 +282,15 @@ notify_telegram_userid: "" # 接收通知的用户或群组 ID。
 notify_telegram_api_url: "" # 可选参数，Telegram API 的自定义URL，默认为空。
 notify_telegram_proxies: "" # 可选参数，请求 Telegram API 是否使用代理。默认为空(使用系统PAC或不使用)。例如："127.0.0.1:10808", "socks5://127.0.0.1:1080/" 。
 
+notify_matrix_enable: false # 是否启用 Matrix 通知。true 开启，false 关闭。
+notify_matrix_homeserver: "" # 服务器的地址 例如官方地址 https://matrix.org
+notify_matrix_device_id: "" # 一个十位大写字母或数字组成的设备ID，登录后由服务器分发 例如： 01234ABCDE
+notify_matrix_user_id: "" # 用户ID 例如：@user_id:matrix.org
+notify_matrix_access_token: "" # access_token，登录后由服务器分发，例如 syt_abcdefghijk_lmNOPQRSTUVWXYZ_123Ab4
+notify_matrix_room_id: "" # 消息发到哪个房间， 例如: !abcdefGHIJKlmnoPQRST:matrix.org
+notify_matrix_proxy: "" # 代理，例如 socks5://127.0.0.1:1080 ，不填则用系统PAC代理
+notify_matrix_separately_text_media: true # 分开发送文字和媒体（图片），如果你使用FluffyChat可能不显示Caption，可以设置为True避免不显示文字。 
+
 # Server酱·Turbo版通知配置
 # 微信推送，适合小白，免费版每天限制五条
 # https://sct.ftqq.com/

module/notification/__init__.py

@@ -1,5 +1,6 @@
 from module.config import cfg
 from module.logger import log
+from module.notification.matrix import MatrixNotifier
 from module.notification.notification import Notification
 # 导入所有通知器类型
 from module.notification.onepush import OnepushNotifier
@@ -19,6 +20,7 @@ class NotifierFactory:
     notifier_classes = {
         "winotify": WinotifyNotifier,
         "telegram": TelegramNotifier,
+        "matrix": MatrixNotifier,
         "onebot": OnebotNotifier,
         "smtp": SMTPNotifier,
         "gocqhttp": GocqhttpNotifier,

module/notification/matrix.py

@@ -0,0 +1,99 @@
+from io import BytesIO
+from .notifier import Notifier
+
+from .pac import match_proxy, match_proxy_url
+from .notifier import Notifier
+from nio import AsyncClient # 如果想使用代理则必须使用AsyncClient
+from nio.responses import RoomSendError, UploadError
+import asyncio
+from PIL import Image
+from module.logger import log
+
+async def send_text_image_msg(client, room_id, msg, image_io):
+    v = image_io.getvalue()
+    image_stream = BytesIO(v)
+    im = Image.open(image_stream)
+    (width, height) = im.size 
+    im.close()
+    resp, _maybe_keys = await client.upload(
+            BytesIO(v),
+            content_type="image/png", 
+            filename="img.png",
+            filesize=len(v),
+    )
+    if isinstance(resp, UploadError):
+        raise RuntimeError(f"message img upload error : {resp}")
+
+    content = {
+            "msgtype": "m.image",
+            "info": {
+                "size": len(v),
+                "mimetype": "image/png",
+                "thumbnail_info": None, 
+                "w": width,  # width in pixel
+                "h": height,  # height in pixel
+                "thumbnail_url": None,
+            },
+            "url": resp.content_uri,
+    }
+
+    if msg is not None and msg != "":
+        content["body"] = msg
+        content["filename"] = "img.png"
+    else:
+        content["body"] = "img.png"
+
+
+    rsp = await client.room_send(
+        room_id=room_id,
+        message_type="m.room.message",
+        content = content,
+    )
+    if isinstance(rsp, RoomSendError):
+        raise RuntimeError(f"message send error : {rsp}")
+
+
+async def send_text_msg(client, room_id, msg):
+    if msg is None or msg == '':
+        return
+    rsp = await client.room_send(
+        # Watch out! If you join an old room you'll see lots of old messages
+        room_id=room_id,
+        message_type="m.room.message",
+        content={"msgtype": "m.text", "body": msg},
+    )
+    if isinstance(rsp, RoomSendError):
+        raise RuntimeError(f"message send error : {rsp}")
+
+class MatrixNotifier(Notifier):
+    def _get_supports_image(self):
+        return True
+
+    def send(self, title: str, content: str, image_io=None):
+        # cfg
+        homeserver = self.params["homeserver"]
+        device_id = self.params["device_id"]
+        user_id = self.params["user_id"]
+        access_token = self.params["access_token"]
+        room_id = self.params["room_id"]
+        proxy = match_proxy_url(self.params.get("proxy", None), homeserver)
+        separately_text_media = self.params["separately_text_media"]
+        # client
+        client = AsyncClient(homeserver)
+        client.user_id = user_id
+        client.access_token = access_token
+        client.device_id = device_id
+        if proxy is not None:
+            client.proxy = proxy
+        # 构建消息文本
+        message = title if not content else f'{title}\n{content}' if title else content
+        loop = asyncio.get_event_loop()
+        if image_io:
+            if separately_text_media:
+                loop.run_until_complete(send_text_msg(client, room_id, message))
+            loop.run_until_complete(send_text_image_msg(client, room_id, None if separately_text_media else message, image_io))
+        else:
+            # 只发送文本消息
+            loop.run_until_complete(send_text_msg(client, room_id, message))
+
+

module/notification/pac.py

@@ -0,0 +1,87 @@
+
+import pypac
+import winreg
+
+PAC_REG_KEY = "Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings"
+
+
+def query_system_pac_settings() -> str | None:
+    r"""
+    Query system pac settings from registry.
+    :return: pac url or None
+    """
+    try:
+        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, PAC_REG_KEY)
+        value, _ = winreg.QueryValueEx(key, "AutoConfigURL")
+        winreg.CloseKey(key)
+        if value:
+            return value
+        else:
+            return None
+    except FileNotFoundError:
+        return None
+
+
+def macth_pac_settings(url: str, pac_url: str):
+    r"""
+    Match pac settings from pac url.
+    :param url: url
+    :param pac_url: pac url
+    :return: proxy or None
+    """
+    pac = pypac.get_pac(url=pac_url)
+    if pac is None:
+        pac_result = None
+    else:
+        pac_result = pac.find_proxy_for_url(url=url, host="0.0.0.0")
+    if isinstance(pac_result, str):
+        pac_result = pac_result.split(";")
+        pac_result = map(lambda x: x.strip(), pac_result)
+        pac_result = filter(lambda x: x, pac_result)
+        pac_result = list(pac_result)
+        if len(pac_result) > 0:
+            pac_result = pac_result[0]
+            if pac_result == "DIRECT":
+                return None
+            if pac_result.startswith("PROXY"):
+                pac_result = pac_result.split(" ")
+                if len(pac_result) == 2:
+                    pac_result = pac_result[1]
+                    pac_result = pac_result.split(":")
+                    if len(pac_result) == 2:
+                        return f"{pac_result[0]}:{pac_result[1]}"
+
+
+def match_proxy(proxies_param: str | None, api_url: str) -> dict | None:
+    r"""
+    Match proxy from system pac settings or proxies param
+    :param proxies_param: proxies_param
+    :param url: url
+    :return: proxies
+    """
+    if proxies_param is not None:
+        return {"http": proxies_param, "https": proxies_param}
+    pac_url = query_system_pac_settings()
+    if pac_url is not None:
+        proxy = macth_pac_settings(api_url, pac_url)
+        if proxy is not None:
+            return {"http": proxy, "https": proxy}
+    return None
+
+
+def match_proxy_url(proxy: str | None, api_url: str) -> dict | None:
+    r"""
+    Match proxy from system pac settings or proxies param
+    :param proxies_param: proxies_param
+    :param url: url
+    :return: proxies
+    """
+    if proxy is not None:
+        return proxy
+    pac_url = query_system_pac_settings()
+    if pac_url is not None:
+        proxy = macth_pac_settings(api_url, pac_url)
+        if proxy is not None:
+            proxy
+    return None
+

module/notification/telegram.py

@@ -1,74 +1,7 @@
 import requests
-from .notifier import Notifier
-import pypac
-import winreg
-
-PAC_REG_KEY = "Software\\Microsoft\\Windows\\CurrentVersion\\Internet Settings"
-
-
-def query_system_pac_settings() -> str | None:
-    r"""
-    Query system pac settings from registry.
-    :return: pac url or None
-    """
-    try:
-        key = winreg.OpenKey(winreg.HKEY_CURRENT_USER, PAC_REG_KEY)
-        value, _ = winreg.QueryValueEx(key, "AutoConfigURL")
-        winreg.CloseKey(key)
-        if value:
-            return value
-        else:
-            return None
-    except FileNotFoundError:
-        return None
-
-
-def macth_pac_settings(url: str, pac_url: str):
-    r"""
-    Match pac settings from pac url.
-    :param url: url
-    :param pac_url: pac url
-    :return: proxy or None
-    """
-    pac = pypac.get_pac(url=pac_url)
-    if pac is None:
-        pac_result = None
-    else:
-        pac_result = pac.find_proxy_for_url(url=url, host="0.0.0.0")
-    if isinstance(pac_result, str):
-        pac_result = pac_result.split(";")
-        pac_result = map(lambda x: x.strip(), pac_result)
-        pac_result = filter(lambda x: x, pac_result)
-        pac_result = list(pac_result)
-        if len(pac_result) > 0:
-            pac_result = pac_result[0]
-            if pac_result == "DIRECT":
-                return None
-            if pac_result.startswith("PROXY"):
-                pac_result = pac_result.split(" ")
-                if len(pac_result) == 2:
-                    pac_result = pac_result[1]
-                    pac_result = pac_result.split(":")
-                    if len(pac_result) == 2:
-                        return f"{pac_result[0]}:{pac_result[1]}"
-
-
-def match_proxy(proxies_param: str | None, api_url: str) -> dict | None:
-    r"""
-    Match proxy from system pac settings or proxies param
-    :param proxies_param: proxies_param
-    :param url: url
-    :return: proxies
-    """
-    if proxies_param is not None:
-        return {"http": proxies_param, "https": proxies_param}
-    pac_url = query_system_pac_settings()
-    if pac_url is not None:
-        proxy = macth_pac_settings(api_url, pac_url)
-        if proxy is not None:
-            return {"http": proxy, "https": proxy}
-    return None
 
+from .pac import match_proxy
+from .notifier import Notifier
 
 class TelegramNotifier(Notifier):
     def _get_supports_image(self):
@@ -78,8 +11,7 @@ def send(self, title: str, content: str, image_io=None):
         token = self.params["token"]
         chat_id = self.params["userid"]
         api_url = self.params.get("api_url", "api.telegram.org")
-        proxies = match_proxy(self.params.get(
-            "proxies", None), f"https://{api_url}/")
+        proxies = match_proxy(self.params.get("proxies", None), f"https://{api_url}/")
 
         # 构建消息文本
         message = title if not content else f'{title}\n{content}' if title else content

requirements.txt

@@ -24,4 +24,7 @@ qrcode
 pysocks
 pypac
 requests_toolbelt
-desktopmagic
+desktopmagic
+matrix-nio
+python-socks[asyncio]
+asyncio