1243 lines
50 KiB
Python
1243 lines
50 KiB
Python
# matrix_service.py
|
||
|
||
from nio import (
|
||
AsyncClient, LoginResponse, RoomMessageText, InviteMemberEvent,
|
||
ClientConfig, MegolmEvent, exceptions, KeyVerificationEvent,
|
||
KeyVerificationStart, KeyVerificationCancel, KeyVerificationKey,
|
||
KeyVerificationMac, ToDeviceError
|
||
)
|
||
import config
|
||
import asyncio
|
||
import schedule
|
||
import requests
|
||
from datetime import datetime, timedelta
|
||
import os
|
||
import json
|
||
from typing import Optional, Tuple
|
||
import aiofiles
|
||
from openai import OpenAI
|
||
|
||
STORE_FOLDER = "store/"
|
||
SESSION_DETAILS_FILE = "credentials.json"
|
||
MODEL_SELECT_RETRIES = 3 # 模型选择最大重试次数
|
||
|
||
# 初始化XAI客户端
|
||
xai_client = OpenAI(
|
||
api_key=config.XAI_API_KEY,
|
||
base_url=config.XAI_BASE_URL,
|
||
)
|
||
|
||
|
||
class MatrixService:
|
||
_instance = None
|
||
|
||
def __new__(cls):
|
||
if cls._instance is None:
|
||
cls._instance = super(MatrixService, cls).__new__(cls)
|
||
|
||
# 確保存儲目錄存在
|
||
if not os.path.exists(STORE_FOLDER):
|
||
os.makedirs(STORE_FOLDER)
|
||
|
||
# 創建客戶端配置
|
||
client_config = ClientConfig(
|
||
store_sync_tokens=True,
|
||
encryption_enabled=True
|
||
)
|
||
|
||
# 初始化客戶端
|
||
cls._instance.client = AsyncClient(
|
||
config.MATRIX_SERVER,
|
||
config.USERNAME,
|
||
device_id=config.DEVICE_ID,
|
||
store_path=STORE_FOLDER,
|
||
config=client_config,
|
||
)
|
||
|
||
cls._instance.is_logged_in = False
|
||
cls._instance.start_time = datetime.now()
|
||
|
||
# 用戶AI對話狀態和會話歷史
|
||
# (room_id, user_id) -> (last_activity_time, messages, model_key)
|
||
cls._instance.ai_chat_users = {}
|
||
|
||
# 用户等待AI响应状态
|
||
# (room_id, user_id) -> (waiting_since, last_question)
|
||
cls._instance.ai_waiting_users = {}
|
||
|
||
# 用户模型使用记录
|
||
# user_id -> {model_key -> (last_reset_time, count)}
|
||
cls._instance.model_usage = {}
|
||
|
||
# 模型选择状态
|
||
# (room_id, user_id) -> (retry_count)
|
||
cls._instance.model_select_state = {}
|
||
|
||
# 已發送歡迎消息的房間記錄
|
||
cls._instance.welcomed_rooms = set()
|
||
|
||
# 记录所有私聊房间信息
|
||
# room_id -> user_id
|
||
cls._instance.private_chat_rooms = {}
|
||
|
||
# 添加事件回調
|
||
cls._instance.client.add_event_callback(
|
||
cls._instance.message_callback,
|
||
RoomMessageText
|
||
)
|
||
cls._instance.client.add_event_callback(
|
||
cls._instance.encrypted_message_callback,
|
||
MegolmEvent
|
||
)
|
||
cls._instance.client.add_event_callback(
|
||
cls._instance.invite_callback,
|
||
InviteMemberEvent
|
||
)
|
||
# 添加设备验证回调
|
||
cls._instance.client.add_to_device_callback(
|
||
cls._instance.to_device_callback,
|
||
(KeyVerificationEvent,)
|
||
)
|
||
|
||
# 啟動AI超時檢查任務
|
||
asyncio.create_task(cls._instance.check_ai_chat_timeout())
|
||
# 启动AI响应超时检查任务
|
||
asyncio.create_task(cls._instance.check_ai_response_timeout())
|
||
|
||
return cls._instance
|
||
|
||
async def to_device_callback(self, event):
|
||
"""处理设备验证事件，自动接受所有验证请求"""
|
||
try:
|
||
if isinstance(event, KeyVerificationStart):
|
||
print(f"收到验证请求，来自 {event.sender}")
|
||
# 自动接受所有验证请求，不限制验证方式
|
||
resp = await self.client.accept_key_verification(event.transaction_id)
|
||
if isinstance(resp, ToDeviceError):
|
||
print(f"接受验证请求失败: {resp}")
|
||
return
|
||
|
||
sas = self.client.key_verifications[event.transaction_id]
|
||
todevice_msg = sas.share_key()
|
||
resp = await self.client.to_device(todevice_msg)
|
||
if isinstance(resp, ToDeviceError):
|
||
print(f"发送设备密钥失败: {resp}")
|
||
|
||
elif isinstance(event, KeyVerificationCancel):
|
||
print(f"验证被 {event.sender} 取消，原因: {event.reason}")
|
||
|
||
elif isinstance(event, KeyVerificationKey):
|
||
print(f"收到验证密钥，自动接受验证")
|
||
resp = await self.client.confirm_short_auth_string(event.transaction_id)
|
||
if isinstance(resp, ToDeviceError):
|
||
print(f"确认验证失败: {resp}")
|
||
|
||
elif isinstance(event, KeyVerificationMac):
|
||
sas = self.client.key_verifications[event.transaction_id]
|
||
try:
|
||
todevice_msg = sas.get_mac()
|
||
resp = await self.client.to_device(todevice_msg)
|
||
if isinstance(resp, ToDeviceError):
|
||
print(f"发送MAC失败: {resp}")
|
||
else:
|
||
print(f"验证完成，设备已验证: {sas.verified_devices}")
|
||
# 重新信任所有已验证的设备
|
||
for device_id in sas.verified_devices:
|
||
if event.sender in self.client.device_store:
|
||
if device_id in self.client.device_store[event.sender]:
|
||
olm_device = (
|
||
self.client.device_store[event.sender][device_id]
|
||
)
|
||
self.client.verify_device(olm_device)
|
||
print(
|
||
f"已信任设备 {device_id} 来自用户 {event.sender}")
|
||
except Exception as e:
|
||
print(f"处理MAC时出错: {e}")
|
||
|
||
except Exception as e:
|
||
print(f"处理设备验证事件时出错: {e}")
|
||
|
||
def trust_devices(self, user_id: str) -> None:
|
||
"""信任用戶的所有設備，包括自己的设备"""
|
||
print(f"信任用戶 {user_id} 的所有設備")
|
||
try:
|
||
if user_id not in self.client.device_store:
|
||
print(f"找不到用户 {user_id} 的设备信息")
|
||
return
|
||
|
||
for device_id, olm_device in self.client.device_store[user_id].items():
|
||
# 不跳过任何设备，包括自己的设备
|
||
print(f"信任設備: {device_id}")
|
||
self.client.verify_device(olm_device)
|
||
print(f"已信任设备 {device_id} 来自用户 {user_id}")
|
||
except Exception as e:
|
||
print(f"信任设备时出错: {e}")
|
||
|
||
async def check_ai_response_timeout(self):
|
||
"""检查AI响应超时"""
|
||
while True:
|
||
try:
|
||
current_time = datetime.now()
|
||
timeout_users = []
|
||
|
||
for (room_id, user_id), (waiting_since, _) in (
|
||
self.ai_waiting_users.items()
|
||
):
|
||
timeout = timedelta(
|
||
minutes=config.AI_RESPONSE_TIMEOUT_MINUTES
|
||
)
|
||
if current_time - waiting_since > timeout:
|
||
timeout_users.append((room_id, user_id))
|
||
|
||
# 处理超时的用户
|
||
for room_id, user_id in timeout_users:
|
||
chat_key = (room_id, user_id)
|
||
if chat_key in self.ai_waiting_users:
|
||
del self.ai_waiting_users[chat_key]
|
||
if chat_key in self.ai_chat_users:
|
||
del self.ai_chat_users[chat_key]
|
||
await self.send_message(
|
||
"AI系统貌似并没有反应了，目前先退出这次AI对话，"
|
||
"可能有其他问题，并反馈给管理员知道并等待修复",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
|
||
await asyncio.sleep(60) # 每分钟检查一次
|
||
except Exception as e:
|
||
print(f"检查AI响应超时时出错: {e}")
|
||
await asyncio.sleep(60)
|
||
|
||
def get_model_list_message(self, user_id: str) -> str:
|
||
"""生成模型列表消息"""
|
||
message = "请问你需要使用下面模型？\n\n"
|
||
|
||
# 首先添加XAI模型作为第一个选项
|
||
# 获取XAI使用统计
|
||
xai_usage = self.get_model_usage(user_id, "xai")
|
||
if config.XAI_DAILY_LIMIT == -1:
|
||
xai_limit_text = "无限制"
|
||
else:
|
||
xai_limit_text = f"每天最多{config.XAI_DAILY_LIMIT}条"
|
||
if xai_usage > 0:
|
||
xai_limit_text += f"（已使用{xai_usage}条）"
|
||
message += f"(1){config.XAI_MODEL_DIYNAME} {xai_limit_text}\n"
|
||
|
||
# 然后添加其他OpenRouter模型
|
||
for i, (key, model) in enumerate(config.OPENROUTER_MODELS.items(), 2):
|
||
# 获取使用统计
|
||
usage = self.get_model_usage(user_id, key)
|
||
|
||
# 构建显示文本
|
||
if model["daily_limit"] == -1:
|
||
limit_text = "无限制"
|
||
else:
|
||
limit_text = f"每天最多{model['daily_limit']}条"
|
||
if usage > 0:
|
||
limit_text += f"（已使用{usage}条）"
|
||
|
||
message += f"({i}){model['name']} {limit_text}\n"
|
||
|
||
message += (
|
||
"\n如果你想使用第一个默认模型，请回复1或者'好的'就开始AI对话，"
|
||
"如果使用其他模型请输入对应的数字就可以啦"
|
||
)
|
||
return message
|
||
|
||
def get_model_key_by_index(self, index: int) -> Optional[str]:
|
||
"""根据索引获取模型key"""
|
||
if index == 1: # XAI模型
|
||
return "xai"
|
||
try:
|
||
return list(config.OPENROUTER_MODELS.keys())[index - 2]
|
||
except IndexError:
|
||
return None
|
||
|
||
async def chat_with_xai(self, message: str, history: list) -> tuple[str, Optional[str]]:
|
||
"""与XAI API交互获取回复"""
|
||
try:
|
||
if not config.XAI_API_KEY:
|
||
return "抱歉，XAI服务暂时不可用，请选择其他模型。", None
|
||
|
||
completion = xai_client.chat.completions.create(
|
||
model="grok-beta",
|
||
messages=history + [{"role": "user", "content": message}],
|
||
)
|
||
content = completion.choices[0].message.content
|
||
|
||
# 检查是否有富文本格式
|
||
formatted_content = None
|
||
if "```" in content:
|
||
formatted_content = content.replace(
|
||
"```", "<pre><code>", 1
|
||
).replace("```", "</code></pre>", 1)
|
||
|
||
return content, formatted_content
|
||
except Exception as e:
|
||
print(f"XAI API调用出错: {e}")
|
||
return "抱歉，XAI服务暂时无法使用，请稍后再试。", None
|
||
|
||
def get_model_usage(self, user_id: str, model_key: str) -> int:
|
||
"""获取用户对特定模型的当天使用次数"""
|
||
if user_id not in self.model_usage:
|
||
self.model_usage[user_id] = {}
|
||
|
||
if model_key not in self.model_usage[user_id]:
|
||
self.model_usage[user_id][model_key] = (datetime.now(), 0)
|
||
|
||
last_reset, count = self.model_usage[user_id][model_key]
|
||
|
||
# 检查是否需要重置计数
|
||
if datetime.now().date() > last_reset.date():
|
||
self.model_usage[user_id][model_key] = (datetime.now(), 0)
|
||
return 0
|
||
|
||
return count
|
||
|
||
def increment_model_usage(self, user_id: str, model_key: str) -> int:
|
||
"""增加用户对特定模型的使用次数，返回新的使用次数"""
|
||
current_usage = self.get_model_usage(user_id, model_key)
|
||
self.model_usage[user_id][model_key] = (
|
||
self.model_usage[user_id][model_key][0],
|
||
current_usage + 1
|
||
)
|
||
return current_usage + 1
|
||
|
||
def check_model_limit(
|
||
self,
|
||
user_id: str,
|
||
model_key: str
|
||
) -> Tuple[bool, Optional[str]]:
|
||
"""检查模型使用限制，返回(是否可用, 提示消息)"""
|
||
if model_key == "xai": # XAI模型使用XAI_DAILY_LIMIT
|
||
daily_limit = config.XAI_DAILY_LIMIT
|
||
else:
|
||
model_info = config.OPENROUTER_MODELS[model_key]
|
||
daily_limit = model_info["daily_limit"]
|
||
|
||
if daily_limit == -1:
|
||
return True, None
|
||
|
||
current_usage = self.get_model_usage(user_id, model_key)
|
||
|
||
if current_usage >= daily_limit:
|
||
model_name = config.XAI_MODEL_DIYNAME if model_key == "xai" else config.OPENROUTER_MODELS[
|
||
model_key]["name"]
|
||
return False, (
|
||
f"感谢你使用{model_name}，已经到达当天最大条数，"
|
||
f"欢迎你明天之后继续使用该模型！你也可以使用其他的模型！"
|
||
)
|
||
|
||
# 检查是否需要发送提醒
|
||
remaining = daily_limit - current_usage
|
||
if remaining in [6, 2, 1]:
|
||
return True, f"（温馨提示：今天还剩余{remaining-1}条对话次数）"
|
||
|
||
return True, None
|
||
|
||
def process_model_selection(
|
||
self,
|
||
user_id: str,
|
||
room_id: str,
|
||
message: str
|
||
) -> Optional[str]:
|
||
"""处理模型选择，返回选择的模型key或None"""
|
||
chat_key = (room_id, user_id)
|
||
|
||
# 检查是否是默认选择
|
||
if message.lower() in ['1', '好的']:
|
||
return "xai" # 返回XAI作为默认模型
|
||
|
||
# 尝试解析数字选择
|
||
try:
|
||
index = int(message)
|
||
model_key = self.get_model_key_by_index(index)
|
||
if model_key:
|
||
return model_key
|
||
except ValueError:
|
||
pass
|
||
|
||
# 更新重试次数
|
||
if chat_key not in self.model_select_state:
|
||
self.model_select_state[chat_key] = 1
|
||
else:
|
||
self.model_select_state[chat_key] += 1
|
||
|
||
return None
|
||
|
||
async def reestablish_room_session(self, room_id: str, user_id: str):
|
||
"""重新建立房间的加密会话"""
|
||
try:
|
||
print(f"正在重新建立与用户 {user_id} 在房间 {room_id} 的加密会话...")
|
||
|
||
# 清除现有的房间密钥
|
||
if room_id in self.client.rooms:
|
||
self.client.rooms[room_id].encrypted = True
|
||
|
||
# 上传新的设备密钥
|
||
if self.client.should_upload_keys:
|
||
await self.client.keys_upload()
|
||
|
||
# 信任所有设备，包括自己的设备
|
||
self.trust_devices(self.client.user_id)
|
||
self.trust_devices(user_id)
|
||
|
||
# 强制同步以获取最新状态
|
||
await self.client.sync()
|
||
|
||
# 发送测试消息以触发新的密钥交换
|
||
test_content = {
|
||
"msgtype": "m.text",
|
||
"body": "正在重新建立加密会话..."
|
||
}
|
||
|
||
await self.client.room_send(
|
||
room_id=room_id,
|
||
message_type="m.room.message",
|
||
content=test_content
|
||
)
|
||
|
||
print(f"已重新建立与用户 {user_id} 的加密会话")
|
||
return True
|
||
|
||
except Exception as e:
|
||
print(f"重新建立加密会话失败: {e}")
|
||
return False
|
||
|
||
def trust_devices(self, user_id: str) -> None:
|
||
"""信任用戶的所有設備"""
|
||
print(f"信任用戶 {user_id} 的所有設備")
|
||
try:
|
||
for device_id, olm_device in self.client.device_store[user_id].items():
|
||
if user_id == self.client.user_id and device_id == self.client.device_id:
|
||
continue
|
||
print(f"信任設備: {device_id}")
|
||
self.client.verify_device(olm_device)
|
||
|
||
print(f"已经信任设备:{device_id}")
|
||
except Exception as e:
|
||
print(f"信任用户设备时出错:{e}")
|
||
|
||
async def trust_room_devices(self, room_id: str):
|
||
"""信任房間中所有用戶的所有設備"""
|
||
try:
|
||
room = self.client.rooms[room_id]
|
||
for user_id in room.users:
|
||
self.trust_devices(user_id)
|
||
except Exception as e:
|
||
print(f"信任房間設備時出錯: {e}")
|
||
|
||
async def login(self):
|
||
"""登錄並處理默認房間"""
|
||
if not self.is_logged_in:
|
||
try:
|
||
# 嘗試恢復會話
|
||
if os.path.exists(SESSION_DETAILS_FILE):
|
||
async with aiofiles.open(SESSION_DETAILS_FILE) as f:
|
||
print("密钥文件存在，开始读取")
|
||
contents = await f.read()
|
||
session_data = json.loads(contents)
|
||
self.client.access_token = session_data["access_token"]
|
||
self.client.user_id = session_data["user_id"]
|
||
self.client.device_id = session_data["device_id"]
|
||
|
||
print("准备加载STORE")
|
||
self.client.load_store()
|
||
print(f"使用已保存的憑證登錄: {self.client.user_id}")
|
||
self.is_logged_in = True
|
||
|
||
# 上传设备密钥
|
||
if self.client.should_upload_keys:
|
||
await self.client.keys_upload()
|
||
|
||
# 信任自己的设备
|
||
self.trust_devices(self.client.user_id)
|
||
|
||
print("正在同步房間狀態...")
|
||
await self.client.sync()
|
||
return True
|
||
|
||
# 如果沒有保存的會話，使用密碼登錄
|
||
print("准备开始使用账号密码登陆")
|
||
response = await self.client.login(
|
||
password=config.PASSWORD,
|
||
device_name=config.DEVICE_NAME
|
||
)
|
||
print("Login response:", response)
|
||
|
||
if isinstance(response, LoginResponse):
|
||
self.is_logged_in = True
|
||
print("登錄成功!")
|
||
|
||
# 保存會話信息
|
||
async with aiofiles.open(SESSION_DETAILS_FILE, "w") as f:
|
||
await f.write(json.dumps({
|
||
"access_token": response.access_token,
|
||
"device_id": response.device_id,
|
||
"user_id": response.user_id,
|
||
}))
|
||
|
||
# 上传设备密钥
|
||
if self.client.should_upload_keys:
|
||
await self.client.keys_upload()
|
||
|
||
# 信任自己的设备
|
||
self.trust_devices(self.client.user_id)
|
||
|
||
print("正在同步房間狀態...")
|
||
await self.client.sync()
|
||
|
||
if hasattr(config, 'ROOM_IDS') and config.ROOM_IDS:
|
||
for room_id in config.ROOM_IDS:
|
||
if room_id in self.client.rooms:
|
||
print(f"已在房間 {room_id} 中")
|
||
else:
|
||
print(f"嘗試加入房間 {room_id}")
|
||
await self.ensure_joined(room_id)
|
||
else:
|
||
print("未配置默認房間，等待用戶邀請...")
|
||
|
||
return True
|
||
else:
|
||
print("登錄失敗!")
|
||
return False
|
||
|
||
except Exception as e:
|
||
print(f"登錄過程出錯: {e}")
|
||
return False
|
||
return True
|
||
|
||
async def check_ai_chat_timeout(self):
|
||
"""定期檢查AI對話超時"""
|
||
while True:
|
||
try:
|
||
current_time = datetime.now()
|
||
timeout_sessions = []
|
||
|
||
for (room_id, user_id), (last_activity, _, _) in (
|
||
self.ai_chat_users.items()
|
||
):
|
||
if isinstance(last_activity, datetime):
|
||
timeout = timedelta(
|
||
minutes=config.AI_CHAT_TIMEOUT_MINUTES
|
||
)
|
||
if current_time - last_activity > timeout:
|
||
timeout_sessions.append((room_id, user_id))
|
||
|
||
# 移除超時會話
|
||
for room_id, user_id in timeout_sessions:
|
||
chat_key = (room_id, user_id)
|
||
if chat_key in self.ai_chat_users:
|
||
del self.ai_chat_users[chat_key]
|
||
if chat_key in self.ai_waiting_users:
|
||
del self.ai_waiting_users[chat_key]
|
||
print(f"用戶 {user_id} 在房間 {room_id} 的AI對話已超時")
|
||
|
||
await asyncio.sleep(60) # 每分鐘檢查一次
|
||
except Exception as e:
|
||
print(f"檢查AI對話超時時出錯: {e}")
|
||
await asyncio.sleep(60)
|
||
|
||
def get_welcome_message(self) -> str:
|
||
"""獲取歡迎消息"""
|
||
return (
|
||
"你好！我是一個自動回覆機器人。\n"
|
||
"你可以使用以下命令：\n"
|
||
"- !ping: 測試我是否在線\n"
|
||
"- 加密货币: 獲取當前加密貨幣價格\n"
|
||
"- 人工智能: 開始AI對話"
|
||
)
|
||
|
||
async def invite_callback(self, room, event):
|
||
"""當 BOT 被邀請到房間時自動加入並處理驗證"""
|
||
try:
|
||
print(f"收到來自 {event.sender} 的邀請")
|
||
|
||
# 自動加入房間
|
||
join_response = await self.client.join(room.room_id)
|
||
if join_response:
|
||
print(f"已加入與 {event.sender} 的房間")
|
||
|
||
# 同步以獲取房間狀態
|
||
await self.client.sync()
|
||
|
||
# 信任房間中的設備
|
||
await self.trust_room_devices(room.room_id)
|
||
|
||
# 检查是否为私聊房间（只有两个用户）
|
||
room_obj = self.client.rooms[room.room_id]
|
||
if len(room_obj.users) == 2:
|
||
# 记录私聊房间信息
|
||
self.private_chat_rooms[room.room_id] = event.sender
|
||
print(f"记录新的私聊房间: {room.room_id} - 用户: {event.sender}")
|
||
|
||
# 發送歡迎消息
|
||
await self.send_message(
|
||
self.get_welcome_message(),
|
||
room_id=room.room_id,
|
||
encrypted=True
|
||
)
|
||
|
||
# 記錄已發送歡迎消息
|
||
self.welcomed_rooms.add(room.room_id)
|
||
|
||
except Exception as e:
|
||
print(f"處理邀請時出錯: {e}")
|
||
|
||
def is_room_encrypted(self, room_id: str) -> bool:
|
||
"""檢查房間是否加密"""
|
||
try:
|
||
if room_id not in self.client.rooms:
|
||
return False
|
||
room = self.client.rooms[room_id]
|
||
return getattr(room, "encrypted", False)
|
||
except Exception as e:
|
||
print(f"檢查房間加密狀態時出錯: {e}")
|
||
return False
|
||
|
||
async def ensure_joined(self, room_id: str):
|
||
"""確保加入指定房間"""
|
||
try:
|
||
if room_id not in self.client.rooms:
|
||
response = await self.client.join(room_id)
|
||
if isinstance(response, LoginResponse):
|
||
print(f"加入房間失敗: {response.message}")
|
||
return
|
||
print(f"成功加入房間: {room_id}")
|
||
await self.client.sync()
|
||
# 信任房間中的設備
|
||
await self.trust_room_devices(room_id)
|
||
except Exception as e:
|
||
print(f"確保加入房間時出錯: {e}")
|
||
|
||
async def get_crypto_prices(self):
|
||
"""使用 CoinGecko API 獲取加密貨幣價格和漲跌幅信息"""
|
||
try:
|
||
# CoinGecko API 基礎URL
|
||
base_url = "https://api.coingecko.com/api/v3"
|
||
|
||
# 獲取BTC、ETH和ADA的數據
|
||
coins = {
|
||
"bitcoin": "BTCUSDT",
|
||
"ethereum": "ETHUSDT" # ,
|
||
# "cardano": "ADAUSDT"
|
||
}
|
||
|
||
# 獲取當前價格和24小時漲跌幅
|
||
current_prices_url = (
|
||
f"{base_url}/simple/price"
|
||
f"?ids={','.join(coins.keys())}"
|
||
f"&vs_currencies=usd"
|
||
f"&include_24hr_change=true"
|
||
)
|
||
|
||
response = requests.get(
|
||
current_prices_url,
|
||
headers={"accept": "application/json"}
|
||
)
|
||
response.raise_for_status()
|
||
current_data = response.json()
|
||
|
||
# 獲取7天前的日期
|
||
seven_days_ago = (
|
||
datetime.now() - timedelta(days=7)
|
||
).strftime('%d-%m-%Y')
|
||
|
||
# 獲取30天前的日期
|
||
thirty_days_ago = (
|
||
datetime.now() - timedelta(days=30)
|
||
).strftime('%d-%m-%Y')
|
||
|
||
prices = {}
|
||
for coin_id, symbol in coins.items():
|
||
try:
|
||
# 獲取當前價格和24小時漲跌幅
|
||
current_price = current_data[coin_id]["usd"]
|
||
price_change_24h = current_data[coin_id]["usd_24h_change"]
|
||
|
||
# 獲取7天前的價格
|
||
history_url_7d = (
|
||
f"{base_url}/coins/{coin_id}/history?date={seven_days_ago}"
|
||
)
|
||
response = requests.get(
|
||
history_url_7d,
|
||
headers={"accept": "application/json"}
|
||
)
|
||
response.raise_for_status()
|
||
history_data_7d = response.json()
|
||
price_7d_ago = (
|
||
history_data_7d["market_data"]["current_price"]["usd"]
|
||
)
|
||
|
||
# 獲取30天前的價格
|
||
history_url_30d = (
|
||
f"{base_url}/coins/{coin_id}/history?date={thirty_days_ago}"
|
||
)
|
||
response = requests.get(
|
||
history_url_30d,
|
||
headers={"accept": "application/json"}
|
||
)
|
||
response.raise_for_status()
|
||
history_data_30d = response.json()
|
||
price_30d_ago = (
|
||
history_data_30d["market_data"]["current_price"]["usd"]
|
||
)
|
||
|
||
# 計算7天和30天漲跌幅
|
||
price_change_7d = (
|
||
(current_price - price_7d_ago) / price_7d_ago) * 100
|
||
price_change_30d = (
|
||
(current_price - price_30d_ago) / price_30d_ago) * 100
|
||
|
||
prices[symbol] = {
|
||
"current_price": current_price,
|
||
"price_change_percent_24h": price_change_24h,
|
||
"price_change_percent_7d": price_change_7d,
|
||
"price_change_percent_30d": price_change_30d
|
||
}
|
||
|
||
except Exception as e:
|
||
print(f"處理 {coin_id} 數據時出錯: {e}")
|
||
continue
|
||
|
||
return prices if prices else None
|
||
|
||
except requests.exceptions.HTTPError as e:
|
||
print(f"HTTP API請求錯誤: {e}")
|
||
if e.response.status_code == 429:
|
||
return "获取加密货币的接口请求过于频繁，请等一段时间之后再获取！"
|
||
else:
|
||
return "网络接口出现问题不能连通，请稍后再尝试或者等待修复之后再使用！"
|
||
|
||
except requests.exceptions.RequestException as e:
|
||
err_msg = (f"API請求錯誤: {e}")
|
||
return err_msg
|
||
except KeyError as e:
|
||
print(f"數據解析錯誤: {e}")
|
||
return None
|
||
except Exception as e:
|
||
print(f"獲取加密貨幣價格時出錯: {e}")
|
||
return None
|
||
|
||
async def send_message(
|
||
self,
|
||
message: str,
|
||
room_id: str = config.ROOM_ID,
|
||
formatted_message: Optional[str] = None,
|
||
encrypted: bool = False
|
||
):
|
||
"""發送消息（支持加密和未加密房間）"""
|
||
if not self.is_logged_in:
|
||
print("Error: Not logged in!")
|
||
return
|
||
|
||
content = {"msgtype": "m.text", "body": message}
|
||
if formatted_message:
|
||
content["format"] = "org.matrix.custom.html"
|
||
content["formatted_body"] = formatted_message
|
||
|
||
try:
|
||
# 如果需要加密或房間是加密的
|
||
if encrypted or self.is_room_encrypted(room_id):
|
||
# 获取房间中的所有用户
|
||
room = self.client.rooms[room_id]
|
||
for user_id in room.users:
|
||
if user_id != self.client.user_id:
|
||
# 确保信任所有设备
|
||
self.trust_devices(user_id)
|
||
|
||
await self.client.room_send(
|
||
room_id=room_id,
|
||
message_type="m.room.message",
|
||
content=content
|
||
)
|
||
print(f"Message sent successfully to room {room_id}")
|
||
|
||
except exceptions.OlmUnverifiedDeviceError as e:
|
||
print(f"房間有未驗證的設備: {e}")
|
||
try:
|
||
# 获取房间中的所有用户
|
||
room = self.client.rooms[room_id]
|
||
for user_id in room.users:
|
||
if user_id != self.client.user_id:
|
||
# 重新信任设备
|
||
self.trust_devices(user_id)
|
||
|
||
# 重试发送
|
||
await self.client.room_send(
|
||
room_id=room_id,
|
||
message_type="m.room.message",
|
||
content=content
|
||
)
|
||
print("重试发送消息成功")
|
||
except Exception as retry_e:
|
||
print(f"重试发送消息失败: {retry_e}")
|
||
except Exception as e:
|
||
print(f"Error sending message: {e}")
|
||
# else:
|
||
# print("Error: Not logged in!")
|
||
|
||
async def sync(self):
|
||
print("Start Sync...")
|
||
if self.is_logged_in:
|
||
while True:
|
||
print("Sync task running...")
|
||
await self.client.sync_forever(timeout=60000, full_state=True)
|
||
|
||
def schedule_daily_message(self, message):
|
||
schedule.every().day.at("09:00").do(
|
||
lambda: asyncio.run(self.send_message(message))
|
||
)
|
||
|
||
def schedule_hourly_updates(self):
|
||
schedule.every().hour.do(self.fetch_and_send_news)
|
||
|
||
async def send_crypto_prices(self, room_id=config.ROOM_ID):
|
||
"""發送加密貨幣價格信息"""
|
||
crypto_prices = await self.get_crypto_prices()
|
||
if crypto_prices and not isinstance(crypto_prices, str):
|
||
# 準備富文本消息
|
||
formatted_message = (
|
||
"<h4>📊 加密貨幣市場報告</h4>"
|
||
"<table style='width:100%;border-collapse:collapse;'>"
|
||
"<tr style='background-color:#f0f0f0;'>"
|
||
"<th style='padding:8px;text-align:left;'>幣種</th>"
|
||
"<th style='padding:8px;text-align:right;'>當前價格</th>"
|
||
"<th style='padding:8px;text-align:right;'>24h漲跌</th>"
|
||
"<th style='padding:8px;text-align:right;'>7d漲跌</th>"
|
||
"<th style='padding:8px;text-align:right;'>30d漲跌</th>"
|
||
"</tr>"
|
||
)
|
||
|
||
# 準備純文本消息
|
||
plain_message = "📊 加密貨幣市場報告\n\n"
|
||
|
||
# 幣種名稱映射
|
||
coin_names = {
|
||
"BTCUSDT": "比特幣 (BTC)",
|
||
"ETHUSDT": "以太坊 (ETH)" # ,
|
||
# "ADAUSDT": "卡爾達諾 (ADA)"
|
||
}
|
||
|
||
for symbol, info in crypto_prices.items():
|
||
# 獲取價格變化的顏色
|
||
color_24h = (
|
||
"green" if info["price_change_percent_24h"] >= 0
|
||
else "red"
|
||
)
|
||
color_7d = (
|
||
"green" if info["price_change_percent_7d"] >= 0
|
||
else "red"
|
||
)
|
||
color_30d = (
|
||
"green" if info["price_change_percent_30d"] >= 0
|
||
else "red"
|
||
)
|
||
|
||
# 添加到富文本消息
|
||
formatted_message += (
|
||
f"<tr>"
|
||
f"<td style='padding:8px;'>{coin_names[symbol]}</td>"
|
||
f"<td style='padding:8px;text-align:right;'>"
|
||
f"${info['current_price']:,.2f}</td>"
|
||
f"<td style='padding:8px;text-align:right;color:{color_24h};'>"
|
||
f"{info['price_change_percent_24h']:+.2f}%</td>"
|
||
f"<td style='padding:8px;text-align:right;color:{color_7d};'>"
|
||
f"{info['price_change_percent_7d']:+.2f}%</td>"
|
||
f"<td style='padding:8px;text-align:right;color:{color_30d};'>"
|
||
f"{info['price_change_percent_30d']:+.2f}%</td>"
|
||
f"</tr>"
|
||
)
|
||
|
||
# 添加到純文本消息
|
||
plain_message += (
|
||
f"{coin_names[symbol]}:\n"
|
||
f" 當前價格: ${info['current_price']:,.2f}\n"
|
||
f" 24h漲跌: {info['price_change_percent_24h']:+.2f}%\n"
|
||
f" 7d漲跌: {info['price_change_percent_7d']:+.2f}%\n"
|
||
f" 30d漲跌: {info['price_change_percent_30d']:+.2f}%\n\n"
|
||
)
|
||
|
||
formatted_message += "</table>"
|
||
|
||
# 發送消息
|
||
await self.send_message(
|
||
plain_message,
|
||
room_id=room_id,
|
||
formatted_message=formatted_message,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
if not crypto_prices:
|
||
error_msg = "抱歉，暫時無法獲取加密貨幣價格信息。"
|
||
elif crypto_prices and isinstance(crypto_prices, str):
|
||
error_msg = crypto_prices
|
||
await self.send_message(
|
||
error_msg,
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
|
||
async def chat_with_ai(
|
||
self,
|
||
message: str,
|
||
history: list,
|
||
model_key: str
|
||
) -> tuple[str, Optional[str]]:
|
||
"""統一的AI對話接口"""
|
||
if model_key == "xai":
|
||
return await self.chat_with_xai(message, history)
|
||
else:
|
||
# 原有的OpenRouter API交互逻辑
|
||
try:
|
||
headers = {
|
||
"Authorization": f"Bearer {config.OPENROUTER_API_KEY}",
|
||
"Content-Type": "application/json",
|
||
"HTTP-Referer": "https://github.com",
|
||
"X-Title": "Bot"
|
||
}
|
||
|
||
messages = history + [{"role": "user", "content": message}]
|
||
|
||
data = {
|
||
"model": config.OPENROUTER_MODELS[model_key]["model"],
|
||
"messages": messages
|
||
}
|
||
|
||
response = requests.post(
|
||
config.OPENROUTER_API_URL,
|
||
headers=headers,
|
||
json=data
|
||
)
|
||
response.raise_for_status()
|
||
|
||
result = response.json()
|
||
content = result["choices"][0]["message"]["content"]
|
||
|
||
# 檢查是否有富文本格式
|
||
formatted_content = None
|
||
if "```" in content:
|
||
# 將markdown代碼塊轉換為HTML格式
|
||
formatted_content = content.replace(
|
||
"```", "<pre><code>", 1
|
||
).replace("```", "</code></pre>", 1)
|
||
|
||
return content, formatted_content
|
||
|
||
except requests.exceptions.HTTPError as e:
|
||
if e.response.status_code == 402:
|
||
return "对不起，由于系统后台余额不足，请等待管理员充值之后再重新进行对话", None
|
||
print(f"AI API調用出錯: {e}")
|
||
return "抱歉，AI服務暫時無法使用，請稍後再試。", None
|
||
except Exception as e:
|
||
print(f"AI API調用出錯: {e}")
|
||
return "抱歉，AI服務暫時無法使用，請稍後再試。", None
|
||
|
||
async def encrypted_message_callback(self, room, event):
|
||
"""處理加密消息"""
|
||
print(f"收到加密消息，房間ID: {room.room_id}")
|
||
# print(f"输出event{event}")
|
||
try:
|
||
# 对于MegolmEvent，我们需要从解密后的内容中获取消息
|
||
if hasattr(event, "decrypted_data"):
|
||
# 从解密数据中获取消息内容
|
||
message_content = event.decrypted_data.get("content", {})
|
||
# 确保event对象有body属性供process_message使用
|
||
event.body = message_content.get("body", "")
|
||
await self.process_message(room, event)
|
||
else:
|
||
# 检查是否是因为缺少会话ID导致的解密失败
|
||
error_msg = str(event.decryption_error) if hasattr(
|
||
event, "decryption_error") else ""
|
||
if "no session found with session id" in error_msg:
|
||
print(f"解密失败，尝试重新建立会话: {error_msg}")
|
||
# 获取发送者ID
|
||
sender_id = event.sender
|
||
# 尝试重新建立会话
|
||
success = await self.reestablish_room_session(room.room_id, sender_id)
|
||
if success:
|
||
# 通知用户重新建立了会话
|
||
await self.send_message(
|
||
"已重新建立加密会话，请重新发送您的消息。",
|
||
room_id=room.room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
print("重新建立会话失败")
|
||
else:
|
||
print("无法处理未解密的消息")
|
||
except Exception as e:
|
||
print(f"处理加密消息时出错: {e}")
|
||
|
||
async def message_callback(self, room, event):
|
||
"""處理未加密消息"""
|
||
print(f"收到未加密消息，房間ID: {room.room_id}")
|
||
await self.process_message(room, event)
|
||
|
||
async def process_message(self, room, event):
|
||
"""統一處理消息的邏輯"""
|
||
# 忽略自己發送的消息
|
||
if event.sender == self.client.user_id:
|
||
return
|
||
|
||
print(
|
||
f"處理來自房間 {room.room_id} 的消息: {event.body}"
|
||
f", 事件ID: {event.event_id}"
|
||
)
|
||
|
||
# 將消息時間戳轉換為datetime
|
||
message_time = datetime.fromtimestamp(event.server_timestamp / 1000)
|
||
if message_time >= self.start_time: # 只響應程序啟動後的消息
|
||
user_id = event.sender
|
||
room_id = room.room_id
|
||
chat_key = (room_id, user_id)
|
||
|
||
# 檢查是否為群聊房間
|
||
is_group_chat = room_id in config.ROOM_IDS
|
||
|
||
# 如果是私聊房间且尚未记录，则记录下来
|
||
if not is_group_chat and room_id not in self.private_chat_rooms:
|
||
print(f"新的私聊房间：{room.users}")
|
||
if len(room.users) == 2: # 确认是私聊房间
|
||
self.private_chat_rooms[room_id] = user_id
|
||
print(f"记录新的私聊房间: {room_id} - 用户: {user_id}")
|
||
|
||
# 處理命令
|
||
if event.body == "!ping":
|
||
await self.send_message(
|
||
"Pong!",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
elif event.body == "加密货币":
|
||
await self.send_crypto_prices(room_id)
|
||
elif event.body == "人工智能":
|
||
if is_group_chat:
|
||
# 在群聊中提示私聊
|
||
await self.send_message(
|
||
"如果想要AI对话，請你點擊我頭像，向我進行私聊，"
|
||
"就重新进行指令就可以进行AI對話",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
# 显示模型列表
|
||
await self.send_message(
|
||
self.get_model_list_message(user_id),
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
# 初始化模型选择状态
|
||
self.model_select_state[chat_key] = 0
|
||
elif chat_key in self.model_select_state:
|
||
# 处理模型选择
|
||
model_key = self.process_model_selection(
|
||
user_id, room_id, event.body)
|
||
if model_key:
|
||
# 检查模型使用限制
|
||
can_use, limit_msg = self.check_model_limit(
|
||
user_id, model_key)
|
||
if can_use:
|
||
# 初始化AI对话
|
||
self.ai_chat_users[chat_key] = (
|
||
datetime.now(),
|
||
[], # 空的会话历史
|
||
model_key # 记录选择的模型
|
||
)
|
||
# 获取模型名称
|
||
model_name = (config.XAI_MODEL_DIYNAME if model_key == "xai"
|
||
else config.OPENROUTER_MODELS[model_key]["name"])
|
||
welcome_msg = (
|
||
f"歡迎你使用AI智哥哥功能1.02，請你回覆我你想要知道的內容"
|
||
f"[如果是更大的模型的回复会更详细，可能需要很多的时间来准备内容，"
|
||
f"所以发送后请耐心等候回复，不要重复发送]，"
|
||
f"閒置{config.AI_CHAT_TIMEOUT_MINUTES}分鐘之後功能會自動關閉，"
|
||
f"或者也可以手动输入'退出'就可以手动退出AI对话功能"
|
||
f"[目前选择的大模型是{model_name}]"
|
||
)
|
||
# 如果有使用限制提示，添加到欢迎消息
|
||
if limit_msg:
|
||
welcome_msg += f"\n{limit_msg}"
|
||
await self.send_message(
|
||
welcome_msg,
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
# 发送限制消息并清除选择状态
|
||
await self.send_message(
|
||
limit_msg,
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
# 清除模型选择状态
|
||
del self.model_select_state[chat_key]
|
||
else:
|
||
# 更新重试次数
|
||
self.model_select_state[chat_key] += 1
|
||
if self.model_select_state[chat_key] >= MODEL_SELECT_RETRIES:
|
||
# 超过重试次数，取消操作
|
||
del self.model_select_state[chat_key]
|
||
await self.send_message(
|
||
"已超过重试次数，如需使用AI对话请重新输入'人工智能'",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
# 重新显示模型列表
|
||
await self.send_message(
|
||
self.get_model_list_message(user_id),
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
elif chat_key in self.ai_chat_users:
|
||
if event.body == "退出":
|
||
# 清除所有相关状态
|
||
if chat_key in self.ai_chat_users:
|
||
del self.ai_chat_users[chat_key]
|
||
if chat_key in self.ai_waiting_users:
|
||
del self.ai_waiting_users[chat_key]
|
||
await self.send_message(
|
||
"已经退出了AI对话啦！欢迎你的使用！",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
# 检查用户是否正在等待AI响应
|
||
if chat_key in self.ai_waiting_users:
|
||
await self.send_message(
|
||
"AI已经接收到你的问题啦，不过还需要时间进行反应，请你耐心等待...",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
return
|
||
|
||
# 获取会话状态
|
||
last_activity, history, model_key = self.ai_chat_users[chat_key]
|
||
|
||
# 检查模型使用限制
|
||
can_use, limit_msg = self.check_model_limit(
|
||
user_id, model_key)
|
||
if not can_use:
|
||
# 达到限制，结束对话
|
||
del self.ai_chat_users[chat_key]
|
||
await self.send_message(
|
||
limit_msg,
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
return
|
||
|
||
# 记录用户正在等待AI响应
|
||
self.ai_waiting_users[chat_key] = (
|
||
datetime.now(), event.body)
|
||
|
||
# 发送等待提示
|
||
await self.send_message(
|
||
"已经收到你的请求,请耐心等待AI智哥哥的思考...",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
|
||
# 更新用户消息到历史
|
||
history.append({"role": "user", "content": event.body})
|
||
|
||
try:
|
||
# 获取AI回复
|
||
response, formatted_response = await self.chat_with_ai(
|
||
event.body,
|
||
history,
|
||
model_key
|
||
)
|
||
|
||
# 更新AI回复到历史
|
||
history.append(
|
||
{"role": "assistant", "content": response})
|
||
|
||
# 增加使用次数
|
||
self.increment_model_usage(user_id, model_key)
|
||
|
||
# 检查是否需要添加剩余次数提示
|
||
if limit_msg:
|
||
response += f"\n{limit_msg}"
|
||
|
||
# 更新会话状态
|
||
self.ai_chat_users[chat_key] = (
|
||
datetime.now(),
|
||
history,
|
||
model_key
|
||
)
|
||
|
||
# 发送回复
|
||
await self.send_message(
|
||
response,
|
||
room_id=room_id,
|
||
formatted_message=formatted_response,
|
||
encrypted=True
|
||
)
|
||
except Exception as e:
|
||
print(f"处理AI回复时出错: {e}")
|
||
await self.send_message(
|
||
"抱歉，处理你的请求时出现了错误，请稍后重试。",
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
finally:
|
||
# 无论成功还是失败，都清除等待状态
|
||
if chat_key in self.ai_waiting_users:
|
||
del self.ai_waiting_users[chat_key]
|
||
|
||
else:
|
||
# 如果不是AI對話且不是命令，發送歡迎消息
|
||
await self.send_message(
|
||
self.get_welcome_message(),
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
else:
|
||
print(
|
||
f"忽略消息，發送時間 {message_time}，"
|
||
f"早於程序啟動時間 {self.start_time}"
|
||
)
|
||
|
||
async def run_schedule(self):
|
||
while True:
|
||
schedule.run_pending()
|
||
await asyncio.sleep(1)
|
||
|
||
async def send_shutdown_notice(self):
|
||
"""向所有私聊房间发送关闭通知"""
|
||
shutdown_message = (
|
||
"服务器准备进行升级，当服务器升级完毕之后，这个房间可能会无法正常解密消息。"
|
||
"如果发送内容没有任何反应,请长按头像'离开'这个私聊房间，重新从公共房间中跟AI智哥哥"
|
||
"再发送私聊信息，这样就可以重新建立连接啦，就可以使用啦！"
|
||
)
|
||
|
||
for room_id, user_id in self.private_chat_rooms.items():
|
||
try:
|
||
print(f"正在向房间 {room_id} 发送关闭通知...")
|
||
await self.send_message(
|
||
shutdown_message,
|
||
room_id=room_id,
|
||
encrypted=True
|
||
)
|
||
# 等待一小段时间确保消息发送成功
|
||
await asyncio.sleep(0.5)
|
||
except Exception as e:
|
||
print(f"向房间 {room_id} 发送关闭通知失败: {e}")
|
||
|
||
async def close(self):
|
||
"""关闭服务前发送通知并清理资源"""
|
||
try:
|
||
if self.private_chat_rooms:
|
||
# 暂时不发送关闭时候的通知
|
||
print("程序正在关闭...")
|
||
# print("正在发送关闭通知...")
|
||
# await self.send_shutdown_notice()
|
||
# 等待一段时间确保所有通知都发送完成
|
||
# await asyncio.sleep(2)
|
||
except Exception as e:
|
||
print(f"发送关闭通知时出错: {e}")
|
||
finally:
|
||
await self.client.close()
|