(self, web_data_params: str)
| 79 | |
| 80 | |
| 81 | async def login(self, web_data_params: str): |
| 82 | web_data = parse_qs(web_data_params) |
| 83 | user = loads(web_data.get("user", ['{}'])[0]) |
| 84 | auth_web_data = { |
| 85 | "query": web_data_params, |
| 86 | "username": user.get("username", get_random_letters(user.get("id", ""))), |
| 87 | "referralToken": get_referral_token().split('_')[1] |
| 88 | } |
| 89 | for _ in range(3): |
| 90 | try: |
| 91 | await sleep(0.1) |
| 92 | data = await self.auth_with_web_data(auth_web_data) |
| 93 | except (UsernameNotAvailableError, AlreadyConnectError): |
| 94 | auth_web_data = {"query": web_data_params} |
| 95 | continue |
| 96 | except InvalidUsernameError as e: |
| 97 | self._log.warning(f"Invalid username from TG account... Error: {e}") |
| 98 | auth_web_data.update({"username": get_random_letters()}) |
| 99 | self._log.warning(f'Try using username for auth - {auth_web_data.get("username")}') |
| 100 | continue |
| 101 | token = data.get("token", {}) |
| 102 | return self.set_tokens(token) |
| 103 | raise AuthError("Auth tries ended. Result Failed!") |
| 104 | |
| 105 | def set_tokens(self, token_data: dict): |
| 106 | self._refresh_token = token_data.get('refresh', '') |
no test coverage detected