Add id command

This commit is contained in:
2026-07-01 21:47:01 +02:00
parent 62ea12033f
commit 12434ec557
2 changed files with 191 additions and 0 deletions

View File

@@ -73,6 +73,117 @@ def _display_command(command: str) -> str:
return f"..{command}" return f"..{command}"
def _value_or_none(value: Any) -> str:
if value is None or value == "":
return "none"
return str(value)
def _message_text_preview(message: Any, *, limit: int = 180) -> str | None:
text = getattr(message, "text", None) or getattr(message, "caption", None)
if not text:
return None
text = str(text).replace("\n", " ").strip()
if len(text) <= limit:
return text
return text[: limit - 1].rstrip() + "..."
def _topic_id(message: Any) -> Any | None:
topic = getattr(message, "topic", None)
topic_id = getattr(topic, "id", None)
if topic_id is not None:
return topic_id
for field in (
"message_thread_id",
"direct_messages_topic_id",
"reply_to_top_message_id",
):
value = getattr(message, field, None)
if value is not None:
return value
created_topic = getattr(message, "forum_topic_created", None)
return getattr(created_topic, "id", None)
def _topic_title(message: Any) -> str | None:
topic = getattr(message, "topic", None)
title = getattr(topic, "title", None)
if title:
return str(title)
created_topic = getattr(message, "forum_topic_created", None)
title = getattr(created_topic, "title", None)
return str(title) if title else None
def _append_chat_lines(lines: list[str], chat: Any, *, prefix: str = "chat") -> None:
lines.append(f"{prefix}_id: {_value_or_none(getattr(chat, 'id', None))}")
chat_type = getattr(chat, "type", None)
if chat_type is not None:
value = getattr(chat_type, "value", chat_type)
lines.append(f"{prefix}_type: {_value_or_none(value)}")
title = getattr(chat, "title", None)
if title:
lines.append(f"{prefix}_title: {title}")
username = getattr(chat, "username", None)
if username:
lines.append(f"{prefix}_username: @{username}")
def _append_user_lines(lines: list[str], user: Any | None, *, prefix: str = "user") -> None:
lines.append(f"{prefix}_id: {_value_or_none(getattr(user, 'id', None))}")
if user is None:
return
username = getattr(user, "username", None)
if username:
lines.append(f"{prefix}_username: @{username}")
lines.append(f"{prefix}_name: {_user_label(user)}")
def _append_sender_chat_lines(lines: list[str], chat: Any | None, *, prefix: str = "sender_chat") -> None:
if chat is None:
return
lines.append(f"{prefix}_id: {_value_or_none(getattr(chat, 'id', None))}")
title = getattr(chat, "title", None)
if title:
lines.append(f"{prefix}_title: {title}")
username = getattr(chat, "username", None)
if username:
lines.append(f"{prefix}_username: @{username}")
def _format_id_report(ctx: "CommandContext") -> str:
message = ctx.message
lines: list[str] = ["Current message:"]
_append_chat_lines(lines, getattr(message, "chat", None))
lines.append(f"message_id: {_value_or_none(getattr(message, 'id', None))}")
lines.append(f"thread_id: {_value_or_none(_topic_id(message))}")
topic_title = _topic_title(message)
if topic_title:
lines.append(f"thread_title: {topic_title}")
_append_user_lines(lines, getattr(message, "from_user", None), prefix="user")
_append_sender_chat_lines(lines, getattr(message, "sender_chat", None))
reply = getattr(message, "reply_to_message", None)
if reply is not None:
lines.extend(["", "Replied message:"])
lines.append(f"reply_message_id: {_value_or_none(getattr(reply, 'id', None))}")
lines.append(f"reply_thread_id: {_value_or_none(_topic_id(reply))}")
reply_topic_title = _topic_title(reply)
if reply_topic_title:
lines.append(f"reply_thread_title: {reply_topic_title}")
_append_user_lines(lines, getattr(reply, "from_user", None), prefix="reply_user")
_append_sender_chat_lines(lines, getattr(reply, "sender_chat", None), prefix="reply_sender_chat")
media = getattr(reply, "media", None)
if media is not None:
lines.append(f"reply_media: {_value_or_none(getattr(media, 'value', media))}")
preview = _message_text_preview(reply)
if preview:
lines.append(f"reply_text: {preview}")
return "\n".join(lines)
def _chunk_text(text: str, limit: int = 3900) -> list[str]: def _chunk_text(text: str, limit: int = 3900) -> list[str]:
if len(text) <= limit: if len(text) <= limit:
return [text] return [text]
@@ -265,6 +376,16 @@ async def status(ctx: "CommandContext") -> None:
await ctx.reply("\n".join(lines)) await ctx.reply("\n".join(lines))
@command_registry.command(
"id",
"Show chat, thread, user, and replied message IDs",
usage=".id",
category="system",
)
async def id_command(ctx: "CommandContext") -> None:
await ctx.reply(_format_id_report(ctx))
@command_registry.command( @command_registry.command(
"help", "help",
"Show command catalog", "Show command catalog",

70
tests/test_id_command.py Normal file
View File

@@ -0,0 +1,70 @@
from __future__ import annotations
import unittest
from types import SimpleNamespace
from unittest.mock import AsyncMock
from app.commands.builtins import _format_id_report, id_command
class IdCommandTests(unittest.IsolatedAsyncioTestCase):
def _ctx(self):
reply = SimpleNamespace(
id=777,
topic=SimpleNamespace(id=55, title="Deploys"),
from_user=SimpleNamespace(
id=202,
first_name="Reply",
last_name="User",
username="reply_user",
),
sender_chat=None,
media=SimpleNamespace(value="photo"),
text="hello from reply",
caption=None,
)
message = SimpleNamespace(
id=123,
chat=SimpleNamespace(
id=-1001234567890,
type=SimpleNamespace(value="channel"),
title="Build Channel",
username="builds",
),
topic=SimpleNamespace(id=42, title="Main Thread"),
from_user=SimpleNamespace(
id=101,
first_name="Owner",
last_name=None,
username="owner",
),
sender_chat=None,
reply_to_message=reply,
reply_text=AsyncMock(),
)
return SimpleNamespace(message=message, reply=message.reply_text)
async def test_id_report_includes_current_and_replied_message_data(self) -> None:
report = _format_id_report(self._ctx())
self.assertIn("chat_id: -1001234567890", report)
self.assertIn("chat_type: channel", report)
self.assertIn("thread_id: 42", report)
self.assertIn("user_id: 101", report)
self.assertIn("reply_message_id: 777", report)
self.assertIn("reply_thread_id: 55", report)
self.assertIn("reply_user_id: 202", report)
self.assertIn("reply_media: photo", report)
self.assertIn("reply_text: hello from reply", report)
async def test_id_command_replies_with_report(self) -> None:
ctx = self._ctx()
await id_command(ctx)
ctx.reply.assert_awaited_once()
self.assertIn("chat_id: -1001234567890", ctx.reply.await_args.args[0])
if __name__ == "__main__":
unittest.main()