兼容pyd2

README.md

@@ -9,7 +9,7 @@
 
 ![License](https://img.shields.io/github/license/eya46/nonebot_plugin_enatfrp)
 ![Python](https://img.shields.io/badge/python-3.8+-blue.svg)
-![NoneBot](https://img.shields.io/badge/nonebot-2.1.0+-red.svg)
+![NoneBot](https://img.shields.io/badge/nonebot-2.2.0+-red.svg)
 </div>
 
 ## 安装方式
@@ -86,6 +86,5 @@ natfrp_cmd_showPCs=
 
 ## 依赖项
 
-- [nonebot2](https://github.com/nonebot/nonebot2) >=2.1.0
-- [plugin-alconna](https://github.com/nonebot/plugin-alconna) >=0.35.0,<=0.40.0
-- httpx >=0.20.0,<1.0.0
+- [nonebot2](https://github.com/nonebot/nonebot2)[httpx] ^2.2.0
+- [plugin-alconna](https://github.com/nonebot/plugin-alconna) >=0.35.0,<1.0.0

nonebot_plugin_enatfrp/__init__.py

@@ -5,8 +5,7 @@
 from typing_extensions import Annotated
 from typing import Literal
 
-from arclet.alconna import Args
-from nonebot_plugin_alconna import Alconna, Match, AlconnaMatch
+from nonebot_plugin_alconna import Alconna, Match, AlconnaMatch, Args
 
 from .api import API
 from .tools import (
@@ -15,25 +14,25 @@
 )
 from .config import config, Config
 
-api = API(config.natfrp_api, config.natfrp_token)
+api = API(config.api, config.token)
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_help, Args["index?", int]))
+@wrap_cmd(Alconna(config.cmd_help, Args["index?", int]))
 async def help_cmd_handle(index: Annotated[Match[int], AlconnaMatch("index")]):
     if index.available:
         await send(f"{index.result}.{HELPS[index.result]}")
     else:
         await send("\n".join(f"{i}.{h}" for i, h in enumerate(HELPS)))
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_powerOn, Args["id", int]["password?", str, ""]))
+@wrap_cmd(Alconna(config.cmd_powerOn, Args["id", int]["password?", str, ""]))
 async def power_on_cmd_handle(
         id_: Annotated[Match[int], AlconnaMatch("id")], password: Annotated[Match[str], AlconnaMatch("password")]
 ):
     await send(await api.computer_poweron(id_.result, password.result))
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_tunnels))
+@wrap_cmd(Alconna(config.cmd_tunnels))
 async def tunnels_cmd_handle(cmd: Annotated[str, RawCommand()]):
     tunnels = await api.get_tunnels()
     msg = cmd + "：\n状态 ID 名称\n"
@@ -43,7 +42,7 @@ async def tunnels_cmd_handle(cmd: Annotated[str, RawCommand()]):
 
 
 @wrap_cmd(Alconna(
-    config.natfrp_cmd_trafficHistory,
+    config.cmd_trafficHistory,
     Args["id?", int]["type?", Literal["day", "week", "month", "日", "周", "月"], "day"]
 ))
 async def traffic_history_cmd_handle(
@@ -60,7 +59,7 @@ async def traffic_history_cmd_handle(
     await send(msg)
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_trafficPlans, Args[
+@wrap_cmd(Alconna(config.cmd_trafficPlans, Args[
     "type", Literal["valid", "invalid", "all", "可用", "不可用", "全部"], "all"
 ]))
 async def traffic_plans_cmd_handle(
@@ -76,7 +75,7 @@ async def traffic_plans_cmd_handle(
     await send(msg)
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_userInfo))
+@wrap_cmd(Alconna(config.cmd_userInfo))
 async def user_info_cmd_handle():
     u = await api.user_info()
     await send(
@@ -95,22 +94,22 @@ async def user_info_cmd_handle():
     )
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_announcement, Args["index", int, 0]))
+@wrap_cmd(Alconna(config.cmd_announcement, Args["index", int, 0]))
 async def announcement_cmd_handle(index: Annotated[Match[int], AlconnaMatch("index")]):
     announcements = await api.system_bulletin()
     index = min(max(len(announcements) - 1, 0), index.result)
     a = announcements[index]
     await send(unescape_html(f"{a['icon']}{a['title']}{a['icon']}\n{a['time']}\n{a['content']}"))
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_auth, Args["id", int]["ip", str]))
+@wrap_cmd(Alconna(config.cmd_auth, Args["id", int]["ip", str]))
 async def auth_cmd_handle(
         id_: Annotated[Match[int], AlconnaMatch("id")], ip: Annotated[Match[str], AlconnaMatch("ip")]
 ):
     await send("IP:" + await api.tunnel_auth(id_.result, ip.result))
 
 
-@wrap_cmd(Alconna(config.natfrp_cmd_showPCs))
+@wrap_cmd(Alconna(config.cmd_showPCs))
 async def show_pcs_cmd_handle():
     pcs = await api.get_computers()
     await send("\n".join(f"{pc['id']}.{pc['name']}" + ("<加密>" if pc["type"] != 1 else "") for pc in pcs))

nonebot_plugin_enatfrp/api.py

@@ -11,8 +11,8 @@ async def __call__(self, **data: Any) -> Any:
 
 
 class API:
-    def __init__(self, api: str, token: str):
-        self.api = URL(api)
+    def __init__(self, api: Any, token: str):
+        self.api = URL(str(api))
         self.headers = {"Authorization": f"Bearer {token}"}
 
     async def call_api(

nonebot_plugin_enatfrp/config.py

@@ -1,28 +1,28 @@
 from typing import Optional, List
 
-from nonebot import get_driver
+from nonebot.plugin import get_plugin_config
 from pydantic import BaseModel, Field, AnyUrl
 
 
 class Config(BaseModel):
-    natfrp_api: AnyUrl = Field(default="https://api.natfrp.com/v4")
-    natfrp_token: Optional[str] = Field(default=None)
-    natfrp_at: bool = Field(default=False)
-    natfrp_reply: bool = Field(default=False)
-    natfrp_use_start: bool = Field(default=True)
-    natfrp_use_sep: bool = Field(default=False)
+    api: AnyUrl = Field(default="https://api.natfrp.com/v4", alias="natfrp_api")
+    token: Optional[str] = Field(default=None, alias="natfrp_token")
+    at: bool = Field(default=False, alias="natfrp_at")
+    reply: bool = Field(default=False, alias="natfrp_reply")
+    use_start: bool = Field(default=True, alias="natfrp_use_start")
+    use_sep: bool = Field(default=False, alias="natfrp_use_sep")
 
-    natfrp_cmd_help: List[str] = Field(default=["frp帮助"])
-    natfrp_cmd_powerOn: List[str] = Field(default=["开机"])
-    natfrp_cmd_tunnels: List[str] = Field(default=["隧道状态"])
-    natfrp_cmd_trafficHistory: List[str] = Field(default=["流量", "流量历史"])
-    natfrp_cmd_trafficPlans: List[str] = Field(default=["流量包", "流量套餐"])
-    natfrp_cmd_userInfo: List[str] = Field(default=["我的信息", "用户信息"])
-    natfrp_cmd_announcement: List[str] = Field(default=["公告"])
-    natfrp_cmd_auth: List[str] = Field(default=["授权"])
-    natfrp_cmd_showPCs: List[str] = Field(default=["计算机列表"])
+    cmd_help: List[str] = Field(default=["frp帮助"], alias="natfrp_cmd_help")
+    cmd_powerOn: List[str] = Field(default=["开机"], alias="natfrp_cmd_powerOn")
+    cmd_tunnels: List[str] = Field(default=["隧道状态"], alias="natfrp_cmd_tunnels")
+    cmd_trafficHistory: List[str] = Field(default=["流量", "流量历史"], alias="natfrp_cmd_trafficHistory")
+    cmd_trafficPlans: List[str] = Field(default=["流量包", "流量套餐"], alias="natfrp_cmd_trafficPlans")
+    cmd_userInfo: List[str] = Field(default=["我的信息", "用户信息"], alias="natfrp_cmd_userInfo")
+    cmd_announcement: List[str] = Field(default=["公告"], alias="natfrp_cmd_announcement")
+    cmd_auth: List[str] = Field(default=["授权"], alias="natfrp_cmd_auth")
+    cmd_showPCs: List[str] = Field(default=["计算机列表"], alias="natfrp_cmd_showPCs")
 
 
-config: Config = Config(**get_driver().config.dict())
+config: Config = get_plugin_config(Config)
 
 __all__ = ["Config", "config"]

nonebot_plugin_enatfrp/exception.py

@@ -1,8 +1,8 @@
-class eNatFrpException(Exception):
+class NatFrpException(Exception):
     pass
 
 
-class eNatFrpAPIException(eNatFrpException):
+class NatFrpAPIException(NatFrpException):
     def __init__(self, code: int, message: str):
         self.code = code
         self.message = message

nonebot_plugin_enatfrp/tools.py

@@ -15,7 +15,7 @@
 from nonebot_plugin_alconna.uniseg import Receipt
 
 from .config import config
-from .exception import eNatFrpAPIException
+from .exception import NatFrpAPIException
 
 if isinstance(driver := get_driver(), HTTPClientMixin):  # type:ignore
     driver: HTTPClientMixin
@@ -31,7 +31,7 @@ async def request(method: str, url: str, headers, **data) -> Any:
 
         res = loads(resp.content)
         if resp.status_code == 500:
-            raise eNatFrpAPIException(res.get("code", -1), res.get("msg", str(resp.content)))
+            raise NatFrpAPIException(res.get("code", -1), res.get("msg", str(resp.content)))
         return res
 else:
     from httpx import AsyncClient
@@ -45,7 +45,7 @@ async def request(method: str, url: str, headers, **data) -> Any:
             )
             res = loads(resp.content)
             if resp.status_code == 500:
-                raise eNatFrpAPIException(res.get("code", -1), res.get("msg", str(resp.content)))
+                raise NatFrpAPIException(res.get("code", -1), res.get("msg", str(resp.content)))
             return res
 
 # 日_周_月
@@ -73,7 +73,7 @@ async def request(method: str, url: str, headers, **data) -> Any:
 async def send(msg: Union[Any, UniMessage]) -> Receipt:
     if not isinstance(msg, UniMessage):
         msg = UniMessage(msg)
-    return await msg.send(at_sender=config.natfrp_at, reply_to=config.natfrp_reply)
+    return await msg.send(at_sender=config.at, reply_to=config.reply)
 
 
 def bytes_to_human(data: int) -> str:
@@ -103,15 +103,15 @@ def wrapper(func: T_Handler) -> T_Handler:
         async def _catch(*args, **kwargs):
             try:
                 await func(*args, **kwargs)
-            except eNatFrpAPIException as e:
+            except NatFrpAPIException as e:
                 await send(
                     f"命令 {_raw_command()} 失败<{e.code}> ->\n"
                     f"{e.message}"
                 )
 
         HELPS.append(acl.get_help().replace("\nUnknown", "").strip())
         on_alconna(
-            acl, permission=SUPERUSER, use_cmd_sep=config.natfrp_use_sep, use_cmd_start=config.natfrp_use_start
+            acl, permission=SUPERUSER, use_cmd_sep=config.use_sep, use_cmd_start=config.use_start
         ).handle()(_catch)
         return func