init

.gitignore

@@ -157,4 +157,4 @@ cython_debug/
 #  be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
 #  and can be added to the global gitignore or merged into this file.  For a more nuclear
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
-#.idea/
+.idea/

__init__.py

@@ -0,0 +1,44 @@
+import logging
+
+from aiohttp import web
+from telethon import TelegramClient
+from telethon.errors import ChannelPrivateError
+
+from .group_data import get_group_data
+
+logger = logging.getLogger('luoxu_plugins.analytics')
+
+
+class GroupAnalyticsHandler:
+    def __init__(self, client):
+        self.client = client
+
+    async def get(self, request: web.Request):
+        if cid_str := request.match_info.get('cid'):
+            try:
+                uid = int(cid_str)
+                await self.client.get_entity(uid)
+            except (ValueError, ChannelPrivateError):
+                raise web.HTTPForbidden(headers={
+                    'Cache-Control': 'public, max-age=86400',
+                })
+            return web.json_response(await get_group_data(uid, self.client))
+        else:
+            raise web.HTTPNotFound
+
+
+async def register(indexer, client: TelegramClient):
+    port: int = int(indexer.config['plugin_analytics']['port'])
+
+    handler = GroupAnalyticsHandler(client)
+
+    app = web.Application()
+    app.router.add_get('/api/group_analytics', handler.get)
+
+    runner = web.AppRunner(app)
+    await runner.setup()
+    site = web.TCPSite(
+        runner,
+        '127.0.0.1', port,
+    )
+    await site.start()

config.toml.example

@@ -0,0 +1,2 @@
+[plugin_analytics]
+port = 9010

group_data.py

@@ -0,0 +1,19 @@
+from typing import List, Dict
+
+from cashews import cache
+from telethon import TelegramClient
+from telethon.tl.functions.stats import GetMegagroupStatsRequest
+from telethon.utils import get_input_channel
+from telethon.tl.types.stats import MegagroupStats
+
+cache.setup("mem://")
+
+
+@cache(ttl="1h", key="{cid}")
+async def get_group_data(cid: int, client: TelegramClient) -> List[Dict]:
+    group = get_input_channel(await client.get_input_entity(cid))
+    result: MegagroupStats = await client(GetMegagroupStatsRequest(
+        channel=group,
+        dark=True
+    ))
+    return [i.to_dict() for i in result.top_posters]

requirements.txt

@@ -0,0 +1,3 @@
+telethon
+aiohttp
+cashews==6.2.0