gardel/Mofox-Core
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

v0.2.2

Browse Source 
优化了CQ码获取逻辑和图片下载逻辑
去除硬编码的“麦麦”()
This commit is contained in:
SengokuCola
2025-02-28 22:56:05 +08:00
parent 7b9b604811
commit 7494fff1ed
13 changed files with 271 additions and 207 deletions
Download Patch File Download Diff File Expand all files Collapse all files

237
src/plugins/chat/cq_code.py
View File

@@ -1,5 +1,5 @@
from dataclasses import dataclass
from typing import Dict, Optional
from typing import Dict, Optional, List, Union
import html
import requests
import base64
@@ -12,6 +12,13 @@ from nonebot.adapters.onebot.v11 import Bot
from .config import global_config, llm_config
import time
import asyncio
#解析各种CQ码
#包含CQ码类
@dataclass
class CQCode:
"""
@@ -25,13 +32,14 @@ class CQCode:
"""
type: str
params: Dict[str, str]
raw_code: str
# raw_code: str
group_id: int
user_id: int
group_name: str = ""
user_nickname: str = ""
translated_plain_text: Optional[str] = None
reply_message: Dict = None # 存储回复消息
image_base64: Optional[str] = None
@classmethod
def from_cq_code(cls, cq_code: str, reply: Dict = None) -> 'CQCode':
@@ -39,6 +47,9 @@ class CQCode:
从CQ码字符串创建CQCode对象
例如：[CQ:image,file=1.jpg,url=http://example.com/1.jpg]
"""
if not cq_code.startswith('[CQ:'):
return cls('text', {'text': cq_code}, cq_code, group_id=0, user_id=0)
# 移除前后的[]
content = cq_code[1:-1]
# 分离类型和参数部分
@@ -69,7 +80,10 @@ class CQCode:
if self.type == 'text':
self.translated_plain_text = self.params.get('text', '')
elif self.type == 'image':
self.translated_plain_text = self.translate_image()
if self.params.get('sub_type') == '0':
self.translated_plain_text = self.translate_image()
else:
self.translated_plain_text = self.translate_emoji()
elif self.type == 'at':
from .message import Message
message_obj = Message(
@@ -87,16 +101,8 @@ class CQCode:
else:
self.translated_plain_text = f"[{self.type}]"
def translate_image(self) -> str:
"""处理图片类型的CQ码，区分普通图片和表情包"""
if 'url' not in self.params:
return '[图片]'
# 获取子类型，默认为普通图片(0)
sub_type = int(self.params.get('sub_type', '0'))
is_emoji = (sub_type == 1)
# 添加更多请求头
def get_img(self):
'''
headers = {
'User-Agent': 'QQ/8.9.68.11565 CFNetwork/1220.1 Darwin/20.3.0',
'Accept': 'image/*;q=0.8',
@@ -105,64 +111,71 @@ class CQCode:
'Cache-Control': 'no-cache',
'Pragma': 'no-cache'
}
'''
# 处理URL编码问题
headers = {
'User-Agent': 'Mozilla/5.0 (Windows NT 6.1; WOW64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/50.0.2661.87 Safari/537.36',
'Accept': 'text/html, application/xhtml xml, */*',
'Accept-Encoding': 'gbk, GB2312',
'Accept-Language': 'zh-cn',
'Content-Type': 'application/x-www-form-urlencoded',
'Cache-Control': 'no-cache'
}
url = html.unescape(self.params['url'])
if not url.startswith(('http://', 'https://')):
return '[图片]' # 直接返回而不是抛出异常
return None # 直接返回而不是抛出异常
try:
# 下载图片，增加重试机制
max_retries = 3
for retry in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=10, verify=False)
if response.status_code == 200:
break
elif response.status_code == 400 and 'multimedia.nt.qq.com.cn' in url:
# 对于腾讯多媒体服务器的链接，直接返回图片描述
if sub_type == 1:
return '[QQ表情]'
return '[图片]'
time.sleep(1) # 重试前等待1秒
except requests.RequestException:
if retry == max_retries - 1:
raise
time.sleep(1)
if response.status_code != 200:
print(f"\033[1;31m[警告]\033[0m 图片下载失败: HTTP {response.status_code}, URL: {url}")
return '[图片]' # 直接返回而不是抛出异常
# 检查响应内容类型
content_type = response.headers.get('content-type', '')
if not content_type.startswith('image/'):
print(f"\033[1;31m[警告]\033[0m 非图片类型响应: {content_type}")
return '[图片]' # 直接返回而不是抛出异常
content = response.content
image_base64 = base64.b64encode(content).decode('utf-8')
# 根据子类型选择不同的处理方式
if sub_type == 1: # 表情包
try:
return self.get_emoji_description(image_base64)
except Exception as e:
print(f"\033[1;31m[警告]\033[0m 表情描述生成失败: {str(e)}")
return '[QQ表情]'
elif sub_type == 0: # 普通图片
try:
return self.get_image_description(image_base64)
except Exception as e:
print(f"\033[1;31m[警告]\033[0m 图片描述生成失败: {str(e)}")
return '[图片]'
else: # 其他类型都按普通图片处理
return '[图片]'
except Exception as e:
print(f"\033[1;31m[警告]\033[0m 图片处理失败: {str(e)}")
return '[图片]' # 出现任何错误都返回默认文本而不是抛出异常
max_retries = 3
for retry in range(max_retries):
try:
response = requests.get(url, headers=headers, timeout=10, verify=False)
if response.status_code == 200:
break
elif response.status_code == 400 and 'multimedia.nt.qq.com.cn' in url:
# 对于腾讯多媒体服务器的链接，直接返回图片描述
return None
time.sleep(1) # 重试前等待1秒
except requests.RequestException:
if retry == max_retries - 1:
raise
time.sleep(1)
if response.status_code != 200:
print(f"\033[1;31m[警告]\033[0m 图片下载失败: HTTP {response.status_code}, URL: {url}")
return None # 直接返回而不是抛出异常
#检查是否为图片
content_type = response.headers.get('content-type', '')
if not content_type.startswith('image/'):
print(f"\033[1;31m[警告]\033[0m 非图片类型响应: {content_type}")
return None # 直接返回而不是抛出异常
content = response.content
image_base64 = base64.b64encode(content).decode('utf-8')
if image_base64:
self.image_base64 = image_base64
return image_base64
else:
return None
def translate_emoji(self) -> str:
"""处理表情包类型的CQ码"""
if 'url' not in self.params:
return '[表情包]'
base64 = self.get_img()
if base64:
return self.get_image_description(base64)
else:
return '[表情包]'
def translate_image(self) -> str:
"""处理图片类型的CQ码，区分普通图片和表情包"""
#没有url，直接返回默认文本
if 'url' not in self.params:
return '[图片]'
base64 = self.get_img()
if base64:
return self.get_image_description(base64)
else:
return '[图片]'
def get_emoji_description(self, image_base64: str) -> str:
"""调用AI接口获取表情包描述"""
@@ -254,53 +267,6 @@ class CQCode:
raise ValueError(f"AI接口调用失败: {response.text}")
def get_image_description_is_setu(self, image_base64: str) -> str:
"""调用AI接口获取普通图片描述"""
headers = {
"Content-Type": "application/json",
"Authorization": f"Bearer {llm_config.SILICONFLOW_API_KEY}"
}
payload = {
"model": "deepseek-ai/deepseek-vl2",
"messages": [
{
"role": "user",
"content": [
{
"type": "text",
"text": "请回答我这张图片是否涉及涩情、情色、裸露或性暗示，请严格判断，有任何涩情迹象就回答是，请用是或否回答"
},
{
"type": "image_url",
"image_url": {
"url": f"data:image/jpeg;base64,{image_base64}"
}
}
]
}
],
"max_tokens": 300,
"temperature": 0.6
}
response = requests.post(
f"{llm_config.SILICONFLOW_BASE_URL}chat/completions",
headers=headers,
json=payload,
timeout=30
)
if response.status_code == 200:
result_json = response.json()
if "choices" in result_json and len(result_json["choices"]) > 0:
description = result_json["choices"][0]["message"]["content"]
# 如果描述中包含"否"，返回否，其他情况返回是
return "否" if "否" in description else "是"
raise ValueError(f"AI接口调用失败: {response.text}")
def translate_forward(self) -> str:
"""处理转发消息"""
try:
@@ -391,7 +357,7 @@ class CQCode:
group_id=self.group_id
)
if message_obj.user_id == global_config.BOT_QQ:
return f"[回复 麦麦 的消息: {message_obj.processed_plain_text}]"
return f"[回复 {global_config.BOT_NICKNAME} 的消息: {message_obj.processed_plain_text}]"
else:
return f"[回复 {self.reply_message.sender.nickname} 的消息: {message_obj.processed_plain_text}]"
@@ -424,7 +390,41 @@ class CQCode:
.replace(',', ',')
# 生成CQ码，设置sub_type=1表示这是表情包
return f"[CQ:image,file=file:///{escaped_path},sub_type=1]"
class CQCode_tool:
@staticmethod
def cq_from_dict_to_class(cq_code: Dict, reply: Optional[Dict] = None) -> CQCode:
"""
将CQ码字典转换为CQCode对象
Args:
cq_code: CQ码字典
reply: 回复消息的字典（可选）
Returns:
CQCode对象
"""
# 处理字典形式的CQ码
# 从cq_code字典中获取type字段的值,如果不存在则默认为'text'
cq_type = cq_code.get('type', 'text')
params = {}
if cq_type == 'text':
params['text'] = cq_code.get('data', {}).get('text', '')
else:
params = cq_code.get('data', {})
instance = CQCode(
type=cq_type,
params=params,
group_id=0,
user_id=0,
reply_message=reply
)
# 进行翻译处理
instance.translate()
return instance
@staticmethod
def create_reply_cq(message_id: int) -> str:
"""
@@ -434,4 +434,7 @@ class CQCode:
Returns:
回复CQ码字符串
"""
return f"[CQ:reply,id={message_id}]"
return f"[CQ:reply,id={message_id}]"
cq_code_tool = CQCode_tool()