feat(telegram): candidate list/delete/clear commands

私聊新增 /candidate [页码](待决策列表)、/candi_del(标题按钮逐条删除,删后原地重建键盘、可翻页连续删)、/candi_clear(全部/待决策/取消)。重构 handle_callback 为前缀分派(cdel/cpage/cclr,保留 add/ign/blk/fav);repository 加 list_pending_candidates/delete_all_candidates;LibraryService 加 pending_candidates/delete_candidate/clear_candidates 并刷看板缓存。
This commit is contained in:
kairee committed 2026-11-01 20:45:57 +08:00
1 parent c6790f0ba1
commit 4a9c37f2f3
9 files changed
+473 -23

No files matched your search

+152 -2
View File
@@ -11,11 +11,16 @@ sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
sys.stdout.reconfigure(encoding="utf-8")
from reaction_board.notify.telegram import ( # noqa: E402
CANDIDATE_DELETE_HEADER,
HELP_TEXT,
TelegramClient,
TelegramError,
candidate_delete_keyboard,
clear_candidates_keyboard,
format_candidate_list,
format_duration,
format_manual_reply,
truncate_title,
)
from reaction_board.services.library import ( # noqa: E402
ActionResult,
@@ -147,7 +152,9 @@ class FakeBot:
self.fail_answer = fail_answer
self.answers = []
self.sent = []
self.markups = []
self.edits = []
self.edit_markups = []
def get_updates(self, offset=None):
return list(self.updates)
@@ -157,17 +164,21 @@ class FakeBot:
raise RuntimeError("HTTP Error 400: Bad Request")
self.answers.append((callback_id, text))
def send_text(self, text):
def send_text(self, text, reply_markup=None):
self.sent.append(text)
self.markups.append(reply_markup)
def edit_reply_markup(self, message_id):
def edit_reply_markup(self, message_id, reply_markup=None):
self.edits.append(message_id)
self.edit_markups.append(reply_markup)
class FakeActions:
def __init__(self):
self.calls = []
self.manual = []
self.candidates = []
self.clears = []
def ingest_candidate(self, aid):
self.calls.append(("add", aid))
@@ -190,6 +201,27 @@ class FakeActions:
results=[IngestResult(True, 9, "added", title="咻咻满 reaction", message="已入库")],
)
def pending_candidates(self):
return list(self.candidates)
def delete_candidate(self, aid):
self.calls.append(("cdel", aid))
for index, candidate in enumerate(self.candidates):
if candidate["aid"] == aid:
return self.candidates.pop(index)
return None
def clear_candidates(self, *, pending_only=False):
self.clears.append(pending_only)
total = len(self.candidates)
if pending_only:
self.candidates = [
c for c in self.candidates if c.get("decision") is not None
]
return total - len(self.candidates)
self.candidates = []
return total
ADD_REPLY = "✅ 已入库《咻咻满 reaction》(aid 9)"
@@ -266,6 +298,118 @@ def test_answer_failure_falls_back_to_text():
assert bot.edits == [10]
def candidates(n, *, start=1):
return [
{"aid": 100 + i, "title": f"咻咻满 reaction {i}"}
for i in range(start, start + n)
]
def test_truncate_and_keyboard_helpers():
assert truncate_title("短标题") == "短标题"
long_title = "咻咻满" + "反" * 40
out = truncate_title(long_title, 20)
assert len(out) == 20 and out.endswith("…")
assert truncate_title("") == "(无标题)"
items = candidates(12)
text = format_candidate_list(items, 1)
assert "共 12 条 · 第 1/2 页" in text
assert "咻咻满 reaction 1" in text
assert "咻咻满 reaction 11" not in text
assert "第 2/2 页" in format_candidate_list(items, 2)
assert "第 2/2 页" in format_candidate_list(items, 99) # 页码越界夹取到末页
assert format_candidate_list([], 1) == "暂无待决策候选。"
keyboard = candidate_delete_keyboard(items, 1)
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "cdel:1:101"
assert keyboard["inline_keyboard"][-1][0]["callback_data"] == "cpage:del:0"
clear_kb = clear_candidates_keyboard()
assert [b["callback_data"] for b in clear_kb["inline_keyboard"][0]] == [
"cclr:all", "cclr:pending", "cclr:cancel",
]
def test_candidate_list_command_pages():
bot = FakeBot([
message_update(1, "/candidate"),
message_update(2, "/candidate 2"),
message_update(3, "/candidate@re_xxm_bot"),
])
actions = FakeActions()
actions.candidates = candidates(12)
TelegramPoller(bot, actions).run_once(None)
assert "共 12 条 · 第 1/2 页" in bot.sent[0]
assert "第 2/2 页" in bot.sent[1]
assert "第 1/2 页" in bot.sent[2] # @BotName 后缀也能解析
def test_candi_del_keyboard_and_rebuild():
bot = FakeBot([
message_update(1, "/candi_del"),
update(2, "cdel:1:101"),
update(3, "cpage:del:2"),
])
actions = FakeActions()
actions.candidates = candidates(12)
TelegramPoller(bot, actions).run_once(None)
assert bot.sent[0] == CANDIDATE_DELETE_HEADER
keyboard = bot.markups[0]
assert keyboard["inline_keyboard"][0][0]["text"] == "咻咻满 reaction 1"
assert keyboard["inline_keyboard"][0][0]["callback_data"] == "cdel:1:101"
assert ("cdel", 101) in actions.calls
assert bot.answers[0] == ("cb2", "✅ 已删除:咻咻满 reaction 1")
# 删除 101 后重建第一页键盘(新首条 102)
assert bot.edit_markups[0]["inline_keyboard"][0][0]["callback_data"] == "cdel:1:102"
# 翻到第 2 页(剩余 11 条:102..112 → 第 2 页仅 112)
assert bot.answers[1] == ("cb3", "第 2 页")
assert bot.edit_markups[1]["inline_keyboard"][0][0]["callback_data"] == "cdel:2:112"
def test_candi_del_empty():
bot = FakeBot([message_update(1, "/candi_del")])
TelegramPoller(bot, FakeActions()).run_once(None)
assert bot.sent == ["暂无待决策候选。"]
assert bot.markups == [None]
def test_candi_clear_buttons():
bot = FakeBot([
message_update(1, "/candi_clear"),
update(2, "cclr:pending"),
update(3, "cclr:cancel"),
])
actions = FakeActions()
actions.candidates = candidates(3) + [
{"aid": 999, "title": "已决策", "decision": "rejected"}
]
TelegramPoller(bot, actions).run_once(None)
keyboard = bot.markups[0]
assert [b["callback_data"] for b in keyboard["inline_keyboard"][0]] == [
"cclr:all", "cclr:pending", "cclr:cancel",
]
assert actions.clears == [True] # 只清待决策
assert bot.answers[0] == ("cb2", "已清空 3 条")
assert bot.edit_markups[0] == {"inline_keyboard": []}
assert bot.answers[1] == ("cb3", "已取消")
def test_candidate_commands_unauthorized_and_unknown():
bot = FakeBot([message_update(1, "/candidate", chat_id=999)])
actions = FakeActions()
actions.candidates = candidates(3)
TelegramPoller(bot, actions).run_once(None)
assert bot.sent == []
bot2 = FakeBot([message_update(1, "/nope")])
TelegramPoller(bot2, FakeActions()).run_once(None)
assert bot2.sent == [HELP_TEXT]
class Recorder:
def __init__(self):
self.events = []
@@ -322,6 +466,12 @@ def main():
test_poller_dispatch_and_authorization()
test_poller_manual_message()
test_answer_failure_falls_back_to_text()
test_truncate_and_keyboard_helpers()
test_candidate_list_command_pages()
test_candi_del_keyboard_and_rebuild()
test_candi_del_empty()
test_candi_clear_buttons()
test_candidate_commands_unauthorized_and_unknown()
test_poller_backoff_and_recovery()
print("OK: Telegram 消息、分片、长轮询与回调授权测试全部通过")