diff --git a/.gitignore b/.gitignore index 68bc17f..2dc53ca 100644 --- a/.gitignore +++ b/.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/ diff --git a/__init__.py b/__init__.py new file mode 100644 index 0000000..ed76dc7 --- /dev/null +++ b/__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() diff --git a/config.toml.example b/config.toml.example new file mode 100644 index 0000000..587a712 --- /dev/null +++ b/config.toml.example @@ -0,0 +1,2 @@ +[plugin_analytics] +port = 9010 diff --git a/group_data.py b/group_data.py new file mode 100644 index 0000000..6a90ed1 --- /dev/null +++ b/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] diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..6e3b181 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,3 @@ +telethon +aiohttp +cashews==6.2.0