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

超级Ruff

Browse Source
This commit is contained in:
墨梓柒
2025-09-09 19:25:12 +08:00
parent ac2936d5fc
commit 163dbb6b90
68 changed files with 1092 additions and 1043 deletions
Download Patch File Download Diff File Expand all files Collapse all files

185
src/mais4u/mais4u_chat/context_web_manager.py
View File

@@ -14,31 +14,31 @@ logger = get_logger("context_web")
class ContextMessage:
"""上下文消息类"""
def __init__(self, message: MessageRecv):
self.user_name = message.message_info.user_info.user_nickname
self.user_id = message.message_info.user_info.user_id
self.content = message.processed_plain_text
self.timestamp = datetime.now()
self.group_name = message.message_info.group_info.group_name if message.message_info.group_info else "私聊"
# 识别消息类型
self.is_gift = getattr(message, 'is_gift', False)
self.is_superchat = getattr(message, 'is_superchat', False)
self.is_gift = getattr(message, "is_gift", False)
self.is_superchat = getattr(message, "is_superchat", False)
# 添加礼物和SC相关信息
if self.is_gift:
self.gift_name = getattr(message, 'gift_name', '')
self.gift_count = getattr(message, 'gift_count', '1')
self.gift_name = getattr(message, "gift_name", "")
self.gift_count = getattr(message, "gift_count", "1")
self.content = f"送出了 {self.gift_name} x{self.gift_count}"
elif self.is_superchat:
self.superchat_price = getattr(message, 'superchat_price', '0')
self.superchat_message = getattr(message, 'superchat_message_text', '')
self.superchat_price = getattr(message, "superchat_price", "0")
self.superchat_message = getattr(message, "superchat_message_text", "")
if self.superchat_message:
self.content = f"[¥{self.superchat_price}] {self.superchat_message}"
else:
self.content = f"[¥{self.superchat_price}] {self.content}"
def to_dict(self):
return {
"user_name": self.user_name,
@@ -47,13 +47,13 @@ class ContextMessage:
"timestamp": self.timestamp.strftime("%m-%d %H:%M:%S"),
"group_name": self.group_name,
"is_gift": self.is_gift,
"is_superchat": self.is_superchat
"is_superchat": self.is_superchat,
}
class ContextWebManager:
"""上下文网页管理器"""
def __init__(self, max_messages: int = 10, port: int = 8765):
self.max_messages = max_messages
self.port = port
@@ -63,53 +63,53 @@ class ContextWebManager:
self.runner = None
self.site = None
self._server_starting = False # 添加启动标志防止并发
async def start_server(self):
"""启动web服务器"""
if self.site is not None:
logger.debug("Web服务器已经启动，跳过重复启动")
return
if self._server_starting:
logger.debug("Web服务器正在启动中，等待启动完成...")
# 等待启动完成
while self._server_starting and self.site is None:
await asyncio.sleep(0.1)
return
self._server_starting = True
try:
self.app = web.Application()
# 设置CORS
cors = aiohttp_cors.setup(self.app, defaults={
"*": aiohttp_cors.ResourceOptions(
allow_credentials=True,
expose_headers="*",
allow_headers="*",
allow_methods="*"
)
})
cors = aiohttp_cors.setup(
self.app,
defaults={
"*": aiohttp_cors.ResourceOptions(
allow_credentials=True, expose_headers="*", allow_headers="*", allow_methods="*"
)
},
)
# 添加路由
self.app.router.add_get('/', self.index_handler)
self.app.router.add_get('/ws', self.websocket_handler)
self.app.router.add_get('/api/contexts', self.get_contexts_handler)
self.app.router.add_get('/debug', self.debug_handler)
self.app.router.add_get("/", self.index_handler)
self.app.router.add_get("/ws", self.websocket_handler)
self.app.router.add_get("/api/contexts", self.get_contexts_handler)
self.app.router.add_get("/debug", self.debug_handler)
# 为所有路由添加CORS
for route in list(self.app.router.routes()):
cors.add(route)
self.runner = web.AppRunner(self.app)
await self.runner.setup()
self.site = web.TCPSite(self.runner, 'localhost', self.port)
self.site = web.TCPSite(self.runner, "localhost", self.port)
await self.site.start()
logger.info(f"🌐 上下文网页服务器启动成功在 http://localhost:{self.port}")
except Exception as e:
logger.error(f"❌ 启动Web服务器失败: {e}")
# 清理部分启动的资源
@@ -121,7 +121,7 @@ class ContextWebManager:
raise
finally:
self._server_starting = False
async def stop_server(self):
"""停止web服务器"""
if self.site:
@@ -132,10 +132,11 @@ class ContextWebManager:
self.runner = None
self.site = None
self._server_starting = False
async def index_handler(self, request):
"""主页处理器"""
html_content = '''
html_content = (
"""
<!DOCTYPE html>
<html>
<head>
@@ -286,7 +287,9 @@ class ContextWebManager:
function connectWebSocket() {
console.log('正在连接WebSocket...');
ws = new WebSocket('ws://localhost:''' + str(self.port) + '''/ws');
ws = new WebSocket('ws://localhost:"""
+ str(self.port)
+ """/ws');
ws.onopen = function() {
console.log('WebSocket连接已建立');
@@ -470,47 +473,48 @@ class ContextWebManager:
</script>
</body>
</html>
'''
return web.Response(text=html_content, content_type='text/html')
"""
)
return web.Response(text=html_content, content_type="text/html")
async def websocket_handler(self, request):
"""WebSocket处理器"""
ws = web.WebSocketResponse()
await ws.prepare(request)
self.websockets.append(ws)
logger.debug(f"WebSocket连接建立，当前连接数: {len(self.websockets)}")
# 发送初始数据
await self.send_contexts_to_websocket(ws)
async for msg in ws:
if msg.type == WSMsgType.ERROR:
logger.error(f'WebSocket错误: {ws.exception()}')
logger.error(f"WebSocket错误: {ws.exception()}")
break
# 清理断开的连接
if ws in self.websockets:
self.websockets.remove(ws)
logger.debug(f"WebSocket连接断开，当前连接数: {len(self.websockets)}")
return ws
async def get_contexts_handler(self, request):
"""获取上下文API"""
all_context_msgs = []
for _chat_id, contexts in self.contexts.items():
all_context_msgs.extend(list(contexts))
# 按时间排序，最新的在最后
all_context_msgs.sort(key=lambda x: x.timestamp)
# 转换为字典格式
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages:]]
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages :]]
logger.debug(f"返回上下文数据，共 {len(contexts_data)} 条消息")
return web.json_response({"contexts": contexts_data})
async def debug_handler(self, request):
"""调试信息处理器"""
debug_info = {
@@ -519,7 +523,7 @@ class ContextWebManager:
"total_chats": len(self.contexts),
"total_messages": sum(len(contexts) for contexts in self.contexts.values()),
}
# 构建聊天详情HTML
chats_html = ""
for chat_id, contexts in self.contexts.items():
@@ -528,15 +532,15 @@ class ContextWebManager:
timestamp = msg.timestamp.strftime("%H:%M:%S")
content = msg.content[:50] + "..." if len(msg.content) > 50 else msg.content
messages_html += f'<div class="message">[{timestamp}] {msg.user_name}: {content}</div>'
chats_html += f'''
chats_html += f"""
<div class="chat">
<h3>聊天 {chat_id} ({len(contexts)} 条消息)</h3>
{messages_html}
</div>
'''
html_content = f'''
"""
html_content = f"""
<!DOCTYPE html>
<html>
<head>
@@ -578,74 +582,78 @@ class ContextWebManager:
</script>
</body>
</html>
'''
return web.Response(text=html_content, content_type='text/html')
"""
return web.Response(text=html_content, content_type="text/html")
async def add_message(self, chat_id: str, message: MessageRecv):
"""添加新消息到上下文"""
if chat_id not in self.contexts:
self.contexts[chat_id] = deque(maxlen=self.max_messages)
logger.debug(f"为聊天 {chat_id} 创建新的上下文队列")
context_msg = ContextMessage(message)
self.contexts[chat_id].append(context_msg)
# 统计当前总消息数
total_messages = sum(len(contexts) for contexts in self.contexts.values())
logger.info(f"✅ 添加消息到上下文 [总数: {total_messages}]: [{context_msg.group_name}] {context_msg.user_name}: {context_msg.content}")
logger.info(
f"✅ 添加消息到上下文 [总数: {total_messages}]: [{context_msg.group_name}] {context_msg.user_name}: {context_msg.content}"
)
# 调试：打印当前所有消息
logger.info("📝 当前上下文中的所有消息：")
for cid, contexts in self.contexts.items():
logger.info(f" 聊天 {cid}: {len(contexts)} 条消息")
for i, msg in enumerate(contexts):
logger.info(f" {i+1}. [{msg.timestamp.strftime('%H:%M:%S')}] {msg.user_name}: {msg.content[:30]}...")
logger.info(
f" {i + 1}. [{msg.timestamp.strftime('%H:%M:%S')}] {msg.user_name}: {msg.content[:30]}..."
)
# 广播更新给所有WebSocket连接
await self.broadcast_contexts()
async def send_contexts_to_websocket(self, ws: web.WebSocketResponse):
"""向单个WebSocket发送上下文数据"""
all_context_msgs = []
for _chat_id, contexts in self.contexts.items():
all_context_msgs.extend(list(contexts))
# 按时间排序，最新的在最后
all_context_msgs.sort(key=lambda x: x.timestamp)
# 转换为字典格式
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages:]]
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages :]]
data = {"contexts": contexts_data}
await ws.send_str(json.dumps(data, ensure_ascii=False))
async def broadcast_contexts(self):
"""向所有WebSocket连接广播上下文更新"""
if not self.websockets:
logger.debug("没有WebSocket连接，跳过广播")
return
all_context_msgs = []
for _chat_id, contexts in self.contexts.items():
all_context_msgs.extend(list(contexts))
# 按时间排序，最新的在最后
all_context_msgs.sort(key=lambda x: x.timestamp)
# 转换为字典格式
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages:]]
contexts_data = [msg.to_dict() for msg in all_context_msgs[-self.max_messages :]]
data = {"contexts": contexts_data}
message = json.dumps(data, ensure_ascii=False)
logger.info(f"广播 {len(contexts_data)} 条消息到 {len(self.websockets)} 个WebSocket连接")
# 创建WebSocket列表的副本，避免在遍历时修改
websockets_copy = self.websockets.copy()
removed_count = 0
for ws in websockets_copy:
if ws.closed:
if ws in self.websockets:
@@ -660,7 +668,7 @@ class ContextWebManager:
if ws in self.websockets:
self.websockets.remove(ws)
removed_count += 1
if removed_count > 0:
logger.debug(f"清理了 {removed_count} 个断开的WebSocket连接")
@@ -681,5 +689,4 @@ async def init_context_web_manager():
"""初始化上下文网页管理器"""
manager = get_context_web_manager()
await manager.start_server()
return manager
return manager

78
src/mais4u/mais4u_chat/gift_manager.py
View File

@@ -11,6 +11,7 @@ logger = get_logger("gift_manager")
@dataclass
class PendingGift:
"""等待中的礼物消息"""
message: MessageRecvS4U
total_count: int
timer_task: asyncio.Task
@@ -19,71 +20,68 @@ class PendingGift:
class GiftManager:
"""礼物管理器，提供防抖功能"""
def __init__(self):
"""初始化礼物管理器"""
self.pending_gifts: Dict[Tuple[str, str], PendingGift] = {}
self.debounce_timeout = 5.0 # 3秒防抖时间
async def handle_gift(self, message: MessageRecvS4U, callback: Optional[Callable[[MessageRecvS4U], None]] = None) -> bool:
async def handle_gift(
self, message: MessageRecvS4U, callback: Optional[Callable[[MessageRecvS4U], None]] = None
) -> bool:
"""处理礼物消息，返回是否应该立即处理
Args:
message: 礼物消息
callback: 防抖完成后的回调函数
Returns:
bool: False表示消息被暂存等待防抖，True表示应该立即处理
"""
if not message.is_gift:
return True
# 构建礼物的唯一键：(发送人ID, 礼物名称)
gift_key = (message.message_info.user_info.user_id, message.gift_name)
# 如果已经有相同的礼物在等待中，则合并
if gift_key in self.pending_gifts:
await self._merge_gift(gift_key, message)
return False
# 创建新的等待礼物
await self._create_pending_gift(gift_key, message, callback)
return False
async def _merge_gift(self, gift_key: Tuple[str, str], new_message: MessageRecvS4U) -> None:
"""合并礼物消息"""
pending_gift = self.pending_gifts[gift_key]
# 取消之前的定时器
if not pending_gift.timer_task.cancelled():
pending_gift.timer_task.cancel()
# 累加礼物数量
try:
new_count = int(new_message.gift_count)
pending_gift.total_count += new_count
# 更新消息为最新的（保留最新的消息，但累加数量）
pending_gift.message = new_message
pending_gift.message.gift_count = str(pending_gift.total_count)
pending_gift.message.gift_info = f"{pending_gift.message.gift_name}:{pending_gift.total_count}"
except ValueError:
logger.warning(f"无法解析礼物数量: {new_message.gift_count}")
# 如果无法解析数量，保持原有数量不变
# 重新创建定时器
pending_gift.timer_task = asyncio.create_task(
self._gift_timeout(gift_key)
)
pending_gift.timer_task = asyncio.create_task(self._gift_timeout(gift_key))
logger.debug(f"合并礼物: {gift_key}, 总数量: {pending_gift.total_count}")
async def _create_pending_gift(
self,
gift_key: Tuple[str, str],
message: MessageRecvS4U,
callback: Optional[Callable[[MessageRecvS4U], None]]
self, gift_key: Tuple[str, str], message: MessageRecvS4U, callback: Optional[Callable[[MessageRecvS4U], None]]
) -> None:
"""创建新的等待礼物"""
try:
@@ -91,56 +89,51 @@ class GiftManager:
except ValueError:
initial_count = 1
logger.warning(f"无法解析礼物数量: {message.gift_count}，默认设为1")
# 创建定时器任务
timer_task = asyncio.create_task(self._gift_timeout(gift_key))
# 创建等待礼物对象
pending_gift = PendingGift(
message=message,
total_count=initial_count,
timer_task=timer_task,
callback=callback
)
pending_gift = PendingGift(message=message, total_count=initial_count, timer_task=timer_task, callback=callback)
self.pending_gifts[gift_key] = pending_gift
logger.debug(f"创建等待礼物: {gift_key}, 初始数量: {initial_count}")
async def _gift_timeout(self, gift_key: Tuple[str, str]) -> None:
"""礼物防抖超时处理"""
try:
# 等待防抖时间
await asyncio.sleep(self.debounce_timeout)
# 获取等待中的礼物
if gift_key not in self.pending_gifts:
return
pending_gift = self.pending_gifts.pop(gift_key)
logger.info(f"礼物防抖完成: {gift_key}, 最终数量: {pending_gift.total_count}")
message = pending_gift.message
message.processed_plain_text = f"用户{message.message_info.user_info.user_nickname}送出了礼物{message.gift_name} x{pending_gift.total_count}"
# 执行回调
if pending_gift.callback:
try:
pending_gift.callback(message)
except Exception as e:
logger.error(f"礼物回调执行失败: {e}", exc_info=True)
except asyncio.CancelledError:
# 定时器被取消，不需要处理
pass
except Exception as e:
logger.error(f"礼物防抖处理异常: {e}", exc_info=True)
def get_pending_count(self) -> int:
"""获取当前等待中的礼物数量"""
return len(self.pending_gifts)
async def flush_all(self) -> None:
"""立即处理所有等待中的礼物"""
for gift_key in list(self.pending_gifts.keys()):
@@ -152,4 +145,3 @@ class GiftManager:
# 创建全局礼物管理器实例
gift_manager = GiftManager()

11
src/mais4u/mais4u_chat/internal_manager.py
View File

@@ -1,14 +1,15 @@
class InternalManager:
def __init__(self):
self.now_internal_state = str()
def set_internal_state(self,internal_state:str):
def set_internal_state(self, internal_state: str):
self.now_internal_state = internal_state
def get_internal_state(self):
return self.now_internal_state
def get_internal_state_str(self):
return f"你今天的直播内容是直播QQ水群，你正在一边回复弹幕，一边在QQ群聊天，你在QQ群聊天中产生的想法是：{self.now_internal_state}"
internal_manager = InternalManager()
internal_manager = InternalManager()

119
src/mais4u/mais4u_chat/s4u_chat.py
View File

@@ -16,7 +16,6 @@ import json
from .s4u_mood_manager import mood_manager
from src.mais4u.s4u_config import s4u_config
from src.person_info.person_info import get_person_id
from .super_chat_manager import get_super_chat_manager
from .yes_or_no import yes_or_no_head
logger = get_logger("S4U_chat")
@@ -33,15 +32,12 @@ class MessageSenderContainer:
self._task: Optional[asyncio.Task] = None
self._paused_event = asyncio.Event()
self._paused_event.set() # 默认设置为非暂停状态
self.msg_id = ""
self.last_msg_id = ""
self.voice_done = ""
self.msg_id = ""
self.last_msg_id = ""
self.voice_done = ""
async def add_message(self, chunk: str):
"""向队列中添加一个消息块。"""
@@ -131,7 +127,7 @@ class MessageSenderContainer:
reply_to=f"{self.original_message.message_info.user_info.platform}:{self.original_message.message_info.user_info.user_id}",
)
await bot_message.process()
await self.storage.store_message(bot_message, self.chat_stream)
except Exception as e:
@@ -198,12 +194,12 @@ class S4UChat:
self.gpt = S4UStreamGenerator()
self.gpt.chat_stream = self.chat_stream
self.interest_dict: Dict[str, float] = {} # 用户兴趣分
self.internal_message :List[MessageRecvS4U] = []
self.internal_message: List[MessageRecvS4U] = []
self.msg_id = ""
self.voice_done = ""
logger.info(f"[{self.stream_name}] S4UChat with two-queue system initialized.")
def _get_priority_info(self, message: MessageRecv) -> dict:
@@ -226,7 +222,7 @@ class S4UChat:
def _get_interest_score(self, user_id: str) -> float:
"""获取用户的兴趣分，默认为1.0"""
return self.interest_dict.get(user_id, 1.0)
def go_processing(self):
if self.voice_done == self.last_msg_id:
return True
@@ -237,14 +233,14 @@ class S4UChat:
为消息计算基础优先级分数。分数越高，优先级越高。
"""
score = 0.0
# 加上消息自带的优先级
score += priority_info.get("message_priority", 0.0)
# 加上用户的固有兴趣分
score += self._get_interest_score(message.message_info.user_info.user_id)
return score
def decay_interest_score(self):
for person_id, score in self.interest_dict.items():
if score > 0:
@@ -252,15 +248,14 @@ class S4UChat:
else:
self.interest_dict[person_id] = 0
async def add_message(self, message: MessageRecvS4U|MessageRecv) -> None:
async def add_message(self, message: MessageRecvS4U | MessageRecv) -> None:
self.decay_interest_score()
"""根据VIP状态和中断逻辑将消息放入相应队列。"""
user_id = message.message_info.user_info.user_id
platform = message.message_info.platform
person_id = get_person_id(platform, user_id)
_person_id = get_person_id(platform, user_id)
# try:
# is_gift = message.is_gift
# is_superchat = message.is_superchat
@@ -276,7 +271,7 @@ class S4UChat:
# # 安全地增加兴趣分，如果person_id不存在则先初始化为1.0
# current_score = self.interest_dict.get(person_id, 1.0)
# self.interest_dict[person_id] = current_score + 0.1 * float(message.superchat_price)
# # 添加SuperChat到管理器
# super_chat_manager = get_super_chat_manager()
# await super_chat_manager.add_superchat(message)
@@ -284,16 +279,19 @@ class S4UChat:
# await self.relationship_builder.build_relation(20)
# except Exception:
# traceback.print_exc()
logger.info(f"[{self.stream_name}] 消息处理完毕，消息内容：{message.processed_plain_text}")
priority_info = self._get_priority_info(message)
is_vip = self._is_vip(priority_info)
new_priority_score = self._calculate_base_priority_score(message, priority_info)
should_interrupt = False
if (s4u_config.enable_message_interruption and
self._current_generation_task and not self._current_generation_task.done()):
if (
s4u_config.enable_message_interruption
and self._current_generation_task
and not self._current_generation_task.done()
):
if self._current_message_being_replied:
current_queue, current_priority, _, current_msg = self._current_message_being_replied
@@ -344,39 +342,45 @@ class S4UChat:
"""清理普通队列中不在最近N条消息范围内的消息"""
if not s4u_config.enable_old_message_cleanup or self._normal_queue.empty():
return
# 计算阈值：保留最近 recent_message_keep_count 条消息
cutoff_counter = max(0, self._entry_counter - s4u_config.recent_message_keep_count)
# 临时存储需要保留的消息
temp_messages = []
removed_count = 0
# 取出所有普通队列中的消息
while not self._normal_queue.empty():
try:
item = self._normal_queue.get_nowait()
neg_priority, entry_count, timestamp, message = item
# 如果消息在最近N条消息范围内，保留它
logger.info(f"检查消息:{message.processed_plain_text},entry_count:{entry_count} cutoff_counter:{cutoff_counter}")
logger.info(
f"检查消息:{message.processed_plain_text},entry_count:{entry_count} cutoff_counter:{cutoff_counter}"
)
if entry_count >= cutoff_counter:
temp_messages.append(item)
else:
removed_count += 1
self._normal_queue.task_done() # 标记被移除的任务为完成
except asyncio.QueueEmpty:
break
# 将保留的消息重新放入队列
for item in temp_messages:
self._normal_queue.put_nowait(item)
if removed_count > 0:
logger.info(f"消息{message.processed_plain_text}超过{s4u_config.recent_message_keep_count}条，现在counter:{self._entry_counter}被移除")
logger.info(f"[{self.stream_name}] Cleaned up {removed_count} old normal messages outside recent {s4u_config.recent_message_keep_count} range.")
logger.info(
f"消息{message.processed_plain_text}超过{s4u_config.recent_message_keep_count}条，现在counter:{self._entry_counter}被移除"
)
logger.info(
f"[{self.stream_name}] Cleaned up {removed_count} old normal messages outside recent {s4u_config.recent_message_keep_count} range."
)
async def _message_processor(self):
"""调度器：优先处理VIP队列，然后处理普通队列。"""
@@ -385,7 +389,7 @@ class S4UChat:
# 等待有新消息的信号，避免空转
await self._new_message_event.wait()
self._new_message_event.clear()
# 清理普通队列中的过旧消息
self._cleanup_old_normal_messages()
@@ -396,7 +400,6 @@ class S4UChat:
queue_name = "vip"
# 其次处理普通队列
elif not self._normal_queue.empty():
neg_priority, entry_count, timestamp, message = self._normal_queue.get_nowait()
priority = -neg_priority
# 检查普通消息是否超时
@@ -411,13 +414,15 @@ class S4UChat:
if self.internal_message:
message = self.internal_message[-1]
self.internal_message = []
priority = 0
neg_priority = 0
entry_count = 0
queue_name = "internal"
logger.info(f"[{self.stream_name}] normal/vip 队列都空，触发 internal_message 回复: {getattr(message, 'processed_plain_text', str(message))[:20]}...")
logger.info(
f"[{self.stream_name}] normal/vip 队列都空，触发 internal_message 回复: {getattr(message, 'processed_plain_text', str(message))[:20]}..."
)
else:
continue # 没有消息了，回去等事件
@@ -457,23 +462,21 @@ class S4UChat:
except Exception as e:
logger.error(f"[{self.stream_name}] Message processor main loop error: {e}", exc_info=True)
await asyncio.sleep(1)
def get_processing_message_id(self):
self.last_msg_id = self.msg_id
self.msg_id = f"{time.time()}_{random.randint(1000, 9999)}"
async def _generate_and_send(self, message: MessageRecv):
"""为单个消息生成文本回复。整个过程可以被中断。"""
self._is_replying = True
total_chars_sent = 0 # 跟踪发送的总字符数
self.get_processing_message_id()
# 视线管理：开始生成回复时切换视线状态
chat_watching = watching_manager.get_watching_by_chat_id(self.stream_id)
if message.is_internal:
await chat_watching.on_internal_message_start()
else:
@@ -516,16 +519,19 @@ class S4UChat:
total_chars_sent = len("麦麦不知道哦")
mood = mood_manager.get_mood_by_chat_id(self.stream_id)
await yes_or_no_head(text = total_chars_sent,emotion = mood.mood_state,chat_history=message.processed_plain_text,chat_id=self.stream_id)
await yes_or_no_head(
text=total_chars_sent,
emotion=mood.mood_state,
chat_history=message.processed_plain_text,
chat_id=self.stream_id,
)
# 等待所有文本消息发送完成
await sender_container.close()
await sender_container.join()
await chat_watching.on_thinking_finished()
start_time = time.time()
logged = False
while not self.go_processing():
@@ -536,7 +542,7 @@ class S4UChat:
logger.info(f"[{self.stream_name}] 等待消息发送完成...")
logged = True
await asyncio.sleep(0.2)
logger.info(f"[{self.stream_name}] 所有文本块处理完毕。")
except asyncio.CancelledError:
@@ -548,11 +554,11 @@ class S4UChat:
# 回复生成实时展示：清空内容（出错时）
finally:
self._is_replying = False
# 视线管理：回复结束时切换视线状态
chat_watching = watching_manager.get_watching_by_chat_id(self.stream_id)
await chat_watching.on_reply_finished()
# 确保发送器被妥善关闭（即使已关闭，再次调用也是安全的）
sender_container.resume()
if not sender_container._task.done():
@@ -576,4 +582,3 @@ class S4UChat:
await self._processing_task
except asyncio.CancelledError:
logger.info(f"处理任务已成功取消: {self.stream_name}")

79
src/mais4u/mais4u_chat/s4u_msg_processor.py
View File

@@ -40,7 +40,7 @@ async def _calculate_interest(message: MessageRecv) -> Tuple[float, bool]:
if global_config.memory.enable_memory:
with Timer("记忆激活"):
interested_rate,_ ,_= await hippocampus_manager.get_activate_from_text(
interested_rate, _, _ = await hippocampus_manager.get_activate_from_text(
message.processed_plain_text,
fast_retrieval=True,
)
@@ -49,7 +49,7 @@ async def _calculate_interest(message: MessageRecv) -> Tuple[float, bool]:
text_len = len(message.processed_plain_text)
# 根据文本长度分布调整兴趣度，采用分段函数实现更精确的兴趣度计算
# 基于实际分布：0-5字符(26.57%), 6-10字符(27.18%), 11-20字符(22.76%), 21-30字符(10.33%), 31+字符(13.86%)
if text_len == 0:
base_interest = 0.01 # 空消息最低兴趣度
elif text_len <= 5:
@@ -73,7 +73,7 @@ async def _calculate_interest(message: MessageRecv) -> Tuple[float, bool]:
else:
# 100+字符：对数增长 0.26 -> 0.3，增长率递减
base_interest = 0.26 + (0.3 - 0.26) * (math.log10(text_len - 99) / math.log10(901)) # 1000-99=901
# 确保在范围内
base_interest = min(max(base_interest, 0.01), 0.3)
@@ -117,36 +117,32 @@ class S4UMessageProcessor:
user_info=userinfo,
group_info=groupinfo,
)
if await self.handle_internal_message(message):
return
if await self.hadle_if_voice_done(message):
return
# 处理礼物消息，如果消息被暂存则停止当前处理流程
if not skip_gift_debounce and not await self.handle_if_gift(message):
return
await self.check_if_fake_gift(message)
# 处理屏幕消息
if await self.handle_screen_message(message):
return
await self.storage.store_message(message, chat)
s4u_chat = get_s4u_chat_manager().get_or_create_chat(chat)
await s4u_chat.add_message(message)
_interested_rate, _ = await _calculate_interest(message)
await mood_manager.start()
# 一系列llm驱动的前处理
chat_mood = mood_manager.get_mood_by_chat_id(chat.stream_id)
asyncio.create_task(chat_mood.update_mood_by_message(message))
@@ -164,61 +160,56 @@ class S4UMessageProcessor:
logger.info(f"[S4U-礼物] {userinfo.user_nickname} 送出了 {message.gift_name} x{message.gift_count}")
else:
logger.info(f"[S4U]{userinfo.user_nickname}:{message.processed_plain_text}")
async def handle_internal_message(self, message: MessageRecvS4U):
if message.is_internal:
group_info = GroupInfo(platform = "amaidesu_default",group_id = 660154,group_name = "内心")
chat = await get_chat_manager().get_or_create_stream(
platform = "amaidesu_default",
user_info = message.message_info.user_info,
group_info = group_info
group_info = GroupInfo(platform="amaidesu_default", group_id=660154, group_name="内心")
chat = await get_chat_manager().get_or_create_stream(
platform="amaidesu_default", user_info=message.message_info.user_info, group_info=group_info
)
s4u_chat = get_s4u_chat_manager().get_or_create_chat(chat)
message.message_info.group_info = s4u_chat.chat_stream.group_info
message.message_info.platform = s4u_chat.chat_stream.platform
s4u_chat.internal_message.append(message)
s4u_chat._new_message_event.set()
logger.info(f"[{s4u_chat.stream_name}] 添加内部消息-------------------------------------------------------: {message.processed_plain_text}")
logger.info(
f"[{s4u_chat.stream_name}] 添加内部消息-------------------------------------------------------: {message.processed_plain_text}"
)
return True
return False
async def handle_screen_message(self, message: MessageRecvS4U):
if message.is_screen:
screen_manager.set_screen(message.screen_info)
return True
return False
async def hadle_if_voice_done(self, message: MessageRecvS4U):
if message.voice_done:
s4u_chat = get_s4u_chat_manager().get_or_create_chat(message.chat_stream)
s4u_chat.voice_done = message.voice_done
return True
return False
async def check_if_fake_gift(self, message: MessageRecvS4U) -> bool:
"""检查消息是否为假礼物"""
if message.is_gift:
return False
gift_keywords = ["送出了礼物", "礼物", "送出了","投喂"]
gift_keywords = ["送出了礼物", "礼物", "送出了", "投喂"]
if any(keyword in message.processed_plain_text for keyword in gift_keywords):
message.is_fake_gift = True
return True
return False
async def handle_if_gift(self, message: MessageRecvS4U) -> bool:
"""处理礼物消息
Returns:
bool: True表示应该继续处理消息，False表示消息已被暂存不需要继续处理
"""
@@ -228,37 +219,37 @@ class S4UMessageProcessor:
"""礼物防抖完成后的回调"""
# 创建异步任务来处理合并后的礼物消息，跳过防抖处理
asyncio.create_task(self.process_message(merged_message, skip_gift_debounce=True))
# 交给礼物管理器处理，并传入回调函数
# 对于礼物消息，handle_gift 总是返回 False（消息被暂存）
await gift_manager.handle_gift(message, gift_callback)
return False # 消息被暂存，不继续处理
return True # 非礼物消息，继续正常处理
async def _handle_context_web_update(self, chat_id: str, message: MessageRecv):
"""处理上下文网页更新的独立task
Args:
chat_id: 聊天ID
message: 消息对象
"""
try:
logger.debug(f"🔄 开始处理上下文网页更新: {message.message_info.user_info.user_nickname}")
context_manager = get_context_web_manager()
# 只在服务器未启动时启动（避免重复启动）
if context_manager.site is None:
logger.info("🚀 首次启动上下文网页服务器...")
await context_manager.start_server()
# 添加消息到上下文并更新网页
await asyncio.sleep(1.5)
await context_manager.add_message(chat_id, message)
logger.debug(f"✅ 上下文网页更新完成: {message.message_info.user_info.user_nickname}")
except Exception as e:
logger.error(f"❌ 处理上下文网页更新失败: {e}", exc_info=True)

10
src/mais4u/mais4u_chat/s4u_prompt.py
View File

@@ -176,7 +176,7 @@ class PromptBuilder:
message_list_before_now = get_raw_msg_before_timestamp_with_chat(
chat_id=chat_stream.stream_id,
timestamp=time.time(),
# sourcery skip: lift-duplicated-conditional, merge-duplicate-blocks, remove-redundant-if
# sourcery skip: lift-duplicated-conditional, merge-duplicate-blocks, remove-redundant-if
limit=300,
)
@@ -228,13 +228,17 @@ class PromptBuilder:
last_speaking_user_id = start_speaking_user_id
msg_seg_str = "对方的发言：\n"
msg_seg_str += f"{time.strftime('%H:%M:%S', time.localtime(first_msg.time))}: {first_msg.processed_plain_text}\n"
msg_seg_str += (
f"{time.strftime('%H:%M:%S', time.localtime(first_msg.time))}: {first_msg.processed_plain_text}\n"
)
all_msg_seg_list = []
for msg in core_dialogue_list[1:]:
speaker = msg.user_info.user_id
if speaker == last_speaking_user_id:
msg_seg_str += f"{time.strftime('%H:%M:%S', time.localtime(msg.time))}: {msg.processed_plain_text}\n"
msg_seg_str += (
f"{time.strftime('%H:%M:%S', time.localtime(msg.time))}: {msg.processed_plain_text}\n"
)
else:
msg_seg_str = f"{msg_seg_str}\n"
all_msg_seg_list.append(msg_seg_str)

23
src/mais4u/mais4u_chat/s4u_stream_generator.py
View File

@@ -14,11 +14,8 @@ logger = get_logger("s4u_stream_generator")
class S4UStreamGenerator:
def __init__(self):
# 使用LLMRequest替代AsyncOpenAIClient
self.llm_request = LLMRequest(
model_set=model_config.model_task_config.replyer,
request_type="s4u_replyer"
)
self.llm_request = LLMRequest(model_set=model_config.model_task_config.replyer, request_type="s4u_replyer")
self.current_model_name = "unknown model"
self.partial_response = ""
@@ -89,16 +86,16 @@ class S4UStreamGenerator:
async def _generate_response_with_llm_request(self, prompt: str) -> AsyncGenerator[str, None]:
"""使用LLMRequest进行流式响应生成"""
# 构建消息
message_builder = MessageBuilder()
message_builder.add_text_content(prompt)
messages = [message_builder.build()]
# 选择模型
model_info, api_provider, client = self.llm_request._select_model()
self.current_model_name = model_info.name
# 如果模型支持强制流式模式，使用真正的流式处理
if model_info.force_stream_mode:
# 简化流式处理：直接使用LLMRequest的流式功能
@@ -111,14 +108,14 @@ class S4UStreamGenerator:
model_info=model_info,
message_list=messages,
)
# 处理响应内容
content = response.content or ""
if content:
# 将内容按句子分割并输出
async for chunk in self._process_content_streaming(content):
yield chunk
except Exception as e:
logger.error(f"流式请求执行失败: {e}")
# 如果流式请求失败，回退到普通模式
@@ -132,7 +129,7 @@ class S4UStreamGenerator:
content = response.content or ""
async for chunk in self._process_content_streaming(content):
yield chunk
else:
# 如果不支持流式，使用普通方式然后模拟流式输出
response = await self.llm_request._execute_request(
@@ -142,7 +139,7 @@ class S4UStreamGenerator:
model_info=model_info,
message_list=messages,
)
content = response.content or ""
async for chunk in self._process_content_streaming(content):
yield chunk
@@ -163,7 +160,7 @@ class S4UStreamGenerator:
"""处理内容进行流式输出（用于非流式模型的模拟流式输出）"""
buffer = content
punctuation_buffer = ""
# 使用正则表达式匹配句子
last_match_end = 0
for match in self.sentence_split_pattern.finditer(buffer):

11
src/mais4u/mais4u_chat/s4u_watching_manager.py
View File

@@ -1,4 +1,3 @@
from src.common.logger import get_logger
from src.plugin_system.apis import send_api
@@ -47,6 +46,7 @@ HEAD_CODE = {
"看向正前方": "(0,0,0)",
}
class ChatWatching:
def __init__(self, chat_id: str):
self.chat_id: str = chat_id
@@ -56,13 +56,13 @@ class ChatWatching:
await send_api.custom_to_stream(
message_type="state", content="start_thinking", stream_id=self.chat_id, storage_message=False
)
async def on_reply_finished(self):
"""生成回复完毕时调用"""
await send_api.custom_to_stream(
message_type="state", content="finish_reply", stream_id=self.chat_id, storage_message=False
)
async def on_thinking_finished(self):
"""思考完毕时调用"""
await send_api.custom_to_stream(
@@ -74,14 +74,14 @@ class ChatWatching:
await send_api.custom_to_stream(
message_type="state", content="start_viewing", stream_id=self.chat_id, storage_message=False
)
async def on_internal_message_start(self):
"""收到消息时调用"""
await send_api.custom_to_stream(
message_type="state", content="start_internal_thinking", stream_id=self.chat_id, storage_message=False
)
class WatchingManager:
def __init__(self):
self.watching_list: list[ChatWatching] = []
@@ -100,6 +100,7 @@ class WatchingManager:
return new_watching
# 全局视线管理器实例
watching_manager = WatchingManager()
"""全局视线管理器"""

11
src/mais4u/mais4u_chat/screen_manager.py
View File

@@ -1,14 +1,15 @@
class ScreenManager:
def __init__(self):
self.now_screen = str()
def set_screen(self,screen_str:str):
def set_screen(self, screen_str: str):
self.now_screen = screen_str
def get_screen(self):
return self.now_screen
def get_screen_str(self):
return f"你可以看见面前的屏幕，目前屏幕的内容是:现在千石可乐在和你一起直播，这是他正在操作的屏幕内容：{self.now_screen}"
screen_manager = ScreenManager()
screen_manager = ScreenManager()

109
src/mais4u/mais4u_chat/super_chat_manager.py
View File

@@ -4,6 +4,7 @@ from dataclasses import dataclass
from typing import Dict, List, Optional
from src.common.logger import get_logger
from src.chat.message_receive.message import MessageRecvS4U
# 全局SuperChat管理器实例
from src.mais4u.s4u_config import s4u_config
@@ -13,7 +14,7 @@ logger = get_logger("super_chat_manager")
@dataclass
class SuperChatRecord:
"""SuperChat记录数据类"""
user_id: str
user_nickname: str
platform: str
@@ -23,15 +24,15 @@ class SuperChatRecord:
timestamp: float
expire_time: float
group_name: Optional[str] = None
def is_expired(self) -> bool:
"""检查SuperChat是否已过期"""
return time.time() > self.expire_time
def remaining_time(self) -> float:
"""获取剩余时间（秒）"""
return max(0, self.expire_time - time.time())
def to_dict(self) -> dict:
"""转换为字典格式"""
return {
@@ -44,19 +45,19 @@ class SuperChatRecord:
"timestamp": self.timestamp,
"expire_time": self.expire_time,
"group_name": self.group_name,
"remaining_time": self.remaining_time()
"remaining_time": self.remaining_time(),
}
class SuperChatManager:
"""SuperChat管理器，负责管理和跟踪SuperChat消息"""
def __init__(self):
self.super_chats: Dict[str, List[SuperChatRecord]] = {} # chat_id -> SuperChat列表
self._cleanup_task: Optional[asyncio.Task] = None
self._is_initialized = False
logger.info("SuperChat管理器已初始化")
def _ensure_cleanup_task_started(self):
"""确保清理任务已启动（延迟启动）"""
if self._cleanup_task is None or self._cleanup_task.done():
@@ -68,7 +69,7 @@ class SuperChatManager:
except RuntimeError:
# 没有运行的事件循环，稍后再启动
logger.debug("当前没有运行的事件循环，将在需要时启动清理任务")
def _start_cleanup_task(self):
"""启动清理任务（已弃用，保留向后兼容）"""
self._ensure_cleanup_task_started()
@@ -78,39 +79,36 @@ class SuperChatManager:
while True:
try:
total_removed = 0
for chat_id in list(self.super_chats.keys()):
original_count = len(self.super_chats[chat_id])
# 移除过期的SuperChat
self.super_chats[chat_id] = [
sc for sc in self.super_chats[chat_id]
if not sc.is_expired()
]
self.super_chats[chat_id] = [sc for sc in self.super_chats[chat_id] if not sc.is_expired()]
removed_count = original_count - len(self.super_chats[chat_id])
total_removed += removed_count
if removed_count > 0:
logger.info(f"从聊天 {chat_id} 中清理了 {removed_count} 个过期的SuperChat")
# 如果列表为空，删除该聊天的记录
if not self.super_chats[chat_id]:
del self.super_chats[chat_id]
if total_removed > 0:
logger.info(f"总共清理了 {total_removed} 个过期的SuperChat")
# 每30秒检查一次
await asyncio.sleep(30)
except Exception as e:
logger.error(f"清理过期SuperChat时出错: {e}", exc_info=True)
await asyncio.sleep(60) # 出错时等待更长时间
def _calculate_expire_time(self, price: float) -> float:
"""根据SuperChat金额计算过期时间"""
current_time = time.time()
# 根据金额阶梯设置不同的存活时间
if price >= 500:
# 500元以上：保持4小时
@@ -133,27 +131,27 @@ class SuperChatManager:
else:
# 10元以下：保持5分钟
duration = 5 * 60
return current_time + duration
async def add_superchat(self, message: MessageRecvS4U) -> None:
"""添加新的SuperChat记录"""
# 确保清理任务已启动
self._ensure_cleanup_task_started()
if not message.is_superchat or not message.superchat_price:
logger.warning("尝试添加非SuperChat消息到SuperChat管理器")
return
try:
price = float(message.superchat_price)
except (ValueError, TypeError):
logger.error(f"无效的SuperChat价格: {message.superchat_price}")
return
user_info = message.message_info.user_info
group_info = message.message_info.group_info
chat_id = getattr(message, 'chat_stream', None)
chat_id = getattr(message, "chat_stream", None)
if chat_id:
chat_id = chat_id.stream_id
else:
@@ -161,9 +159,9 @@ class SuperChatManager:
chat_id = f"{message.message_info.platform}_{user_info.user_id}"
if group_info:
chat_id = f"{message.message_info.platform}_{group_info.group_id}"
expire_time = self._calculate_expire_time(price)
record = SuperChatRecord(
user_id=user_info.user_id,
user_nickname=user_info.user_nickname,
@@ -173,44 +171,44 @@ class SuperChatManager:
message_text=message.superchat_message_text or "",
timestamp=message.message_info.time,
expire_time=expire_time,
group_name=group_info.group_name if group_info else None
group_name=group_info.group_name if group_info else None,
)
# 添加到对应聊天的SuperChat列表
if chat_id not in self.super_chats:
self.super_chats[chat_id] = []
self.super_chats[chat_id].append(record)
# 按价格降序排序（价格高的在前）
self.super_chats[chat_id].sort(key=lambda x: x.price, reverse=True)
logger.info(f"添加SuperChat记录: {user_info.user_nickname} - {price}元 - {message.superchat_message_text}")
def get_superchats_by_chat(self, chat_id: str) -> List[SuperChatRecord]:
"""获取指定聊天的所有有效SuperChat"""
# 确保清理任务已启动
self._ensure_cleanup_task_started()
if chat_id not in self.super_chats:
return []
# 过滤掉过期的SuperChat
valid_superchats = [sc for sc in self.super_chats[chat_id] if not sc.is_expired()]
return valid_superchats
def get_all_valid_superchats(self) -> Dict[str, List[SuperChatRecord]]:
"""获取所有有效的SuperChat"""
# 确保清理任务已启动
self._ensure_cleanup_task_started()
result = {}
for chat_id, superchats in self.super_chats.items():
valid_superchats = [sc for sc in superchats if not sc.is_expired()]
if valid_superchats:
result[chat_id] = valid_superchats
return result
def build_superchat_display_string(self, chat_id: str, max_count: int = 10) -> str:
"""构建SuperChat显示字符串"""
superchats = self.get_superchats_by_chat(chat_id)
@@ -226,7 +224,9 @@ class SuperChatManager:
remaining_minutes = int(sc.remaining_time() / 60)
remaining_seconds = int(sc.remaining_time() % 60)
time_display = f"{remaining_minutes}分{remaining_seconds}秒" if remaining_minutes > 0 else f"{remaining_seconds}秒"
time_display = (
f"{remaining_minutes}分{remaining_seconds}秒" if remaining_minutes > 0 else f"{remaining_seconds}秒"
)
line = f"{i}. 【{sc.price}元】{sc.user_nickname}: {sc.message_text}"
if len(line) > 100: # 限制单行长度
@@ -238,7 +238,7 @@ class SuperChatManager:
lines.append(f"... 还有{len(superchats) - max_count}条SuperChat")
return "\n".join(lines)
def build_superchat_summary_string(self, chat_id: str) -> str:
"""构建SuperChat摘要字符串"""
superchats = self.get_superchats_by_chat(chat_id)
@@ -261,30 +261,24 @@ class SuperChatManager:
if lines:
final_str += "\n" + "\n".join(lines)
return final_str
def get_superchat_statistics(self, chat_id: str) -> dict:
"""获取SuperChat统计信息"""
superchats = self.get_superchats_by_chat(chat_id)
if not superchats:
return {
"count": 0,
"total_amount": 0,
"average_amount": 0,
"highest_amount": 0,
"lowest_amount": 0
}
return {"count": 0, "total_amount": 0, "average_amount": 0, "highest_amount": 0, "lowest_amount": 0}
amounts = [sc.price for sc in superchats]
return {
"count": len(superchats),
"total_amount": sum(amounts),
"average_amount": sum(amounts) / len(amounts),
"highest_amount": max(amounts),
"lowest_amount": min(amounts)
"lowest_amount": min(amounts),
}
async def shutdown(self): # sourcery skip: use-contextlib-suppress
"""关闭管理器，清理资源"""
if self._cleanup_task and not self._cleanup_task.done():
@@ -296,15 +290,14 @@ class SuperChatManager:
logger.info("SuperChat管理器已关闭")
# sourcery skip: assign-if-exp
if s4u_config.enable_s4u:
super_chat_manager = SuperChatManager()
else:
super_chat_manager = None
def get_super_chat_manager() -> SuperChatManager:
"""获取全局SuperChat管理器实例"""
return super_chat_manager
return super_chat_manager

20
src/mais4u/s4u_config.py
View File

@@ -10,10 +10,12 @@ from src.common.logger import get_logger
logger = get_logger("s4u_config")
# 新增：兼容dict和tomlkit Table
def is_dict_like(obj):
return isinstance(obj, (dict, Table))
# 新增：递归将Table转为dict
def table_to_dict(obj):
if isinstance(obj, Table):
@@ -25,6 +27,7 @@ def table_to_dict(obj):
else:
return obj
# 获取mais4u模块目录
MAIS4U_ROOT = os.path.dirname(__file__)
CONFIG_DIR = os.path.join(MAIS4U_ROOT, "config")
@@ -190,7 +193,7 @@ class S4UModelConfig(S4UConfigBase):
@dataclass
class S4UConfig(S4UConfigBase):
"""S4U聊天系统配置类"""
enable_s4u: bool = False
"""是否启用S4U聊天系统"""
@@ -229,12 +232,12 @@ class S4UConfig(S4UConfigBase):
enable_streaming_output: bool = True
"""是否启用流式输出，false时全部生成后一次性发送"""
max_context_message_length: int = 20
"""上下文消息最大长度"""
max_core_message_length: int = 30
"""核心消息最大长度"""
"""核心消息最大长度"""
# 模型配置
models: S4UModelConfig = field(default_factory=S4UModelConfig)
@@ -243,7 +246,6 @@ class S4UConfig(S4UConfigBase):
# 兼容性字段，保持向后兼容
@dataclass
class S4UGlobalConfig(S4UConfigBase):
"""S4U总配置类"""
@@ -256,7 +258,7 @@ def update_s4u_config():
"""更新S4U配置文件"""
# 创建配置目录（如果不存在）
os.makedirs(CONFIG_DIR, exist_ok=True)
# 检查模板文件是否存在
if not os.path.exists(TEMPLATE_PATH):
logger.error(f"S4U配置模板文件不存在: {TEMPLATE_PATH}")
@@ -354,13 +356,13 @@ def load_s4u_config(config_path: str) -> S4UGlobalConfig:
logger.critical("S4U配置文件解析失败")
raise e
# 初始化S4U配置
logger.info(f"S4U当前版本: {S4U_VERSION}")
update_s4u_config()
s4u_config_main = load_s4u_config(config_path=CONFIG_PATH)
logger.info("S4U配置文件加载完成！")
s4u_config: S4UConfig = s4u_config_main.s4u
s4u_config: S4UConfig = s4u_config_main.s4u