添加日志输出以指示插件功能开发状态;移除未使用的函数注册模块

This commit is contained in:
远野千束(神羽) 2024-12-15 02:59:06 +08:00
parent 0704285cde
commit 6c6b45a168
5 changed files with 22 additions and 175 deletions

View File

@ -1,7 +1,11 @@
"""该功能目前正在开发中,暂时不可用,受影响的文件夹 `plugin`, `plugins`
"""
from nonebot import logger
from .func_call import *
from .load import *
from .models import *
from .utils import *
logger.opt(colors=True).info("<y>MarshoAI 插件功能开发中,用户请忽略相关日志</y>")

View File

@ -1,3 +1,2 @@
from .caller import *
from .params import *
from .register import *

View File

@ -1,8 +1,11 @@
from typing import Generic
from typing import Generic, TypeVar
from nonebot import logger
from ..typing import FUNCTION_CALL_FUNC
from .params import P
from .register import F
F = TypeVar("F", bound=FUNCTION_CALL_FUNC)
class Caller(Generic[P]):
@ -60,6 +63,17 @@ class Caller(Generic[P]):
return self
def __call__(self, func: F) -> F:
"""装饰函数注册为一个可被AI调用的function call函数
Args:
func (F): 函数对象
Returns:
F: 函数对象
"""
logger.opt(colors=True).info(
f"<y>加载函数 {func.__name__} {self._description}</y>"
)
self.func = func
return func
@ -73,4 +87,5 @@ def on_function_call(name: str | None = None, description: str | None = None) ->
Returns:
Caller: Caller对象
"""
return Caller(name=name, description=description)

View File

@ -1,126 +0,0 @@
"""此模块用于获取function call中函数定义信息以及注册函数
"""
from typing import TypeVar
from nonebot import logger
from ..docstring.parser import parse
from ..typing import (
ASYNC_FUNCTION_CALL_FUNC,
FUNCTION_CALL_FUNC,
SYNC_FUNCTION_CALL_FUNC,
)
from .params import *
F = TypeVar("F", bound=FUNCTION_CALL_FUNC)
_loaded_functions: dict[str, FUNCTION_CALL_FUNC] = {}
def async_wrapper(func: SYNC_FUNCTION_CALL_FUNC) -> ASYNC_FUNCTION_CALL_FUNC:
"""将同步函数包装为异步函数,但是不会真正异步执行,仅用于统一调用及函数签名
Args:
func: 同步函数
Returns:
ASYNC_FUNCTION_CALL: 异步函数
"""
async def wrapper(*args, **kwargs) -> str:
return func(*args, **kwargs)
return wrapper
def function_call(func: F) -> F:
"""返回一个装饰器,装饰一个函数, 使其注册为一个可被AI调用的function call函数
Args:
func: 函数对象要有完整的 Google Style Docstring
Returns:
str: 函数定义信息
"""
# TODO
# pre check docstring
if not func.__doc__:
logger.error(f"函数 {func.__name__} 没有文档字串,不被加载")
return func
else:
# 解析函数文档字串
result = parse(docstring=func.__doc__)
logger.debug(result.reduction())
return func
def caller(
description: str | None = None,
parameters: dict[str, P] | None = None,
):
"""返回一个装饰器,装饰一个函数, 使其注册为一个可被AI调用的function call函数
Args:
description: 函数描述
parameters: 函数参数
Returns:
str: 函数定义信息
"""
def decorator(func: FUNCTION_CALL_FUNC) -> FUNCTION_CALL_FUNC:
# TODO
# pre check docstring
if not func.__doc__:
logger.error(f"函数 {func.__name__} 没有文档字串,不被加载")
return func
else:
# 解析函数文档字串
result = parse(docstring=func.__doc__)
logger.debug(result.reduction())
return func
return decorator
# TODO 草案
# @caller(
# description="这个函数用来给你算命",
# parameters={
# "birthday": String(description="生日"),
# "gender": String(enum=["男", "女"], description="性别"),
# "name": String(description="姓名"),
# },
# )
# async def tell_fortune(birthday: str, name: str, gender: str) -> str:
# """这个函数用来给你算命
# Args:
# birthday: 生日
# name: 姓名
# Returns:
# str: 算命结果
# """
# return f"{name},你的生日是{birthday},你的运势是大吉大利"
@caller(
description="这个函数用来给你算命",
).parameters(
birthday=String(description="生日"),
name=String(enum=["", ""], description="性别"),
gender=String(description="姓名"),
)
async def tell_fortune(birthday: str, name: str, gender: str) -> str:
"""这个函数用来给你算命
Args:
birthday: 生日
name: 姓名
Returns:
str: 算命结果
"""
return f"{name},你的生日是{birthday},你的运势是大吉大利"

View File

@ -2,13 +2,7 @@ import traceback
import httpx
from nonebot_plugin_marshoai.plugin import (
Integer,
PluginMetadata,
String,
function_call,
on_function_call,
)
from nonebot_plugin_marshoai.plugin import PluginMetadata
__marsho_meta__ = PluginMetadata(
name="Bangumi 番剧信息",
@ -17,42 +11,3 @@ __marsho_meta__ = PluginMetadata(
author="Liteyuki",
homepage="",
)
@function_call
async def fetch_calendar():
"""获取今天日期"""
url = "https://api.bgm.tv/calendar"
headers = {
"User-Agent": "LiteyukiStudio/nonebot-plugin-marshoai (https://github.com/LiteyukiStudio/nonebot-plugin-marshoai)"
}
async with httpx.AsyncClient() as client:
response = await client.get(url, headers=headers)
# print(response.text)
return response.json()
@function_call
async def get_bangumi_news() -> str:
"""获取今天的新番(动漫)列表,在调用之前,你需要知道今天星期几。"""
result = await fetch_calendar()
info = ""
try:
for i in result:
weekday = i["weekday"]["cn"]
# print(weekday)
info += f"{weekday}:"
items = i["items"]
for item in items:
name = item["name_cn"]
info += f"{name}"
info += "\n"
return info
except Exception as e:
traceback.print_exc()
return ""
@function_call
def test_sync() -> str:
return "sync"