From 3dd749f1ba0485fcc4554535c089e4102da0ef0e Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 15 Jun 2025 15:24:36 +0300 Subject: [PATCH 1/5] Refactor library structure --- README.md | 7 +- setup.py | 4 +- tele2api/__init__.py | 4 + tele2api/client.py | 252 ++++++++++++++++++++++++++++++++++++++++++ tele2api/tele2_api.py | 236 +-------------------------------------- 5 files changed, 264 insertions(+), 239 deletions(-) create mode 100644 tele2api/client.py diff --git a/README.md b/README.md index f5a0bc5..f3c9e3a 100644 --- a/README.md +++ b/README.md @@ -1,4 +1,5 @@ # tele2api -API Tele2 -API для работы с маркетом Теле2. -Работает авторизация как по постоянному паролю, так и по смс. +Python client for Tele2 market API. + +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/setup.py b/setup.py index bfa30a1..0decaa1 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.0.0', + version='1.1.0', # Chose a license from here: https: // # help.github.com / articles / licensing - a - # repository. For example: MIT @@ -43,6 +43,6 @@ classifiers=[], entry_points={ 'console_scripts': - ['tele2api = tele2api.tele2_api:Tele2Api'] + ['tele2api = tele2api.client:Tele2Api'] } ) diff --git a/tele2api/__init__.py b/tele2api/__init__.py index e69de29..c97e5bc 100644 --- a/tele2api/__init__.py +++ b/tele2api/__init__.py @@ -0,0 +1,4 @@ +from .client import Tele2Api + +__all__ = ["Tele2Api"] +__version__ = "1.1.0" diff --git a/tele2api/client.py b/tele2api/client.py new file mode 100644 index 0000000..e19b481 --- /dev/null +++ b/tele2api/client.py @@ -0,0 +1,252 @@ +from __future__ import annotations + +import json +import random +from dataclasses import dataclass, field +from typing import Dict, List, Optional, Tuple, Union + +import requests + + +HEADERS: Dict[str, str] = { + "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.14.0"; "unknown"; "Android/11"; "Build/164755374"', + "X-API-Version": "1", + "User-Agent": "okhttp/4.2.0", + "Accept-Encoding": "gzip, deflate", + "Accept": "application/json, text/plain, */*", + "Content-Type": "application/json", + "Connection": "keep-alive", +} + +MAIN_API = "https://my.tele2.ru/api/subscribers/" +URL_VALIDATION = "https://my.tele2.ru/api/validation/number/" +URL_AUTH = "https://my.tele2.ru/auth/realms/tele2-b2c/protocol/openid-connect/token" +URL_RESET_OPTION = ( + "https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-options?username=" +) +URL_RESET_PASS = ( + "https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-password?username=" +) + + +def _is_success(response: requests.Response) -> bool: + """Return ``True`` if response status code is 200.""" + + return response.status_code == 200 + + +@dataclass +class Tele2Api: + """Client for Tele2 personal account API.""" + + phone_number: str + access_token: str = "" + refresh_token: str = "" + session: requests.Session = field(default_factory=requests.Session, init=False) + + def __post_init__(self) -> None: + base_api = f"{MAIN_API}{self.phone_number}" + self.market_api = f"{base_api}/exchange/lots/created" + self.bought_api = f"{base_api}/exchange/lots/bought" + self.rests_api = f"{base_api}/rests" + self.profile_api = f"{base_api}/profile" + self.balance_api = f"{base_api}/balance" + 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}) + + # ------------------------------------------------------------------ + # Context manager helpers + # ------------------------------------------------------------------ + def close(self) -> None: + self.session.close() + + def __enter__(self) -> "Tele2Api": + return self + + def __exit__(self, exc_type, exc, tb) -> None: + self.close() + + # ------------------------------------------------------------------ + # API methods + # ------------------------------------------------------------------ + def get_sms_code(self, operation: Optional[str] = None) -> str: + """Request a one-time SMS code for authorization.""" + + data: Dict[str, str] = {"sender": "Tele2"} + if operation: + data["operation"] = operation + response = self.session.post(self.url_validation, json=data) + if not _is_success(response): + return response.json().get("detail", "error") + return "OK" + + def reset_password(self) -> str: + """Reset the permanent password via SMS.""" + + 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 "OK" + + def authorization(self, sms_code: str, password_type: str = "sms_code") -> Union[str, Tuple[str, str]]: + """Authorize using an SMS code or a permanent password.""" + + payload = { + "client_id": "digital-suite-web-app", + "grant_type": "password", + "username": self.phone_number, + "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) + 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}" + return self.access_token, self.refresh_token + return response.json().get("error_description", "error") + + def refresh_access_token(self, refresh_token: str) -> Union[str, Tuple[str, str]]: + """Refresh authorization token using ``refresh_token``.""" + + payload = { + "client_id": "digital-suite-web-app", + "grant_type": "refresh_token", + "refresh_token": refresh_token, + } + 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}" + return self.access_token, self.refresh_token + return response.json().get("error_description", "error") + + def get_balance(self) -> Optional[int]: + """Return current balance.""" + + response = self.session.get(self.balance_api) + if _is_success(response): + return response.json()["data"]["value"] + return None + + def get_rests(self) -> Dict[str, int]: + """Return remaining data and voice minutes available for sale.""" + + response = self.session.get(self.rests_api) + data = response.json() + rests = list(data["data"]["rests"]) + sellable = [r for r in rests if r["type"] == "tariff" and not r["rollover"]] + return { + "data": int(sum(r["remain"] for r in sellable if r["uom"] == "mb") / 1024), + "voice": int(sum(r["remain"] for r in sellable if r["uom"] == "min")), + } + + def create_lot( + self, + traffic_type: str, + value: int, + amount: int, + emojis: Union[str, List[str], None] = None, + ) -> str: + """Create a new market lot.""" + + payload = { + "trafficType": traffic_type, + "cost": {"amount": amount, "currency": "rub"}, + "volume": {"value": value, "uom": "min" if traffic_type == "voice" else "gb"}, + } + response = self.session.put(self.market_api, json=payload) + if not _is_success(response): + return response.json()["meta"]["status"] + lot_id = response.json()["data"]["id"] + if emojis: + if emojis == "random": + all_emojis = ["cat", "scream", "bomb", "rich", "zipped", "tongue", "cool", "devil"] + list_emojis = random.sample(all_emojis, k=3) + elif isinstance(emojis, str): + list_emojis = [emojis] + else: + list_emojis = list(emojis) + self.session.patch( + f"{self.market_api}/{lot_id}", + json={"showSellerName": True, "emojis": list_emojis, "cost": {"amount": amount, "currency": "rub"}}, + ) + return lot_id + + def patch_lot(self, lot_id: str, amount: int) -> str: + """Change lot price.""" + + response = self.session.patch( + f"{self.market_api}/{lot_id}", + json={"cost": {"amount": amount, "currency": "rub"}}, + ) + if not _is_success(response): + return response.json()["meta"]["status"] + return "OK" + + def bought_lot(self, sms_code: str, lot: Dict[str, any]) -> str: + """Purchase a lot.""" + + response = self.session.put( + f"{self.bought_api}?validationCode={sms_code}", + json={ + "volume": {"value": lot["volume"]["value"], "uom": lot["volume"]["uom"]}, + "cost": {"amount": lot["cost"]["amount"], "currency": "rub"}, + "lotId": lot["id"], + "hash": lot["hash"], + "trafficType": lot["trafficType"], + }, + ) + if not _is_success(response): + return response.json()["meta"]["status"] + return "OK" + + def delete_lot(self, lot_id: str) -> str: + """Remove a lot from the market.""" + + response = self.session.delete(f"{self.market_api}/{lot_id}") + if not _is_success(response): + return response.json()["meta"]["status"] + return "OK" + + def get_active_lots(self) -> Optional[List[Dict[str, any]]]: + """Return list of active lots.""" + + response = self.session.get(self.market_api) + if _is_success(response): + data = response.json() + lots = list(data["data"]) + return [lot for lot in lots if lot.get("status") == "active"] + return None + + def mixx_update_subscribe(self, action: str = "enable") -> str: + """Enable or disable mixx notifications.""" + + json_data = { + "operationType": "change_service", + "changedServices": [ + { + "billingServiceId": "31299", + "action": action, + } + ], + } + self.session.post(f"{self.service_api}/notifications/check", json=json_data) + if action == "enable": + response = self.session.put(f"{self.service_api}/31299") + else: + response = self.session.delete(f"{self.service_api}/31299") + if not _is_success(response): + return response.json() + return "OK" diff --git a/tele2api/tele2_api.py b/tele2api/tele2_api.py index c1870d2..bd85117 100644 --- a/tele2api/tele2_api.py +++ b/tele2api/tele2_api.py @@ -1,235 +1,3 @@ -import json -import random - -import requests - -HEADERS = { - '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.14.0"; "unknown"; "Android/11"; "Build/164755374"', - 'X-API-Version': '1', - 'User-Agent': 'okhttp/4.2.0', - 'Accept-Encoding': 'gzip, deflate', - 'Accept': 'application/json, text/plain, */*', - 'Content-Type': 'application/json', - 'Connection': 'keep-alive' - } - -MAIN_API = 'https://my.tele2.ru/api/subscribers/' -URL_VALIDATION = 'https://my.tele2.ru/api/validation/number/' -URL_AUTH = 'https://my.tele2.ru/auth/realms/tele2-b2c/protocol/openid-connect/token' -URL_RESET_OPTION = 'https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-options?username=' -URL_RESET_PASS = 'https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-password?username=' - - -def _get_status_code(response): - return response.status_code == 200 - - -class Tele2Api: - def __init__(self, phone_number: str, access_token: str = '', refresh_token: str = ''): - self._phone_number = phone_number - base_api = MAIN_API + phone_number - self.market_api = f'{base_api}/exchange/lots/created' - self.bought_api = f'{base_api}/exchange/lots/bought' - self.rests_api = f'{base_api}/rests' - self.profile_api = f'{base_api}/profile' - self.balance_api = f'{base_api}/balance' - 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.access_token = access_token - self.refresh_token = refresh_token - self.session = requests.Session() - self.session.headers = { - 'Authorization': f'Bearer {self.access_token}', - **HEADERS - } - - def get_sms_code(self, operation=None): - """ - Получение одноразового sms-кода для авторизации - :return: - """ - data = {"sender": "Tele2"} - if operation is not None: - data['operation'] = operation - data = json.dumps(data) - response = self.session.post(self.url_validation, data=data) - if not _get_status_code(response): - return response.json().get('detail') - return 'OK' - - def reset_password(self): - """ - Получение нового постоянного пароля - :return: - """ - data = json.dumps({}) - response_option = self.session.get(self.url_reset_option) - response_pass = self.session.post(self.url_reset_pass, data=data) - if not _get_status_code(response_option): - return _get_status_code(response_option) - return 'OK' - - def authorization(self, sms_code: str, password_type: str = 'sms_code' or 'password'): - """ - Авторизация - :param sms_code: одноразовый sms-код, либо постоянный пароль для входа - :param password_type: выбор способа авторизации: 'sms_code' или 'password' - :return: Получаем access_token и refresh_token - """ - data_auth = {"client_id": "digital-suite-web-app", "grant_type": "password", "username": self._phone_number, - "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_auth, verify=False) - if _get_status_code(response): - self.access_token = response.json()['access_token'] - self.refresh_token = response.json()['refresh_token'] - return response.json()['access_token'], response.json()['refresh_token'] - return response.json()['error_description'] - - def refresh_token(self, refresh_token: str): - """ - Обновления токена для авторизации - :param refresh_token: нужно передать в качестве параметра refresh_token, полученный в прошлой сессии - :return: - """ - response = self.session.post(self.url_auth, data={ - 'client_id': 'digital-suite-web-app', - 'grant_type': 'refresh_token', - 'refresh_token': refresh_token, - }) - if _get_status_code(response): - return response.json()['access_token'], response.json()['refresh_token'] - return response.json()['error_description'] - - def get_balance(self): - """ - Получение данных о балансе - :return: - """ - response = self.session.get(self.balance_api) - if _get_status_code(response): - return response.json()['data']['value'] - - def get_rests(self): - """ - Получение остатков, доступных для продаж на Маркете - :return: - """ - response = self.session.get(self.rests_api) - response_json = response.json() - rests = list(response_json['data']['rests']) - sellable = [a for a in rests if a['type'] == 'tariff' and a['rollover'] == False] - return { - 'data': int( - sum(a['remain'] for a in sellable if a['uom'] == 'mb') / 1024), - 'voice': int( - sum(a['remain'] for a in sellable if a['uom'] == 'min')) - } - - def create_lot(self, traffic_type, value, amount, emojis='None'): - """ - Создание нового лота - :param emojis: если не хотим использовать, оставляем 'None', либо передаем значение 'random', - либо список необходимых эмоджи: 'cat', 'scream', 'bomb', 'rich', 'zipped', 'tongue', 'cool', 'devil' - :param traffic_type: тип трафика ('voice' или 'data') - :param value: число минут или Гб одного лота - :param amount: стоимость лота - :return: - """ - response = self.session.put(self.market_api, json={ - 'trafficType': traffic_type, - 'cost': {'amount': amount, 'currency': 'rub'}, - 'volume': {'value': value, - 'uom': 'min' if traffic_type == 'voice' else 'gb'} - }) - if not _get_status_code(response): - print(response.json()) - return response.json()['meta']['status'] - id_lot = response.json()["data"]["id"] - if emojis != 'None': - if emojis == 'random': - all_emojis = ['cat', 'scream', 'bomb', 'rich', 'zipped', 'tongue', 'cool', 'devil'] - list_emojis = [] - for _ in range(3): - list_emojis.append(random.choice(all_emojis)) - else: - list_emojis = emojis - response = self.session.patch(self.market_api + '/{}'.format(id_lot), json={ - "showSellerName": True, "emojis": list_emojis, - "cost": {"amount": amount, "currency": "rub"} - }) - - return id_lot - - def patch_lot(self, id_lot, amount): - """ - Изменение цены лота - :param id_lot: необходимо передать id лота - :param amount: новая цена лота - :return: - """ - response = self.session.patch(self.market_api + '/{}'.format(id_lot), json={ - 'cost': {'amount': amount, 'currency': 'rub'} - }) - if not _get_status_code(response): - return response.json()['meta']['status'] - return 'OK' - - def bought_lot(self, sms_code, lot): - response = self.session.put(self.bought_api + '?validationCode={}'.format(sms_code), - json={"volume": {"value": lot['volume']['value'], "uom": lot['volume']['uom']}, - "cost": {"amount": lot['cost']['amount'], "currency": "rub"}, - "lotId": lot['id'], - "hash": lot['hash'], "trafficType": lot['trafficType']}) - if not _get_status_code(response): - return response.json()['meta']['status'] - return 'OK' - - def delete_lot(self, id_lot): - """ - :param id_lot: необходимо передать id лота - :return: Снимаем с продажи лот - """ - response = self.session.delete(self.market_api + '/{}'.format(id_lot)) - if not _get_status_code(response): - return response.json()['meta']['status'] - return 'OK' - - def get_active_lots(self): - """ - :return: Получаем список активных лотов - """ - response = self.session.get(self.market_api) - # print(response.json()) - if _get_status_code(response): - response_json = response.json() - lots = list(response_json['data']) - active_lots = [a for a in lots if a['status'] == 'active'] - return active_lots - - def mixx_update_subscribe(self, action="enable"): - json_data = { - 'operationType': 'change_service', - 'changedServices': [ - { - 'billingServiceId': '31299', - 'action': action, - }, - ], - } - response = self.session.post(f'{self.service_api}/notifications/check', - json=json_data) - if action == "enable": - response = self.session.put(f'{self.service_api}/31299') - elif action == "disable": - response = self.session.delete(f'{self.service_api}/31299') - if not _get_status_code(response): - return response.json() - return 'OK' - +from .client import Tele2Api +__all__ = ["Tele2Api"] From a65bd3600e0c28b6764c6b387c0ef037d417e322 Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 15 Jun 2025 15:29:00 +0300 Subject: [PATCH 2/5] docs: add usage examples --- README.md | 17 +++++++++++++++++ 1 file changed, 17 insertions(+) diff --git a/README.md b/README.md index f3c9e3a..4098cbc 100644 --- a/README.md +++ b/README.md @@ -3,3 +3,20 @@ Python client for Tele2 market API. 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. + +## Usage + +```python +from tele2api import Tele2Api + +client = Tele2Api("79001234567") +client.get_sms_code() +token, refresh = client.authorization("123456") + +balance = client.get_balance() +print(balance) + +# or use a context manager +with Tele2Api("79001234567", access_token=token) as api: + print(api.get_active_lots()) +``` From 8c0e116314f3bc0d5790d8cd36b9c4ccc9c95849 Mon Sep 17 00:00:00 2001 From: Sergei Date: Sun, 15 Jun 2025 16:06:46 +0300 Subject: [PATCH 3/5] changed tele2 -> t2 --- README.md | 7 ++++--- tele2api/client.py | 28 ++++++++++++++-------------- 2 files changed, 18 insertions(+), 17 deletions(-) diff --git a/README.md b/README.md index 4098cbc..12cbbdb 100644 --- a/README.md +++ b/README.md @@ -9,14 +9,15 @@ Basic operations for creating and managing lots are supported. ```python from tele2api import Tele2Api -client = Tele2Api("79001234567") +phone = "79001234567" +client = Tele2Api(phone) client.get_sms_code() -token, refresh = client.authorization("123456") +token, refresh = client.authorization(input("Enter SMS code: ")) balance = client.get_balance() print(balance) # or use a context manager -with Tele2Api("79001234567", access_token=token) as api: +with Tele2Api(phone, access_token=token) as api: print(api.get_active_lots()) ``` diff --git a/tele2api/client.py b/tele2api/client.py index e19b481..87ea7d0 100644 --- a/tele2api/client.py +++ b/tele2api/client.py @@ -9,25 +9,25 @@ HEADERS: Dict[str, str] = { - "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.14.0"; "unknown"; "Android/11"; "Build/164755374"', - "X-API-Version": "1", - "User-Agent": "okhttp/4.2.0", - "Accept-Encoding": "gzip, deflate", - "Accept": "application/json, text/plain, */*", - "Content-Type": "application/json", - "Connection": "keep-alive", + 'Tele2-User-Agent': '"mytele2-app/4.17.0"; "unknown"; "Android/11"; "Build/165135449"', + 'X-API-Version': '1', + 'User-Agent': 'okhttp/4.9.2', + 'Accept-Encoding': 'gzip, deflate', + 'Accept': 'application/json, text/plain, */*', + 'Content-Type': 'application/json', + 'Connection': 'keep-alive' } -MAIN_API = "https://my.tele2.ru/api/subscribers/" -URL_VALIDATION = "https://my.tele2.ru/api/validation/number/" -URL_AUTH = "https://my.tele2.ru/auth/realms/tele2-b2c/protocol/openid-connect/token" +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" URL_RESET_OPTION = ( - "https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-options?username=" + "https://msk.t2.ru/auth/realms/tele2-b2c/credential-management/reset-options?username=" ) URL_RESET_PASS = ( - "https://my.tele2.ru/auth/realms/tele2-b2c/credential-management/reset-password?username=" + "https://msk.t2.ru/auth/realms/tele2-b2c/credential-management/reset-password?username=" ) @@ -78,7 +78,7 @@ def __exit__(self, exc_type, exc, tb) -> None: def get_sms_code(self, operation: Optional[str] = None) -> str: """Request a one-time SMS code for authorization.""" - data: Dict[str, str] = {"sender": "Tele2"} + data: Dict[str, str] = {"sender": "t2.ru"} if operation: data["operation"] = operation response = self.session.post(self.url_validation, json=data) From dc4d21726e4f5a27a2dae83d9006a815d51ce56a Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 22 Jun 2025 12:17:44 +0300 Subject: [PATCH 4/5] Improve context manager with automatic authorization --- README.md | 6 +++++- setup.py | 2 +- tele2api/__init__.py | 2 +- tele2api/client.py | 30 ++++++++++++++++++++++++++++++ 4 files changed, 37 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index 12cbbdb..1af76e4 100644 --- a/README.md +++ b/README.md @@ -9,7 +9,11 @@ Basic operations for creating and managing lots are supported. ```python from tele2api import Tele2Api -phone = "79001234567" +# use a context manager with saved tokens + +# or let the context manager handle authorization +with Tele2Api("79001234567") as api: + print(api.get_balance()) client = Tele2Api(phone) client.get_sms_code() token, refresh = client.authorization(input("Enter SMS code: ")) diff --git a/setup.py b/setup.py index 0decaa1..ff444d8 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.1.0', + version='1.2.0', # Chose a license from here: https: // # help.github.com / articles / licensing - a - # repository. For example: MIT diff --git a/tele2api/__init__.py b/tele2api/__init__.py index c97e5bc..b7e0529 100644 --- a/tele2api/__init__.py +++ b/tele2api/__init__.py @@ -1,4 +1,4 @@ from .client import Tele2Api __all__ = ["Tele2Api"] -__version__ = "1.1.0" +__version__ = "1.2.0" diff --git a/tele2api/client.py b/tele2api/client.py index 87ea7d0..1fa6370 100644 --- a/tele2api/client.py +++ b/tele2api/client.py @@ -3,12 +3,42 @@ import json import random from dataclasses import dataclass, field +from pathlib import Path from typing import Dict, List, Optional, Tuple, Union +import os +import pickle import requests HEADERS: Dict[str, str] = { + token_file: Optional[Path] = None + if self.token_file is None: + self.token_file = Path(f".tele2api_{self.phone_number}.pickle") + if not self.access_token and self.token_file and self.token_file.exists(): + try: + with open(self.token_file, "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.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', "Cache-Control": "max-age=0", 'Tele2-User-Agent': '"mytele2-app/4.17.0"; "unknown"; "Android/11"; "Build/165135449"', From 7d8586158e76bec2759d46bd323d295cca038c08 Mon Sep 17 00:00:00 2001 From: Sergei Mikheev <54879069+Muxee4ka@users.noreply.github.com> Date: Sun, 22 Jun 2025 12:17:50 +0300 Subject: [PATCH 5/5] Update headers and endpoints --- tele2api/client.py | 16 ++++++++-------- 1 file changed, 8 insertions(+), 8 deletions(-) diff --git a/tele2api/client.py b/tele2api/client.py index 1fa6370..dc4d235 100644 --- a/tele2api/client.py +++ b/tele2api/client.py @@ -12,13 +12,13 @@ HEADERS: Dict[str, str] = { - token_file: Optional[Path] = None - if self.token_file is None: - self.token_file = Path(f".tele2api_{self.phone_number}.pickle") - if not self.access_token and self.token_file and self.token_file.exists(): - try: - with open(self.token_file, "rb") as fh: - data = pickle.load(fh) + "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: @@ -83,7 +83,7 @@ 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" - self.service_api = f"{base_api}/services" + data: Dict[str, str] = {"sender": "t2.ru"} self.url_validation = URL_VALIDATION + self.phone_number self.url_auth = URL_AUTH self.url_reset_option = URL_RESET_OPTION + self.phone_number