Tracking Shipments with Shiprocket API and Telegram Notifications

Tracking shipments and keeping up with their activities can be quite a challenge, especially when the notification system doesn’t provide detailed updates throughout the package’s journey. If you are a paranoid like me, you might be interested in exploring some extra options to enhance the situation.

In this blog post, we will explore a Python code snippet that leverages the Shiprocket API to track shipments and sends real-time notifications through Telegram. This integration allows you to automate the tracking process and receive timely updates on shipment activities.

Prerequisites

Before we dive into the code, make sure you have the following prerequisites in place:

• Shiprocket API credentials: To interact with the Shiprocket API, you need to have valid API credentials. If you don’t have them already, you can obtain them by signing up on the Shiprocket platform.

• Telegram bot and token: To send notifications via Telegram, you need to create a Telegram bot and obtain an API token. You can create a bot and obtain the token by following the instructions provided in the Telegram Bot API documentation.

• Required Python libraries: The code snippet utilizes several Python libraries. Make sure you have the following libraries installed:

  • requests: For making HTTP requests to the Shiprocket API and Telegram API.
  • json: For working with JSON data.
  • time: For introducing delays between API calls.
  • datetime, timedelta: For handling date and time calculations.
  • dateutil.parser: For parsing date strings.
  • os: For accessing environment variables.
  • humanize: For human-readable time and date representations.

  Once you have fulfilled these prerequisites, you are ready to proceed with the code implementation.

Setting up the Code

Let’s start by importing the required libraries and setting up the necessary variables:

import requests
import json
import time
from datetime import datetime, timedelta
import dateutil.parser
import os
import humanize

token = 'if-you-are-worried-about-auth-error-on-initial-request-from-this-script-find-some-method-to-bypass-it-i-am-oka-hence-this-is-how-it-is'
telegram_bot_token = 'bot<the_actual_bot_token>'
shiprocket_api_user_email = '[email protected]'
shiprocket_api_user_password = 'W'
awb = 'fff'
sleep_time = 120

In this code snippet, we initialize variables such as token, telegram_bot_token, shiprocket_api_user_email, shiprocket_api_user_password, awb, and sleep_time. These variables store the authentication tokens, Shiprocket API credentials, the shipment’s AWB (Air Waybill) number, and the duration between successive API calls, respectively. Please note that you need to replace the placeholder values with your actual credentials.

don’t be silly like me by hardcoding sensitive details, at least use env variables.

Implementing Helper Functions

Next, we define some helper functions that facilitate the code execution:

def format_text(string):
    if len(string) > 40:
        return string[:36] + "..."
    else:
        return string

def send_telegram_message(message):
    parameters = {
        'chat_id': 616727329,
        'text': message,
        'parse_mode': 'HTML'
    }
    response = requests.post(url=f"https://api.telegram.org/{telegram_bot_token}/sendmessage", data=parameters)
    return response.status_code in [200, 201, 202, 203, 204]

def get_shiprocket_token():
    auth_url = "https://apiv2.shiprocket.in/v1/external/auth/login"
    auth_payload = json.dumps({
        "email": shiprocket_api_user_email,
        "password": shiprocket_api_user_password
    })
    auth_headers = {'Content-Type': 'application/json'}
    response = requests.post(auth_url, headers=auth_headers, data=auth_payload)
    if response.status_code == 200:
        token_data = json.loads(response.text)
        return token_data.get('token', '')
    else:
        raise ValueError(f"Failed to get Shiprocket token: {response.status_code}")

def get_shipment_activities():
    payload = {}
    headers = {
        'Content-Type': 'application/json',
        'Authorization': 'Bearer ' + token
    }
    response = requests.get(url, headers=headers, data=payload)
    if response.status_code in [200, 202]:
        result = json.loads(response.text)
        return result.get('tracking_data', {}).get('shipment_track_activities', [])
    elif response.status_code == 401:
        new_token = get_shiprocket_token()
        if new_token:
            return get_shipment_activities()
        else:
            raise ValueError("Failed to get a new Shiprocket token")
    elif response.status_code == 429:
        raise ValueError("Too many requests to Shiprocket API")
    else:
        raise ValueError(f"Failed to fetch shipment activities: {response.reason}")

