82 lines
1.9 KiB
Python
82 lines
1.9 KiB
Python
|
import asyncio
|
||
|
|
import random
|
||
|
|
from PIL import Image
|
||
|
|
from io import BytesIO
|
||
|
|
|
||
|
|
from telethon import TelegramClient
|
||
|
|
from telethon.tl.functions.messages import GetAllStickersRequest, GetStickerSetRequest
|
||
|
|
from telethon.tl.types import InputStickerSetID
|
||
|
|
from aiohttp import web
|
||
|
|
|
||
|
|
loop = asyncio.get_event_loop()
|
||
|
|
|
||
|
|
api_id = 0 # Put you api id here
|
||
|
|
api_hash = '' # Put your api hash here
|
||
|
|
client = TelegramClient('devenv', api_id, api_hash, loop=loop)
|
||
|
|
|
||
|
|
|
||
|
|
async def get_sticker():
|
||
|
|
sticker_sets = await client(GetAllStickersRequest(0))
|
||
|
|
sticker_set = random.choice(sticker_sets.sets)
|
||
|
|
stickers = await client(
|
||
|
|
GetStickerSetRequest(
|
||
|
|
stickerset=InputStickerSetID(
|
||
|
|
id=sticker_set.id,
|
||
|
|
access_hash=sticker_set.access_hash
|
||
|
|
)
|
||
|
|
)
|
||
|
|
)
|
||
|
|
|
||
|
|
sticker = BytesIO()
|
||
|
|
await client.download_media(random.choice(stickers.documents), file=sticker)
|
||
|
|
|
||
|
|
return sticker
|
||
|
|
|
||
|
|
|
||
|
|
async def get_handler(request):
|
||
|
|
sticker = Image.open(
|
||
|
|
await get_sticker()
|
||
|
|
).convert('RGBA')
|
||
|
|
|
||
|
|
png = BytesIO()
|
||
|
|
sticker.save(png, 'png')
|
||
|
|
png.seek(0)
|
||
|
|
|
||
|
|
return web.Response(body=png.read(), content_type='image/png')
|
||
|
|
|
||
|
|
|
||
|
|
async def start_site(app, address='0.0.0.0', port=8080):
|
||
|
|
runner = web.AppRunner(app)
|
||
|
|
await runner.setup()
|
||
|
|
site = web.TCPSite(runner, address, port)
|
||
|
|
await site.start()
|
||
|
|
return runner
|
||
|
|
|
||
|
|
|
||
|
|
async def main():
|
||
|
|
await client.start()
|
||
|
|
|
||
|
|
web_app = web.Application()
|
||
|
|
web_app.add_routes(
|
||
|
|
[
|
||
|
|
web.get('/', get_handler)
|
||
|
|
]
|
||
|
|
)
|
||
|
|
|
||
|
|
runner = await start_site(web_app)
|
||
|
|
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
await asyncio.sleep(0.01)
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
pass
|
||
|
|
|
||
|
|
await runner.cleanup()
|
||
|
|
|
||
|
|
if __name__ == '__main__':
|
||
|
|
try:
|
||
|
|
loop.run_until_complete(main())
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
pass
|
||
|
|
loop.close()
|