liuwei_x_x/abot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

图片内容，将图片地址缓存到redis，然后通过随机redis快速提取路径。减少等待时间

Browse Source
This commit is contained in:
liuwei
2025-05-20 11:12:49 +08:00
parent edc556ad16
commit b436d08f5d
3 changed files with 193 additions and 51 deletions
Download Patch File Download Diff File Expand all files Collapse all files

8
main.py
View File

@@ -5,6 +5,7 @@ import threading
from async_job import async_job
from configuration import Config
from plugins.xiuren_image.images_cache import ImageCacheManager
from robot import Robot
from loguru import logger
@@ -111,6 +112,13 @@ def jobs(robot: Robot):
async def login_check_job():
await asyncio.to_thread(robot.login_twice_auto_auth)
@async_job.at_times(["11:150"])
async def update_image_cache_job():
logger.info("开始执行图片缓存更新任务")
manager = ImageCacheManager("/mnt/nfs_share") # 替换为你的图片目录
await manager.update_image_cache()
logger.info("图片缓存更新完成")
if __name__ == "__main__":
main()

114
plugins/xiuren_image/images_cache.py Normal file
View File

@@ -0,0 +1,114 @@
import os
import time
import random
import asyncio
from typing import Optional, List
import logging
from db.connection import DBConnectionManager
logger = logging.getLogger(__name__)
class ImageCacheManager:
IMAGE_KEY_PREFIX = "group:images:"
LAST_UPDATE_TIME_KEY = "group:images:last_update_time"
BATCH_SIZE = 500
def __init__(self, image_folder: str):
self.image_folder = image_folder
self.redis = DBConnectionManager.get_instance().get_redis_connection()
self.image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
def _get_last_update_time(self) -> float:
ts = self.redis.get(self.LAST_UPDATE_TIME_KEY)
if ts:
try:
return float(ts)
except Exception as e:
logger.warning(f"解析最后更新时间失败: {e}")
return 0.0
def _set_last_update_time(self, ts: float) -> None:
self.redis.set(self.LAST_UPDATE_TIME_KEY, ts)
def should_update_index(self) -> bool:
try:
folder_mtime = os.path.getmtime(self.image_folder)
last_ts = self._get_last_update_time()
if folder_mtime <= last_ts:
logger.info("图片目录未更新，无需重新索引")
return False
return True
except Exception as e:
logger.error(f"判断图片目录更新时间失败: {e}")
return True # 出错则默认更新
def _scan_new_images(self, last_update_ts: float) -> List[str]:
"""扫描目录获取新增图片文件路径"""
new_images = []
for root, _, files in os.walk(self.image_folder):
for file in files:
try:
_, ext = os.path.splitext(file)
if ext.lower() in self.image_extensions:
full_path = os.path.join(root, file)
# 只收录修改时间大于上次更新的文件
if os.path.getmtime(full_path) > last_update_ts:
if os.access(full_path, os.R_OK):
new_images.append(full_path)
except Exception as e:
logger.warning(f"处理文件时异常 {file}: {e}")
return new_images
def _redis_batch_write(self, keys_values: List[tuple]):
pipeline = self.redis.pipeline()
for key, value in keys_values:
pipeline.sadd(key, value)
pipeline.execute()
async def update_image_cache(self):
"""异步更新Redis图片缓存，分批写入，避免一次写入压力过大"""
if not self.should_update_index():
return
last_update_ts = self._get_last_update_time()
new_images = self._scan_new_images(last_update_ts)
if not new_images:
logger.info("无新增图片，无需更新缓存")
# 也更新时间戳防止重复扫描
self._set_last_update_time(time.time())
return
logger.info(f"新增图片数量: {len(new_images)}, 开始写入 Redis 分批")
total = len(new_images)
batch_size = self.BATCH_SIZE
# Redis key 固定为 set，方便随机取成员
redis_key = self.IMAGE_KEY_PREFIX + "all"
for i in range(0, total, batch_size):
batch = new_images[i:i + batch_size]
kvs = [(redis_key, path) for path in batch]
try:
self._redis_batch_write(kvs)
logger.info(f"写入 Redis 批次 {i // batch_size + 1} 成功，数量: {len(batch)}")
except Exception as e:
logger.error(f"Redis 写入失败: {e}")
# 这里可选择是否继续或退出，暂继续
# 更新最后更新时间戳
self._set_last_update_time(time.time())
def get_random_image(self) -> Optional[str]:
redis_key = self.IMAGE_KEY_PREFIX + "all"
try:
img = self.redis.srandmember(redis_key)
if img:
# redis 返回字节，转字符串
return img.decode('utf-8') if isinstance(img, bytes) else img
except Exception as e:
logger.error(f"获取随机图片失败: {e}")
return None

122
plugins/xiuren_image/main.py
View File

