♻️ 使用local_store获取插件目录

Author: shadow3aaa

src/nonebot_plugin_nyaturingtest/__init__.py

@@ -9,7 +9,7 @@
 
 import anyio
 import httpx
-from nonebot import logger, on_command, on_message
+from nonebot import logger, on_command, on_message, require
 from nonebot.adapters import Message
 from nonebot.adapters.onebot.v11 import (
     Bot,
@@ -29,6 +29,8 @@
 from .mem import Message as MMessage
 from .session import Session
 
+require("nonebot_plugin_localstore")
+
 __plugin_meta__ = PluginMetadata(
     name="NYATuringTest",
     description="群聊特化llm聊天机器人，具有长期记忆和情绪模拟能力",
@@ -564,7 +566,9 @@ async def message2BotMessage(bot_name: str, group_id: int, message: Message, bot
 
                 is_sticker = seg.data.get("sub_type") == 1
                 image_base64 = base64.b64encode(image_bytes).decode("utf-8")
-                description = image_manager.get_image_description(image_base64=image_base64, is_sticker=is_sticker)
+                description = await image_manager.get_image_description(
+                    image_base64=image_base64, is_sticker=is_sticker
+                )
                 if description:
                     if is_sticker:
                         message_content += f"\n[表情包] [情感:{description.emotion}] [内容:{description.description}]\n"

src/nonebot_plugin_nyaturingtest/image_manager.py

@@ -5,15 +5,16 @@
 import json
 from pathlib import Path
 
+import anyio
 from nonebot import logger
+import nonebot_plugin_localstore as store
 import numpy as np
 from PIL import Image
 
-from nonebot_plugin_nyaturingtest.vlm import SiliconFlowVLM
-
 from .config import plugin_config
+from .vlm import SiliconFlowVLM
 
-IMAGE_CACHE_DIR = Path("image_cache")
+IMAGE_CACHE_DIR = Path(f"{store.get_plugin_cache_dir()}/image_cache")
 
 
 @dataclass
@@ -80,7 +81,7 @@ def __init__(self):
             IMAGE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
             self._initialized = True
 
-    def get_image_description(self, image_base64: str, is_sticker: bool) -> ImageWithDescription | None:
+    async def get_image_description(self, image_base64: str, is_sticker: bool) -> ImageWithDescription | None:
         """
         获取图片描述
         """
@@ -90,20 +91,21 @@ def get_image_description(self, image_base64: str, is_sticker: bool) -> ImageWit
         # 检查缓存
         cache = IMAGE_CACHE_DIR.joinpath(f"{image_hash}.json")
         if cache.exists():
-            with open(cache, encoding="utf-8") as f:
-                image_with_desc_raw = f.read()
+            async with await anyio.open_file(cache, encoding="utf-8") as f:
+                image_with_desc_raw = await f.read()
                 try:
                     image_with_desc = ImageWithDescription.from_json(image_with_desc_raw)
                     if image_with_desc.is_sticker != is_sticker:
                         image_with_desc.is_sticker = is_sticker
                         # 修改缓存文件
-                        with open(cache, "w", encoding="utf-8") as f:
-                            f.write(image_with_desc.to_json())
+                        async with await anyio.open_file(cache, "w", encoding="utf-8") as f:
+                            await f.write(image_with_desc.to_json())
                     return image_with_desc
                 except ValueError as e:
                     logger.error(f"缓存文件({cache})格式错误，重新生成")
                     logger.error(e)
                     cache.unlink()  # 删除缓存文件
+                    cache.unlink()  # 删除缓存文件
 
         # 获取图片描述
         # 获取图片类型
@@ -120,30 +122,30 @@ def get_image_description(self, image_base64: str, is_sticker: bool) -> ImageWit
                 return None
             prompt = """这是一个动态图，每一张图代表了动态图的某一帧，黑色背景代表透明。"请用中文描述这张图片的内容。如
             果有文字，请把文字都描述出来。并尝试猜测这个图片的含义。最多100个字"""
-            description = self._vlm.request(
+            description = await self._vlm.request(
                 prompt=prompt,
                 image_base64=gif_transfromed,
                 image_format="jpeg",
             )
             # 分析表达的情感
             prompt = """这是一个动态图，每一张图代表了动态图的某一帧，黑色背景代表透明。请分析这个表情包表达的情感，
             用中文给出'情感，类型，含义'的三元式描述，要求每个描述都是一个简单的词语"""
-            description_emotion = self._vlm.request(
+            description_emotion = await self._vlm.request(
                 prompt=prompt,
                 image_base64=gif_transfromed,
                 image_format="jpeg",
             )
         else:
             prompt = "请用中文描述这张图片的内容。如果有文字，请把文字都描述出来。并尝试猜测这个图片的含义。最多100个字"
-            description = self._vlm.request(
+            description = await self._vlm.request(
                 prompt=prompt,
                 image_base64=image_base64,
                 image_format=image_format,
             )
             # 分析表达的情感
             prompt = """请分析这个表情包表达的情感，用中文给出'情感，类型，含义'的三元式描述，要求每个描述都是一个简单的
             词语"""
-            description_emotion = self._vlm.request(
+            description_emotion = await self._vlm.request(
                 prompt=prompt,
                 image_base64=image_base64,
                 image_format="jpeg",
@@ -158,10 +160,9 @@ def get_image_description(self, image_base64: str, is_sticker: bool) -> ImageWit
             emotion=description_emotion,
             is_sticker=is_sticker,
         )
-
         # 缓存结果
-        with open(cache, "w", encoding="utf-8") as f:
-            f.write(result.to_json())
+        async with await anyio.open_file(cache, "w", encoding="utf-8") as f:
+            await f.write(result.to_json())
 
         return result
 

src/nonebot_plugin_nyaturingtest/mem.py

@@ -105,7 +105,7 @@ def related_users(self) -> list[str]:
         """
         return list({message.user_name for message in self.__messages})
 
-    def compress_message(self):
+    async def compress_message(self):
         """
         压缩历史消息
         """
@@ -129,7 +129,7 @@ def compress_message(self):
 {history_messages}
 """
         try:
-            response = self.__llm_client.generate_response(prompt, model="Qwen/Qwen3-8B")
+            response = await self.__llm_client.generate_response(prompt, model="Qwen/Qwen3-8B")
             if response:
                 self.__compressed_message = response
                 logger.info(f"压缩消息成功: {response}")

src/nonebot_plugin_nyaturingtest/presets.py

@@ -3,6 +3,7 @@
 import os
 
 from nonebot import logger
+import nonebot_plugin_localstore as store
 
 
 @dataclass
@@ -59,7 +60,7 @@ class RolePreset:
 PRESETS: dict[str, RolePreset] = {}
 
 
-def _load_presets_from_directory(directory: str = "./nya_presets"):
+def _load_presets_from_directory(directory: str = f"{store.get_plugin_config_dir()}/nya_presets"):
     # 如果文件夹不存在就创建并且写入例子(_猫娘预设)
     if not os.path.exists(directory):
         os.makedirs(directory)

src/nonebot_plugin_nyaturingtest/session.py

@@ -11,6 +11,7 @@
 import traceback
 
 from nonebot import logger
+import nonebot_plugin_localstore as store
 from openai import AsyncOpenAI
 
 from .client import LLMClient
@@ -85,7 +86,7 @@ def __init__(self, siliconflow_api_key: str, id: str = "global", name: str = "te
             llm_api_key=plugin_config.nyaturingtest_chat_openai_api_key,
             llm_base_url=plugin_config.nyaturingtest_chat_openai_base_url,
             embedding_api_key=siliconflow_api_key,
-            persist_directory=f"./hippo_index_{id}",
+            persist_directory=f"{store.get_plugin_data_dir()}/hippo_index_{id}",
         )
         """
         对聊天记录的长期记忆 (基于HippoRAG)
@@ -176,8 +177,8 @@ def get_session_file_path(self) -> str:
         获取会话文件路径
         """
         # 确保会话目录存在
-        os.makedirs("yaturningtest_sessions", exist_ok=True)
-        return f"yaturningtest_sessions/session_{self.id}.json"
+        os.makedirs(f"{store.get_plugin_data_dir()}/yaturningtest_sessions", exist_ok=True)
+        return f"{store.get_plugin_data_dir()}/yaturningtest_sessions/session_{self.id}.json"
 
     def save_session(self):
         """
@@ -925,7 +926,7 @@ async def update(self, messages_chunk: list[Message], llm: Callable[[str], Await
                 texts=[f"'{self.__name}':'{msg}'" for msg in reply_messages],
             )
         # 压缩，索引记忆
-        self.global_memory.compress_message()
+        await self.global_memory.compress_message()
         self.long_term_memory.index()
 
         # 保存会话状态