feat(core): 实现了 wxpusher 消息推送

.gitignore

@@ -11,6 +11,7 @@ invoke.yml
 pyproject.toml
 tmp/
 res/sprites_compiled/
+messages/
 ##########################
 
 # Byte-compiled / optimized / DLL files

kotonebot/ui/pushkit/__init__.py

@@ -0,0 +1,3 @@
+from .protocol import PushkitProtocol
+from .wxpusher import Wxpusher
+

kotonebot/ui/pushkit/image_host.py

@@ -0,0 +1,87 @@
+import os
+import tempfile
+from pathlib import Path
+from typing import Sequence
+
+import cv2
+import requests
+import numpy as np
+from cv2.typing import MatLike
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def _save_temp_image(image: MatLike) -> Path:
+    """将OpenCV图片保存为临时文件"""
+    temp_file = Path(tempfile.mktemp(suffix='.jpg'))
+    cv2.imwrite(str(temp_file), image)
+    return temp_file
+
+def _upload_single(image: MatLike | str) -> str:
+    """
+    上传单张图片到freeimage.host
+    
+    :param image: OpenCV MatLike 或本地图片文件路径
+    """
+    api_url = 'https://freeimage.host/api/1/upload'
+    api_key = os.getenv('FREEIMAGEHOST_KEY')
+    
+    if not api_key:
+        raise RuntimeError('Environment variable FREEIMAGEHOST_KEY is not set')
+    
+    # 处理输入
+    temp_file = None
+    if isinstance(image, str):
+        # 本地文件路径
+        files = {'source': open(image, 'rb')}
+    else:
+        # 保存OpenCV图片为临时文件
+        temp_file = _save_temp_image(image)
+        files = {'source': open(temp_file, 'rb')}
+    
+    data = {
+        'key': api_key,
+        'action': 'upload',
+        'format': 'json'
+    }
+    
+    try:
+        # 发送POST请求
+        response = requests.post(api_url, data=data, files=files)
+        
+        if response.status_code != 200:
+            raise RuntimeError(f'Upload failed: HTTP {response.status_code}')
+            
+        result = response.json()
+        
+        if result['status_code'] != 200:
+            raise RuntimeError(f'Upload failed: API {result["status_txt"]}')
+            
+        return result['image']['url']
+        
+    finally:
+        # 清理临时文件
+        files['source'].close()
+        if temp_file and temp_file.exists():
+            temp_file.unlink()
+
+def upload(images: MatLike | str | Sequence[MatLike | str]) -> list[str]:
+    """上传一张或多张图片到freeimage.host
+    
+    Args:
+        images: 单张图片或图片列表。每个图片可以是OpenCV图片对象或本地图片文件路径
+        
+    Returns:
+        上传后的图片URL列表
+    """
+    if isinstance(images, (str, np.ndarray)):
+        _images = [images]
+    elif isinstance(images, Sequence):
+        _images = [img for img in images]
+    else:
+        raise ValueError("Invalid input type")
+        
+    return [_upload_single(img) for img in _images]
+
+if __name__ == "__main__":
+    print(upload(cv2.imread("res/sprites/jp/common/button_close.png")))

kotonebot/ui/pushkit/protocol.py

@@ -0,0 +1,13 @@
+from typing import Protocol
+
+from cv2.typing import MatLike
+
+class PushkitProtocol(Protocol):
+    def push(
+        self,
+        title: str,
+        message: str,
+        *,
+        images: list[str | MatLike] | None = None,
+    ) -> None:
+        ...

kotonebot/ui/pushkit/wxpusher.py

