My Shatel API - Get traffic details

apis.md

Get Token

Request:

curl --request GET \
  --url 'https://account-api.shatel.ir/connect/authorize?redirect_uri=com.shatel.myshatel%3A%2F%2Fredirect&client_id=AppMyShatel&response_type=code&state=STATE&nonce=NONCE&scope=openid%20offline_access%20shatel.gateway.apiscope%20idsrv.local.apiscope&code_challenge=CODE_CHALLENGE&code_challenge_method=S256' \
  --compressed \
  --header 'host: account-api.shatel.ir' \
  --header 'referer: android-app://com.shatel.myshatel/' \
  --header 'sec-ch-ua-mobile: ?1' \
  --header 'sec-ch-ua-platform: "Android"' \
  --header 'sec-fetch-dest: document' \
  --header 'sec-fetch-mode: navigate' \
  --header 'sec-fetch-site: cross-site' \
  --header 'upgrade-insecure-requests: 1'

Inspect the callback from network tab -> capture the code manually -> insert to get the token

Request to get a token:

curl --request POST \
  --url https://account-api.shatel.ir/connect/token \
  --compressed \
  --header 'accept: application/json' \
  --header 'content-type: application/x-www-form-urlencoded' \
  --header 'host: account-api.shatel.ir' \
  --data code=CODE \
  --data grant_type=authorization_code \
  --data redirect_uri=com.shatel.myshatel://redirect \
  --data code_verifier=CODE_VERIFIER \
  --data client_id=AppMyShatel

Response:

{
  "id_token": "LONG_TOKEN",
  "access_token": "LONG_TOKEN",
  "expires_in": 2592000,
  "token_type": "Bearer",
  "refresh_token": "REFRESH_TOKEN",
  "scope": "openid offline_access shatel.gateway.apiscope idsrv.local.apiscope"
}

Refresh Token

Request:

curl --request POST \
  --url https://account-api.shatel.ir/connect/token \
  --compressed \
  --header 'accept: application/json' \
  --header 'accept-charset: UTF-8' \
  --header 'accept-encoding: gzip' \
  --header 'connection: Keep-Alive' \
  --header 'content-type: application/x-www-form-urlencoded; charset=UTF-8' \
  --header 'host: account-api.shatel.ir' \
  --header 'user-agent: Ktor client' \
  --data 'grant_type=refresh_token&client_id=AppMyShatel&refresh_token=REFRESH_TOKEN'

Response:

{
	"id_token": "LONG_TOKEN",
	"access_token": "LONG_TOKEN",
	"expires_in": 2592000,
	"token_type": "Bearer",
	"refresh_token": "REFRESH_TOKEN",
	"scope": "openid shatel.gateway.apiscope idsrv.local.apiscope offline_access"
}

Get Traffic

Request:

curl --request GET \
  --url https://gateway.shatel.ir/api/v1.0/appmyshatel/traffic \
  --compressed \
  --header 'accept: application/json' \
  --header 'accept-charset: UTF-8' \
  --header 'accept-encoding: gzip' \
  --header 'accept-language: fa,fa' \
  --header 'authorization: Bearer ACCESS_TOKEN' \
  --header 'connection: Keep-Alive' \
  --header 'host: gateway.shatel.ir' \
  --header 'user-agent: Ktor client'

Response:

{
  "result": [
    {
      "creditKB": 41943040,
      "expirationDate": "2025-06-05T00:00:00",
      "id": 0,
      "lastUsage": "9999-12-31T23:59:59.9999999",
      "receiveKB": -1,
      "refId": 0,
      "sendKB": -1,
      "startDate": "2025-05-05T00:00:00+03:30",
      "trafficDomainId": 1,
      "usageKB": 41943040,
      "trafficDomainName": "اینترنت",
      "trafficPackageTypeId": 1,
      "trafficPackageTypeName": "پایه",
      "requestTime": "2025-06-02T21:07:04.1978736+03:30"
    },
    {
      "creditKB": 104857600,
      "expirationDate": "2026-02-12T11:48:11.3153118+03:30",
      "id": 70192149,
      "lastUsage": "2025-06-02T21:04:34+03:30",
      "receiveKB": 3564573,
      "refId": 101799604,
      "sendKB": 228883,
      "startDate": "2025-02-12T11:48:11.3153118+03:30",
      "trafficDomainId": 1,
      "usageKB": 3629348,
      "trafficDomainName": "اینترنت",
      "trafficPackageTypeId": 4,
      "trafficPackageTypeName": "روزشمار",
      "requestTime": "2025-06-02T21:07:04.1978739+03:30"
    }
  ],
  "isSuccess": true,
  "messages": [],
  "statusCode": 200
}