@@ -5,8 +5,10 @@ import os
import random
from typing import Dict, Any, List, Optional, Tuple
from db.connection import DBConnectionManager
from plugin_common.message_plugin_interface import MessagePluginInterface
from plugin_common.plugin_interface import PluginStatus
from plugins.xiuren_image.images_cache import ImageCacheManager
from utils.decorator.plugin_decorators import plugin_stats_decorator
from utils.revoke.message_auto_revoke import MessageAutoRevoke
from utils.robot_cmd.robot_command import Feature, PermissionStatus, GroupBotManager
@@ -50,19 +52,19 @@ class XiurenImagePlugin(MessagePluginInterface):
"""初始化插件"""
self.LOG = logger
self.LOG.info(f"正在初始化 {self.name} 插件...")
# 保存上下文对象
self.event_system = context.get("event_system")
self._commands = self._config.get("XiurenImage", {}).get("command", ["图来", "秀人"])
self.command_format = self._config.get("XiurenImage", {}).get("command-format", "图来")
self.enable = self._config.get("XiurenImage", {}).get("enable", True)
# 从配置获取图片文件夹，如果配置中有值则使用配置值
config_image_folder = self._config.get("XiurenImage", {}).get("image_folder")
if config_image_folder:
self.image_folder = config_image_folder
# 检查图片文件夹是否存在
try:
if not os.path.exists(self.image_folder):
@@ -70,7 +72,7 @@ class XiurenImagePlugin(MessagePluginInterface):
os.makedirs(self.image_folder, exist_ok=True)
except Exception as e:
self.LOG.error(f"创建图片文件夹失败: {e}")
self.LOG.info(f"[{self.name}] 插件初始化完成，指令：{self._commands}，图片目录：{self.image_folder}")
return True
@@ -114,6 +116,7 @@ class XiurenImagePlugin(MessagePluginInterface):
try:
# 获取随机图片
pic_path = self._get_random_pic()
self.LOG.info(f"返回图片地址：{pic_path}")
if not pic_path:
client_msg_id, create_time, new_msg_id = await bot.send_text_message((roomid if roomid else sender),
f"❌未找到图片资源",
@@ -135,54 +138,71 @@ class XiurenImagePlugin(MessagePluginInterface):
return False, f"处理出错: {e}"
def _get_random_pic(self) -> Optional[str]:
"""获取随机图片路径"""
"""
从 Redis 随机获取图片路径
"""
redis_key = ImageCacheManager.IMAGE_KEY_PREFIX + "all"
try:
# 获取图片文件夹中的所有图片
if not os.path.exists(self.image_folder):
self.LOG.error(f"图片文件夹不存在: {self.image_folder}")
img = DBConnectionManager.get_instance().get_redis_connection().srandmember(redis_key)
if img:
# redis 返回 bytes，转字符串
return img.decode('utf-8') if isinstance(img, bytes) else img
else:
self.LOG.warning("Redis 中没有图片数据")
return None
# 检查读取权限
if not os.access(self.image_folder, os.R_OK):
self.LOG.error(f"没有图片文件夹的读取权限: {self.image_folder}")
return None
self.LOG.debug(f"扫描图片目录: {self.image_folder} (包括子目录)")
image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
image_files = []
# 使用 os.walk() 递归遍历所有子目录
try:
for root, _, files in os.walk(self.image_folder):
for file in files:
try:
_, ext = os.path.splitext(file)
if ext.lower() in image_extensions:
full_path = os.path.join(root, file)
# 检查文件是否可读
if os.access(full_path, os.R_OK):
image_files.append(full_path)
else:
self.LOG.warning(f"文件无法读取: {full_path}")
except Exception as file_err:
self.LOG.warning(f"处理文件时出错: {file} - {file_err}")
except Exception as walk_err:
self.LOG.error(f"遍历目录时出错: {walk_err}")
return None
if not image_files:
self.LOG.warning("在目录中未找到图片文件(包括子目录)")
return None
# 随机选择一张图片
try:
random_pic = random.choice(image_files)
return random_pic # 直接返回路径，不需要再调用os.path.abspath
except Exception as choice_err:
self.LOG.error(f"选择随机图片时出错: {choice_err}")
return None
except Exception as e:
self.LOG.error(f"获取随机图片出错: {e}")
self.LOG.error(f"从 Redis 获取随机图片失败: {e}")
return None
# def _get_random_pic(self) -> Optional[str]:
# """获取随机图片路径"""
# try:
# # 获取图片文件夹中的所有图片
# if not os.path.exists(self.image_folder):
# self.LOG.error(f"图片文件夹不存在: {self.image_folder}")
# return None
#
# # 检查读取权限
# if not os.access(self.image_folder, os.R_OK):
# self.LOG.error(f"没有图片文件夹的读取权限: {self.image_folder}")
# return None
#
# self.LOG.debug(f"扫描图片目录: {self.image_folder} (包括子目录)")
#
# image_extensions = {'.jpg', '.jpeg', '.png', '.gif', '.bmp', '.tiff', '.webp'}
# image_files = []
#
# # 使用 os.walk() 递归遍历所有子目录
# try:
# for root, _, files in os.walk(self.image_folder):
# for file in files:
# try:
# _, ext = os.path.splitext(file)
# if ext.lower() in image_extensions:
# full_path = os.path.join(root, file)
# # 检查文件是否可读
# if os.access(full_path, os.R_OK):
# image_files.append(full_path)
# else:
# self.LOG.warning(f"文件无法读取: {full_path}")
# except Exception as file_err:
# self.LOG.warning(f"处理文件时出错: {file} - {file_err}")
# except Exception as walk_err:
# self.LOG.error(f"遍历目录时出错: {walk_err}")
# return None
#
# if not image_files:
# self.LOG.warning("在目录中未找到图片文件(包括子目录)")
# return None
#
# # 随机选择一张图片
# try:
# random_pic = random.choice(image_files)
# return random_pic # 直接返回路径，不需要再调用os.path.abspath
# except Exception as choice_err:
# self.LOG.error(f"选择随机图片时出错: {choice_err}")
# return None
#
# except Exception as e:
# self.LOG.error(f"获取随机图片出错: {e}")
# return None