diff --git a/CHANGELOG.md b/CHANGELOG.md index 915c566..633b64c 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -11,6 +11,12 @@ ## [Unreleased] +### Poke Interaction + +#### Fixed + +- **`poke_reply_enabled` 配置未生效**:戳一戳事件处理现在会检查开关,关闭后不再戳回或发送吐槽文本 + ### Persona Arc — 情感图鉴与离线反刍 #### Added diff --git a/main.py b/main.py index e585561..3e5bb64 100644 --- a/main.py +++ b/main.py @@ -920,7 +920,7 @@ async def on_message_listener(self, event: AstrMessageEvent): sender_id = str(getattr(msg_obj.sender, "user_id", "")) group_id = str(getattr(msg_obj, "group_id", "") or "") bot_id = str(getattr(msg_obj, "self_id", "") or "") - if target_id == bot_id: + if target_id == bot_id and self.cfg.poke_reply_enabled: asyncio.create_task(_poke_reply_async(self, target_id, sender_id, group_id, sender_id)) event.stop_event() return diff --git a/tests/test_config_contract.py b/tests/test_config_contract.py index 43fb6ef..0160064 100644 --- a/tests/test_config_contract.py +++ b/tests/test_config_contract.py @@ -88,6 +88,39 @@ def test_main_dead_helpers_and_constants_removed(self): self.assertNotIn("def _clean_messages(", main_text) self.assertNotIn("def _post_init(", main_text) + def test_poke_reply_toggle_guards_poke_reply_task(self): + import ast + + main_text = (ROOT / "main.py").read_text(encoding="utf-8") + tree = ast.parse(main_text) + + guarded_poke_reply_tasks = [] + for node in ast.walk(tree): + if not isinstance(node, ast.If): + continue + for child in node.body: + if not isinstance(child, ast.Expr) or not isinstance(child.value, ast.Call): + continue + child = child.value + if not ( + isinstance(child.func, ast.Attribute) + and child.func.attr == "create_task" + and isinstance(child.func.value, ast.Name) + and child.func.value.id == "asyncio" + and child.args + and isinstance(child.args[0], ast.Call) + and isinstance(child.args[0].func, ast.Name) + and child.args[0].func.id == "_poke_reply_async" + ): + continue + guarded_poke_reply_tasks.append(node) + + self.assertEqual(len(guarded_poke_reply_tasks), 1) + condition = guarded_poke_reply_tasks[0].test + self.assertIsInstance(condition, ast.BoolOp) + self.assertIsInstance(condition.op, ast.And) + self.assertEqual(ast.unparse(condition), "target_id == bot_id and self.cfg.poke_reply_enabled") + def test_eavesdropping_dead_state_removed(self): eavesdropping_text = (ROOT / "engine" / "eavesdropping.py").read_text(encoding="utf-8") self.assertNotIn("_current_boredom_state", eavesdropping_text)