优化代码

plugins/guess_song/main.py

@@ -20,6 +20,8 @@ import aiohttp
 from pydub import AudioSegment
 from io import BytesIO
 
+from wechat_ipad.models.appmsg_xml import MUSIC_XML
+
 
 class GuessSongRedisDB:
     """猜歌名游戏Redis相关操作"""
@@ -135,17 +137,17 @@ class GuessSongPlugin(MessagePluginInterface):
         self._commands = guess_song_config.get("command", ["猜歌名"])
         self.command_format = guess_song_config.get("command-format", "猜歌名 [歌手名]/[歌名]")
         self.enable = guess_song_config.get("enable", True)
-        
+
         # 加载API配置
         api_config = guess_song_config.get("api", {})
         self.music_list_api = api_config.get("music_list_api", "")
         self.music_single_api = api_config.get("music_single_api", "")
-        
+
         # 加载游戏配置
         game_config = guess_song_config.get("game", {})
         self.clip_duration = game_config.get("clip_duration", 10)
         self.hint_timeout = game_config.get("hint_timeout", 30)
-        
+
         # 加载热门歌手列表
         singers_config = guess_song_config.get("popular_singers", {})
         self.popular_singers = singers_config.get("list", [
@@ -438,52 +440,8 @@ class GuessSongPlugin(MessagePluginInterface):
             singer_pic = song_info.get("singer_pic", "")
             data_url = song_info.get("data_url", "")
 
-            xml_message = f"""<appmsg appid="wx904fb3ecf62c7dea" sdkver="0">
-    <title>{song_name}</title>
-    <des>{singer_name}-点击三角直接播放</des>
-    <action>view</action>
-    <type>3</type>
-    <showtype>0</showtype>
-    <content />
-    <url>{data_url}</url>
-    <dataurl>{play_url}</dataurl>
-    <lowurl/>
-    <lowdataurl/>
-    <recorditem />
-    <thumburl />
-    <messageaction />
-    <laninfo />
-    <extinfo />
-    <sourceusername />
-    <sourcedisplayname />
-    <commenturl />
-    <appattach>
-        <totallen>0</totallen>
-        <attachid />
-        <emoticonmd5></emoticonmd5>
-        <fileext />
-        <aeskey></aeskey>
-    </appattach>
-    <webviewshared>
-        <publisherId />
-        <publisherReqId>0</publisherReqId>
-    </webviewshared>
-    <weappinfo>
-        <pagepath />
-        <username />
-        <appid />
-        <appservicetype>0</appservicetype>
-    </weappinfo>
-    <websearch />
-    <songalbumurl>{singer_pic}</songalbumurl>
-</appmsg>
-<scene>0</scene>
-<appinfo>
-    <version>49</version>
-    <appname>汽水音乐</appname>
-</appinfo>
-<commenturl />"""
-
+            xml_message = f"{MUSIC_XML}".format(song_name=song_name, singer_name=singer_name, play_url=play_url,
+                                                data_url=data_url, singer_pic=singer_pic)
             self.LOG.info(f"发送音乐消息：{xml_message}")
             res = await bot.send_app_message(wxid=receiver, xml=xml_message, type=0)
             self.LOG.info(f"发送音乐消息 res:{res}")

utils/video_utils.py

@@ -1,13 +1,11 @@
-import io
 import os
 import tempfile
 
 import cv2
-
 from loguru import logger
 
 
-def _get_first_frame(video_path, output_path):
+def get_first_frame(video_path, output_path):
     """
     提取视频的第一帧并保存为图片
     :param video_path: 视频文件路径
@@ -50,7 +48,7 @@ def _get_first_frame(video_path, output_path):
         cap.release()
 
 
-def _get_first_frame_bytes(video_bytes, output_path):
+def get_first_frame_bytes(video_bytes, output_path):
     """
     提取视频的第一帧并保存为图片
     :param video_bytes: 视频文件流
@@ -65,7 +63,7 @@ def _get_first_frame_bytes(video_bytes, output_path):
             temp_file.write(video_bytes)
             temp_file_path = temp_file.name  # 获取临时文件路径
         # 打开视频文件
-        return _get_first_frame(temp_file_path, output_path)
+        return get_first_frame(temp_file_path, output_path)
     except Exception as e:
         logger.error(f"提取视频首帧时出错: {e}")
         return None

wechat_ipad/client/message.py

@@ -9,13 +9,12 @@ from pathlib import Path
 from typing import Union
 
 import aiohttp
+import pysilk
 from loguru import logger
+from pydub import AudioSegment
 from pymediainfo import MediaInfo
 
-import pysilk
-from pydub import AudioSegment
-
-from utils.video_utils import _get_first_frame, _get_first_frame_bytes
+from utils.video_utils import get_first_frame, get_first_frame_bytes
 from wechat_ipad import UserLoggedOut
 from wechat_ipad.client.base import WechatAPIClientBase
 
@@ -233,7 +232,7 @@ class MessageMixin(WechatAPIClientBase):
             media_info = MediaInfo.parse(BytesIO(video))
             # 如果没有传入首帧，则自己提取一次
             if not has_image:
-                first_frame = _get_first_frame_bytes(video, f"frame_{int(time.time())}.jpg")
+                first_frame = get_first_frame_bytes(video, f"frame_{int(time.time())}.jpg")
                 if first_frame:
                     image = Path(first_frame)
         elif isinstance(video, os.PathLike):
@@ -247,7 +246,7 @@ class MessageMixin(WechatAPIClientBase):
             media_info = MediaInfo.parse(video_path)
             # 如果没有传入首帧，则自己提取一次
             if not has_image:
-                first_frame = _get_first_frame(video_path, f"frame_{int(time.time())}.jpg")
+                first_frame = get_first_frame(video_path, f"frame_{int(time.time())}.jpg")
                 if first_frame:
                     image = Path(first_frame)
         else:
@@ -542,10 +541,17 @@ class MessageMixin(WechatAPIClientBase):
         if not self.wxid:
             raise UserLoggedOut("请先登录")
 
-        async with aiohttp.ClientSession() as session:
-            json_param = {"Wxid": self.wxid, "ToWxid": wxid, "CardWxid": card_wxid, "CardAlias": card_alias,
-                          "CardNickname": card_nickname}
-            response = await session.post(f'http://{self.ip}:{self.port}/SendCardMsg', json=json_param)
+        async with (aiohttp.ClientSession() as session):
+
+            json_param = {
+                "CardAlias": card_alias,
+                "CardNickName": card_nickname,
+                "CardWxId": card_wxid,
+                "ToWxid": wxid,
+                "Wxid": self.wxid
+            }
+
+            response = await session.post(f'http://{self.ip}:{self.port}/api/Msg/ShareCard', json=json_param)
             json_resp = await response.json()
 
             if json_resp.get("Success"):
@@ -577,7 +583,7 @@ class MessageMixin(WechatAPIClientBase):
         """
         return await self._queue_message(self._send_app_message, wxid, xml, type)
 
-    async def _send_app_message(self, wxid: str, xml: str, type: int) -> tuple[int, int, int]:
+    async def _send_app_message(self, wxid: str, xml: str, type: int=0) -> tuple[int, int, int]:
         if not self.wxid:
             raise UserLoggedOut("请先登录")
         # {