PR-Testing process

.gitignore

@@ -8,6 +8,9 @@ __pycache__/
 # C extensions
 *.so
 
+# Telegram stuff
+*.telegramFile
+*.telegramToken
 
 # ignore .pem file
 *.pem
@@ -154,3 +157,4 @@ crashlytics.properties
 crashlytics-build.properties
 fabric.properties
 
+/asg-config.json

Dockerfile

@@ -1,7 +1,6 @@
-FROM python:3.8.12-slim-buster
-
-# YOUR COMMANDS HERE
-# ....
-# ....
-
+FROM python:3.8-slim-bullseye
+WORKDIR /botapp
+LABEL app=bot
+COPY . .
+RUN pip install -r requirements.txt
 CMD ["python3", "bot.py"]

PR.Jenkinsfile

@@ -0,0 +1,39 @@
+pipeline {
+    agent any
+
+    stages {
+        stage('Unittest') {
+            steps {
+                sh '''
+                pip3 install -r requirements.txt
+                python3 -m pytest --junitxml results.xml tests
+                '''
+            }
+            post {
+                always {
+                    junit allowEmptyResults: true, testResults: 'results.xml'
+                }
+            }
+        }
+        stage('Functional test') {
+            steps {
+                echo "testing"
+            }
+        }
+        stage('Static code linting') {
+            steps {
+              sh 'python3 -m pylint -f parseable --reports=no *.py > pylint.log'
+            }
+            post {
+              always {
+                sh 'cat pylint.log'
+                recordIssues (
+                  enabledForFailure: true,
+                  aggregatingResults: true,
+                  tools: [pyLint(name: 'Pylint', pattern: '**/pylint.log')]
+                )
+              }
+            }
+        }
+    }
+}

bot.py

@@ -1,6 +1,12 @@
+import json
+import threading
+import botocore
 from telegram.ext import Updater, MessageHandler, Filters
-from utils import search_download_youtube_video
 from loguru import logger
+import boto3
+from boto3.dynamodb.conditions import Key
+from utils import calc_backlog_per_instance, search_youtube_video, send_videos_from_queue2
+
 
 
 class Bot:
@@ -26,10 +32,13 @@ def send_video(self, update, context, file_path):
         """Sends video to a chat"""
         context.bot.send_video(chat_id=update.message.chat_id, video=open(file_path, 'rb'), supports_streaming=True)
 
-    def send_text(self, update,  text, quote=False):
+    def send_text(self, update, text, chat_id=None, quote=False):
         """Sends text to a chat"""
-        # retry https://github.com/python-telegram-bot/python-telegram-bot/issues/1124
-        update.message.reply_text(text, quote=quote)
+        if chat_id:
+            self.updater.bot.send_message(chat_id, text=text)
+        else:
+            # retry https://github.com/python-telegram-bot/python-telegram-bot/issues/1124
+            update.message.reply_text(text, quote=quote)
 
 
 class QuoteBot(Bot):
@@ -42,14 +51,68 @@ def _message_handler(self, update, context):
         self.send_text(update, f'Your original message: {update.message.text}', quote=to_quote)
 
 
-class YoutubeBot(Bot):
-    pass
+class YoutubeObjectDetectBot(Bot):
+    def __init__(self, token):
+        super().__init__(token)
+        threading.Thread(
+            target=calc_backlog_per_instance,
+            args=(workers_queue, asg, config.get("autoscaling_group_name"), config.get('aws_region'))
+        ).start()
+        threading.Thread(
+            target=send_videos_from_queue2,
+            args=(worker_to_bot_queue, config.get('bucket_name'))
+        ).start()
+
+    def _message_handler(self, update, context):
+        try:
+            chat_id = str(update.effective_message.chat_id)
+            if update.message.text.startswith('/myvideos'):
+                response = table.query(KeyConditionExpression=Key('chatId').eq(chat_id))
+                for key, value in response.items():
+                    if isinstance(value, list):
+                        array_length = len(value)
+                        for i in range(array_length):
+                            temp_dict = value[i]
+                            video_url = temp_dict['url']
+                            video = search_youtube_video(None, video_url)
+                            self.send_text(update, f'Video Name: {video["title"]}, Video Link: {video["webpage_url"]}', chat_id=chat_id)
+                logger.info(f'sent videos information to client, chat_id: {chat_id}')
+            else:
+                response = workers_queue.send_message(
+                    MessageBody=update.message.text,
+                    MessageAttributes={
+                        'chat_id': {'StringValue': chat_id, 'DataType': 'String'}
+                    }
+                )
+                logger.info(f'msg {response.get("MessageId")} has been sent to queue')
+                self.send_text(update, f'Your message is being processed...', chat_id=chat_id)
+                for video in search_youtube_video(update.message.text, None):
+                    item = {
+                        'chatId': chat_id,
+                        'videoId': video['id'],
+                        'url': video['webpage_url'],
+                        'title': video['title']
+                    }
+                    response2 = table.put_item(Item=item)
+
+        except botocore.exceptions.ClientError as error:
+            logger.error(error)
+            self.send_text(update, f'Something went wrong, please try again...')
 
 
 if __name__ == '__main__':
     with open('.telegramToken') as f:
         _token = f.read()
 
