🚀v1.0.0b1

.pre-commit-config.yaml

@@ -5,10 +5,10 @@ ci:
   autoupdate_branch: main
   autoupdate_schedule: monthly
   autoupdate_commit_msg: ':arrow_up: auto update by pre-commit hooks'
-
+  
 repos:
   - repo: https://github.com/astral-sh/ruff-pre-commit
-    rev: v0.0.291
+    rev: v0.5.6
     hooks:
       - id: ruff
         args: [--fix, --exit-non-zero-on-fix]
@@ -19,7 +19,7 @@ repos:
   #     - id: pyright
 
   - repo: https://github.com/psf/black
-    rev: 23.7.0
+    rev: 24.8.0
     hooks:
       - id: black
         stages: [commit]

nonebot_plugin_l4d2_server/__main__.py

@@ -26,7 +26,7 @@
 
 from .config import config
 from .l4_help import get_l4d2_core_help
-from .l4_local import *  # noqa: F401, F403
+from .l4_local import *  # noqa: F403
 from .l4_request import (
     COMMAND,
     get_all_server_detail,

nonebot_plugin_l4d2_server/config.py

@@ -1,5 +1,4 @@
 from pathlib import Path
-from typing import List
 
 from nonebot import get_plugin_config
 from pydantic import BaseModel
@@ -12,6 +11,9 @@
 
 ICONPATH = DATAPATH / "icon"
 
+global map_index
+map_index = 0
+
 
 class ConfigModel(BaseModel):
     l4_anne: bool = False

nonebot_plugin_l4d2_server/l4_anne/__init__.py

@@ -1,4 +1,5 @@
-from nonebot import log as log, on_command  # noqa: N999
+from nonebot import log as log
+from nonebot import on_command
 from nonebot.adapters import Event, Message
 from nonebot.log import logger
 from nonebot.params import CommandArg

nonebot_plugin_l4d2_server/l4_help/Help.json

@@ -83,6 +83,14 @@
         "need_ck": false,
         "need_sk": false,
         "need_admin": true
+      },      
+      {
+        "name": "l4地图上传",
+        "desc": "进入交互上传地图文件",
+        "eg": "l4地图上传",
+        "need_ck": false,
+        "need_sk": false,
+        "need_admin": true
       }
     ]
   },

nonebot_plugin_l4d2_server/l4_help/__init__.py

@@ -10,7 +10,7 @@
 from ..l4_image.model import PluginHelp
 from .draw import get_help
 
-__version__ = "1.0.0a2"
+__version__ = "1.0.0b1"
 TEXT_PATH = Path(__file__).parent / "texture2d"
 HELP_DATA = Path(__file__).parent / "Help.json"
 

nonebot_plugin_l4d2_server/l4_image/anne_pil.py

@@ -1,16 +1,17 @@
 from pathlib import Path
 
-from nonebot_plugin_l4d2_server.utils.api.models import AnnePlayer2
 from PIL import Image, ImageFont
 
+from nonebot_plugin_l4d2_server.utils.api.models import AnnePlayer2
+
 from ..config import config
 
 font = ImageFont.truetype(config.l4_font)
 
 anne_path = Path(__file__).parent / "img" / "anne"
 
 
-async def anne_player_info(msg: AnnePlayer2):
+async def anne_player_info(msg: AnnePlayer2):  # noqa: RUF029
     back_img = Image.open(anne_path / "back1.jpg")
     base_img = Image.new("RGBA", (back_img.size), (255, 255, 255, 50))
     back_img.paste(base_img, (0, 0), base_img)

nonebot_plugin_l4d2_server/l4_image/convert.py