requirements.txt

requests>=2.25.0

shate_auth.py

#!/usr/bin/env python3
"""
Shatel Authentication Script
This script handles the OAuth2 flow for Shatel API authentication.
"""

import requests
import urllib.parse
import webbrowser
import base64
import hashlib
import secrets
import json
from urllib.parse import parse_qs, urlparse


class ShatelAuth:
    def __init__(self):
        self.client_id = "AppMyShatel"
        self.redirect_uri = "com.shatel.myshatel://redirect"
        self.auth_base_url = "https://account-api.shatel.ir/connect/authorize"
        self.token_url = "https://account-api.shatel.ir/connect/token"
        self.scope = "openid offline_access shatel.gateway.apiscope idsrv.local.apiscope"
        
        # Generate PKCE parameters
        self.code_verifier = self._generate_code_verifier()
        self.code_challenge = self._generate_code_challenge(self.code_verifier)
        self.state = self._generate_state()
        self.nonce = self._generate_nonce()

    def _generate_code_verifier(self):
        """Generate a random code verifier for PKCE"""
        return base64.urlsafe_b64encode(secrets.token_bytes(32)).decode('utf-8').rstrip('=')

    def _generate_code_challenge(self, verifier):
        """Generate code challenge from verifier"""
        digest = hashlib.sha256(verifier.encode('utf-8')).digest()
        return base64.urlsafe_b64encode(digest).decode('utf-8').rstrip('=')

    def _generate_state(self):
        """Generate a random state parameter"""
        return base64.urlsafe_b64encode(secrets.token_bytes(16)).decode('utf-8').rstrip('=')

    def _generate_nonce(self):
        """Generate a random nonce parameter"""
        return base64.urlsafe_b64encode(secrets.token_bytes(16)).decode('utf-8').rstrip('=')

    def get_authorization_url(self):
        """Generate the authorization URL"""
        params = {
            'redirect_uri': self.redirect_uri,
            'client_id': self.client_id,
            'response_type': 'code',
            'state': self.state,
            'nonce': self.nonce,
            'scope': self.scope,
            'code_challenge': self.code_challenge,
            'code_challenge_method': 'S256'
        }
        
        url = f"{self.auth_base_url}?{urllib.parse.urlencode(params)}"
        return url

    def exchange_code_for_token(self, authorization_code):
        """Exchange authorization code for access token"""
        headers = {
            'accept': 'application/json',
            'content-type': 'application/x-www-form-urlencoded',
            'host': 'account-api.shatel.ir'
        }
        
        data = {
            'code': authorization_code,
            'grant_type': 'authorization_code',
            'redirect_uri': self.redirect_uri,
            'code_verifier': self.code_verifier,
            'client_id': self.client_id
        }
        
        try:
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Error exchanging code for token: {e}")
            return None

    def run_auth_flow(self):
        """Run the complete authentication flow"""
        print("🔐 Shatel Authentication Flow")
        print("=" * 50)
        
        # Step 1: Generate and display authorization URL
        auth_url = self.get_authorization_url()
        print(f"\n📋 Generated authorization URL:")
        print(f"{auth_url}")
        print(f"\n🔗 Opening authorization page in your browser...")
        
        # Open the URL in the default browser
        try:
            webbrowser.open(auth_url)
            print("✅ Browser opened successfully")
        except Exception as e:
            print(f"❌ Could not open browser: {e}")
            print("Please manually copy and paste the URL above into your browser")
        
        print(f"\n📝 Instructions:")
        print(f"1. Right click on the page and select 'Inspect' and go to the 'Network' tab")
        print(f"2. Now complete the authentication process in the website")
        print(f"3. Look for a request to '{self.redirect_uri}' in the Network tab")
        print(f"4. Copy the value of the 'code' parameter from the redirect URL")
        print(f"5. Paste it below")
        
        # Step 2: Get authorization code from user
        print(f"\n" + "─" * 50)
        authorization_code = input("🔑 Enter the authorization code: ").strip()
        
        if not authorization_code:
            print("❌ No authorization code provided. Exiting.")
            return
        
        # Step 3: Exchange code for tokens
        print(f"\n🔄 Exchanging authorization code for tokens...")
        token_response = self.exchange_code_for_token(authorization_code)
        
        if not token_response:
            print("❌ Failed to get tokens. Please try again.")
            return
        
        # Step 4: Display tokens
        print(f"\n✅ Tokens obtained successfully!")
        print(f"=" * 50)
        print(f"📄 Token Response:")
        print(json.dumps(token_response, indent=2))
        
        # Step 6: Save tokens to file for future use
        try:
            with open('shatel_tokens.json', 'w') as f:
                json.dump(token_response, f, indent=2)
            print(f"\n💾 Tokens saved to 'shatel_tokens.json'")
        except Exception as e:
            print(f"⚠️  Could not save tokens to file: {e}")
        
        print(f"\n🎉 Authentication flow completed successfully!")
        return token_response


