Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 23 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -207,6 +207,29 @@ skland__background_source = '{"uri": "/imgs/image.jpg"}'
| 萨卡兹肉鸽 | `skland rogue --topic 萨卡兹` |
| 萨米肉鸽 | `skland rogue --topic 萨米` |

#### 🪄 自定义快捷指令

> 该特性依赖于 [Alconna 快捷指令](https://nonebot.dev/docs/best-practice/alconna/command#command%E7%9A%84%E4%BD%BF%E7%94%A8)。自定义指令不带 `COMMAND_START`,若有必要需手动填写
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion: 澄清关于自定义命令的 COMMAND_START 的说明。

“自定义指令不带 COMMAND_START” 一句不太清晰,因为示例别名 /兔兔签到 已包含 /。请说明“自定义指令”指的是快捷别名还是目标命令字符串,并举例说明何时需手动添加 COMMAND_START

Original comment in English

suggestion: Clarify the note about COMMAND_START for custom commands.

“自定义指令不带 COMMAND_START” 一句不太清晰,因为示例别名 /兔兔签到 已包含 /。请说明“自定义指令”指的是快捷别名还是目标命令字符串,并举例说明何时需手动添加 COMMAND_START


```bash
# 增加
/skland --shortcut <自定义指令> /skland
# 删除
/skland --shortcut delete <自定义指令>
# 列出
/skland --shortcut list
```

> [!NOTE]
> 自定义指令中包含空格,需要用引号`""`包裹。

例子:

```bash
user: /skland --shortcut /兔兔签到 "/skland arksign --all"
bot: skland::skland 的快捷指令: "/兔兔签到" 添加成功
```

### 🫣 暗语表

> [!NOTE]
Expand Down
40 changes: 34 additions & 6 deletions nonebot_plugin_skland/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,7 @@
require("nonebot_plugin_alconna")
require("nonebot_plugin_localstore")
require("nonebot_plugin_htmlrender")
from arclet.alconna import config as alc_config
from nonebot_plugin_orm import async_scoped_session
from nonebot_plugin_user import UserSession, get_user
from nonebot_plugin_argot import Text, Argot, Image, ArgotExtension
Expand All @@ -26,6 +27,7 @@
Alconna,
Arparma,
MsgTarget,
Namespace,
Subcommand,
UniMessage,
CommandMeta,
Expand Down Expand Up @@ -63,6 +65,9 @@
},
)

ns = Namespace("skland", disable_builtin_options=set())
alc_config.namespaces["skland"] = ns