-    my_bot = Bot(_token)
-    my_bot.start()
+    with open('config.json') as f:
+        config = json.load(f)
+
+    sqs = boto3.resource('sqs', region_name=config.get('aws_region'))
+    workers_queue = sqs.get_queue_by_name(QueueName=config.get('bot_to_worker_queue_name'))
+    worker_to_bot_queue = sqs.get_queue_by_name(QueueName=config.get('worker_to_bot_queue_name'))
+    asg = boto3.client('autoscaling', region_name=config.get('aws_region'))
+    dynamodb = boto3.resource('dynamodb', region_name=config.get('aws_region'))
+    table = dynamodb.Table(config.get('table_name'))
 
+    my_bot = YoutubeObjectDetectBot(_token)
+    my_bot.start()

config.json

@@ -0,0 +1,9 @@
+{
+  "aws_region": "eu-central-1",
+  "bot_to_worker_queue_name": "daniel-reuven-aws-ex1-polybot-queue",
+  "worker_to_bot_queue_name": "daniel-reuven-aws-ex1-polybot-queue2",
+  "autoscaling_group_name": "daniel-reuven-polybot-asg",
+  "cloudwatch_namespace": "cloudwatch-namespace",
+  "bucket_name": "daniel-reuven-aws-ex1-polybot-bucket",
+  "table_name": "daniel-reuven-aws-ex1-polybot-ddb-table1"
+}

requirements.txt

@@ -1,3 +1,9 @@
 python-telegram-bot>=13.11
-youtube-dl>=2021.12.17
-loguru
+yt-dlp>=2022.6.29
+loguru~=0.6.0
+botocore~=1.27.13
+boto3~=1.24.13
+requests~=2.28.1
+unittest2~=1.1.0
+pytest
+pylint

tests/test_autoscaling_metric.py

@@ -0,0 +1,50 @@
+import unittest2 as unittest
+from unittest.mock import Mock
+from utils import calc_backlog_per_instance
+
+# run by `PYTHONPATH=. python3 -m pytest --junitxml results.xml tests`
+
+
+class TestBacklogPerInstanceMetric(unittest.TestCase):
+    def setUp(self):
+        self.sqs_queue_client = Mock()
+        self.asg_client = Mock()
+
+    def test_no_worker_full_queue(self):
+        self.sqs_queue_client.attributes = {
+            'ApproximateNumberOfMessages': '100'
+        }
+
+        self.asg_client.describe_auto_scaling_groups = Mock(return_value={
+            'AutoScalingGroups': [{
+                'DesiredCapacity': 0
+            }]
+        })
+
+        self.assertEqual(calc_backlog_per_instance(self.sqs_queue_client, self.asg_client, None, None), 99)
+
+    def test_no_workers_empty_queue(self):
+        self.sqs_queue_client.attributes = {
+            'ApproximateNumberOfMessages': '0'
+        }
+
+        self.asg_client.describe_auto_scaling_groups = Mock(return_value={
+            'AutoScalingGroups': [{
+                'DesiredCapacity': 0
+            }]
+        })
+
+        self.assertEqual(calc_backlog_per_instance(self.sqs_queue_client, self.asg_client, None, None), 0)
+
+    def test_2_workers_100_msgs_in_queue(self):
+        self.sqs_queue_client.attributes = {
+            'ApproximateNumberOfMessages': '100'
+        }
+
+        self.asg_client.describe_auto_scaling_groups = Mock(return_value={
+            'AutoScalingGroups': [{
+                'DesiredCapacity': 2
+            }]
+        })
+
+        self.assertEqual(calc_backlog_per_instance(self.sqs_queue_client, self.asg_client, None, None), 50)