@@ -21,32 +21,28 @@ def core_font(size: int) -> ImageFont.FreeTypeFont:
 async def convert_img(
     img: Image.Image,
     is_base64: bool = False,
-) -> bytes:
-    ...
+) -> bytes: ...
 
 
 @overload
 async def convert_img(
     img: Image.Image,
-    is_base64: bool = True,
-) -> str:
-    ...
+    is_base64: bool = True,  # noqa: FBT001
+) -> str: ...
 
 
 @overload
 async def convert_img(
     img: bytes,
     is_base64: bool = False,
-) -> str:
-    ...
+) -> str: ...
 
 
 @overload
 async def convert_img(
     img: Path,
     is_base64: bool = False,
-) -> str:
-    ...
+) -> str: ...
 
 
 async def convert_img(
@@ -78,7 +74,7 @@ async def convert_img(
         async with aiofiles.open(img, "rb") as fp:
             img = await fp.read()
 
-    logger.success("图片处理完成！")
+    logger.success("图片处理完成!")
 
     return f"base64://{b64encode(img).decode()}"
 
@@ -90,7 +86,7 @@ def convert_img_sync(img_path: Path):
     return f"base64://{b64encode(img).decode()}"
 
 
-async def str_lenth(r: str, size: int, limit: int = 540) -> str:
+async def str_lenth(r: str, size: int, limit: int = 540) -> str:  # noqa: RUF029
     result = ""
     temp = 0
     for i in r:

nonebot_plugin_l4d2_server/l4_image/download.py

@@ -27,20 +27,22 @@ async def download_url(url: str) -> bytes:
                 logger.warning(f"Error downloading {url}, retry {i}/3: {e}")
                 await asyncio.sleep(3)
 
-    raise Exception(f"{url} 下载失败！")
+    raise Exception(f"{url} 下载失败!")
 
 
 async def download_head(user_id: str) -> bytes:
     url = f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=640"
     data = await download_url(url)
-    if hashlib.md5(data).hexdigest() == "acef72340ac0e914090bd35799f5594e":  # noqa: S324
+    if (
+        hashlib.md5(data).hexdigest() == "acef72340ac0e914090bd35799f5594e"
+    ):  # noqa: S324
         url = f"http://q1.qlogo.cn/g?b=qq&nk={user_id}&s=100"
         data = await download_url(url)
     return data
 
 
 def square_to_circle(im: ImageS):
-    """im是正方形，变圆形"""
+    """im是正方形,变圆形"""
     size = im.size
     mask = Image.new("L", size, 0)
     draw = ImageDraw.Draw(mask)

nonebot_plugin_l4d2_server/l4_image/html_img.py

@@ -23,7 +23,7 @@
 
 async def server_ip_pic(server_dict: List[OutServer]):
     """
-    输入一个字典列表，输出图片
+    输入一个字典列表,输出图片
     msg_dict:folder/name/map_/players/max_players/Players/[Name]
     """
     for server_info in server_dict:
@@ -92,7 +92,7 @@ async def get_server_img(plugins: List[OutServer]) -> Optional[bytes]:
 
 # async def server_group_ip_pic(msg_list: List[ServerGroup]) -> bytes:
 #     """
-#     输入一个群组字典列表，输出图片
+#     输入一个群组字典列表,输出图片
 #     msg_dict:folder/name/map_/players/max_players/Players/[Name]
 #     """
 #     template = env.get_template("group_ip.html")

nonebot_plugin_l4d2_server/l4_image/image_tools.py

@@ -44,7 +44,10 @@ def get_v4_bg(w: int, h: int, is_dark: bool = False, is_blur: bool = False):
     return img.convert("RGBA")
 
 
-async def shift_image_hue(img: Image.Image, angle: float = 30) -> Image.Image:
+async def shift_image_hue(
+    img: Image.Image,
+    angle: float = 30,
+) -> Image.Image:  # noqa: RUF029
     alpha = img.getchannel("A")
     img = img.convert("HSV")
 
@@ -198,7 +201,7 @@ def easy_paste(
     """
     inplace method
     快速粘贴, 自动获取被粘贴图像的坐标。
-    pos应当是粘贴点坐标，direction指定粘贴点方位，例如lt为左上
+    pos应当是粘贴点坐标,direction指定粘贴点方位,例如lt为左上
     """
     x, y = pos
     size_x, size_y = im_paste.size
@@ -379,11 +382,11 @@ def get_bg_color(
         for i in range(color):
             bg = tuple(
                 q.getpalette()[  # type:ignore
-                    i * 3 : (i * 3) + 3  # noqa:E203
+                    i * 3 : (i * 3) + 3
                 ],
             )
             light_value = bg[0] * 0.3 + bg[1] * 0.6 + bg[2] * 0.1
-            if abs(light_value - based_light) < temp:  # noqa:E203
+            if abs(light_value - based_light) < temp:
                 bg_color = bg
                 temp = abs(light_value - based_light)
         return bg_color  # type:ignore

nonebot_plugin_l4d2_server/l4_local/__init__.py

@@ -1,33 +1,42 @@
-
 from pathlib import Path
 
-from nonebot import on_command
+from nonebot.adapters import Event
 from nonebot.log import logger
-from nonebot_plugin_alconna import UniMessage
+from nonebot.matcher import Matcher
+from nonebot_plugin_alconna import File, UniMessage, UniMsg, on_alconna
 
-from ..config import config
+from ..config import config, map_index
 from ..l4_image.convert import text2pic
+from ..utils.utils import mes_list
+from .file import updown_l4d2_vpk
+
+vpk_path = config.l4_local[map_index]
 
 local_path_list = config.l4_local
 if not local_path_list:
-    logger.warning("未填写本地服务器路径,如果想要使用本地服务器功能,请填写本地服务器路径")
+    logger.warning(
+        "未填写本地服务器路径,如果想要使用本地服务器功能,请填写本地服务器路径",
+    )
 else:
     local_path: list[Path] = []
     for folder_path in local_path_list:
-        path = Path(folder_path)  
-        
-        if path.is_dir():  
-            for nextdir in path.iterdir():  
-                # 如果找到了名为left4dead2的目录，返回True  
-                if nextdir.name == "left4dead2" and nextdir.is_dir():  
+        path = Path(folder_path)
+
+        if path.is_dir():
+            for nextdir in path.iterdir():
+                # 如果找到了名为left4dead2的目录,返回True
+                if nextdir.name == "left4dead2" and nextdir.is_dir():
                     local_path.append(nextdir)
             continue
-    logger.info(f"本地服务器路径列表:{local_path}")
-
-    global map_index
-    map_index = 0
+    logger.debug(f"本地服务器路径列表:{local_path}")
+
+    search_map = on_alconna(
+        "l4map",
+        aliases={"l4地图查询", "l4地图"},
+        priority=20,
+        block=True,
+    )
 
-    search_map = on_command("l4map", aliases={"l4地图查询", "l4地图"}, priority=5, block=True)
     @search_map.handle()
     async def _():
         supath = local_path[map_index] / "addons"
@@ -40,7 +49,46 @@ async def _():
                     vpk_list.append(sudir.name)
         if not vpk_list:
             await UniMessage.text("未找到可用的VPK文件").finish()
-        out_msg = "\n".join(f"{index + 1}、{line}" for index, line in enumerate(vpk_list)) 
-
+        out_msg = "\n".join(
+            f"{index + 1}、{line}" for index, line in enumerate(vpk_list)
+        )
+
         img = await text2pic(f"服务器地图:\n{out_msg}")
-        await UniMessage.image(raw=img).send()
+        await UniMessage.image(raw=img).send()
+
+    up = on_alconna(
+        "l4upload",
+        aliases={"l4地图上传"},
+        priority=5,
+        block=True,
+    )
+
+    @up.handle()
+    async def _():
+        await UniMessage.text("请发送地图文件").finish()
+
+    @up.got("map_url", prompt="图来")
+    async def _(ev: Event, msg: UniMsg, matcher: Matcher):
+        if not msg.has(File):
+            await UniMessage.text("不是文件,退出交互").finish()
+        args = ev.dict()
+        if args["notice_type"] != "offline_file":
+            matcher.set_arg("txt", args)  # type: ignore
+            return
+        l4_file_path = config.l4_local[map_index]
+        map_path = Path(l4_file_path, vpk_path)  # type: ignore
+        # 检查下载路径是否存在
+        if not Path(l4_file_path).exists():  # type: ignore
+            await UniMessage.text("你填写的路径不存在辣").finish()
+        if not Path(map_path).exists():
+            await UniMessage.text("这个路径并不是求生服务器的路径,请再看看罢").finish()
+        url: str = args["file"]["url"]
+        name: str = args["file"]["name"]
+        # 如果不符合格式则忽略
+        await up.send("已收到文件,开始下载")
+        vpk_files = await updown_l4d2_vpk(map_path, name, url)
+        if vpk_files:
+            mes = "解压成功,新增以下几个vpk文件"
+            await UniMessage.text(mes_list(mes, vpk_files)).finish()
+        else:
+            await UniMessage.text("你可能上传了相同的文件,或者解压失败了捏").finish()

old_project/l4d2_file/utils.py → nonebot_plugin_l4d2_server/l4_local/file.py

@@ -1,4 +1,5 @@
 import io
+import platform
 import zipfile
 from pathlib import Path
 from typing import Callable, Dict, List
@@ -8,8 +9,9 @@
 from nonebot.log import logger
 from pyunpack import Archive
 
-from ..l4d2_utils.config import systems
-from ..l4d2_utils.utils import get_file, get_vpk
+from ..utils.utils import get_file, get_vpk
+
+systems = platform.system()
 
 
 async def updown_l4d2_vpk(map_paths: Path, name: str, url: str):
@@ -52,9 +54,8 @@ def unpack_rarfile(down_file: Path, down_path: Path):
 def open_packet(name: str, down_file: Path) -> str:
     """解压压缩包"""
     down_path = down_file.parent
-    logger.info("文件名为：" + name)
+    logger.info("文件名为:" + name)
     logger.info(f"系统为{systems}")
-
     if name.endswith(".vpk"):
         return "vpk文件已下载"
 
@@ -79,7 +80,7 @@ def support_gbk(zip_file: ZipFile):
     """
     压缩包中文恢复
     """
-    if type(zip_file) == ZipFile:
+    if isinstance(zip_file, ZipFile):
         name_to_info = zip_file.NameToInfo
         # copy map first
         for name, info in name_to_info.copy().items():
@@ -91,7 +92,7 @@ def support_gbk(zip_file: ZipFile):
     return zip_file
 
 
-async def all_zip_to_one(data_list: List[bytes]):
+async def all_zip_to_one(data_list: List[bytes]):  # noqa: RUF029
     """多压缩包文件合并"""
     file_list = [io.BytesIO(data).getbuffer() for data in data_list]
     data_file = io.BytesIO()

nonebot_plugin_l4d2_server/l4_request/draw_msg.py

@@ -52,7 +52,7 @@ async def get_much_server(server_json: List[NserverOut], command: str):
     out_server: List[OutServer] = []
     search_list: List[Tuple[str, int]] = []
     for i in server_json:
-        search_list.append((i["host"], i["port"]))  # noqa: PERF401
+        search_list.append((i["host"], i["port"]))
 
     all_server = await L4API.a2s_info(search_list, is_player=True)