Losita/mai-bot
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix：图片存储问题

Browse Source
This commit is contained in:
SengokuCola
2026-04-09 14:32:05 +08:00
parent ad720f42bc
commit 4395e3aad7
5 changed files with 98 additions and 353 deletions
Download Patch File Download Diff File Expand all files Collapse all files

5
AGENTS.md
View File

@@ -42,4 +42,7 @@
如果你需要改动配置文件，不需要修改实际的bot_config.toml或者model_config.toml，只需要修改配置文件模版，并新增一个版本号即可，也不必要为配置改动创建测试文件。
# 关于webui修改
不要修改dashboard下的内容，因为这部分内容由另一个仓库build
不要修改dashboard下的内容，因为这部分内容由另一个仓库build
# maibot插件开发文档
https://github.com/Mai-with-u/maibot-plugin-sdk/blob/main/docs/guide.md

44
plugins/emoji_manage_plugin/_manifest.json
View File

@@ -1,44 +0,0 @@
{
"manifest_version": 2,
"version": "2.0.0",
"name": "BetterEmoji",
"description": "更好的表情包管理插件",
"author": {
"name": "SengokuCola",
"url": "https://github.com/SengokuCola"
},
"license": "GPL-v3.0-or-later",
"urls": {
"repository": "https://github.com/SengokuCola/BetterEmoji",
"homepage": "https://github.com/SengokuCola/BetterEmoji",
"documentation": "https://github.com/SengokuCola/BetterEmoji",
"issues": "https://github.com/SengokuCola/BetterEmoji/issues"
},
"host_application": {
"min_version": "1.0.0",
"max_version": "1.0.0"
},
"sdk": {
"min_version": "2.0.0",
"max_version": "2.99.99"
},
"dependencies": [],
"capabilities": [
"emoji.get_random",
"emoji.get_count",
"emoji.get_info",
"emoji.get_all",
"emoji.register_emoji",
"emoji.delete_emoji",
"send.text",
"send.forward"
],
"i18n": {
"default_locale": "zh-CN",
"locales_path": "_locales",
"supported_locales": [
"zh-CN"
]
},
"id": "sengokucola.betteremoji"
}

238
plugins/emoji_manage_plugin/plugin.py
View File

