(client: Client, log: SessionLogger)
| 14 | |
| 15 | |
| 16 | async def get_tg_web_data(client: Client, log: SessionLogger) -> str: |
| 17 | try: |
| 18 | if not client.is_connected: |
| 19 | await client.connect() |
| 20 | acc = await client.get_me() |
| 21 | log.trace(f"TG Account Login: {acc.username} ({acc.first_name}) {acc.last_name})") |
| 22 | |
| 23 | peer = await client.resolve_peer('BlumCryptoBot') |
| 24 | web_view = await client.invoke(RequestAppWebView( |
| 25 | peer=peer, |
| 26 | app=InputBotAppShortName(bot_id=peer, short_name="app"), |
| 27 | platform='android', |
| 28 | write_allowed=True, |
| 29 | start_param=get_referral_token() |
| 30 | )) |
| 31 | data = unquote(string=web_view.url.split('#tgWebAppData=', maxsplit=1)[1].split('&tgWebAppVersion', maxsplit=1)[0]) |
| 32 | return data |
| 33 | |
| 34 | except (Unauthorized, UserDeactivated, AuthKeyUnregistered, UserDeactivatedBan, AuthKeyDuplicated, |
| 35 | SessionExpired, SessionRevoked): |
| 36 | raise TelegramInvalidSessionException(f"Telegram session is invalid. Client: {client.name}") |
| 37 | except AttributeError as e: |
| 38 | raise TelegramProxyError(e) |
| 39 | finally: |
| 40 | if client.is_connected: |
| 41 | await client.disconnect() |
no test coverage detected