@@ -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字符串。
|
||||
|
||||
Reference in New Issue
Block a user