feat: add telegram bot for cowin alert

Signed-off-by: Ameya Shenoy <shenoy.ameya@gmail.com>

Dockerfile

@@ -0,0 +1,11 @@
+FROM python:3.9.4-alpine3.13
+
+COPY requirements.txt /
+
+RUN set -ex \
+  && pip install -r requirements.txt
+
+COPY app.py /
+
+CMD python app.py
+

app.py

@@ -0,0 +1,67 @@
+# standard imports
+import os
+import time
+import logging
+from datetime import datetime
+
+# third-party imports
+import pytz
+import requests
+import telegram
+
+IST = pytz.timezone('Asia/Kolkata')
+
+logging.basicConfig(format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+                    level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+# customizable vars
+AGE = os.environ.get('AGE', 18)
+DIST_CODE = int(os.environ.get('DIST_CODE'))
+TELEGRAM_BOT_TOKEN = os.environ.get('TELEGRAM_BOT_TOKEN')
+TELEGRAM_CHAT_ID = int(os.environ.get('TELEGRAM_CHAT_ID'))
+SLEEP_TIME_SEC = int(os.environ.get('SLEEP_TIME_SEC', 60))
+
+
+def main():
+    message_header = "Atleast 1 new slot seems to have just opened up for vaccination.\n\n"
+    message_footer = "Visit [this link](https://selfregistration.cowin.gov.in/appointment) to book a slot for yourself."
+
+    logger.info('Initiating loop')
+    while True:
+        logger.info('Hitting CoWIN API')
+        date_str = datetime.now(IST).strftime("%d-%m-%Y")
+
+        url = "https://cdn-api.co-vin.in/api/v2/appointment/sessions/public/calendarByDistrict?district_id={}&date={}".format(DIST_CODE, date_str)
+        response = requests.get(url)
+
+        available_center_list = []
+        if response.ok:
+            resp_json = response.json()
+            for center in resp_json.get('centers'):
+                for session in center.get('sessions'):
+                    if session.get('available_capacity') > 0:
+                        curr = {
+                            'location': center.get('name'),
+                            'available_capacity': session.get('available_capacity'),
+                            'available_slots': ', '.join(session.get('slots')),
+                            'vaccine': session.get('vaccine'),
+                            'fee_type': center.get('fee_type'),
+                        }
+                        available_center_list.append(curr)
+
+        bot = telegram.Bot(token=TELEGRAM_BOT_TOKEN)
+        message_body = ''
+        if available_center_list:
+            for center in available_center_list:
+                msg = f"Location: {center.get('location')}\nAvailable Capacity: {center.get('available_capacity')}\nAvailable Slots: {center.get('available_slots')}\nVaccine: {center.get('vaccine')}\nFee: {center.get('fee_type')}\n\n"
+                message_body += msg
+                break  # to prevent too long messages
+            msg = message_header + message_body + message_footer
+            bot.sendMessage(chat_id=TELEGRAM_CHAT_ID, text=msg, parse_mode='markdown')
+        time.sleep(SLEEP_TIME_SEC)
+
+if __name__ == '__main__':
+    main()
+

requirements.txt

@@ -0,0 +1,3 @@
+python-telegram-bot==13.5
+requests==2.25.1
+pytz==2021.1