Psiphon_key_check_bot.ipynb

{
  "nbformat": 4,
  "nbformat_minor": 0,
  "metadata": {
    "colab": {
      "provenance": [],
      "collapsed_sections": [],
      "authorship_tag": "ABX9TyNwPGaYxnHDTVmyB4tj3cAc",
      "include_colab_link": true
    },
    "kernelspec": {
      "name": "python3",
      "display_name": "Python 3"
    },
    "language_info": {
      "name": "python"
    }
  },
  "cells": [
    {
      "cell_type": "markdown",
      "metadata": {
        "id": "view-in-github",
        "colab_type": "text"
      },
      "source": [
        "<a href=\"https://colab.research.google.com/github/iam-muni-baa/Psiphon_key_check_bot/blob/main/Psiphon_key_check_bot.ipynb\" target=\"_parent\"><img src=\"https://colab.research.google.com/assets/colab-badge.svg\" alt=\"Open In Colab\"/></a>"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Final Script\n",
        "Required modules"
      ],
      "metadata": {
        "id": "0c50jrUTphHs"
      }
    },
    {
      "cell_type": "code",
      "execution_count": 3,
      "metadata": {
        "id": "ZIGpNp_UpXPP"
      },
      "outputs": [],
      "source": [
        "import requests\n",
        "import json\n",
        "import base64\n",
        "from datetime import datetime\n",
        "from pytz import timezone\n",
        "from getpass import getpass"
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Variables\n"
      ],
      "metadata": {
        "id": "ODNLyP9zpwCI"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "base_url = \"https://api.telegram.org/bot\"\n",
        "\n",
        "secret = getpass('Enter Code')\n",
        "api_key= secret #your key\n",
        "\n",
        "# Static variables\n",
        "base_url_api_key = base_url+api_key\n",
        "\n",
        "# Time Format\n",
        "time_format = \"%Y-%m-%dT%H:%M:%S\"\n",
        "\n",
        "# Dynamic variables \n",
        "owner_credits=\"Credits @iam_muni_baa\"\n",
        "\n",
        "match_var = \"check\""
      ],
      "metadata": {
        "colab": {
          "base_uri": "https://localhost:8080/"
        },
        "id": "1wEsUtEGpvgl",
        "outputId": "99eeaacc-9206-4c40-bee0-14db26b5b4de"
      },
      "execution_count": 4,
      "outputs": [
        {
          "name": "stdout",
          "output_type": "stream",
          "text": [
            "Enter Code··········\n"
          ]
        }
      ]
    },
    {
      "cell_type": "markdown",
      "source": [
        "#Reciving messages or Reading data"
      ],
      "metadata": {
        "id": "yeR_pyIXqCmR"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "\n",
        "#Reciving messages or reading data\n",
        "def read_message(offset):\n",
        "  try:\n",
        "    base_url=base_url_api_key+\"/getUpdates\"\n",
        "    parameters = {\n",
        "        \"offset\":offset,\n",
        "    }\n",
        "    res = requests.get(base_url,data = parameters)\n",
        "\n",
        "    data = res.json() # REsponse data converted into json data\n",
        "    if data[\"result\"]:\n",
        "      chat_id = data[\"result\"][-1][\"message\"][\"chat\"][\"id\"] # Getting the chat_id from the read dat or recived message\n",
        "      if \"reply_to_message\" in res.text :\n",
        "        key_in_chat=data[\"result\"][-1][\"message\"][\"reply_to_message\"][\"text\"]\n",
        "        replay_message_id = data[\"result\"][-1][\"message\"][\"reply_to_message\"][\"message_id\"] # Getting replay data\n",
        "        replaying_text = key_check(key_in_chat[2:-2])\n",
        "        for result in data[\"result\"]:\n",
        "          if match_var in (result[\"message\"][\"text\"]): # Checking the recived messages matches with ours or not\n",
        "            send_message(chat_id,replay_message_id,replaying_text) # Calling Send Function\n",
        "\n",
        "      # # Debugging prints\n",
        "      # print(chat_id,  replay_message_id  )\n",
        "  except:\n",
        "    pass\n",
        "  if data[\"result\"]:\n",
        "\n",
        "    return data[\"result\"][-1][\"update_id\"] + 1 # Return the updated offset "
      ],
      "metadata": {
        "id": "rELqk6MjqFOd"
      },
      "execution_count": 11,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Sending Message code"
      ],
      "metadata": {
        "id": "sPtguvyMqui5"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "#sending Message code\n",
        "def send_message(chat_id,replay_message_id,replaying_text):\n",
        "  base_url=base_url_api_key+\"/sendMessage\"\n",
        "  # Parameters that we are used for \n",
        "  parameters = {\n",
        "      \"chat_id\":chat_id, # For message we need to replay\n",
        "      \"text\":replaying_text, #Text Message replaying here\n",
        "      \"reply_to_message_id\": replay_message_id\n",
        "  }\n",
        "  res = requests.get(base_url,data = parameters)"
      ],
      "metadata": {
        "id": "8LMfdDjUqvF6"
      },
      "execution_count": 12,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Psiphon key decode\n",
        "Here its take the base64 value and return the decoded json data with date and time."
      ],
      "metadata": {
        "id": "hALPfDi9q-qM"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def key_check(pkey):\n",
        "  try:\n",
        "    key_dt = json.loads(base64.b64decode(pkey))\n",
        "    ExpirationDate= datetime.strptime(key_dt[\"Authorization\"][\"Expires\"][0:19],time_format)\n",
        "   \n",
        "    now = datetime.strptime(present_time(),time_format) # For getting the local time\n",
        "    \n",
        "    # print(ExpirationDate)\n",
        "    if ExpirationDate <= now:\n",
        "      return (f\"Expired {owner_credits}\")\n",
        "    elif ExpirationDate > now:\n",
        "      return (f\"{ExpirationDate - now} Reaming {owner_credits}\")\n",
        "\n",
        "  except :\n",
        "     return \"Not a valid data\""
      ],
      "metadata": {
        "id": "23c9_4vNrY2W"
      },
      "execution_count": 13,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# For getting present time \n",
        "By using the 'PYTZ' timezones\n"
      ],
      "metadata": {
        "id": "rsDVVeLmrxb5"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "def present_time():\n",
        "  now_utc = datetime.now(timezone('UTC'))\n",
        "  return now_utc.strftime(time_format)\n",
        "  # print(now_utc.strftime(format))"
      ],
      "metadata": {
        "id": "qKs7euJwr0gk"
      },
      "execution_count": 14,
      "outputs": []
    },
    {
      "cell_type": "markdown",
      "source": [
        "# Finally run code here"
      ],
      "metadata": {
        "id": "erILsY9zsMMx"
      }
    },
    {
      "cell_type": "code",
      "source": [
        "if __name__ == \"__main__\":\n",
        "  offset = 0\n",
        "  print(\"Iam working\")\n",
        "  print(\"\")\n",
        "  try:\n",
        "    while True: # Its looking for messages always\n",
        "      offset = read_message(offset)\n",
        "      # print(offset)\n",
        "  except KeyboardInterrupt:\n",
        "    pass"
      ],
      "metadata": {
        "id": "5IwUkv-gsQaZ"
      },
      "execution_count": null,
      "outputs": []
    }
  ]
}