From d2e4b64b97bdf7d6ab1e4f5e9b0ed62b20593206 Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 22 Jun 2025 16:42:42 +0300 Subject: [PATCH 1/2] Add Playwright-based client --- README.md | 1 + tele2api/__init__.py | 2 +- tele2api/client.py | 143 ++++++++++++++++++++++++++++++------------- 3 files changed, 103 insertions(+), 43 deletions(-) diff --git a/README.md b/README.md index 1af76e4..ded8459 100644 --- a/README.md +++ b/README.md @@ -1,6 +1,7 @@ # tele2api Python client for Tele2 market API. +Since Tele2 uses an NGINX challenge, requests are performed through Playwright to bypass it. The library allows you to authorise either using a permanent password or via a one time SMS code. Basic operations for creating and managing lots are supported. diff --git a/tele2api/__init__.py b/tele2api/__init__.py index b7e0529..5afbd46 100644 --- a/tele2api/__init__.py +++ b/tele2api/__init__.py @@ -1,4 +1,4 @@ from .client import Tele2Api __all__ = ["Tele2Api"] -__version__ = "1.2.0" +__version__ = "1.3.0" diff --git a/tele2api/client.py b/tele2api/client.py index dc4d235..5cf856e 100644 --- a/tele2api/client.py +++ b/tele2api/client.py @@ -10,36 +10,13 @@ import requests +try: + from playwright.sync_api import sync_playwright +except Exception: + sync_playwright = None HEADERS: Dict[str, str] = { - "Tele2-User-Agent": '"mytele2-app/4.17.0"; "unknown"; "Android/11"; "Build/165135449"', - "User-Agent": "okhttp/4.9.2", -MAIN_API = "https://msk.t2.ru/api/subscribers/" -URL_VALIDATION = "https://msk.t2.ru/api/validation/number/" -URL_AUTH = "https://msk.t2.ru/auth/realms/tele2-b2c/protocol/openid-connect/token" - "https://msk.t2.ru/auth/realms/tele2-b2c/credential-management/reset-options?username=" - "https://msk.t2.ru/auth/realms/tele2-b2c/credential-management/reset-password?username=" - self.access_token = data.get("access_token", "") - self.refresh_token = data.get("refresh_token", "") - if self.access_token: - self.session.headers["Authorization"] = f"Bearer {self.access_token}" - except Exception: - pass - if not self.access_token: - print("Requesting SMS code...") - self.get_sms_code() - sms_code = input("Enter SMS code: ") - result = self.authorization(sms_code) - if isinstance(result, tuple): - if self.token_file: - try: - with open(self.token_file, "wb") as fh: - pickle.dump({"access_token": self.access_token, "refresh_token": self.refresh_token}, fh) - except Exception: - pass - else: - raise RuntimeError(f"Authorization failed: {result}") - 'Accept-Language': 'ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7,de;q=0.6,fr;q=0.5', + "Accept-Language": "ru-RU,ru;q=0.9,en-US;q=0.8,en;q=0.7,de;q=0.6,fr;q=0.5", "Cache-Control": "max-age=0", 'Tele2-User-Agent': '"mytele2-app/4.17.0"; "unknown"; "Android/11"; "Build/165135449"', 'X-API-Version': '1', @@ -47,7 +24,7 @@ 'Accept-Encoding': 'gzip, deflate', 'Accept': 'application/json, text/plain, */*', 'Content-Type': 'application/json', - 'Connection': 'keep-alive' + 'Connection': 'keep-alive', } MAIN_API = "https://msk.t2.ru/api/subscribers/" @@ -61,10 +38,45 @@ ) -def _is_success(response: requests.Response) -> bool: - """Return ``True`` if response status code is 200.""" +def _is_success(response) -> bool: + status = getattr(response, "status_code", getattr(response, "status", 0)) + return status == 200 + + +class PlaywrightSession: + """Minimal wrapper around Playwright request context.""" + + def __init__(self, headers: Optional[Dict[str, str]] = None) -> None: + if sync_playwright is None: + raise RuntimeError("playwright is not installed") + self._playwright = sync_playwright().start() + self._context = self._playwright.request.new_context() + self.headers: Dict[str, str] = headers or {} + if self.headers: + self._context.set_extra_http_headers(self.headers) + + def update_headers(self, headers: Dict[str, str]) -> None: + self.headers.update(headers) + self._context.set_extra_http_headers(self.headers) + + def get(self, *args, **kwargs): + return self._context.get(*args, **kwargs) + + def post(self, *args, **kwargs): + return self._context.post(*args, **kwargs) - return response.status_code == 200 + def put(self, *args, **kwargs): + return self._context.put(*args, **kwargs) + + def patch(self, *args, **kwargs): + return self._context.patch(*args, **kwargs) + + def delete(self, *args, **kwargs): + return self._context.delete(*args, **kwargs) + + def close(self) -> None: + self._context.dispose() + self._playwright.stop() @dataclass @@ -74,7 +86,9 @@ class Tele2Api: phone_number: str access_token: str = "" refresh_token: str = "" - session: requests.Session = field(default_factory=requests.Session, init=False) + token_file: Optional[Union[str, Path]] = None + use_playwright: bool = True + session: object = field(default=None, init=False) def __post_init__(self) -> None: base_api = f"{MAIN_API}{self.phone_number}" @@ -83,18 +97,63 @@ def __post_init__(self) -> None: self.rests_api = f"{base_api}/rests" self.profile_api = f"{base_api}/profile" self.balance_api = f"{base_api}/balance" - data: Dict[str, str] = {"sender": "t2.ru"} + self.service_api = f"{base_api}/services" self.url_validation = URL_VALIDATION + self.phone_number self.url_auth = URL_AUTH self.url_reset_option = URL_RESET_OPTION + self.phone_number self.url_reset_pass = URL_RESET_PASS + self.phone_number - self.session.headers.update({"Authorization": f"Bearer {self.access_token}", **HEADERS}) + + if self.use_playwright: + self.session = PlaywrightSession() + else: + self.session = requests.Session() + + self.update_headers({"Authorization": f"Bearer {self.access_token}", **HEADERS}) + + if self.token_file and not self.access_token: + token_path = Path(self.token_file) + if token_path.exists(): + try: + with token_path.open("rb") as fh: + data = pickle.load(fh) + self.access_token = data.get("access_token", "") + self.refresh_token = data.get("refresh_token", "") + if self.access_token: + self.update_headers({"Authorization": f"Bearer {self.access_token}"}) + except Exception: + pass + if not self.access_token: + print("Requesting SMS code...") + self.get_sms_code() + sms_code = input("Enter SMS code: ") + result = self.authorization(sms_code) + if isinstance(result, tuple): + try: + token_path.parent.mkdir(parents=True, exist_ok=True) + with token_path.open("wb") as fh: + pickle.dump( + {"access_token": self.access_token, "refresh_token": self.refresh_token}, + fh, + ) + except Exception: + pass + else: + raise RuntimeError(f"Authorization failed: {result}") # ------------------------------------------------------------------ - # Context manager helpers + # Helpers # ------------------------------------------------------------------ + def update_headers(self, headers: Dict[str, str]) -> None: + if isinstance(self.session, PlaywrightSession): + self.session.update_headers(headers) + else: + self.session.headers.update(headers) + def close(self) -> None: - self.session.close() + if isinstance(self.session, PlaywrightSession): + self.session.close() + else: + self.session.close() def __enter__(self) -> "Tele2Api": return self @@ -122,7 +181,7 @@ def reset_password(self) -> str: response_option = self.session.get(self.url_reset_option) self.session.post(self.url_reset_pass, json={}) if not _is_success(response_option): - return str(response_option.status_code) + return str(getattr(response_option, "status", response_option.status_code)) return "OK" def authorization(self, sms_code: str, password_type: str = "sms_code") -> Union[str, Tuple[str, str]]: @@ -135,13 +194,13 @@ def authorization(self, sms_code: str, password_type: str = "sms_code") -> Union "password": sms_code, "password_type": password_type, } - self.session.headers["Content-Type"] = "application/x-www-form-urlencoded" - response = self.session.post(self.url_auth, data=payload, verify=False) + self.update_headers({"Content-Type": "application/x-www-form-urlencoded"}) + response = self.session.post(self.url_auth, data=payload) if _is_success(response): data = response.json() self.access_token = data["access_token"] self.refresh_token = data["refresh_token"] - self.session.headers["Authorization"] = f"Bearer {self.access_token}" + self.update_headers({"Authorization": f"Bearer {self.access_token}"}) return self.access_token, self.refresh_token return response.json().get("error_description", "error") @@ -158,7 +217,7 @@ def refresh_access_token(self, refresh_token: str) -> Union[str, Tuple[str, str] data = response.json() self.access_token = data["access_token"] self.refresh_token = data["refresh_token"] - self.session.headers["Authorization"] = f"Bearer {self.access_token}" + self.update_headers({"Authorization": f"Bearer {self.access_token}"}) return self.access_token, self.refresh_token return response.json().get("error_description", "error") From 953f52a42d40e31ffcb431147ffd13b22f496b51 Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 22 Jun 2025 16:54:41 +0300 Subject: [PATCH 2/2] Update version and add Playwright dependency --- setup.py | 4 ++-- 1 file changed, 2 insertions(+), 2 deletions(-) diff --git a/setup.py b/setup.py index ff444d8..905dd58 100644 --- a/setup.py +++ b/setup.py @@ -17,7 +17,7 @@ packages=find_packages('.'), # Start with a small number and increase it with # every change you make https://semver.org - version='1.2.0', + version='1.3.0', # Chose a license from here: https: // # help.github.com / articles / licensing - a - # repository. For example: MIT @@ -38,7 +38,7 @@ # List of keywords keywords=[], # List of packages to install with this one - install_requires=["requests==2.27.1"], + install_requires=["requests==2.27.1", "playwright"], # https://pypi.org/classifiers/ classifiers=[], entry_points={