Revert "完善表情资产后台能力并补充群总结落库"

This reverts commit 2a54650a6f.
This commit is contained in:
liuwei
2026-04-02 17:54:35 +08:00
parent 2a54650a6f
commit 57bb46bb21
11 changed files with 17 additions and 671 deletions

View File

@@ -1,7 +1,6 @@
import base64
import io
import os
import imghdr
import aiofiles
import aiohttp
@@ -13,19 +12,19 @@ from wechat_ipad.client.base import WechatAPIClientBase, Proxy
class ToolMixin(WechatAPIClientBase):
async def download_cdn_file(self, aeskey: str, file_url: str) -> str:
"""通用 CDN 文件下载
async def download_image(self, aeskey: str, cdnmidimgurl: str) -> str:
"""CDN下载高清图片
{
"Wxid": "string",
"FileNo": "string",
"FileAesKey": "string"
}
Args:
aeskey (str): 文件的AES密钥
file_url (str): 文件的CDN URL
aeskey (str): 图片的AES密钥
cdnmidimgurl (str): 图片的CDN URL
Returns:
str: 文件的base64编码字符串
str: 图片的base64编码字符串
Raises:
UserLoggedOut: 未登录时调用
@@ -35,7 +34,7 @@ class ToolMixin(WechatAPIClientBase):
raise UserLoggedOut("请先登录")
async with aiohttp.ClientSession() as session:
json_param = {"Wxid": self.wxid, "FileAesKey": aeskey, "FileNo": file_url}
json_param = {"Wxid": self.wxid, "FileAesKey": aeskey, "FileNo": cdnmidimgurl}
response = await session.post(f'http://{self.ip}:{self.port}/api/Tools/CdnDownloadImage', json=json_param)
json_resp = await response.json()
@@ -44,10 +43,6 @@ class ToolMixin(WechatAPIClientBase):
else:
self.error_handler(json_resp)
async def download_image(self, aeskey: str, cdnmidimgurl: str) -> str:
"""CDN下载高清图片。"""
return await self.download_cdn_file(aeskey, cdnmidimgurl)
async def download_voice(self, msg_id: str, voiceurl: str, length: int) -> str:
"""下载语音文件。
@@ -270,42 +265,6 @@ class ToolMixin(WechatAPIClientBase):
except Exception as e:
return False
@staticmethod
def guess_file_extension(file_bytes: bytes, default_ext: str = ".bin") -> str:
"""根据文件头猜测扩展名。"""
if not file_bytes:
return default_ext
if file_bytes.startswith(b"GIF87a") or file_bytes.startswith(b"GIF89a"):
return ".gif"
if file_bytes.startswith(b"\x89PNG\r\n\x1a\n"):
return ".png"
if file_bytes.startswith(b"RIFF") and file_bytes[8:12] == b"WEBP":
return ".webp"
if file_bytes.startswith(b"\xff\xd8\xff"):
return ".jpg"
detected = imghdr.what(None, h=file_bytes)
if detected:
return f".{detected}"
return default_ext
@staticmethod
async def base64_to_file_autoext(base64_str: str, file_stem: str, file_path: str,
default_ext: str = ".bin") -> str:
"""将base64写入文件并自动识别扩展名。"""
os.makedirs(file_path, exist_ok=True)
if ',' in base64_str:
base64_str = base64_str.split(',')[1]
file_bytes = base64.b64decode(base64_str)
ext = ToolMixin.guess_file_extension(file_bytes, default_ext=default_ext)
full_path = os.path.join(file_path, f"{file_stem}{ext}")
async with aiofiles.open(full_path, 'wb') as f:
await f.write(file_bytes)
return full_path
@staticmethod
async def file_to_base64(file_path: str) -> str:
"""将文件转换为base64字符串。