def main():
    """Main function to run the authentication flow"""
    auth = ShatelAuth()
    tokens = auth.run_auth_flow()
    
    if tokens:
        print(f"\n📋 Summary:")
        print(f"Access Token: {tokens.get('access_token', 'N/A')}")
        print(f"Refresh Token: {tokens.get('refresh_token', 'N/A')}")
        print(f"Expires in: {tokens.get('expires_in', 'N/A')} seconds")
        print(f"Token Type: {tokens.get('token_type', 'N/A')}")


if __name__ == "__main__":
    main()

shatel_traffic.py

#!/usr/bin/env python3
"""
Shatel Traffic Checker
This script manages tokens and fetches traffic data from Shatel API.
"""

import requests
import os
import sys
import shelve
from datetime import datetime, timedelta
import base64
import json
from typing import Optional, Dict, Any


INITIAL_ACCESS_TOKEN = ''
INITIAL_REFRESH_TOKEN = ''

def app_folder(prog: str) -> str:
    """Create and return application data folder path"""
    if sys.platform == "win32":
        directory = os.path.join(os.path.expanduser("~"), "AppData", "Local", prog)
    else:
        directory = os.path.join(os.path.expanduser("~"), "." + prog)
    if not os.path.exists(directory):
        os.makedirs(directory)
    return directory


def save_key(key: str, value: Any) -> None:
    """Save a key-value pair to the database"""
    path = os.path.join(app_folder('script_data'), '.shateldb')
    with shelve.open(path) as db:
        db[key] = value


def load_key(key: str, default: Any = None) -> Any:
    """Load a value from the database by key"""
    path = os.path.join(app_folder('script_data'), '.shateldb')
    with shelve.open(path) as db:
        try:
            value = db[key]
        except KeyError:
            value = None
        return default if value is None else value


