mirror of
https://github.com/AstrBotDevs/AstrBot
synced 2026-07-16 01:40:15 +08:00
clean it
This commit is contained in:
whatevertogo
@@ -1,10 +1,29 @@
|
||||
"""V4 sample plugin used by integration tests.
|
||||
|
||||
This fixture intentionally exercises the currently supported public v4 API:
|
||||
- top-level decorators: on_command/on_message/on_event/on_schedule/require_admin
|
||||
- Context clients: llm, memory, db, platform
|
||||
- MessageEvent helpers: reply/plain_result/target/to_payload
|
||||
- plugin-provided capabilities: normal + stream
|
||||
This fixture exercises the full public v4 API surface:
|
||||
|
||||
Decorators:
|
||||
- on_command: command handling with aliases and description
|
||||
- on_message: message handling with regex, keywords, platforms
|
||||
- on_event: event subscription
|
||||
- on_schedule: scheduled tasks
|
||||
- require_admin: permission control
|
||||
- provide_capability: custom capabilities (normal + stream)
|
||||
|
||||
Context clients:
|
||||
- ctx.llm: LLM client (chat, chat_raw, stream_chat, with images)
|
||||
- ctx.memory: Memory client (save, get, delete, search)
|
||||
- ctx.db: DB client (get, set, delete, list, get_many, set_many, watch)
|
||||
- ctx.platform: Platform client (send, send_image, send_chain, get_members)
|
||||
- ctx.http: HTTP client (register_api, unregister_api, list_apis)
|
||||
- ctx.metadata: Metadata client (get_plugin, list_plugins, get_plugin_config)
|
||||
|
||||
MessageEvent:
|
||||
- reply(), plain_result(), target, to_payload()
|
||||
- user_id, group_id, session_id, platform, text
|
||||
|
||||
Star lifecycle:
|
||||
- on_start, on_stop, on_error
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
@@ -26,89 +45,299 @@ from astrbot_sdk.context import CancelToken
|
||||
|
||||
|
||||
class HelloPlugin(Star):
|
||||
"""Representative v4 plugin fixture."""
|
||||
"""Representative v4 plugin fixture covering all SDK capabilities."""
|
||||
|
||||
# ============================================================
|
||||
# Lifecycle hooks
|
||||
# ============================================================
|
||||
|
||||
async def on_start(self, ctx: Context) -> None:
|
||||
"""Called when the plugin starts."""
|
||||
ctx.logger.info("HelloPlugin starting up")
|
||||
# Store startup timestamp
|
||||
await ctx.db.set("demo:started", {"status": "ok"})
|
||||
|
||||
async def on_stop(self, ctx: Context) -> None:
|
||||
"""Called when the plugin stops."""
|
||||
ctx.logger.info("HelloPlugin shutting down")
|
||||
# Cleanup
|
||||
await ctx.db.delete("demo:started")
|
||||
|
||||
# ============================================================
|
||||
# Command handlers
|
||||
# ============================================================
|
||||
|
||||
@on_command("hello", aliases=["hi"], description="发送问候消息")
|
||||
async def hello(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Basic command with LLM response."""
|
||||
reply = await ctx.llm.chat(event.text)
|
||||
await event.reply(reply)
|
||||
|
||||
chunks: list[str] = []
|
||||
async for chunk in ctx.llm.stream_chat("stream"):
|
||||
chunks.append(chunk)
|
||||
await event.reply("".join(chunks))
|
||||
|
||||
@on_command("raw", description="调用 llm.chat_raw 并返回结构化信息")
|
||||
async def raw(self, event: MessageEvent, ctx: Context):
|
||||
"""LLM chat with full response metadata."""
|
||||
response = await ctx.llm.chat_raw(
|
||||
event.text,
|
||||
system="be concise",
|
||||
history=[{"role": "user", "content": "history"}],
|
||||
)
|
||||
payload = event.to_payload()
|
||||
ctx.logger.info("raw handler for {}", payload.get("session_id"))
|
||||
return event.plain_result(
|
||||
f"raw={response.text}|finish={response.finish_reason}|"
|
||||
f"cancelled={ctx.cancel_token.cancelled}"
|
||||
f"usage={response.usage}"
|
||||
)
|
||||
|
||||
@on_command("remember", description="覆盖 memory/db 全量基本操作")
|
||||
@on_command("stream", description="流式 LLM 调用")
|
||||
async def stream(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Streaming LLM response."""
|
||||
chunks: list[str] = []
|
||||
async for chunk in ctx.llm.stream_chat(event.text or "stream"):
|
||||
chunks.append(chunk)
|
||||
# Real-time feedback
|
||||
await event.reply(f"[streaming...] {chunk}")
|
||||
await event.reply(f"[完成] {''.join(chunks)}")
|
||||
|
||||
@on_command("vision", description="带图片的 LLM 调用")
|
||||
async def vision(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""LLM with image input."""
|
||||
# Extract image URL from message or use default
|
||||
image_url = "https://example.com/demo.png"
|
||||
response = await ctx.llm.chat(
|
||||
event.text or "描述这张图片",
|
||||
image_urls=[image_url],
|
||||
)
|
||||
await event.reply(response)
|
||||
|
||||
# ============================================================
|
||||
# Memory operations
|
||||
# ============================================================
|
||||
|
||||
@on_command("remember", description="记忆操作演示")
|
||||
async def remember(self, event: MessageEvent, ctx: Context):
|
||||
"""Memory client full API demo."""
|
||||
# Save with metadata
|
||||
await ctx.memory.save(
|
||||
"demo:last_message",
|
||||
{"user_id": event.user_id or "", "text": event.text},
|
||||
source="fixture",
|
||||
tags=["demo"],
|
||||
)
|
||||
remembered = await ctx.memory.get("demo:last_message") or {}
|
||||
searched = await ctx.memory.search("fixture")
|
||||
await ctx.memory.delete("demo:last_message")
|
||||
|
||||
await ctx.db.set("demo:last_session", event.session_id)
|
||||
session_value = await ctx.db.get("demo:last_session")
|
||||
keys = await ctx.db.list("demo:")
|
||||
await ctx.db.delete("demo:last_session")
|
||||
# Get exact match
|
||||
remembered = await ctx.memory.get("demo:last_message") or {}
|
||||
|
||||
# Semantic search
|
||||
searched = await ctx.memory.search("demo")
|
||||
|
||||
# Delete
|
||||
await ctx.memory.delete("demo:last_message")
|
||||
|
||||
return event.plain_result(
|
||||
f"remembered={remembered.get('user_id', 'unknown')}|"
|
||||
f"searched={len(searched)}|session={session_value}|keys={len(keys)}"
|
||||
f"searched={len(searched)}"
|
||||
)
|
||||
|
||||
@on_command("platforms", description="覆盖 platform 相关 API")
|
||||
# ============================================================
|
||||
# Database operations
|
||||
# ============================================================
|
||||
|
||||
@on_command("db", description="数据库操作演示")
|
||||
async def db_ops(self, event: MessageEvent, ctx: Context):
|
||||
"""DB client full API demo."""
|
||||
# Basic operations
|
||||
await ctx.db.set("demo:key1", {"value": "data1"})
|
||||
await ctx.db.set("demo:key2", {"value": "data2"})
|
||||
await ctx.db.set("demo:key3", {"value": "data3"})
|
||||
|
||||
value1 = await ctx.db.get("demo:key1")
|
||||
|
||||
# List keys with prefix
|
||||
keys = await ctx.db.list("demo:")
|
||||
|
||||
# Batch operations
|
||||
values = await ctx.db.get_many(["demo:key1", "demo:key2"])
|
||||
await ctx.db.set_many({
|
||||
"demo:batch1": {"batch": True},
|
||||
"demo:batch2": {"batch": True},
|
||||
})
|
||||
|
||||
# Cleanup
|
||||
for key in ["demo:key1", "demo:key2", "demo:key3", "demo:batch1", "demo:batch2"]:
|
||||
await ctx.db.delete(key)
|
||||
|
||||
return event.plain_result(
|
||||
f"value1={value1}|keys={len(keys)}|batch_get={len(values)}"
|
||||
)
|
||||
|
||||
@on_command("watch", description="监听数据库变更")
|
||||
async def watch_db(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Watch for DB changes (demonstration)."""
|
||||
await event.reply("开始监听 demo: 前缀的变更 (5秒)...")
|
||||
|
||||
async def watcher():
|
||||
count = 0
|
||||
async for change in ctx.db.watch("demo:"):
|
||||
count += 1
|
||||
await event.reply(
|
||||
f"变更: {change['op']} {change['key']}"
|
||||
)
|
||||
if count >= 3:
|
||||
break
|
||||
|
||||
# Run watcher with timeout
|
||||
try:
|
||||
await asyncio.wait_for(watcher(), timeout=5.0)
|
||||
except asyncio.TimeoutError:
|
||||
await event.reply("监听超时结束")
|
||||
|
||||
# ============================================================
|
||||
# Platform operations
|
||||
# ============================================================
|
||||
|
||||
@on_command("platforms", description="平台操作演示")
|
||||
async def platforms(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Platform client full API demo."""
|
||||
target = event.target or event.session_id
|
||||
members = await ctx.platform.get_members(target)
|
||||
await ctx.platform.send_image(target, "https://example.com/demo.png")
|
||||
await ctx.platform.send(
|
||||
target,
|
||||
f"members={len(members)} first={members[0]['user_id'] if members else 'none'}",
|
||||
)
|
||||
|
||||
@on_command("announce", description="发送一条富消息链")
|
||||
async def announce(self, event: MessageEvent, ctx: Context) -> None:
|
||||
# Get group members
|
||||
members = await ctx.platform.get_members(target)
|
||||
|
||||
# Send text
|
||||
await ctx.platform.send(target, f"成员数: {len(members)}")
|
||||
|
||||
# Send image
|
||||
await ctx.platform.send_image(target, "https://example.com/demo.png")
|
||||
|
||||
# Send message chain
|
||||
await ctx.platform.send_chain(
|
||||
event.target or event.session_id,
|
||||
target,
|
||||
[
|
||||
{"type": "Plain", "text": "Demo "},
|
||||
{"type": "Plain", "text": "消息链 "},
|
||||
{"type": "Image", "file": "https://example.com/demo.png"},
|
||||
],
|
||||
)
|
||||
|
||||
@on_command("announce", description="发送富消息链")
|
||||
async def announce(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Send rich message chain."""
|
||||
await ctx.platform.send_chain(
|
||||
event.target or event.session_id,
|
||||
[
|
||||
{"type": "Plain", "text": "公告: "},
|
||||
{"type": "Plain", "text": event.text or "无内容"},
|
||||
],
|
||||
)
|
||||
|
||||
# ============================================================
|
||||
# HTTP API operations
|
||||
# ============================================================
|
||||
|
||||
@on_command("register_api", description="注册 HTTP API")
|
||||
async def register_http_api(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Register a custom HTTP API endpoint."""
|
||||
await ctx.http.register_api(
|
||||
route="/demo/api",
|
||||
handler_capability="demo.http_handler",
|
||||
methods=["GET", "POST"],
|
||||
description="Demo HTTP API",
|
||||
)
|
||||
apis = await ctx.http.list_apis()
|
||||
return event.plain_result(f"已注册 API，当前共 {len(apis)} 个")
|
||||
|
||||
@on_command("unregister_api", description="注销 HTTP API")
|
||||
async def unregister_http_api(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Unregister the HTTP API endpoint."""
|
||||
await ctx.http.unregister_api("/demo/api")
|
||||
return event.plain_result("已注销 API")
|
||||
|
||||
# ============================================================
|
||||
# Metadata operations
|
||||
# ============================================================
|
||||
|
||||
@on_command("plugins", description="列出所有插件")
|
||||
async def list_plugins(self, event: MessageEvent, ctx: Context):
|
||||
"""List all loaded plugins."""
|
||||
plugins = await ctx.metadata.list_plugins()
|
||||
names = [p.name for p in plugins]
|
||||
return event.plain_result(f"插件: {', '.join(names)}")
|
||||
|
||||
@on_command("plugin_info", description="获取插件信息")
|
||||
async def plugin_info(self, event: MessageEvent, ctx: Context):
|
||||
"""Get current plugin metadata."""
|
||||
me = await ctx.metadata.get_current_plugin()
|
||||
if me:
|
||||
return event.plain_result(
|
||||
f"name={me.name}|version={me.version}|author={me.author}"
|
||||
)
|
||||
return event.plain_result("无法获取插件信息")
|
||||
|
||||
@on_command("config", description="获取插件配置")
|
||||
async def get_config(self, event: MessageEvent, ctx: Context):
|
||||
"""Get plugin configuration."""
|
||||
config = await ctx.metadata.get_plugin_config()
|
||||
if config:
|
||||
return event.plain_result(f"config={config}")
|
||||
return event.plain_result("无配置")
|
||||
|
||||
# ============================================================
|
||||
# Permission control
|
||||
# ============================================================
|
||||
|
||||
@require_admin
|
||||
@on_command("secure", description="测试 require_admin")
|
||||
@on_command("secure", description="管理员专用命令")
|
||||
async def secure(self, event: MessageEvent):
|
||||
"""Admin-only command."""
|
||||
return event.plain_result(f"secure:{event.user_id or 'unknown'}")
|
||||
|
||||
# ============================================================
|
||||
# Message handlers
|
||||
# ============================================================
|
||||
|
||||
@on_message(regex=r"^ping$", keywords=["ping"], platforms=["test"])
|
||||
async def ping(self, event: MessageEvent):
|
||||
"""Regex and keyword matching."""
|
||||
return event.plain_result("pong")
|
||||
|
||||
@on_message(keywords=["hello"])
|
||||
async def on_hello(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Keyword-based message handler."""
|
||||
await event.reply("检测到 hello 关键词!")
|
||||
|
||||
# ============================================================
|
||||
# Event handlers
|
||||
# ============================================================
|
||||
|
||||
@on_event("group_join")
|
||||
async def on_group_join(self, event: MessageEvent, ctx: Context) -> None:
|
||||
ctx.logger.info("event handler observed {}", event.text)
|
||||
"""Handle group join events."""
|
||||
ctx.logger.info("用户加入群组: {}", event.user_id)
|
||||
await ctx.platform.send(
|
||||
event.session_id,
|
||||
f"欢迎 {event.user_id} 加入群组!"
|
||||
)
|
||||
|
||||
@on_schedule(interval_seconds=60)
|
||||
async def heartbeat(self, ctx: Context) -> None:
|
||||
await ctx.db.set("demo:last_schedule", {"status": "ok"})
|
||||
@on_event("group_leave")
|
||||
async def on_group_leave(self, event: MessageEvent, ctx: Context) -> None:
|
||||
"""Handle group leave events."""
|
||||
ctx.logger.info("用户离开群组: {}", event.user_id)
|
||||
|
||||
# ============================================================
|
||||
# Scheduled tasks
|
||||
# ============================================================
|
||||
|
||||
@on_schedule(interval_seconds=3600)
|
||||
async def hourly_heartbeat(self, ctx: Context) -> None:
|
||||
"""Hourly scheduled task."""
|
||||
await ctx.db.set("demo:last_heartbeat", {"time": "hourly"})
|
||||
ctx.logger.info("执行每小时心跳")
|
||||
|
||||
@on_schedule(cron="0 9 * * *")
|
||||
async def morning_greeting(self, ctx: Context) -> None:
|
||||
"""Cron-based scheduled task (9 AM daily)."""
|
||||
ctx.logger.info("早安问候任务触发")
|
||||
|
||||
# ============================================================
|
||||
# Custom capabilities
|
||||
# ============================================================
|
||||
|
||||
@provide_capability(
|
||||
"demo.echo",
|
||||
@@ -133,14 +362,12 @@ class HelloPlugin(Star):
|
||||
ctx: Context,
|
||||
cancel_token: CancelToken,
|
||||
) -> dict[str, str]:
|
||||
"""Simple echo capability."""
|
||||
cancel_token.raise_if_cancelled()
|
||||
await ctx.db.set(
|
||||
"demo:capability_echo",
|
||||
{"text": str(payload.get("text", ""))},
|
||||
)
|
||||
stored = await ctx.db.get("demo:capability_echo") or {}
|
||||
text = str(payload.get("text", ""))
|
||||
await ctx.db.set("demo:capability_echo", {"text": text})
|
||||
return {
|
||||
"echo": str(stored.get("text", "")),
|
||||
"echo": text,
|
||||
"plugin_id": ctx.plugin_id,
|
||||
}
|
||||
|
||||
@@ -171,9 +398,47 @@ class HelloPlugin(Star):
|
||||
ctx: Context,
|
||||
cancel_token: CancelToken,
|
||||
):
|
||||
"""Streaming echo capability."""
|
||||
text = str(payload.get("text", ""))
|
||||
await ctx.db.set("demo:last_stream", {"text": text})
|
||||
for char in text:
|
||||
cancel_token.raise_if_cancelled()
|
||||
await asyncio.sleep(0)
|
||||
yield {"text": char}
|
||||
|
||||
@provide_capability(
|
||||
"demo.http_handler",
|
||||
description="处理 /demo/api HTTP 请求",
|
||||
input_schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"method": {"type": "string"},
|
||||
"body": {"type": "object"},
|
||||
},
|
||||
},
|
||||
output_schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"status": {"type": "integer"},
|
||||
"body": {"type": "object"},
|
||||
},
|
||||
},
|
||||
)
|
||||
async def http_handler_capability(
|
||||
self,
|
||||
payload: dict[str, object],
|
||||
ctx: Context,
|
||||
cancel_token: CancelToken,
|
||||
) -> dict[str, object]:
|
||||
"""Handle HTTP API requests."""
|
||||
method = payload.get("method", "GET")
|
||||
body = payload.get("body", {})
|
||||
ctx.logger.info(f"HTTP {method} request: {body}")
|
||||
return {
|
||||
"status": 200,
|
||||
"body": {
|
||||
"message": "Hello from plugin!",
|
||||
"method": method,
|
||||
"plugin_id": ctx.plugin_id,
|
||||
},
|
||||
}
|
||||
|
||||
@@ -1,267 +0,0 @@
|
||||
"""旧版插件兼容测试夹具。
|
||||
|
||||
这个样例故意覆盖当前仍被兼容层支持的旧 API：
|
||||
- CommandComponent / AstrMessageEvent / MessageChain
|
||||
- filter.command / regex / permission / permission_type /
|
||||
event_message_type / platform_adapter_type
|
||||
- LegacyContext 的 conversation_manager / llm_generate / tool_loop_agent /
|
||||
send_message / put|get|delete_kv_data / call_context_function
|
||||
- 消息组件及其旧字段别名/工厂方法
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from astrbot_sdk import provide_capability
|
||||
from astrbot_sdk.api.components.command import CommandComponent
|
||||
from astrbot_sdk.api.event import AstrMessageEvent, filter
|
||||
from astrbot_sdk.api.event.filter import (
|
||||
ADMIN,
|
||||
EventMessageType,
|
||||
PermissionType,
|
||||
PlatformAdapterType,
|
||||
)
|
||||
from astrbot_sdk.api.message import MessageChain
|
||||
from astrbot_sdk.api.message_components import (
|
||||
At,
|
||||
AtAll,
|
||||
Face,
|
||||
File,
|
||||
Image,
|
||||
Node,
|
||||
Plain,
|
||||
Record,
|
||||
Reply,
|
||||
Video,
|
||||
)
|
||||
from astrbot_sdk.api.star.context import Context
|
||||
from loguru import logger
|
||||
|
||||
|
||||
class CompatHelper:
|
||||
__compat_component_name__ = "CompatHelper"
|
||||
|
||||
async def shout(self, text: str) -> str:
|
||||
return text.upper()
|
||||
|
||||
|
||||
class HelloCommand(CommandComponent):
|
||||
"""测试旧版 CommandComponent 兼容性。"""
|
||||
|
||||
def __init__(self, context: Context):
|
||||
self.context = context
|
||||
self.context._register_component(CompatHelper())
|
||||
|
||||
@filter.command("hello", alias={"hi"}, priority=10, desc="问候命令")
|
||||
async def hello(self, event: AstrMessageEvent):
|
||||
"""基本命令测试。"""
|
||||
ret = await self.context.conversation_manager.new_conversation("hello")
|
||||
logger.info("New conversation created: {}", ret)
|
||||
yield event.plain_result(f"Hello, Astrbot! Created conversation ID: {ret}")
|
||||
|
||||
@filter.command("echo")
|
||||
async def echo(self, event: AstrMessageEvent):
|
||||
"""测试消息获取和 extra 状态。"""
|
||||
text = event.get_message_str()
|
||||
event.set_extra("last_echo", text)
|
||||
extra = event.get_extra("last_echo")
|
||||
event.clear_extra()
|
||||
group = await event.get_group()
|
||||
yield event.plain_result(
|
||||
f"Echo: {text}|sender={event.get_sender_id()}|sender_name={event.get_sender_name()}|"
|
||||
f"platform={event.get_platform_name()}:{event.get_platform_id()}|"
|
||||
f"type={event.get_message_type().value}|messages={len(event.get_messages())}|"
|
||||
f"self={event.get_self_id()}|private={event.is_private_chat()}|"
|
||||
f"wake={event.is_wake_up()}|group={group is not None}|extra={extra}"
|
||||
)
|
||||
|
||||
@filter.command("chain")
|
||||
async def test_chain(self, event: AstrMessageEvent):
|
||||
"""测试 MessageChain 构建、发送与 react。"""
|
||||
chain = (
|
||||
MessageChain()
|
||||
.message("Hello")
|
||||
.message(" ")
|
||||
.at("user", "12345")
|
||||
.at_all()
|
||||
.message(" check this image: ")
|
||||
.url_image("https://example.com/test.png")
|
||||
.file_image("C:/tmp/test.png")
|
||||
.base64_image("base64://fixture")
|
||||
.use_t2i(True)
|
||||
.squash_plain()
|
||||
)
|
||||
|
||||
await event.send(chain)
|
||||
await event.react(":thumbsup:")
|
||||
yield event.plain_result(
|
||||
f"Chain sent with {len(chain.to_payload())} components and t2i={chain.use_t2i_}"
|
||||
)
|
||||
|
||||
@filter.command("db")
|
||||
async def test_db(self, event: AstrMessageEvent):
|
||||
"""测试 KV 数据库操作。"""
|
||||
await self.context.put_kv_data("test_key", {"value": "test_data"})
|
||||
data = await self.context.get_kv_data("test_key")
|
||||
await self.context.delete_kv_data("test_key")
|
||||
deleted = await self.context.get_kv_data("test_key", "missing")
|
||||
logger.info("Got data from db: {}", data)
|
||||
yield event.plain_result(f"DB test: stored={data} deleted={deleted}")
|
||||
|
||||
@filter.command("conversation")
|
||||
async def test_conversation(self, event: AstrMessageEvent):
|
||||
"""测试会话管理器和 call_context_function。"""
|
||||
umo = f"platform:{event.get_session_id()}"
|
||||
cid = await self.context.conversation_manager.new_conversation(
|
||||
unified_msg_origin=umo,
|
||||
title="Compat Chat",
|
||||
persona_id="assistant",
|
||||
)
|
||||
await self.context.conversation_manager.add_message_pair(cid, "hello", "world")
|
||||
await self.context.conversation_manager.update_conversation(
|
||||
unified_msg_origin=umo,
|
||||
conversation_id=cid,
|
||||
title="Compat Chat Updated",
|
||||
)
|
||||
await self.context.conversation_manager.update_conversation_title(umo, "Compat")
|
||||
await self.context.conversation_manager.update_conversation_persona_id(
|
||||
umo,
|
||||
"compat-persona",
|
||||
)
|
||||
current_id = await self.context.conversation_manager.get_curr_conversation_id(
|
||||
umo
|
||||
)
|
||||
conv = await self.context.conversation_manager.get_conversation(umo, cid)
|
||||
all_conversations = await self.context.conversation_manager.get_conversations(
|
||||
umo
|
||||
)
|
||||
helper_result = await self.context.call_context_function(
|
||||
"CompatHelper.shout",
|
||||
{"text": "compat"},
|
||||
)
|
||||
await self.context.conversation_manager.switch_conversation(umo, cid)
|
||||
await self.context.conversation_manager.delete_conversation(umo, cid)
|
||||
yield event.plain_result(
|
||||
f"conversation={current_id}|messages={len((conv or {}).get('content', []))}|"
|
||||
f"all={len(all_conversations)}|helper={helper_result['data']}"
|
||||
)
|
||||
|
||||
@filter.command("ai")
|
||||
async def test_ai(self, event: AstrMessageEvent):
|
||||
"""测试旧版 AI compat 入口。"""
|
||||
llm_resp = await self.context.llm_generate(
|
||||
chat_provider_id="provider-demo",
|
||||
prompt="legacy hello",
|
||||
contexts=[{"role": "user", "content": "hi"}],
|
||||
)
|
||||
agent_resp = await self.context.tool_loop_agent(
|
||||
chat_provider_id="provider-demo",
|
||||
prompt="legacy hello",
|
||||
tools=[{"name": "search"}],
|
||||
max_steps=3,
|
||||
)
|
||||
yield event.plain_result(f"LLM:{llm_resp.text}|AGENT:{agent_resp.text}")
|
||||
|
||||
@filter.command("sendmsg")
|
||||
async def test_send_message(self, event: AstrMessageEvent):
|
||||
"""测试 LegacyContext.send_message。"""
|
||||
chain = (
|
||||
MessageChain()
|
||||
.message("compat send ")
|
||||
.at("legacy-user", "10001")
|
||||
.url_image("https://example.com/send.png")
|
||||
)
|
||||
await self.context.send_message(event.get_session_id(), chain)
|
||||
yield event.plain_result("send_message invoked")
|
||||
|
||||
@filter.command("components")
|
||||
async def test_components(self, event: AstrMessageEvent):
|
||||
"""测试消息组件和 chain_result/image_result。"""
|
||||
components = [
|
||||
Plain(text="Hello"),
|
||||
At(qq="123", name="legacy_user"),
|
||||
AtAll(),
|
||||
Image.fromURL("https://example.com/img.png"),
|
||||
Image.fromFileSystem("C:/tmp/local.png"),
|
||||
Record.fromFileSystem("C:/tmp/sound.wav"),
|
||||
Video.fromURL("https://example.com/video.mp4"),
|
||||
File(name="demo.txt", file="https://example.com/demo.txt"),
|
||||
Reply(id="reply-1"),
|
||||
Node(uin="10001", name="node_user", content=[Plain(text="node")]),
|
||||
Face(id=1),
|
||||
]
|
||||
component_dicts = [component.to_dict() for component in components]
|
||||
logger.info("Components: {}", component_dicts)
|
||||
yield event.chain_result(components)
|
||||
yield event.image_result("https://example.com/components.png")
|
||||
|
||||
@filter.command("state")
|
||||
async def test_event_state(self, event: AstrMessageEvent):
|
||||
"""测试事件状态方法。"""
|
||||
result = event.make_result().message("state-ok")
|
||||
event.set_result(result)
|
||||
before_stop = event.is_stopped()
|
||||
event.stop_event()
|
||||
after_stop = event.is_stopped()
|
||||
event.continue_event()
|
||||
after_continue = event.is_stopped()
|
||||
event.should_call_llm(True)
|
||||
stored = event.get_result()
|
||||
event.clear_result()
|
||||
yield event.plain_result(
|
||||
f"before={before_stop}|after_stop={after_stop}|after_continue={after_continue}|"
|
||||
f"call_llm={event.call_llm}|stored={stored is not None}"
|
||||
)
|
||||
|
||||
@filter.regex(r"^ping.*")
|
||||
async def ping_regex(self, event: AstrMessageEvent):
|
||||
"""测试正则匹配。"""
|
||||
yield event.plain_result("Pong from regex!")
|
||||
|
||||
@filter.permission(ADMIN)
|
||||
@filter.command("admin")
|
||||
async def admin_only(self, event: AstrMessageEvent):
|
||||
"""测试 permission(ADMIN)。"""
|
||||
yield event.plain_result(f"Admin command executed by {event.is_admin()}")
|
||||
|
||||
@filter.permission_type(PermissionType.ADMIN)
|
||||
@filter.command("admin_type")
|
||||
async def admin_type_only(self, event: AstrMessageEvent):
|
||||
"""测试 permission_type(PermissionType.ADMIN)。"""
|
||||
yield event.plain_result(f"Admin type command executed by {event.is_admin()}")
|
||||
|
||||
@filter.event_message_type(EventMessageType.GROUP_MESSAGE)
|
||||
async def group_only(self, event: AstrMessageEvent):
|
||||
"""测试消息类型过滤。"""
|
||||
yield event.plain_result("Group message received")
|
||||
|
||||
@filter.platform_adapter_type(PlatformAdapterType.AIOCQHTTP)
|
||||
async def cqhttp_only(self, event: AstrMessageEvent):
|
||||
"""测试平台过滤。"""
|
||||
yield event.plain_result("CQHttp platform detected")
|
||||
|
||||
@provide_capability(
|
||||
"compat.echo",
|
||||
description="使用 legacy Context 回显输入文本",
|
||||
input_schema={
|
||||
"type": "object",
|
||||
"properties": {"text": {"type": "string"}},
|
||||
"required": ["text"],
|
||||
},
|
||||
output_schema={
|
||||
"type": "object",
|
||||
"properties": {
|
||||
"echo": {"type": "string"},
|
||||
"plugin_id": {"type": "string"},
|
||||
},
|
||||
"required": ["echo", "plugin_id"],
|
||||
},
|
||||
)
|
||||
async def echo_capability(self, payload: dict):
|
||||
"""测试旧版插件 capability 能走 LegacyContext。"""
|
||||
text = str(payload.get("text", ""))
|
||||
await self.context.put_kv_data("compat_capability", {"text": text})
|
||||
stored = await self.context.get_kv_data("compat_capability", {})
|
||||
return {
|
||||
"echo": str(stored.get("text", "")),
|
||||
"plugin_id": self.context.plugin_id,
|
||||
}
|
||||
@@ -1,24 +0,0 @@
|
||||
_schema_version: 2
|
||||
name: astrbot_plugin_helloworld
|
||||
display_name: HelloWorld 插件
|
||||
desc: 一个简单的问候插件示例
|
||||
author: Soulter
|
||||
version: 0.1.0
|
||||
runtime:
|
||||
python: "3.12"
|
||||
components: # 组件列表，将支持自动生成
|
||||
- class: commands.hello:HelloCommand
|
||||
type: command
|
||||
name: hello
|
||||
description: 发送问候消息
|
||||
subcommands:
|
||||
- name: wow
|
||||
description: 发送 "Hello, Astrbot!" 消息
|
||||
# - class: handlers.tools.echo:EchoTool
|
||||
# type: llm_tool
|
||||
# name: echo_tool
|
||||
# description: 回显输入的消息
|
||||
# - class: handlers.listeners.echo:EchoListener
|
||||
# type: listener
|
||||
# name: message_logger
|
||||
# description: 监听并记录所有消息
|
||||
@@ -1 +0,0 @@
|
||||
|
||||
Reference in New Issue
Block a user