def process_shipment_activities(activities):
    new_activities_count = len(activities) - past_activities_array
    telegram_message = ''
    todays = '<b>Today</b>:'
    yesterdays = '<b>Yesterday</b>:'
    for id, activity in enumerate(activities):
        if id < new_activities_count:
            my_date_string = activity['date']
            datetime_object = datetime.strptime(my_date_string, '%Y-%m-%d %H:%M:%S')
            date_for_telegram = humanize.precisedelta(datetime_object)
            if humanize.naturalday(datetime_object) == 'today':
                diff = datetime.now() - datetime_object
                time_human = humanize.naturaldelta(timedelta(seconds=diff.total_seconds())) + " ago at " + datetime_object.strftime("%H:%M:%S")
                todays += f"\n\t\t\t\t{format_text(activity['activity'])}\n\t\t\t\t\t\t\t\t{activity['location']}\n\t\t\t\t\t\t\t\t{time_human}"
            elif humanize.naturalday(datetime_object) == 'yesterday':
                diff = datetime.now() - datetime_object
                time_human = datetime_object.strftime("%H:%M:%S")
                yesterdays += f"\n\t\t\t\t{format_text(activity['activity'])}\n\t\t\t\t\t\t\t\t{activity['location']}\n\t\t\t\t\t\t\t\t{time_human}"
            else:
                time_human = datetime_object.strftime("%d %b at %H:%M")
                telegram_message += f"<b>{humanize.naturalday(datetime_object)}</b>:\n\t\t\t\t{format_text(activity['activity'])}\n\t\t\t\t\t\t\t\t{activity['location']}\n\t\t\t\t\t\t\t\t{time_human}\n\n"
    if yesterdays != '<b>Yesterday</b>:':
        telegram_message = yesterdays + "\n\n" + telegram_message
    if todays != '<b>Today</b>:':
        telegram_message = todays + "\n\n" + telegram_message
    telegram_message += "#shiprocket #tracking #courier #electronicscomp #neopixel #smps"
    return telegram_message

The format_text function is used to truncate long strings to a manageable length for display purposes.

The send_telegram_message function sends a message via the Telegram API. It takes a message parameter and constructs the necessary request payload to send the message to the specified chat ID.

The get_shiprocket_token function retrieves the Shiprocket API token required for authentication. It sends a request to the Shiprocket API with the provided credentials and returns the token if the request is successful. Otherwise, it raises an error.

The get_shipment_activities function retrieves the shipment activities using the Shiprocket API. It makes a GET request to the API endpoint with the appropriate headers and returns the shipment activities if the request is successful. If the API returns a 401 status code (Unauthorized), it tries to obtain a new token using get_shiprocket_token and recursively calls itself. If the API returns a 429 status code (Too Many Requests), it raises an error. For any other error response, it raises an error with the corresponding reason.

The process_shipment_activities function takes the shipment activities as input and processes them to generate a formatted message for Telegram notifications. It extracts relevant information such as activity, location, and date from the activities and formats them into a human-readable representation. The function categorizes the activities into today’s and yesterday’s activities and constructs the final Telegram message. It also adds relevant hashtags to the message for easy tracking. The function returns the formatted Telegram message.

Tracking Shipments and Sending Notifications

Now that we have defined the helper functions, let’s move on to the main part of the code that tracks the shipments and sends notifications:

while True:
    try:
        activities = get_shipment_activities()
        if activities:
            if past_activities_array != len(activities):
                print("Sending notifications")
                telegram_message = process_shipment_activities(activities)
                if send_telegram_message(telegram_message):
                    past_activities_array = len(activities)
                else:
                    print("Telegram message sending failed")
            else:
                print("Nothing new to notify")
        else:
            print("No shipment activities found")

        sleep_time = 120

    except Exception as e:
        print(str(e))

    next_check = datetime.now() + timedelta(seconds=sleep_time)
    next_check = next_check.strftime('%d %B at %I:%M %p')
    print(f"Next check at {next_check}")
    time.sleep(sleep_time)

This part of the code runs in an infinite loop, continuously tracking the shipments and sending notifications when there are new activities. Here’s how it works:

• It calls the get_shipment_activities function to retrieve the latest shipment activities.

• If activities are found, it compares the number of activities with the past_activities_array variable to determine if there are new activities.

• If there are new activities, it calls the process_shipment_activities function to generate a formatted Telegram message.

• The formatted message is then sent as a Telegram notification using the send_telegram_message function. If the message is successfully sent, the past_activities_array variable is updated to reflect the new activity count.

• If there are no new activities, it prints a message indicating that there is nothing new to notify.

• If no shipment activities are found, it prints a message indicating that no activities were found.

• The sleep_time variable is set to the desired duration between successive API calls (in this case, 120 seconds).

• Any exceptions that occur during the execution of the code are caught, and the corresponding error message is printed.

• The next check time is calculated by adding the sleep_time duration to the current time. It is then formatted into a human-readable string.

• Finally, the code waits for the specified duration using time.sleep(sleep_time) before starting the next iteration of the loop.

Ciao!