:bug 修复语法问题。优化部分逻辑

Author: Agnes4m

nonebot_plugin_l4d2_server/__init__.py

@@ -44,4 +44,3 @@
         "author": "Agnes4m <Z735803792@163.com>",
     },
 )
-from .commands import server_groups

nonebot_plugin_l4d2_server/__main__.py

@@ -16,7 +16,7 @@
 """
 
 from pathlib import Path
-from typing import List, Optional
+from typing import List, Optional, cast
 
 import aiofiles
 import ujson as json
@@ -27,10 +27,12 @@
 from nonebot.params import CommandArg, CommandStart, RawCommand
 from nonebot.permission import SUPERUSER
 from nonebot.plugin import on_command
-from nonebot.rule import command as command_rule
 from nonebot_plugin_alconna import UniMessage
 
+from nonebot_plugin_l4d2_server.l4_ban.utils import refresh_server_command_rule
+
 from .config import config, config_manager
+from .l4_ban import l4_request
 from .l4_help import get_l4d2_core_help
 from .l4_local import *  # noqa: F403
 from .l4_request import (
@@ -54,25 +56,17 @@
 
 reload_ip()
 
+
 l4_help = on_command("l4help", aliases={"l4d2帮助"})
-l4_request = on_command("anne", priority=10)
+
 l4_reload = on_command("l4reload", aliases={"l4刷新,l4重载"})
 l4_all = on_command("l4all", aliases={"l4全服"})
 l4_connect = on_command("connect", aliases={"l4连接"})
 l4_find_player = on_command("l4find", aliases={"l4查找"})
 
 
 config_path = Path(config.l4_path) / "config.json"
-
-
-def refresh_server_command_rule() -> None:
-    """根据最新的服务器组刷新命令别名"""
-    commands = {"anne"}
-    commands.update(COMMAND)
-    l4_request.rule = command_rule(*commands)
-
-
-refresh_server_command_rule()
+refresh_server_command_rule(l4_request)
 
 
 async def sync_sb_pages_groups() -> None:
@@ -81,7 +75,7 @@ async def sync_sb_pages_groups() -> None:
     if not pages:
         logger.info("sb_pages.json 为空，跳过启动时刷新")
         reload_ip()
-        refresh_server_command_rule()
+        refresh_server_command_rule(l4_request)
         return
 
     api = L4D2Api()
@@ -96,7 +90,7 @@ async def sync_sb_pages_groups() -> None:
             failed.append(f"{tag}: {exc}")
 
     reload_ip()
-    refresh_server_command_rule()
+    refresh_server_command_rule(l4_request)
 
     if ok:
         logger.success(f"启动时已刷新 {ok} 个服务器组")
@@ -202,15 +196,24 @@ async def _(
                     logger.info(type(out_msg))
                     if config.l4_connect and isinstance(out_msg, bytes):
                         logger.info(f"connect {one['host']}:{one['port']}")
-                        out_msg = UniMessage.image(raw=out_msg) + UniMessage.text(
+                        out_msgs = UniMessage.image(raw=out_msg) + UniMessage.text(
                             f"\nconnect {one['host']}:{one['port']}",
                         )
+                    elif config.l4_connect and isinstance(out_msg, str):
+                        logger.info(f"connect {one['host']}:{one['port']}")
+                        out_msgs = UniMessage.text(out_msg) + UniMessage.text(
+                            f"\nconnect {one['host']}:{one['port']}",
+                        )
+                    elif isinstance(out_msg, str):
+                        out_msgs = UniMessage.text(out_msg)
                     else:
-                        out_msg = UniMessage.text(out_msg)
+                        out_msgs = UniMessage.image(raw=out_msg)
+                    return await out_msg_out(out_msgs)
+        return None
     if len(tag_list) == 2:
         group, name = tag_list
         await UniMessage.text(f"正在查询{group}组").send()
-        out: List[OutServer] = await server_find(command=group, is_img=True)
+        out = cast(List[OutServer], await server_find(command=group, is_img=True))
         out_msg = Gm.no_player
         for one in out:
             for player in one["player"]:
@@ -220,13 +223,22 @@ async def _(
                     logger.info(type(out_msg))
                     if config.l4_connect and isinstance(out_msg, bytes):
                         logger.info(f"connect {one['host']}:{one['port']}")
-                        out_msg = UniMessage.image(raw=out_msg) + UniMessage.text(
+                        out_msgs = UniMessage.image(raw=out_msg) + UniMessage.text(
+                            f"\nconnect {one['host']}:{one['port']}",
+                        )
+                    elif config.l4_connect and isinstance(out_msg, str):
+                        logger.info(f"connect {one['host']}:{one['port']}")
+                        out_msgs = UniMessage.text(out_msg) + UniMessage.text(
                             f"\nconnect {one['host']}:{one['port']}",
                         )
+                    elif isinstance(out_msg, str):
+                        out_msgs = UniMessage.text(out_msg)
                     else:
-                        out_msg = UniMessage.text(out_msg)
+                        out_msgs = UniMessage.image(raw=out_msg)
 
-    return await out_msg_out(out_msg)
+                    return await out_msg_out(out_msgs)
+        return None
+    return None
 
 
 @l4_all.handle()
@@ -243,7 +255,7 @@ async def _(args: Message = CommandArg()):
             await get_ip_server(ip),
             is_connect=config.l4_connect,
             host=host,
-            port=port,
+            port=str(port),
         )
 
 
@@ -258,7 +270,7 @@ async def _(args: Message = CommandArg()):
             logger.info(f"重载{tag}的ip")
             await L4API.get_sourceban(tag, url)
         reload_ip()
-        refresh_server_command_rule()
+        refresh_server_command_rule(l4_request)
         logger.success("重载ip完成")
         await out_msg_out("重载ip完成")
 
@@ -278,7 +290,7 @@ async def _(args: Message = CommandArg()):
 
     await L4API.get_sourceban(arg[0], arg[1])
     reload_ip()
-    refresh_server_command_rule()
+    refresh_server_command_rule(l4_request)
     await UniMessage.text("添加成功\n组名: {arg[0]}\n网址: {arg[1]}").send()
 
 

nonebot_plugin_l4d2_server/commands/server_groups.py

@@ -1,175 +0,0 @@
-# nonebot_plugin_l4d2_server/commands/server_groups.py
-from __future__ import annotations
-
-import ujson as json
-from nonebot import on_command
-from nonebot.adapters.onebot.v11 import Message
-from nonebot.params import CommandArg
-from nonebot_plugin_alconna import UniMessage
-
-from ..__main__ import refresh_server_command_rule
-from ..l4_request import reload_ip
-from ..utils.api.request import L4D2Api
-from ..utils.group_store import (
-    export_all,
-    get_group,
-    list_groups,
-    remove_group,
-    set_group,
-)
-from ..utils.sb_sources import del_page, get_page, load_pages, set_page
-
-
-async def _delete_group_and_page(tag: str) -> tuple[bool, bool]:
-    """删除服务器组文件及对应的 URL 配置"""
-    group_deleted = await remove_group(tag)
-    page_deleted = await del_page(tag)
-    return group_deleted, page_deleted
-
-
-# l4addban <组名> [SourceBans服务器页URL]
-l4_add_ban = on_command("l4addban", aliases={"l4添加服务器组"})
-
-
-@l4_add_ban.handle()
-async def _(args: Message = CommandArg()):
-    text = args.extract_plain_text().strip()
-    if not text:
-        await UniMessage.text("用法：l4addban <组名> [SourceBans服务器页URL]").finish()
-
-    parts = text.split(None, 1)
-    tag = parts[0]
-    url = parts[1] if len(parts) > 1 else None
-
-    # 如果本次传了 URL，则记入 data/L4D2/sb_pages.json
-    if url:
-        await set_page(tag, url)
-
-    page = url or await get_page(tag)
-    if not page:
-        await UniMessage.text(
-            f"未在 data/L4D2/sb_pages.json 找到组“{tag}”的 URL；请执行：l4addban {tag} <URL>",
-        ).finish()
-
-    api = L4D2Api()
-    try:
-        server_list = await api.get_sourceban(tag, page)
-    except Exception as e:
-        await UniMessage.text(f"抓取失败：{e}").finish()
-
-    path = await set_group(tag, server_list)
-    await UniMessage.text(
-        f"✅ 已更新：{path.name}（共 {len(server_list)} 台）",
-    ).finish()
-
-
-# 批量刷新：遍历 sb_pages.json 中所有组
-l4_reload_sb = on_command("l4reloadsb", aliases={"l4刷新服务器组"})
-
-
-@l4_reload_sb.handle()
-async def _():
-    pages = await load_pages()
-    if not pages:
-        await UniMessage.text(
-            "data/L4D2/sb_pages.json 为空，先用：l4addban <组名> <URL>",
-        ).finish()
-
-    api = L4D2Api()
-    ok, fail = 0, []
-    for tag, page in pages.items():
-        try:
-            servers = await api.get_sourceban(tag, page)
-            await set_group(tag, servers)
-            ok += 1
-        except Exception as e:
-            fail.append(f"{tag}: {e}")
-
-    msg = [f"✅ 刷新完成：成功 {ok} 个组。"]
-    if fail:
-        msg.append("❌ 失败：\n" + "\n".join(fail))
-    await UniMessage.text("\n".join(msg)).send()
-    reload_ip()
-    refresh_server_command_rule()
-
-
-# 列出所有组及数量
-l4_list_groups = on_command("l4listgroup", aliases={"l4listgroups", "l4列服务器组"})
-
-
-@l4_list_groups.handle()
-async def _():
-    names = await list_groups()
-    if not names:
-        await UniMessage.text("暂无服务器组。").finish()
-    lines = []
-    for name in names:
-        items = await get_group(name)
-        lines.append(f"{name}（{len(items)}）")
-    await UniMessage.text("现有服务器组：\n" + "\n".join(lines)).finish()
-
-
-# 删除服务器组（即删除 data/L4D2/l4d2/<tag>.json）
-l4_del_group = on_command("l4delgroup", aliases={"l4删除服务器组"})
-
-
-@l4_del_group.handle()
-async def _(args: Message = CommandArg()):
-    tag = args.extract_plain_text().strip()
-    if not tag:
-        await UniMessage.text("用法：l4delgroup <组名>").finish()
-    group_deleted, page_deleted = await _delete_group_and_page(tag)
-
-    if not group_deleted and not page_deleted:
-        await UniMessage.text("未找到该组文件或 URL 记录").finish()
-        return
-
-    msg_parts = []
-    if group_deleted:
-        msg_parts.append("服务器组文件")
-    if page_deleted:
-        msg_parts.append("SourceBans URL")
-    await UniMessage.text("✅ 已删除 " + "、".join(msg_parts)).send()
-    reload_ip()
-    refresh_server_command_rule()
-
-
-# 删除 sb_pages.json 里的 URL 映射（仅删 URL）
-l4_del_page = on_command("l4delpage", aliases={"l4删除服务器页"})
-
-
-@l4_del_page.handle()
-async def _(args: Message = CommandArg()):
-    tag = args.extract_plain_text().strip()
-    if not tag:
-        await UniMessage.text("用法：l4delpage <组名>").finish()
-    ok = await del_page(tag)
-    await UniMessage.text("✅ 已删除" if ok else "未找到该组的 URL").finish()
-
-
-# 导出指定组的 JSON 片段（直接读取 data/L4D2/l4d2/<tag>.json）
-l4_export_group = on_command("l4exportgroup", aliases={"l4导出服务器组"})
-
-
-@l4_export_group.handle()
-async def _(args: Message = CommandArg()):
-    tag = args.extract_plain_text().strip()
-    if not tag:
-        await UniMessage.text("用法：l4exportgroup <组名>").finish()
-    items = await get_group(tag)
-    if not items:
-        await UniMessage.text("未找到该组。").finish()
-    data = {tag: items}
-    await UniMessage.text(json.dumps(data, ensure_ascii=False, indent=4)).finish()
-
-
-# 导出全部组（仅用于查看，组合成一个对象返回，不写入任何聚合文件）
-l4_export_groups = on_command("l4exportgroups", aliases={"l4导出全部服务器组"})
-
-
-@l4_export_groups.handle()
-async def _():
-    data = await export_all()
-    if not data:
-        await UniMessage.text("暂无服务器组。").finish()
-    await UniMessage.text(json.dumps(data, ensure_ascii=False, indent=4)).finish()

nonebot_plugin_l4d2_server/config.py

@@ -1,10 +1,10 @@
 from pathlib import Path
-from typing import List, Set
+from typing import List
 
 from nonebot import get_plugin_config
 from nonebot.adapters.onebot.v11 import GROUP_ADMIN, GROUP_MEMBER, GROUP_OWNER
 from nonebot.log import logger
-from nonebot.permission import SUPERUSER
+from nonebot.permission import SUPERUSER, Permission
 from pydantic import BaseModel, Field, field_validator
 
 # 常量定义
@@ -81,10 +81,10 @@ def update_map_index(self, index: int) -> None:
         """更新地图索引配置"""
         if index < 0:
             raise ValueError("地图索引不能小于0")
-        self._config.map_index = index
+        self.map_index = index
 
     @property
-    def l4_permission_set(self) -> Set[int]:
+    def l4_permission_set(self) -> Permission:
         permissions = {
             1: SUPERUSER,
             2: SUPERUSER | GROUP_OWNER,
@@ -139,15 +139,19 @@ def update(self, **kwargs) -> None:
             if key not in valid_keys:
                 raise ValueError(f"无效的配置项: {key}")
 
-            field_type = ConfigModel[key].type_
+            field_info = ConfigModel.model_fields[key]
+            field_type = field_info.annotation
+            if not field_type:
+                continue
+
             if not isinstance(value, field_type):
                 raise TypeError(f"{key} 必须是 {field_type.__name__} 类型")
 
             setattr(self._config, key, value)
 
         # 验证更新后的配置
         try:
-            self._config = ConfigModel(**self._config.dict())
+            self._config = ConfigModel(**self._config.model_dump())
         except ValueError as e:
             logger.error(f"配置更新失败: {e!s}")