Support images in webp format.master
@@ -9,7 +9,6 @@ import json | |||||
import os | import os | ||||
import threading | import threading | ||||
import time | import time | ||||
import requests | import requests | ||||
from bridge.context import * | from bridge.context import * | ||||
@@ -21,6 +20,7 @@ from common.expired_dict import ExpiredDict | |||||
from common.log import logger | from common.log import logger | ||||
from common.singleton import singleton | from common.singleton import singleton | ||||
from common.time_check import time_checker | from common.time_check import time_checker | ||||
from common.utils import convert_webp_to_png | |||||
from config import conf, get_appdata_dir | from config import conf, get_appdata_dir | ||||
from lib import itchat | from lib import itchat | ||||
from lib.itchat.content import * | from lib.itchat.content import * | ||||
@@ -229,6 +229,12 @@ class WechatChannel(ChatChannel): | |||||
image_storage.write(block) | image_storage.write(block) | ||||
logger.info(f"[WX] download image success, size={size}, img_url={img_url}") | logger.info(f"[WX] download image success, size={size}, img_url={img_url}") | ||||
image_storage.seek(0) | image_storage.seek(0) | ||||
if ".webp" in img_url: | |||||
try: | |||||
image_storage = convert_webp_to_png(image_storage) | |||||
except Exception as e: | |||||
logger.error(f"Failed to convert image: {e}") | |||||
return | |||||
itchat.send_image(image_storage, toUserName=receiver) | itchat.send_image(image_storage, toUserName=receiver) | ||||
logger.info("[WX] sendImage url={}, receiver={}".format(img_url, receiver)) | logger.info("[WX] sendImage url={}, receiver={}".format(img_url, receiver)) | ||||
elif reply.type == ReplyType.IMAGE: # 从文件读取图片 | elif reply.type == ReplyType.IMAGE: # 从文件读取图片 | ||||
@@ -266,6 +272,7 @@ def _send_login_success(): | |||||
except Exception as e: | except Exception as e: | ||||
pass | pass | ||||
def _send_logout(): | def _send_logout(): | ||||
try: | try: | ||||
from common.linkai_client import chat_client | from common.linkai_client import chat_client | ||||
@@ -274,6 +281,7 @@ def _send_logout(): | |||||
except Exception as e: | except Exception as e: | ||||
pass | pass | ||||
def _send_qr_code(qrcode_list: list): | def _send_qr_code(qrcode_list: list): | ||||
try: | try: | ||||
from common.linkai_client import chat_client | from common.linkai_client import chat_client | ||||
@@ -281,3 +289,4 @@ def _send_qr_code(qrcode_list: list): | |||||
chat_client.send_qrcode(qrcode_list) | chat_client.send_qrcode(qrcode_list) | ||||
except Exception as e: | except Exception as e: | ||||
pass | pass | ||||
@@ -17,7 +17,7 @@ from channel.wechatcom.wechatcomapp_client import WechatComAppClient | |||||
from channel.wechatcom.wechatcomapp_message import WechatComAppMessage | from channel.wechatcom.wechatcomapp_message import WechatComAppMessage | ||||
from common.log import logger | from common.log import logger | ||||
from common.singleton import singleton | from common.singleton import singleton | ||||
from common.utils import compress_imgfile, fsize, split_string_by_utf8_length | |||||
from common.utils import compress_imgfile, fsize, split_string_by_utf8_length, convert_webp_to_png | |||||
from config import conf, subscribe_msg | from config import conf, subscribe_msg | ||||
from voice.audio_convert import any_to_amr, split_audio | from voice.audio_convert import any_to_amr, split_audio | ||||
@@ -99,6 +99,12 @@ class WechatComAppChannel(ChatChannel): | |||||
image_storage = compress_imgfile(image_storage, 10 * 1024 * 1024 - 1) | image_storage = compress_imgfile(image_storage, 10 * 1024 * 1024 - 1) | ||||
logger.info("[wechatcom] image compressed, sz={}".format(fsize(image_storage))) | logger.info("[wechatcom] image compressed, sz={}".format(fsize(image_storage))) | ||||
image_storage.seek(0) | image_storage.seek(0) | ||||
if ".webp" in img_url: | |||||
try: | |||||
image_storage = convert_webp_to_png(image_storage) | |||||
except Exception as e: | |||||
logger.error(f"Failed to convert image: {e}") | |||||
return | |||||
try: | try: | ||||
response = self.client.media.upload("image", image_storage) | response = self.client.media.upload("image", image_storage) | ||||
logger.debug("[wechatcom] upload image response: {}".format(response)) | logger.debug("[wechatcom] upload image response: {}".format(response)) | ||||
@@ -2,7 +2,7 @@ import io | |||||
import os | import os | ||||
from urllib.parse import urlparse | from urllib.parse import urlparse | ||||
from PIL import Image | from PIL import Image | ||||
from common.log import logger | |||||
def fsize(file): | def fsize(file): | ||||
if isinstance(file, io.BytesIO): | if isinstance(file, io.BytesIO): | ||||
@@ -54,3 +54,17 @@ def split_string_by_utf8_length(string, max_length, max_split=0): | |||||
def get_path_suffix(path): | def get_path_suffix(path): | ||||
path = urlparse(path).path | path = urlparse(path).path | ||||
return os.path.splitext(path)[-1].lstrip('.') | return os.path.splitext(path)[-1].lstrip('.') | ||||
def convert_webp_to_png(webp_image): | |||||
from PIL import Image | |||||
try: | |||||
webp_image.seek(0) | |||||
img = Image.open(webp_image).convert("RGBA") | |||||
png_image = io.BytesIO() | |||||
img.save(png_image, format="PNG") | |||||
png_image.seek(0) | |||||
return png_image | |||||
except Exception as e: | |||||
logger.error(f"Failed to convert WEBP to PNG: {e}") | |||||
raise |
@@ -55,7 +55,7 @@ class Keyword(Plugin): | |||||
reply_text = self.keyword[content] | reply_text = self.keyword[content] | ||||
# 判断匹配内容的类型 | # 判断匹配内容的类型 | ||||
if (reply_text.startswith("http://") or reply_text.startswith("https://")) and any(reply_text.endswith(ext) for ext in [".jpg", ".jpeg", ".png", ".gif", ".img"]): | |||||
if (reply_text.startswith("http://") or reply_text.startswith("https://")) and any(reply_text.endswith(ext) for ext in [".jpg", ".webp", ".jpeg", ".png", ".gif", ".img"]): | |||||
# 如果是以 http:// 或 https:// 开头,且".jpg", ".jpeg", ".png", ".gif", ".img"结尾,则认为是图片 URL。 | # 如果是以 http:// 或 https:// 开头,且".jpg", ".jpeg", ".png", ".gif", ".img"结尾,则认为是图片 URL。 | ||||
reply = Reply() | reply = Reply() | ||||
reply.type = ReplyType.IMAGE_URL | reply.type = ReplyType.IMAGE_URL | ||||