@@ -1,238 +0,0 @@
"""表情包管理插件 — 新 SDK 版本
通过 /emoji 命令管理表情包的添加、列表和删除。
"""
from maibot_sdk import Command, MaiBotPlugin
import base64
import datetime
import hashlib
import re
class EmojiManagePlugin(MaiBotPlugin):
"""表情包管理插件"""
async def on_load(self) -> None:
"""处理插件加载。"""
async def on_unload(self) -> None:
"""处理插件卸载。"""
# ===== 工具方法 =====
@staticmethod
def _extract_emoji_base64(segments) -> list[str]:
"""从消息 segments 中提取 emoji/image 的 base64 数据。
segments 可以是 dict 列表或 Seg 对象列表（兼容两种格式）。
"""
results: list[str] = []
if not segments:
return results
if isinstance(segments, dict):
seg_type = segments.get("type", "")
if seg_type in ("emoji", "image"):
data = segments.get("data", "")
if data:
results.append(data)
elif seg_type == "seglist":
for child in segments.get("data", []):
results.extend(EmojiManagePlugin._extract_emoji_base64(child))
return results
# 如果有 .type 属性（Seg 对象）
if hasattr(segments, "type"):
seg_type = getattr(segments, "type", "")
if seg_type in ("emoji", "image"):
results.append(getattr(segments, "data", ""))
elif seg_type == "seglist":
for child in getattr(segments, "data", []):
results.extend(EmojiManagePlugin._extract_emoji_base64(child))
return results
# 列表
for seg in segments:
results.extend(EmojiManagePlugin._extract_emoji_base64(seg))
return results
# ===== Command 组件 =====
@Command("add_emoji", description="添加表情包", pattern=r".*/emoji add.*")
async def handle_add_emoji(self, stream_id: str = "", message_segments=None, **kwargs):
"""添加表情包"""
emoji_base64_list = self._extract_emoji_base64(message_segments)
if not emoji_base64_list:
await self.ctx.send.text("未在消息中找到表情包或图片", stream_id)
return False, "未在消息中找到表情包或图片", False
success_count = 0
fail_count = 0
results = []
for i, emoji_b64 in enumerate(emoji_base64_list):
result = await self.ctx.emoji.register_emoji(emoji_b64)
if isinstance(result, dict) and result.get("success"):
success_count += 1
desc = result.get("description", "未知描述")
emotions = result.get("emotions", [])
replaced = result.get("replaced", False)
msg = f"表情包 {i + 1} 注册成功{'(替换旧表情包)' if replaced else '(新增表情包)'}"
if desc:
msg += f"\n描述: {desc}"
if emotions:
msg += f"\n情感标签: {', '.join(emotions)}"
results.append(msg)
else:
fail_count += 1
err = result.get("message", "注册失败") if isinstance(result, dict) else "注册失败"
results.append(f"表情包 {i + 1} 注册失败: {err}")
total = success_count + fail_count
summary = f"表情包注册完成: 成功 {success_count} 个，失败 {fail_count} 个，共处理 {total} 个"
if results:
summary += "\n" + "\n".join(results)
await self.ctx.send.text(summary, stream_id)
return success_count > 0, summary, success_count > 0
@Command("emoji_list", description="列表表情包", pattern=r"^/emoji list(\s+\d+)?$")
async def handle_list_emoji(self, stream_id: str = "", raw_message: str = "", **kwargs):
"""列出表情包"""
max_count = 10
match = re.match(r"^/emoji list(?:\s+(\d+))?$", raw_message)
if match and match.group(1):
max_count = min(int(match.group(1)), 50)
now = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
count_result = await self.ctx.emoji.get_count()
emoji_count = count_result if isinstance(count_result, int) else 0
info_result = await self.ctx.emoji.get_info()
max_emoji = info_result.get("max_count", 0) if isinstance(info_result, dict) else 0
available = info_result.get("available_emojis", 0) if isinstance(info_result, dict) else 0
lines = [
f"📊 表情包统计信息 ({now})",
f"• 总数: {emoji_count} / {max_emoji}",
f"• 可用: {available}",
]
if emoji_count == 0:
lines.append("\n❌ 暂无表情包")
await self.ctx.send.text("\n".join(lines), stream_id)
return True, "\n".join(lines), True
all_result = await self.ctx.emoji.get_all()
all_emojis = all_result if isinstance(all_result, list) else []
if not all_emojis:
lines.append("\n❌ 无法获取表情包列表")
await self.ctx.send.text("\n".join(lines), stream_id)
return False, "\n".join(lines), True
display = all_emojis[:max_count]
lines.append(f"\n📋 显示前 {len(display)} 个表情包:")
for i, emoji in enumerate(display, 1):
if isinstance(emoji, (list, tuple)) and len(emoji) >= 3:
_, desc, emotion = emoji[0], emoji[1], emoji[2]
elif isinstance(emoji, dict):
desc = emoji.get("description", "")
emotion = emoji.get("emotion", "")
else:
desc, emotion = str(emoji), ""
short_desc = desc[:50] + "..." if len(desc) > 50 else desc
lines.append(f"{i}. {short_desc} [{emotion}]")
if len(all_emojis) > max_count:
lines.append(f"\n💡 还有 {len(all_emojis) - max_count} 个表情包未显示")
final = "\n".join(lines)
await self.ctx.send.text(final, stream_id)
return True, final, True
@Command("delete_emoji", description="删除表情包", pattern=r".*/emoji delete.*")
async def handle_delete_emoji(self, stream_id: str = "", message_segments=None, **kwargs):
"""删除表情包"""
emoji_base64_list = self._extract_emoji_base64(message_segments)
if not emoji_base64_list:
await self.ctx.send.text("未在消息中找到表情包或图片", stream_id)
return False, "未找到表情包", False
success_count = 0
fail_count = 0
results = []
for i, emoji_b64 in enumerate(emoji_base64_list):
# 计算哈希
if isinstance(emoji_b64, str):
clean = emoji_b64.encode("ascii", errors="ignore").decode("ascii")
else:
clean = str(emoji_b64)
image_bytes = base64.b64decode(clean)
emoji_hash = hashlib.md5(image_bytes).hexdigest() # noqa: S324
result = await self.ctx.emoji.delete_emoji(emoji_hash)
if isinstance(result, dict) and result.get("success"):
success_count += 1
desc = result.get("description", "未知描述")
emotions = result.get("emotions", [])
before = result.get("count_before", 0)
after = result.get("count_after", 0)
msg = f"表情包 {i + 1} 删除成功"
if desc:
msg += f"\n描述: {desc}"
if emotions:
msg += f"\n情感标签: {', '.join(emotions)}"
msg += f"\n表情包数量: {before} → {after}"
results.append(msg)
else:
fail_count += 1
err = result.get("message", "删除失败") if isinstance(result, dict) else "删除失败"
results.append(f"表情包 {i + 1} 删除失败: {err}")
total = success_count + fail_count
summary = f"表情包删除完成: 成功 {success_count} 个，失败 {fail_count} 个，共处理 {total} 个"
if results:
summary += "\n" + "\n".join(results)
await self.ctx.send.text(summary, stream_id)
return success_count > 0, summary, success_count > 0
@Command("random_emojis", description="发送多张随机表情包", pattern=r"^/random_emojis$")
async def handle_random_emojis(self, stream_id: str = "", **kwargs):
"""发送多张随机表情包"""
emojis = await self.ctx.emoji.get_random(5)
if not emojis:
return False, "未找到表情包", False
messages = [
{"user_id": "0", "nickname": "神秘用户", "segments": [{"type": "image", "content": e.get("base64", "")}]}
for e in emojis
]
await self.ctx.send.forward(messages, stream_id)
return True, "已发送随机表情包", True
async def on_config_update(self, scope: str, config_data: dict[str, object], version: str) -> None:
"""处理配置热重载事件。
Args:
scope: 配置变更范围。
config_data: 最新配置数据。
version: 配置版本号。
"""
del scope
del config_data
del version
def create_plugin() -> EmojiManagePlugin:
"""创建表情包管理插件实例。
Returns:
EmojiManagePlugin: 新的表情包管理插件实例。
"""
return EmojiManagePlugin()

