mttu-developers/konabot
8
2
Fork 3
Code Issues 13 Pull Requests 1 Actions Packages Releases 79 Activity

konaph 接入权限系统
Some checks failed
continuous-integration/drone/push Build is failing
Details

Browse Source
This commit is contained in:
钟晓帕
2026-03-07 18:15:28 +08:00
parent 870e2383d8
commit f5b81319f8
1 changed files with 157 additions and 102 deletions
Download Patch File Download Diff File Expand all files Collapse all files

259
konabot/plugins/kona_ph/manager.py
View File

@ -2,22 +2,39 @@ import datetime
from math import ceil
from nonebot import get_plugin_config
from nonebot_plugin_alconna import (Alconna, Args, Image, Option, Query,
Subcommand, SubcommandResult, UniMessage,
on_alconna)
from nonebot.adapters import Event
from nonebot_plugin_alconna import (
Alconna,
Args,
Image,
Option,
Query,
Subcommand,
SubcommandResult,
UniMessage,
on_alconna,
)
from pydantic import BaseModel
from konabot.common.longtask import DepLongTaskTarget
from konabot.common.nb.exc import BotExceptionMessage
from konabot.common.nb.extract_image import download_image_bytes
from konabot.common.permsys import DepPermManager, require_permission
from konabot.common.username import get_username
from konabot.plugins.kona_ph.core.image import get_image_manager
from konabot.plugins.kona_ph.core.message import (get_puzzle_description, get_puzzle_hint_list,
get_puzzle_info_message,
get_submission_message)
from konabot.plugins.kona_ph.core.storage import (Puzzle, PuzzleHint, PuzzleManager,
get_today_date,
puzzle_manager)
from konabot.plugins.kona_ph.core.message import (
get_puzzle_description,
get_puzzle_hint_list,
get_puzzle_info_message,
get_submission_message,
)
from konabot.plugins.kona_ph.core.storage import (
Puzzle,
PuzzleHint,
PuzzleManager,
get_today_date,
puzzle_manager,
)
from konabot.plugins.poster.service import broadcast
PUZZLE_PAGE_SIZE = 10
@ -32,19 +49,17 @@ class PuzzleConfig(BaseModel):
config = get_plugin_config(PuzzleConfig)
def is_puzzle_manager(target: DepLongTaskTarget):
return target.target_id in config.plugin_puzzle_manager or is_puzzle_admin(target)
def is_puzzle_admin(target: DepLongTaskTarget):
return target.target_id in config.plugin_puzzle_admin
def check_puzzle(manager: PuzzleManager, target: DepLongTaskTarget, raw_id: str) -> Puzzle:
async def check_puzzle(
manager: PuzzleManager,
perm: DepPermManager,
raw_id: str,
event: Event,
target: DepLongTaskTarget,
) -> Puzzle:
if raw_id not in manager.puzzle_data:
raise BotExceptionMessage("没有这个谜题")
puzzle = manager.puzzle_data[raw_id]
if is_puzzle_admin(target):
if await perm.check_has_permission(event, "konaph.admin"):
return puzzle
if target.target_id != puzzle.author_id:
raise BotExceptionMessage("你没有权限查看或编辑这个谜题")
@ -60,7 +75,9 @@ def create_admin_commands():
Subcommand("unready", Args["raw_id", str], dest="unready"),
Subcommand("info", Args["raw_id", str], dest="info"),
Subcommand("my", Args["page?", int], dest="my"),
Subcommand("all", Option("--ready", alias=["-r"]), Args["page?", int], dest="all"),
Subcommand(
"all", Option("--ready", alias=["-r"]), Args["page?", int], dest="all"
),
Subcommand("pin", Args["raw_id?", str], dest="pin"),
Subcommand("unpin", dest="unpin"),
Subcommand(
@ -115,11 +132,11 @@ def create_admin_commands():
dest="hint",
),
),
rule=is_puzzle_manager,
rule=require_permission("konaph.manager"),
)
@cmd_admin.assign("$main")
async def _(target: DepLongTaskTarget):
async def _(target: DepLongTaskTarget, pm: DepPermManager, event: Event):
msg = UniMessage.text("==== [KonaPH] 指令一览 ====\n\n")
msg = msg.text("konaph create - 创建一个新的谜题\n")
msg = msg.text("konaph ready <id> - 准备好一道谜题\n")
@ -132,7 +149,7 @@ def create_admin_commands():
msg = msg.text("konaph test <id> <answer> - 尝试提交一个答案，看回答的效果\n")
msg = msg.text("konaph hint - 查看如何编辑题目的中间答案\n")
if is_puzzle_admin(target):
if await pm.check_has_permission(event, "konaph.admin"):
msg = msg.text("konaph all [--ready] <page?> - 查看所有谜题\n")
msg = msg.text("konaph pin - 查看当前置顶谜题\n")
msg = msg.text("konaph pin <id> - 置顶一个谜题\n")
@ -145,48 +162,54 @@ def create_admin_commands():
async def _(target: DepLongTaskTarget):
async with puzzle_manager() as manager:
puzzle = manager.admin_create_puzzle(target.target_id)
await target.send_message(UniMessage.text(
f"✨ 创建好啦！谜题 ID 为 {puzzle.raw_id}\n\n"
f"- 输入 `konaph info {puzzle.raw_id}` 获得谜题的信息\n"
f"- 输入 `konaph my` 查看你创建的谜题\n"
f"- 输入 `konaph modify` 查看更改谜题的方法"
))
await target.send_message(
UniMessage.text(
f"✨ 创建好啦！谜题 ID 为 {puzzle.raw_id}\n\n"
f"- 输入 `konaph info {puzzle.raw_id}` 获得谜题的信息\n"
f"- 输入 `konaph my` 查看你创建的谜题\n"
f"- 输入 `konaph modify` 查看更改谜题的方法"
)
)
@cmd_admin.assign("ready")
async def _(raw_id: str, target: DepLongTaskTarget):
async def _(
raw_id: str, target: DepLongTaskTarget, event: Event, perm: DepPermManager
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
if p.ready:
return await target.send_message(UniMessage.text(
"题目早就准备好啦！"
))
return await target.send_message(UniMessage.text("题目早就准备好啦！"))
p.ready = True
await target.send_message(UniMessage.text(
f"谜题「{p.title}」已经准备就绪！"
))
await target.send_message(
UniMessage.text(f"谜题「{p.title}」已经准备就绪！")
)
@cmd_admin.assign("unready")
async def _(raw_id: str, target: DepLongTaskTarget):
async def _(
raw_id: str, target: DepLongTaskTarget, event: Event, perm: DepPermManager
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
if not p.ready:
return await target.send_message(UniMessage.text(
f"谜题「{p.title}」已经是未取消状态了！"
))
return await target.send_message(
UniMessage.text(f"谜题「{p.title}」已经是未取消状态了！")
)
if manager.is_puzzle_published(p.raw_id):
return await target.send_message(UniMessage.text(
"已发布的谜题不能取消准备状态！"
))
return await target.send_message(
UniMessage.text("已发布的谜题不能取消准备状态！")
)
p.ready = False
await target.send_message(UniMessage.text(
f"谜题「{p.title}」已经取消准备！"
))
await target.send_message(
UniMessage.text(f"谜题「{p.title}」已经取消准备！")
)
@cmd_admin.assign("info")
async def _(raw_id: str, target: DepLongTaskTarget):
async def _(
raw_id: str, target: DepLongTaskTarget, event: Event, perm: DepPermManager
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
await target.send_message(get_puzzle_info_message(manager, p))
@cmd_admin.assign("my")
@ -194,15 +217,15 @@ def create_admin_commands():
async with puzzle_manager() as manager:
puzzles = manager.get_puzzles_of_user(target.target_id)
if len(puzzles) == 0:
return await target.send_message(UniMessage.text(
"你没有谜题哦，使用 `konaph create` 创建一个吧！"
))
return await target.send_message(
UniMessage.text("你没有谜题哦，使用 `konaph create` 创建一个吧！")
)
count_pages = ceil(len(puzzles) / PUZZLE_PAGE_SIZE)
if page <= 0 or page > count_pages:
return await target.send_message(UniMessage.text(
f"页数只有 1 ~ {count_pages} 啦！"
))
puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE: page * PUZZLE_PAGE_SIZE]
return await target.send_message(
UniMessage.text(f"页数只有 1 ~ {count_pages} 啦！")
)
puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE : page * PUZZLE_PAGE_SIZE]
message = UniMessage.text("==== 我的谜题 ====\n\n")
for p in puzzles:
message = message.text("- ")
@ -220,11 +243,15 @@ def create_admin_commands():
await target.send_message(message)
@cmd_admin.assign("all")
async def _(target: DepLongTaskTarget, ready: Query[bool] = Query("all.ready"), page: int = 1):
if not is_puzzle_admin(target):
return await target.send_message(UniMessage.text(
"你没有权限使用该指令"
))
async def _(
target: DepLongTaskTarget,
event: Event,
perm: DepPermManager,
ready: Query[bool] = Query("all.ready"),
page: int = 1,
):
if not perm.check_has_permission(event, "konaph.admin"):
return await target.send_message(UniMessage.text("你没有权限使用该指令"))
async with puzzle_manager() as manager:
puzzles = [*manager.puzzle_data.values()]
if ready.available:
@ -232,10 +259,10 @@ def create_admin_commands():
puzzles = sorted(puzzles, key=lambda p: p.created_at, reverse=True)
count_pages = ceil(len(puzzles) / PUZZLE_PAGE_SIZE)
if page <= 0 or page > count_pages:
return await target.send_message(UniMessage.text(
f"页数只有 1 ~ {count_pages} 啦！"
))
puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE: page * PUZZLE_PAGE_SIZE]
return await target.send_message(
UniMessage.text(f"页数只有 1 ~ {count_pages} 啦！")
)
puzzles = puzzles[(page - 1) * PUZZLE_PAGE_SIZE : page * PUZZLE_PAGE_SIZE]
message = UniMessage.text("==== 所有谜题 ====\n\n")
for p in puzzles:
message = message.text("- ")
@ -253,32 +280,30 @@ def create_admin_commands():
await target.send_message(message)
@cmd_admin.assign("pin")
async def _(target: DepLongTaskTarget, raw_id: str = ""):
if not is_puzzle_admin(target):
return await target.send_message(UniMessage.text(
"你没有权限使用该指令"
))
async def _(
target: DepLongTaskTarget, event: Event, perm: DepPermManager, raw_id: str = ""
):
if not perm.check_has_permission(event, "konaph.admin"):
return await target.send_message(UniMessage.text("你没有权限使用该指令"))
async with puzzle_manager() as manager:
if raw_id == "":
if manager.puzzle_pinned:
return await target.send_message(UniMessage.text(
f"被 Pin 的谜题 ID = {manager.puzzle_pinned}"
))
return await target.send_message(
UniMessage.text(f"被 Pin 的谜题 ID = {manager.puzzle_pinned}")
)
return await target.send_message("没有置顶谜题")
if raw_id not in manager.unpublished_puzzles:
return await target.send_message(UniMessage.text(
"这个谜题已经发布了，或者还没准备好，或者不存在"
))
return await target.send_message(
UniMessage.text("这个谜题已经发布了，或者还没准备好，或者不存在")
)
manager.admin_pin_puzzle(raw_id)
return await target.send_message(f"已置顶谜题 {raw_id}")
@cmd_admin.assign("unpin")
async def _(target: DepLongTaskTarget):
if not is_puzzle_admin(target):
return await target.send_message(UniMessage.text(
"你没有权限使用该指令"
))
async def _(target: DepLongTaskTarget, event: Event, perm: DepPermManager):
if not perm.check_has_permission(event, "konaph.admin"):
return await target.send_message(UniMessage.text("你没有权限使用该指令"))
async with puzzle_manager() as manager:
manager.admin_pin_puzzle("")
return await target.send_message("已取消所有置顶")
@ -286,6 +311,8 @@ def create_admin_commands():
@cmd_admin.assign("modify")
async def _(
target: DepLongTaskTarget,
event: Event,
perm: DepPermManager,
raw_id: str = "",
title: str | None = None,
description: str | None = None,
@ -306,7 +333,7 @@ def create_admin_commands():
image_manager = get_image_manager()
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
if title is not None:
p.title = title
if description is not None:
@ -329,11 +356,14 @@ def create_admin_commands():
return await target.send_message("修改好啦！看看效果：\n\n" + info2)
@cmd_admin.assign("publish")
async def _(target: DepLongTaskTarget, raw_id: str | None = None):
if not is_puzzle_admin(target):
return await target.send_message(UniMessage.text(
"你没有权限使用该指令"
))
async def _(
target: DepLongTaskTarget,
event: Event,
perm: DepPermManager,
raw_id: str | None = None,
):
if not perm.check_has_permission(event, "konaph.admin"):
return await target.send_message(UniMessage.text("你没有权限使用该指令"))
today = get_today_date()
async with puzzle_manager() as manager:
if today in manager.daily_puzzle_of_date:
@ -348,46 +378,64 @@ def create_admin_commands():
return await target.send_message("Ok!")
@cmd_admin.assign("preview")
async def _(target: DepLongTaskTarget, raw_id: str):
async def _(
target: DepLongTaskTarget, event: Event, perm: DepPermManager, raw_id: str
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
return await target.send_message(get_puzzle_description(p))
@cmd_admin.assign("get-submits")
async def _(target: DepLongTaskTarget, raw_id: str):
async def _(
target: DepLongTaskTarget, event: Event, perm: DepPermManager, raw_id: str
):
async with puzzle_manager() as manager:
puzzle = manager.puzzle_data.get(raw_id)
if puzzle is None:
return await target.send_message("没有这个谜题")
if not is_puzzle_admin(target) and target.target_id != puzzle.author_id:
if (
not perm.check_has_permission(event, "konaph.admin")
and target.target_id != puzzle.author_id
):
return await target.send_message("你没有权限预览这个谜题")
msg = UniMessage.text(f"==== {puzzle.title} 提交记录 ====\n\n")
submits = manager.submissions.get(raw_id, {})
for uid, ls in submits.items():
s = ', '.join((i.flag for i in ls))
s = ", ".join((i.flag for i in ls))
msg = msg.text(f"- {get_username(uid)}：{s}\n")
return await target.send_message(msg)
@cmd_admin.assign("test")
async def _(target: DepLongTaskTarget, raw_id: str, submission: str):
async def _(
target: DepLongTaskTarget,
raw_id: str,
submission: str,
event: Event,
perm: DepPermManager,
):
"""
测试一道谜题的回答，并给出结果
"""
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
result = p.check_submission(submission)
msg = get_submission_message(p, result)
return await target.send_message("[测试提交] " + msg)
@cmd_admin.assign("subcommands.hint")
async def _(target: DepLongTaskTarget, subcommands: Query[SubcommandResult] = Query("subcommands.hint")):
async def _(
target: DepLongTaskTarget,
subcommands: Query[SubcommandResult] = Query("subcommands.hint"),
):
if len(subcommands.result.subcommands) > 0:
return
return await target.send_message(
UniMessage.text("==== 提示/中间答案编辑器 ====\n\n")
.text("- konaph hint list <id>\n - 查看某道题的所有提示 / 中间答案\n")
.text("- konaph hint add <id> <pattern> <hint>\n - 添加一个提示 / 中间答案\n")
.text(
"- konaph hint add <id> <pattern> <hint>\n - 添加一个提示 / 中间答案\n"
)
.text("- konaph hint modify <id> <hint_id>\n")
.text(" - --pattern <pattern>\n - 更改匹配规则\n")
.text(" - --message <message>\n - 更改提示文本\n")
@ -402,9 +450,11 @@ def create_admin_commands():
raw_id: str,
pattern: str,
message: str,
event: Event,
perm: DepPermManager,
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
p.hints[p.hint_id_max + 1] = PuzzleHint(
pattern=pattern,
message=message,
@ -416,9 +466,11 @@ def create_admin_commands():
async def _(
target: DepLongTaskTarget,
raw_id: str,
event: Event,
perm: DepPermManager,
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
await target.send_message(get_puzzle_hint_list(p))
@cmd_admin.assign("subcommands.hint.modify")
@ -426,12 +478,14 @@ def create_admin_commands():
target: DepLongTaskTarget,
raw_id: str,
hint_id: int,
event: Event,
perm: DepPermManager,
pattern: str | None = None,
message: str | None = None,
is_checkpoint: bool | None = None,
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
if hint_id not in p.hints:
raise BotExceptionMessage(
f"没有这个 hint_id。请使用 konaph hint list {raw_id} 了解 hint 清单"
@ -450,9 +504,11 @@ def create_admin_commands():
target: DepLongTaskTarget,
raw_id: str,
hint_id: int,
event: Event,
perm: DepPermManager,
):
async with puzzle_manager() as manager:
p = check_puzzle(manager, target, raw_id)
p = await check_puzzle(manager, perm, raw_id, event, target)
if hint_id not in p.hints:
raise BotExceptionMessage(
f"没有这个 hint_id。请使用 konaph hint list {raw_id} 了解 hint 清单"
@ -460,5 +516,4 @@ def create_admin_commands():
del p.hints[hint_id]
await target.send_message("删除成功！\n\n" + get_puzzle_hint_list(p))
return cmd_admin