mirror of
https://github.com/langbot-app/LangBot.git
synced 2026-10-30 20:35:22 +08:00
114 lines
4.3 KiB
Python
114 lines
4.3 KiB
Python
# Auto generated by LangBot Plugin SDK.
|
|
# Please refer to https://langbot.app/docs/en/plugin/dev/tutor.html for more details.
|
|
from __future__ import annotations
|
|
|
|
from typing import Any
|
|
import time
|
|
import datetime
|
|
|
|
from langbot_plugin.api.definition.components.tool.tool import Tool
|
|
from langbot_plugin.api.entities.builtin.provider import session as provider_session
|
|
|
|
|
|
class ScheduleNotify(Tool):
|
|
"""
|
|
Schedule a notification at a specific time.
|
|
This function should be called after getting the current time using get_current_time_str.
|
|
The program checks for notifications every minute, so precision is to the minute level.
|
|
"""
|
|
|
|
def _get_message(self, key: str, **kwargs) -> str:
|
|
"""
|
|
Get localized message based on plugin language setting.
|
|
|
|
Args:
|
|
key: Message key
|
|
**kwargs: Format arguments
|
|
|
|
Returns:
|
|
Formatted message string
|
|
"""
|
|
config = self.plugin.get_config()
|
|
language = config.get("language", "zh_Hans")
|
|
|
|
messages = {
|
|
"zh_Hans": {
|
|
"time_passed": "计划时间已经过去",
|
|
"success": "计划成功:\n将在{time}提醒您：{message}",
|
|
"no_session": "错误：无法获取会话信息，请先发送一条消息",
|
|
"too_many": "错误：本会话的计划提醒数量已达上限，请先删除一些提醒",
|
|
},
|
|
"en_US": {
|
|
"time_passed": "The scheduled time has already passed",
|
|
"success": "Scheduled successfully:\nWill remind you at {time}: {message}",
|
|
"no_session": "Error: Cannot get session info, please send a message first",
|
|
"too_many": "Error: This installation has reached its scheduled reminder limit; delete some reminders first",
|
|
}
|
|
}
|
|
|
|
template = messages.get(language, messages["zh_Hans"]).get(key, "")
|
|
return template.format(**kwargs) if kwargs else template
|
|
|
|
async def call(self, params: dict[str, Any], session: provider_session.Session, query_id: int) -> str:
|
|
"""
|
|
Schedule a notification.
|
|
|
|
Args:
|
|
params: {
|
|
"time_str": Time string in format "%Y-%m-%d %H:%M:%S",
|
|
"message": Notification message content
|
|
}
|
|
session: Current session information
|
|
query_id: Query ID
|
|
|
|
Returns:
|
|
str: Success message or error message
|
|
"""
|
|
time_str = params.get("time_str")
|
|
message = params.get("message")
|
|
|
|
if not time_str or not message:
|
|
return "Error: time_str and message are required"
|
|
|
|
try:
|
|
# Convert time string to timestamp
|
|
time_stamp = time.mktime(time.strptime(time_str, "%Y-%m-%d %H:%M:%S"))
|
|
# Get current timestamp
|
|
current_time_stamp = time.time()
|
|
|
|
# If scheduled time has already passed, return error
|
|
if time_stamp <= current_time_stamp:
|
|
return self._get_message("time_passed")
|
|
|
|
real_date = datetime.datetime.fromtimestamp(time_stamp)
|
|
|
|
# Get session info. The bot uuid is part of the session identity, so
|
|
# it is taken from the trusted session (falling back to its active
|
|
# conversation) and stored with the target type and id.
|
|
bot_uuid = session.bot_uuid or (
|
|
session.using_conversation.bot_uuid if session.using_conversation else None
|
|
)
|
|
target_type = session.launcher_type.value
|
|
target_id = str(session.launcher_id)
|
|
|
|
if not bot_uuid:
|
|
return self._get_message("no_session")
|
|
|
|
# Add to plugin's scheduled events with session info
|
|
queued = await self.plugin.add_scheduled_event(
|
|
time=real_date,
|
|
message=message,
|
|
bot_uuid=bot_uuid,
|
|
target_type=target_type,
|
|
target_id=target_id
|
|
)
|
|
if not queued:
|
|
return self._get_message("too_many")
|
|
|
|
return self._get_message("success", time=real_date, message=message)
|
|
|
|
except ValueError as e:
|
|
return f"Error: Invalid time format. Expected '%Y-%m-%d %H:%M:%S', got '{time_str}'"
|
|
except Exception as e:
|
|
return f"Error: {str(e)}"
|