17
pytests/image_sys_test/test_image_data_model.py
View File

@@ -31,6 +31,23 @@ async def test_calculate_hash_format_updates_runtime_path_metadata(tmp_path: Pat
assert emoji.dir_path == tmp_path.resolve()
@pytest.mark.asyncio
async def test_calculate_hash_format_reuses_existing_target_file(tmp_path: Path) -> None:
image_bytes = _build_test_image_bytes("JPEG")
tmp_file_path = tmp_path / "emoji.tmp"
target_file_path = tmp_path / "emoji.jpeg"
tmp_file_path.write_bytes(image_bytes)
target_file_path.write_bytes(image_bytes)
emoji = MaiEmoji(full_path=tmp_file_path, image_bytes=image_bytes)
assert await emoji.calculate_hash_format() is True
assert emoji.full_path == target_file_path.resolve()
assert emoji.file_name == target_file_path.name
assert not tmp_file_path.exists()
assert target_file_path.exists()
@pytest.mark.parametrize(
("model_cls", "extra_fields"),
[

147
src/common/data_models/image_data_model.py
View File

@@ -24,125 +24,148 @@ logger = get_logger("emoji")
class BaseImageDataModel(BaseDatabaseDataModel[Images]):
def __init__(self, full_path: str | Path, image_bytes: Optional[bytes] = None):
if not full_path:
# 创建时候即检测文件路径合法性
raise ValueError("表情包路径不能为空")
raise ValueError("图片路径不能为空")
if Path(full_path).is_dir() or not Path(full_path).exists():
raise FileNotFoundError(f"表情包路径无效: {full_path}")
raise FileNotFoundError(f"图片路径无效: {full_path}")
resolved_path = Path(full_path).absolute().resolve()
self.full_path: Path
self.dir_path: Path
self.file_name: str
self._set_full_path(resolved_path)
self.file_hash: str = None # type: ignore
self.image_bytes: Optional[bytes] = image_bytes
self.image_format: str = "" # 图片格式
self.image_format: str = ""
def _set_full_path(self, full_path: Path) -> None:
"""同步更新文件路径相关的运行时元数据。"""
"""同步刷新路径、目录和文件名等运行时元数据。"""
resolved_path = full_path.absolute().resolve()
self.full_path = resolved_path
self.dir_path = resolved_path.parent.resolve()
self.file_name = resolved_path.name
def _restore_image_format_from_path(self) -> None:
"""根据文件扩展名恢复基础图片格式信息。"""
"""根据文件扩展名恢复图片格式信息。"""
self.image_format = self.full_path.suffix.removeprefix(".").lower()
def _build_non_conflicting_path(self, target_path: Path) -> Path:
"""在目标路径被占用时，生成一个可用的新路径。"""
candidate_path = target_path
index = 1
while candidate_path.exists():
candidate_path = target_path.with_name(
f"{target_path.stem}_{self.file_hash[:8]}_{index}{target_path.suffix}"
)
index += 1
return candidate_path
def _rename_file_to_match_format(self) -> None:
"""修正文件扩展名，并处理目标文件已存在的冲突。"""
new_file_name = ".".join(self.file_name.split(".")[:-1] + [self.image_format])
new_full_path = self.dir_path / new_file_name
if new_full_path == self.full_path:
return
if new_full_path.exists():
existing_file_hash = hashlib.sha256(self.read_image_bytes(new_full_path)).hexdigest()
if existing_file_hash == self.file_hash:
logger.info(f"[初始化] {new_full_path.name} 已存在且内容一致，复用已有文件")
self.full_path.unlink()
self._set_full_path(new_full_path)
return
conflict_free_path = self._build_non_conflicting_path(new_full_path)
logger.warning(
f"[初始化] {new_full_path.name} 已存在且内容不同，改为保存到 {conflict_free_path.name}"
)
self.full_path.rename(conflict_free_path)
self._set_full_path(conflict_free_path)
return
self.full_path.rename(new_full_path)
self._set_full_path(new_full_path)
def read_image_bytes(self, path: Path) -> bytes:
"""
同步读取图片文件的字节内容
同步读取图片文件的字节内容。
Args:
path (Path): 图片文件的完整路径
path: 图片文件的完整路径。
Returns:
return (bytes): 图片文件的字节内容
Raises:
FileNotFoundError: 如果文件不存在则抛出该异常
Exception: 其他读取文件时发生的异常
图片文件的字节内容。
"""
try:
with open(path, "rb") as f:
return f.read()
except FileNotFoundError as e:
with open(path, "rb") as file:
return file.read()
except FileNotFoundError as exc:
logger.error(f"[读取图片文件] 文件未找到: {path}")
raise e
except Exception as e:
logger.error(f"[读取图片文件] 读取文件时发生错误: {e}")
raise e
raise exc
except Exception as exc:
logger.error(f"[读取图片文件] 读取文件时发生错误: {exc}")
raise exc
def get_image_format(self, image_bytes: bytes) -> str:
"""
获取图片的格式
获取图片的实际格式。
Args:
image_bytes (bytes): 图片的字节内容
image_bytes: 图片的字节内容。
Returns:
return (str): 图片的格式（小写）
Raises:
ValueError: 如果无法识别图片格式
Exception: 其他读取图片格式时发生的异常
小写格式名，例如 `png`、`jpeg`。
"""
try:
with PILImage.open(io.BytesIO(image_bytes)) as img:
if not img.format:
raise ValueError("无法识别图片格式")
return img.format.lower()
except Exception as e:
logger.error(f"[获取图片格式] 读取图片格式时发生错误: {e}")
raise e
except Exception as exc:
logger.error(f"[获取图片格式] 读取图片格式时发生错误: {exc}")
raise exc
async def calculate_hash_format(self) -> bool:
"""
异步计算表情包的哈希值和格式，初始化后应该执行此方法来确保对象的哈希值和格式正确
计算图片哈希和实际格式，并在需要时修正扩展名。
Returns:
return (bool): 如果成功计算哈希值和格式则返回True，否则返回False
成功返回 `True`，失败返回 `False`。
"""
try:
# 计算哈希值
logger.debug(f"[初始化] 计算 {self.file_name} 的哈希值...")
if not self.image_bytes:
if self.image_bytes is None:
logger.debug(f"[初始化] 正在读取文件: {self.full_path}")
image_bytes = await asyncio.to_thread(self.read_image_bytes, self.full_path)
else:
image_bytes = self.image_bytes
self.image_bytes = image_bytes
self.file_hash = hashlib.sha256(image_bytes).hexdigest()
logger.debug(f"[初始化] {self.file_name} 计算哈希值成功: {self.file_hash}")
# 用PIL读取图片格式
logger.debug(f"[初始化] 读取 {self.file_name} 的图片格式...")
self.image_format = await asyncio.to_thread(self.get_image_format, image_bytes)
logger.debug(f"[初始化] {self.file_name} 读取图片格式成功: {self.image_format}")
# 比对文件扩展名和实际格式
file_ext = self.file_name.split(".")[-1].lower()
if file_ext != self.image_format:
logger.warning(
f"[初始化] {self.file_name} 文件扩展名与实际格式不符: ext`{file_ext}`!=`{self.image_format}`"
)
# 重命名文件以匹配实际格式
new_file_name = ".".join(self.file_name.split(".")[:-1] + [self.image_format])
new_full_path = self.dir_path / new_file_name
self.full_path.rename(new_full_path)
self._set_full_path(new_full_path)
self._rename_file_to_match_format()
return True
except Exception as e:
logger.error(f"[初始化] 初始化图片时发生错误: {e}")
except Exception as exc:
logger.error(f"[初始化] 初始化图片时发生错误: {exc}")
logger.error(traceback.format_exc())
return False
class MaiEmoji(BaseImageDataModel):
"""麦麦的表情包对象，仅当**图片文件存在**时才应该创建此对象，数据库记录如果标记为文件不存在`(no_file_flag = True)`则不应该调用 `from_db_instance` 方法来创建此对象"""
"""表情包数据模型。"""
def __init__(self, full_path: str | Path, image_bytes: Optional[bytes] = None):
# self.embedding = []
self.description: str = ""
self.emotion: List[str] = []
self.query_count = 0
@@ -152,33 +175,26 @@ class MaiEmoji(BaseImageDataModel):
@classmethod
def from_db_instance(cls, db_record: Images):
"""从数据库记录创建 MaiEmoji 对象，如果记录标记为文件不存在则**抛出异常**
调用者应该对数据库记录进行检查，如果 `no_file_flag` 为 True 则不应该调用此方法
Args:
db_record (Images): 数据库中的图片记录
Returns:
return (MaiEmoji): 包含图片信息的 MaiEmoji 对象
Raises:
ValueError: 如果数据库记录标记为文件不存在则抛出该异常
"""
"""从数据库记录构建 `MaiEmoji` 对象。"""
if db_record.no_file_flag:
raise ValueError(f"数据库记录 {db_record.image_hash} 标记为文件不存在，无法创建 MaiEmoji 对象")
obj = cls(db_record.full_path)
obj.file_hash = db_record.image_hash
obj._restore_image_format_from_path()
description = db_record.description or ""
obj.description = description
normalized_tags = [
str(item).strip()
for item in str(description).replace("，", ",").replace("、", ",").replace("；", ",").split(",")
for item in str(description).replace("，", ",").replace("。", ",").replace("、", ",").split(",")
if str(item).strip()
]
deduped_tags: List[str] = []
for item in normalized_tags:
if item not in deduped_tags:
deduped_tags.append(item)
obj.emotion = deduped_tags
obj.query_count = db_record.query_count
obj.last_used_time = db_record.last_used_time
@@ -198,7 +214,7 @@ class MaiEmoji(BaseImageDataModel):
class MaiImage(BaseImageDataModel):
"""麦麦图片数据模型，仅当**图片文件存在**时才应该创建此对象，数据库记录如果标记为文件不存在`(no_file_flag = True)`则不应该调用 `from_db_instance` 方法来创建此对象"""
"""普通图片数据模型。"""
def __init__(self, full_path: str | Path, image_bytes: Optional[bytes] = None):
self.description: str = ""
@@ -207,19 +223,10 @@ class MaiImage(BaseImageDataModel):
@classmethod
def from_db_instance(cls, db_record: Images):
"""从数据库记录创建 MaiImage 对象，如果记录标记为文件不存在则**抛出异常**
调用者应该对数据库记录进行检查，如果 `no_file_flag` 为 True 则不应该调用此方法
Args:
db_record (Images): 数据库中的图片记录
Returns:
return (MaiImage): 包含图片信息的 MaiImage 对象
Raises:
ValueError: 如果数据库记录标记为文件不存在则抛出该异常
"""
"""从数据库记录构建 `MaiImage` 对象。"""
if db_record.no_file_flag:
raise ValueError(f"数据库记录 {db_record.image_hash} 标记为文件不存在，无法创建 MaiImage 对象")
obj = cls(db_record.full_path)
obj.file_hash = db_record.image_hash
obj._set_full_path(Path(db_record.full_path))