skland = on_alconna(
Alconna(
"skland",
Expand Down Expand Up @@ -91,11 +96,16 @@
Args["target?#目标", At | int],
Option(
"-t|--topic|topic",
Args["topic_name?#主题", ["萨米", "萨卡兹"], Field(completion=lambda: "请输入指定topic_id")],
Args[
"topic_name?#主题",
["萨米", "萨卡兹"],
Field(completion=lambda: "请输入指定topic_id"),
],
help_text="指定主题进行肉鸽战绩查询",
),
help_text="肉鸽战绩查询",
),
namespace=alc_config.namespaces["skland"],
meta=CommandMeta(
description=__plugin_meta__.description,
usage=__plugin_meta__.usage,
Expand All @@ -112,7 +122,10 @@
skland.shortcut("森空岛绑定", {"command": "skland bind", "fuzzy": True, "prefix": True})
skland.shortcut("扫码绑定", {"command": "skland qrcode", "fuzzy": False, "prefix": True})
skland.shortcut("明日方舟签到", {"command": "skland arksign --all", "fuzzy": True, "prefix": True})
skland.shortcut("萨卡兹肉鸽", {"command": "skland rogue --topic 萨卡兹", "fuzzy": True, "prefix": True})
skland.shortcut(
"萨卡兹肉鸽",
{"command": "skland rogue --topic 萨卡兹", "fuzzy": True, "prefix": True},
)
skland.shortcut("萨米肉鸽", {"command": "skland rogue --topic 萨米", "fuzzy": True, "prefix": True})
skland.shortcut("角色更新", {"command": "skland char update", "fuzzy": False, "prefix": True})
skland.shortcut("资源更新", {"command": "skland sync", "fuzzy": False, "prefix": True})
Expand Down Expand Up @@ -272,7 +285,12 @@ async def _(


@skland.assign("arksign")
async def _(user_session: UserSession, session: async_scoped_session, uid: Match[str], result: Arparma):
async def _(
user_session: UserSession,
session: async_scoped_session,
uid: Match[str],
result: Arparma,
):
"""明日方舟森空岛签到"""

@refresh_cred_token_if_needed
Expand Down Expand Up @@ -337,7 +355,10 @@ async def _(is_superuser: bool = Depends(SuperUser())):
for route in RESOURCE_ROUTES:
logger.info(f"正在下载: {route}")
await GameResourceDownloader.download_all(
owner="yuanyan3060", repo="ArknightsGameResource", route=route, branch="main"
owner="yuanyan3060",
repo="ArknightsGameResource",
route=route,
branch="main",
)
version = await GameResourceDownloader.get_version()
GameResourceDownloader.update_version_file(version)
Expand All @@ -348,15 +369,22 @@ async def _(is_superuser: bool = Depends(SuperUser())):


@skland.assign("rogue")
async def _(user_session: UserSession, session: async_scoped_session, result: Arparma, target: Match[At | int]):
async def _(
user_session: UserSession,
session: async_scoped_session,
result: Arparma,
target: Match[At | int],
):
"""获取明日方舟肉鸽战绩"""

# Not Finished
@refresh_cred_token_if_needed
@refresh_access_token_if_needed
async def get_rogue_info(user: User, uid: str, topic_id: str):
return await SklandAPI.get_rogue(
CRED(cred=user.cred, token=user.cred_token, userId=str(user.user_id)), uid, topic_id
CRED(cred=user.cred, token=user.cred_token, userId=str(user.user_id)),
uid,
topic_id,
)

if target.available:
Expand Down
28 changes: 24 additions & 4 deletions nonebot_plugin_skland/download.py
Original file line number Diff line number Diff line change
Expand Up @@ -63,7 +63,10 @@ def make_tasks_table(self, tasks: Iterable[Task]) -> Table:
status = self.STATUS_FIN if task.finished else self.STATUS_DL
itable = Table.grid(padding=(0, 1), expand=self.expand)
filename_column = Text(f"{task.fields['filename']}")
itable.add_row(filename_column, *(column(task) for column in [status, *self.STATUS_ROW]))
itable.add_row(
filename_column,
*(column(task) for column in [status, *self.STATUS_ROW]),
)
itable.add_row(*(column(task) for column in self.PROG_ROW))
tasks_table.add_row(itable)
if not task.finished:
Expand All @@ -75,7 +78,14 @@ def make_tasks_table(self, tasks: Iterable[Task]) -> Table:
if all_tasks_finished:
return table
else:
table.add_row(Panel(tasks_table, title="Downloading Files", title_align="left", padding=(1, 2)))
table.add_row(
Panel(
tasks_table,
title="Downloading Files",
title_align="left",
padding=(1, 2),
)
)

return table

Expand Down Expand Up @@ -134,7 +144,10 @@ async def fetch_file_list(cls, url: str, dl_url: str, route: str) -> list[File]:
data = response.json()
route = route.rstrip("/") + "/"
files = [
File(name=item["path"].split("/")[-1], download_url=f"{dl_url}{item['path']}")
File(
name=item["path"].split("/")[-1],
download_url=f"{dl_url}{item['path']}",
)
for item in data.get("tree", [])
if item["type"] == "blob" and item["path"].startswith(route)
]
Expand Down Expand Up @@ -168,7 +181,14 @@ async def worker(file: File):
return
async with cls.SEMAPHORE:
task_id = progress.add_task("Downloading", filename=file.name, total=0)
await cls.download_file(client, file, save_path, progress, task_id=task_id, timeout=300)
await cls.download_file(
client,
file,
save_path,
progress,
task_id=task_id,
timeout=300,
)
progress.remove_task(task_id)
cls.download_count += 1

Expand Down
17 changes: 15 additions & 2 deletions nonebot_plugin_skland/hook.py
Original file line number Diff line number Diff line change
@@ -1,26 +1,39 @@
from nonebot import logger, get_driver
from nonebot_plugin_alconna import command_manager

from .exception import RequestException
from .config import RESOURCE_ROUTES, config
from .download import GameResourceDownloader
from .config import CACHE_DIR, RESOURCE_ROUTES, config

driver = get_driver()
shortcut_cache = CACHE_DIR / "shortcut.db"


@driver.on_startup
async def startup():
command_manager.load_cache(shortcut_cache)
logger.debug("Skland shortcuts cache loaded")
Comment on lines +14 to +15
Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

suggestion (bug_risk): Cache operations on startup are clear.

将启动时的缓存加载和转储操作包装在 try/except 块中,以捕获启动或关闭期间的 IO 错误。

Suggested implementation:

@driver.on_startup
async def startup():
    try:
        command_manager.load_cache(shortcut_cache)
        logger.debug("Skland shortcuts cache loaded")
    except Exception as e:
        logger.error("Failed to load shortcut cache on startup", exc_info=e)

    if config.check_res_update:
        try:
            if version := await GameResourceDownloader.check_update():
                for route in RESOURCE_ROUTES:
                    logger.info(f"正在下载: {route}")
                    await GameResourceDownloader.download_all(
                        owner="yuanyan3060",
                        repo="ArknightsGameResource",
                        route=route,
                        branch="main",
                    )
        except Exception as e:
            logger.error("Error during resource update", exc_info=e)

If there is a cache dump operation in your shutdown function (for example, within a @driver.on_shutdown decorated function), Wrap that operation in a similar try/except block to catch and log any IO errors during shutdown.

Original comment in English

suggestion (bug_risk): Cache operations on startup are clear.

Wrap cache load and dump in a try/except to catch IO errors during startup or shutdown.

Suggested implementation:

@driver.on_startup
async def startup():
    try:
        command_manager.load_cache(shortcut_cache)
        logger.debug("Skland shortcuts cache loaded")
    except Exception as e:
        logger.error("Failed to load shortcut cache on startup", exc_info=e)

    if config.check_res_update:
        try:
            if version := await GameResourceDownloader.check_update():
                for route in RESOURCE_ROUTES:
                    logger.info(f"正在下载: {route}")
                    await GameResourceDownloader.download_all(
                        owner="yuanyan3060",
                        repo="ArknightsGameResource",
                        route=route,
                        branch="main",
                    )
        except Exception as e:
            logger.error("Error during resource update", exc_info=e)

If there is a cache dump operation in your shutdown function (for example, within a @driver.on_shutdown decorated function), Wrap that operation in a similar try/except block to catch and log any IO errors during shutdown.

if config.check_res_update:
try:
if version := await GameResourceDownloader.check_update():
logger.info("开始下载游戏资源")
for route in RESOURCE_ROUTES:
logger.info(f"正在下载: {route}")
await GameResourceDownloader.download_all(
owner="yuanyan3060", repo="ArknightsGameResource", route=route, branch="main"
owner="yuanyan3060",
repo="ArknightsGameResource",
route=route,
branch="main",
)
except RequestException as e:
logger.error(f"下载游戏资源失败: {e}")
return
if version:
GameResourceDownloader.update_version_file(version)
logger.success(f"游戏资源已更新到版本:{version}")


@driver.on_shutdown
async def shutdown():
command_manager.dump_cache(shortcut_cache)
logger.debug("Skland shortcuts cache dumped")