class ShatelTokenManager:
    """Manages Shatel API tokens and handles authentication"""
    
    def __init__(self):
        self.client_id = "AppMyShatel"
        self.token_url = "https://account-api.shatel.ir/connect/token"
        self.traffic_url = "https://gateway.shatel.ir/api/v1.0/appmyshatel/traffic"
        
    def is_token_expired(self, access_token: str) -> bool:
        """Check if the access token is expired"""
        try:
            # JWT format: header.payload.signature
            payload_b64 = access_token.split('.')[1]
            # Add padding if necessary
            padding = '=' * (-len(payload_b64) % 4)
            payload_b64 += padding
            payload_json = base64.urlsafe_b64decode(payload_b64)
            payload = json.loads(payload_json)
            exp_timestamp = payload.get('exp')
            if exp_timestamp:
                exp_datetime = datetime.fromtimestamp(exp_timestamp)
                buffer_time = exp_datetime - timedelta(minutes=5)
                return datetime.now() >= buffer_time
            return True  # If no expiration found, consider it expired
        except Exception as e:
            print(f"⚠️  Error checking token expiration: {e}")
            return True  # If we can't decode, consider it expired
    
    def refresh_access_token(self, refresh_token: str) -> Optional[Dict[str, Any]]:
        """Refresh the access token using refresh token"""
        headers = {
            'accept': 'application/json',
            'accept-charset': 'UTF-8',
            'accept-encoding': 'gzip',
            'connection': 'Keep-Alive',
            'content-type': 'application/x-www-form-urlencoded; charset=UTF-8',
            'host': 'account-api.shatel.ir',
            'user-agent': 'Ktor client'
        }
        
        data = {
            'grant_type': 'refresh_token',
            'client_id': self.client_id,
            'refresh_token': refresh_token
        }
        
        try:
            print("🔄 Refreshing access token...")
            response = requests.post(self.token_url, headers=headers, data=data)
            response.raise_for_status()
            
            token_data = response.json()
            print("✅ Token refreshed successfully!")
            
            # Save new tokens to database
            save_key('access_token', token_data.get('access_token'))
            save_key('refresh_token', token_data.get('refresh_token'))
            
            return token_data
            
        except requests.exceptions.RequestException as e:
            print(f"❌ Error refreshing token: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"Response: {e.response.text}")
            return None
    
    def get_valid_access_token(self) -> Optional[str]:
        """Get a valid access token, refreshing if necessary"""
        access_token = load_key('access_token')
        refresh_token = load_key('refresh_token')
        
        # If no tokens exist, prompt user to enter them
        if not access_token or not refresh_token:
            # check initial tokens
            if INITIAL_ACCESS_TOKEN and INITIAL_REFRESH_TOKEN:
                access_token = INITIAL_ACCESS_TOKEN
                refresh_token = INITIAL_REFRESH_TOKEN
                save_key('access_token', access_token)
                save_key('refresh_token', refresh_token)
                print("✅ Initial tokens saved to database!")
            else:
                print("🔑 No tokens found in database.")
                print("Put tokens in the code file at the start of the file and run it again")
                return None
        
        # Check if token is expired and refresh if needed
        if self.is_token_expired(access_token):
            print("⏰ Access token expired, refreshing...")
            token_data = self.refresh_access_token(refresh_token)
            if token_data:
                access_token = token_data.get('access_token')
            else:
                print("❌ Failed to refresh token. Clearing stored tokens...")
                # Delete invalid tokens from database
                save_key('access_token', None)
                save_key('refresh_token', None)
                print("🗑️  Stored tokens cleared. Please re-authenticate on next run.")
                return None
        
        return access_token
    
    def get_traffic_data(self) -> Optional[Dict[str, Any]]:
        """Fetch traffic data from Shatel API"""
        access_token = self.get_valid_access_token()
        
        if not access_token:
            return None
        
        headers = {
            'accept': 'application/json',
            'accept-charset': 'UTF-8',
            'accept-encoding': 'gzip',
            'accept-language': 'fa,fa',
            'authorization': f'Bearer {access_token}',
            'connection': 'Keep-Alive',
            'host': 'gateway.shatel.ir',
            'user-agent': 'Ktor client'
        }
        
        try:
            print("📊 Fetching traffic data...")
            response = requests.get(self.traffic_url, headers=headers)
            response.raise_for_status()
            
            traffic_data = response.json()
            print("✅ Traffic data retrieved successfully!")
            return traffic_data
            
        except requests.exceptions.RequestException as e:
            print(f"❌ Error fetching traffic data: {e}")
            if hasattr(e, 'response') and e.response is not None:
                print(f"Response: {e.response.text}")
            return None


def format_bytes(bytes_value: int) -> str:
    """Convert bytes to human readable format"""
    if bytes_value < 0:
        return "N/A"
    
    for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
        if bytes_value < 1024.0:
            return f"{bytes_value:.2f} {unit}"
        bytes_value /= 1024.0
    return f"{bytes_value:.2f} PB"


def format_traffic_data(traffic_data: Dict[str, Any]) -> None:
    """Format and display traffic data in a readable way"""
    if not traffic_data.get('isSuccess'):
        print("❌ API returned unsuccessful response")
        return
    
    results = traffic_data.get('result', [])
    if not results:
        print("📭 No traffic data found")
        return
    
    print("\n" + "=" * 80)
    print("📊 SHATEL TRAFFIC SUMMARY")
    print("=" * 80)
    
    total_credit = 0
    total_usage = 0
    
    for i, package in enumerate(results, 1):
        print(f"\n📦 Package {i}: {package.get('trafficDomainName', 'Unknown')} - {package.get('trafficPackageTypeName', 'Unknown')}")
        print("-" * 50)
        
        credit_kb = package.get('creditKB', 0)
        usage_kb = package.get('usageKB', 0)
        remaining_kb = credit_kb - usage_kb if credit_kb > 0 else 0
        
        total_credit += credit_kb
        total_usage += usage_kb
        
        print(f"💳 Total Credit:     {format_bytes(credit_kb * 1024)}")
        print(f"📈 Used:            {format_bytes(usage_kb * 1024)}")
        print(f"💰 Remaining:       {format_bytes(remaining_kb * 1024)}")
        
        # Usage percentage
        if credit_kb > 0:
            usage_percent = (usage_kb / credit_kb) * 100
            print(f"📊 Usage:           {usage_percent:.1f}%")
        
        # Dates
        start_date = package.get('startDate', '')
        exp_date = package.get('expirationDate', '')
        last_usage = package.get('lastUsage', '')
        
        if start_date:
            print(f"📅 Start Date:      {start_date}")
        if exp_date and exp_date != "9999-12-31T23:59:59.9999999":
            print(f"⏰ Expiration:      {exp_date}")
        if last_usage and last_usage != "9999-12-31T23:59:59.9999999":
            print(f"🕐 Last Usage:      {last_usage}")
        
        # Traffic details
        send_kb = package.get('sendKB', -1)
        receive_kb = package.get('receiveKB', -1)
        
        if send_kb >= 0:
            print(f"⬆️  Upload:          {format_bytes(send_kb * 1024)}")
        if receive_kb >= 0:
            print(f"⬇️  Download:        {format_bytes(receive_kb * 1024)}")
    
    # Total summary
    print("\n" + "=" * 80)
    print("📋 TOTAL SUMMARY")
    print("=" * 80)
    print(f"💳 Total Credit:     {format_bytes(total_credit * 1024)}")
    print(f"📈 Total Used:       {format_bytes(total_usage * 1024)}")
    print(f"💰 Total Remaining:  {format_bytes((total_credit - total_usage) * 1024)}")
    
    if total_credit > 0:
        total_usage_percent = (total_usage / total_credit) * 100
        print(f"📊 Total Usage:      {total_usage_percent:.1f}%")
    
    print("=" * 80)


def main():
    """Main function"""
    print("🌐 Shatel Traffic Checker")
    print("=" * 50)
    
    # Initialize token manager
    manager = ShatelTokenManager()
    traffic_data = manager.get_traffic_data()   
    if traffic_data:
        # Display formatted traffic data
        format_traffic_data(traffic_data)

        # Raw JSON Response
        # print(json.dumps(traffic_data, indent=2, ensure_ascii=False))
        
    else:
        print("❌ Failed to retrieve traffic data")
        return 1
    
    return 0


if __name__ == "__main__":
    try:
        exit_code = main()
        sys.exit(exit_code)
    except KeyboardInterrupt:
        print("\n\n⏹️  Interrupted by user")
        sys.exit(1)
    except Exception as e:
        print(f"\n❌ Unexpected error: {e}")
        sys.exit(1)