session 注销/查询已登录的会话

list.json

@@ -329,6 +329,16 @@
             "supported": true,
             "des-short": "仅联系人可以发送语音消息给我。",
             "des": "仅联系人可以发送语音消息给我。"
+        },
+        {
+            "name": "session",
+            "version": "1.0",
+            "section": "profile",
+            "maintainer": "xtaodada",
+            "size": "4.36 kb",
+            "supported": true,
+            "des-short": "注销/查询已登录的会话。",
+            "des": "注销/查询已登录的会话。\n指令：,session"
         }
     ]
 }

session/main.py

@@ -0,0 +1,104 @@
+import contextlib
+
+from typing import List, Optional
+
+from datetime import datetime
+
+from pyrogram.enums import ChatType
+from pyrogram.raw.functions.account import GetAuthorizations, ResetAuthorization
+from pyrogram.raw.types import Authorization
+
+from pagermaid.listener import listener
+from pagermaid.single_utils import Message
+from pagermaid.utils import alias_command
+from pagermaid import bot
+
+
+async def get_all_session() -> List[Authorization]:
+    data = await bot.invoke(GetAuthorizations())
+    return data.authorizations
+
+
+async def filter_session(hash_start: str) -> Optional[Authorization]:
+    try:
+        hash_start = int(hash_start)
+        if len(str(hash_start)) != 6 and hash_start != 0:
+            return None
+    except ValueError:
+        return None
+    return next((session for session in await get_all_session() if str(session.hash).startswith(str(hash_start))), None)
+
+
+async def kick_session(session: Authorization) -> bool:
+    if session.hash != 0:
+        with contextlib.suppress(Exception):
+            return await bot.invoke(ResetAuthorization(hash=session.hash))
+    return False
+
+
+def format_timestamp(timestamp: int) -> str:
+    datetime_obj = datetime.fromtimestamp(timestamp)
+    return datetime_obj.strftime("%Y-%m-%d %H:%M:%S")
+
+
+def format_session(session: Authorization, private: bool = True) -> str:
+    text = f"标识符：<code>{str(session.hash)[:6]}</code>\n" \
+           f"设备型号：<code>{session.device_model}</code>\n" \
+           f"设备平台：<code>{session.platform}</code>\n" \
+           f"系统版本：<code>{session.system_version}</code>\n" \
+           f"应用名称：<code>{session.app_name}</code>\n" \
+           f"应用版本：<code>{session.app_version}</code>\n" \
+           f"官方应用：<code>{'是' if session.official_app else '否'}</code>\n" \
+           f"登录时间：<code>{format_timestamp(session.date_created)}</code>\n" \
+           f"在线时间：<code>{format_timestamp(session.date_active)}</code>"
+    if private:
+        text += f"\nIP：<code>{session.ip}</code>\n" \
+                f"地理位置：<code>{session.country}</code>"
+    if session.hash != 0:
+        text += f"\n\n使用命令 <code>,{alias_command('session')} 注销 {str(session.hash)[:6]}</code> 可以注销此会话。"
+    return text
+
+
+async def count_platform(private: bool = True) -> str:
+    sessions = await get_all_session()
+    if not sessions:
+        return "无任何在线设备？"
+    platform_count = {}
+    text = f"共有 {len(sessions)} 台设备在线，分别是：\n\n"
+    for session in sessions:
+        if session.platform in platform_count:
+            platform_count[session.platform] += 1
+        else:
+            platform_count[session.platform] = 1
+        text += f"<code>{str(session.hash)[:6]}</code> - <code>{session.device_model}</code>"
+        if private:
+            text += f" - <code>{session.app_name}</code>"
+        text += f"\n"
+    text += "\n"
+    text += "\n".join(f"{platform}：{count} 台" for platform, count in platform_count.items())
+    return text
+
+
+@listener(command="session",
+          need_admin=True,
+          parameters="注销/查询",
+          description="注销/查询已登录的会话")
+async def session_manage(message: Message):
+    if not message.arguments:
+        return await message.edit(await count_platform(private=message.chat.type in [ChatType.PRIVATE, ChatType.BOT]))
+    if len(message.parameter) != 2:
+        return await message.edit_text("请输入 `注销/查询 标识符` 来查询或注销会话")
+    if message.parameter[0] == "查询":
+        session = await filter_session(message.parameter[1])
+        if session:
+            return await message.edit(format_session(
+                session,
+                private=message.chat.type in [ChatType.PRIVATE, ChatType.BOT]))
+        return await message.edit_text("请输入正确的标识符！")
+    if message.parameter[0] == "注销":
+        session = await filter_session(message.parameter[1])
+        if session:
+            success = await kick_session(session)
+            return await message.edit("注销成功！" if success else "注销失败！")
+        return await message.edit_text("请输入正确的标识符！")
+    return await message.edit_text("请输入 `注销/查询 标识符` 来查询或注销会话")