@@ -0,0 +1,53 @@
+import os
+import json
+from typing import Sequence
+import requests
+from cv2.typing import MatLike
+from dotenv import dotenv_values
+
+from .image_host import upload
+from .protocol import PushkitProtocol
+
+config = dotenv_values(".env")
+
+class Wxpusher(PushkitProtocol):
+    def __init__(self, app_token: str | None = None, uid: str | None = None):
+        self.app_token = app_token or config["WXPUSHER_APP_TOKEN"]
+        self.uid = uid or config["WXPUSHER_UID"]
+
+    def push(self, title: str, message: str, *, images: Sequence[str | MatLike] | None = None) -> None:
+        summary = title
+        content = message
+
+        if images:
+            image_urls = upload(images)
+            img_md = "\n".join([f"![{img_url}]({img_url})" for img_url in image_urls])
+            content = content + "\n" + img_md
+
+        data = {
+            "appToken": self.app_token,
+            "uid": self.uid,
+            "summary": summary,
+            "content": content,
+            "contentType": 3, # 1: 文本 2: HTML 3: Markdown
+            "uids": [self.uid],
+            "verifyPay": False,
+            "verifyPayType": 0
+        }
+
+        response = requests.post(
+            "http://wxpusher.zjiecode.com/api/send/message",
+            json=data
+        )
+        result = response.json()
+        
+        if result["code"] != 1000 or not result["success"]:
+            raise RuntimeError(f"推送失败: {result['msg']}")
+
+# TODO: 极简推送 https://wxpusher.zjiecode.com/docs/#/?id=spt
+        
+if __name__ == "__main__":
+    import cv2
+    wxpusher = Wxpusher()
+    wxpusher.push("测试图片", "测试图片", images=[cv2.imread("res/sprites/jp/common/button_close.png")])
+

kotonebot/ui/user.py

@@ -1,9 +1,13 @@
 """消息框、通知、推送等 UI 相关函数"""
-import logging
+import os
-from typing import Callable
+import time
 
+import cv2
 from cv2.typing import MatLike
 
+from .pushkit import Wxpusher
+from .. import logging
+
 logger = logging.getLogger(__name__)
 
 def ask(
@@ -17,16 +21,50 @@ def ask(
     """
     raise NotImplementedError
 
+def _save_local(
+    title: str,
+    message: str,
+    images: list[MatLike] | None = None
+):
+    """
+    保存消息到本地
+    """
+    if not os.path.exists('messages'):
+        os.makedirs('messages')
+    file_name = f'messages/{time.strftime("%Y-%m-%d_%H-%M-%S")}'
+    with open(file_name + '.txt', 'w', encoding='utf-8') as f:
+        logger.verbose('saving message to local: %s', file_name + '.txt')
+        f.write(message)
+    if images is not None:
+        for i, image in enumerate(images):
+            logger.verbose('saving image to local: %s', f'{file_name}_{i}.png')
+            cv2.imwrite(f'{file_name}_{i}.png', image)
+
+def push(
+    title: str,
+    message: str,
+    *,
+    images: list[MatLike] | None = None
+):
+    """
+    推送消息
+    """
+    try:
+        logger.verbose('pushing to wxpusher: %s', message)
+        wxpusher = Wxpusher()
+        wxpusher.push(title, message, images=images)
+    except Exception as e:
+        logger.warning('push remote message failed: %s', e)
+        _save_local(title, message, images)
+
 def info(
     message: str,
     images: list[MatLike] | None = None,
     *,
     once: bool = False
 ):
-    """
+    logger.info('user.info: %s', message)
-    信息
+    push("KAA 提示", message, images=images)
-    """
-    logger.debug('user.info: %s', message)
 
 def warning(
     message: str,
@@ -41,3 +79,4 @@ def warning(
     :param once: 每次运行是否只显示一次。
     """
     logger.warning('user.warning: %s', message)
+    push("KAA 警告", message, images=images)

template.env

@@ -1,2 +1,10 @@
+# PyPI
 PYPI_TOKEN=
 PYPI_TEST_TOKEN=
+
+# WxPusher 推送服务
+WXPUSHER_APP_TOKEN=
+WXPUSHER_UID=
+
+# 图床上传
+FREEIMAGEHOST_KEY=