If you see this page, then we are watching you .
+ + diff --git a/nginx/nginx.conf b/nginx/nginx.conf new file mode 100644 index 00000000..8703a7c2 --- /dev/null +++ b/nginx/nginx.conf @@ -0,0 +1,21 @@ +events { + worker_connections 1024; # Adjust this number based on your requirements +} + +http { + include /etc/nginx/mime.types; + default_type application/octet-stream; + + server { + listen 8002; + server_name localhost; + + location / { + root /usr/share/nginx/html; + index index.html; + } + + # Additional location blocks and configurations can be added here + } +} + diff --git a/polybot/Dockerfile b/polybot/Dockerfile new file mode 100644 index 00000000..1b41f0e1 --- /dev/null +++ b/polybot/Dockerfile @@ -0,0 +1,39 @@ +# Install system dependencies and build tools +RUN apt-get update \ + && apt-get install -y --no-install-recommends \ + libglib2.0-0 \ + libgl1-mesa-glx \ + npm \ + zlib1g-dev \ + build-essential \ + python3-dev \ + && apt-get autoremove -y \ + && rm -rf /var/lib/apt/lists/* + +# Install Snyk globally using npm +RUN npm install -g snyk + +# Upgrade pip and install Python dependencies +COPY requirements.txt . +RUN python -m pip install --upgrade pip \ + && python -m pip install --no-cache-dir -r requirements.txt \ + && apt-get autoremove -y \ + && rm -rf /var/lib/apt/lists/* + +COPY .env .env + +# Copy the rest of the application code +COPY . . + + +# Ensure the directory structure is correct +RUN ls -la /app + +# Set the working directory to /app +WORKDIR /app + +# Expose port 5000 +EXPOSE 5000 + +# Command to run the bot when the container launches +CMD ["python3", "-m", "polybot.bot"] diff --git a/polybot/app.py b/polybot/app.py deleted file mode 100644 index e88600d1..00000000 --- a/polybot/app.py +++ /dev/null @@ -1,27 +0,0 @@ -import flask -from flask import request -import os -from bot import Bot, QuoteBot, ImageProcessingBot - -app = flask.Flask(__name__) - -TELEGRAM_TOKEN = os.environ['TELEGRAM_TOKEN'] -TELEGRAM_APP_URL = os.environ['TELEGRAM_APP_URL'] - - -@app.route('/', methods=['GET']) -def index(): - return 'Ok' - - -@app.route(f'/{TELEGRAM_TOKEN}/', methods=['POST']) -def webhook(): - req = request.get_json() - bot.handle_message(req['message']) - return 'Ok' - - -if __name__ == "__main__": - bot = ImageProcessingBot(TELEGRAM_TOKEN, TELEGRAM_APP_URL) - - app.run(host='0.0.0.0', port=8443) diff --git a/polybot/bot.py b/polybot/bot.py index 7fea847b..cdd65ce3 100644 --- a/polybot/bot.py +++ b/polybot/bot.py @@ -1,78 +1,156 @@ -import telebot -from loguru import logger import os -import time -from telebot.types import InputFile -from polybot.img_proc import Img - - -class Bot: - - def __init__(self, token, telegram_chat_url): - # create a new instance of the TeleBot class. - # all communication with Telegram servers are done using self.telegram_bot_client - self.telegram_bot_client = telebot.TeleBot(token) - - # remove any existing webhooks configured in Telegram servers - self.telegram_bot_client.remove_webhook() - time.sleep(0.5) - - # set the webhook URL - self.telegram_bot_client.set_webhook(url=f'{telegram_chat_url}/{token}/', timeout=60) - - logger.info(f'Telegram Bot information\n\n{self.telegram_bot_client.get_me()}') - - def send_text(self, chat_id, text): - self.telegram_bot_client.send_message(chat_id, text) - - def send_text_with_quote(self, chat_id, text, quoted_msg_id): - self.telegram_bot_client.send_message(chat_id, text, reply_to_message_id=quoted_msg_id) - - def is_current_msg_photo(self, msg): - return 'photo' in msg - - def download_user_photo(self, msg): - """ - Downloads the photos that sent to the Bot to `photos` directory (should be existed) - :return: - """ - if not self.is_current_msg_photo(msg): - raise RuntimeError(f'Message content of type \'photo\' expected') - - file_info = self.telegram_bot_client.get_file(msg['photo'][-1]['file_id']) - data = self.telegram_bot_client.download_file(file_info.file_path) - folder_name = file_info.file_path.split('/')[0] - - if not os.path.exists(folder_name): - os.makedirs(folder_name) - - with open(file_info.file_path, 'wb') as photo: - photo.write(data) - - return file_info.file_path - - def send_photo(self, chat_id, img_path): - if not os.path.exists(img_path): - raise RuntimeError("Image path doesn't exist") - - self.telegram_bot_client.send_photo( - chat_id, - InputFile(img_path) - ) - - def handle_message(self, msg): - """Bot Main message handler""" - logger.info(f'Incoming message: {msg}') - self.send_text(msg['chat']['id'], f'Your original message: {msg["text"]}') - - -class QuoteBot(Bot): - def handle_message(self, msg): - logger.info(f'Incoming message: {msg}') - - if msg["text"] != 'Please don\'t quote me': - self.send_text_with_quote(msg['chat']['id'], msg["text"], quoted_msg_id=msg["message_id"]) - - -class ImageProcessingBot(Bot): - pass +import telebot +import cv2 +from dotenv import load_dotenv +from img_proc import Img + +# load environment variables +load_dotenv() + +# Check if TELEGRAM_TOKEN is set in the .env file +TELEGRAM_TOKEN = os.getenv('TELEGRAM_TOKEN') +if not TELEGRAM_TOKEN: + print("Error: TELEGRAM_TOKEN is not set in the .env file.") + exit(1) + +# initialize telegram-bot +bot = telebot.TeleBot(TELEGRAM_TOKEN) + +# Dictionary to store images temporarily +user_images = {} + +# handler for the /start command +@bot.message_handler(commands=['start']) +def handle_start(message): + bot.send_message(message.chat.id, + "Hello!\n\nSend me an image and choose a filter:\n" + "- Blur: Reduce noise and detail.\n" + "- Rotate: Turn the image upside down.\n" + "- Salt and Pepper: Add random bright and dark pixels.\n" + "- Segment: Divide the image based on color.\n" + "- Grayscale: Convert to grayscale.\n" + "- Sharpen: Enhance edges and details.\n" + "- Emboss: Create a raised effect.\n" + "- Invert Colors: Invert the image colors.\n" + "- Oil Painting: Apply an oil painting-like effect.\n" + "- Cartoonize: Create a cartoon-like version.\n") + +# handler for receiving photos +@bot.message_handler(content_types=['photo']) +def handle_image(message): + try: + print("Received a photo message") + # get the photo file id + file_id = message.photo[-1].file_id + # get the file object using the file id + file_info = bot.get_file(file_id) + # download the file + downloaded_file = bot.download_file(file_info.file_path) + + # save the file temporarily with a unique name based on the file id + image_path = f"images/{file_id}.jpg" + with open(image_path, 'wb') as new_file: + new_file.write(downloaded_file) + + # check if this is the first image or the second image for concatenation + if message.chat.id in user_images: + print("User already has an image in memory") + if 'concat_pending' in user_images[message.chat.id]: + print("This is the second image for concatenation") + # this is the second image for concatenation + second_image_path = image_path + first_image_path = user_images[message.chat.id]['concat_pending'] + del user_images[message.chat.id]['concat_pending'] + + # load the images + img_processor = Img(first_image_path) + first_image_data = cv2.imread(first_image_path) + second_image_data = cv2.imread(second_image_path) + + # concatenate the images + concatenated_image = img_processor.concat(second_image_data) + if concatenated_image is not None: + print("Concatenation successful") + # save and send the concatenated image + processed_image_path = img_processor.save_image(concatenated_image, suffix='_concatenated') + with open(processed_image_path, 'rb') as photo_file: + bot.send_photo(message.chat.id, photo_file) + else: + print("Error concatenating images.") + bot.reply_to(message, "Error concatenating images.") + + # clear user history + del user_images[message.chat.id] + else: + # this is the first image + print("This is the first image for concatenation") + user_images[message.chat.id]['concat_pending'] = image_path + bot.reply_to(message, "First image saved successfully! Now please send the second image to concatenate with.") + else: + # this is the first image + print("This is the first image received") + user_images[message.chat.id] = {'concat_pending': image_path} + bot.reply_to(message, "First image saved successfully! To apply the concatenation filter, please send another image or choose a filter from the list at the top of the page to apply a filter.") + except Exception as e: + print(f"Error handling image: {e}") + bot.reply_to(message, f"Error handling image: {e}") + +# handler for filter selection +@bot.message_handler( + func=lambda message: message.text.lower() in ['blur', 'rotate', 'salt and pepper', 'segment', 'grayscale', + 'sharpen', 'emboss', 'invert colors', 'oil painting', 'cartoonize']) +def handle_filter(message): + try: + # Check if the user has previously sent an image + if message.chat.id in user_images: + # Get the image path + if 'concat_pending' in user_images[message.chat.id]: + image_path = user_images[message.chat.id]['concat_pending'] + else: + image_path = user_images[message.chat.id]['first_image'] + + # apply the selected filter + img_processor = Img(image_path) + filter_name = message.text.lower() + + if filter_name == 'blur': + processed_image = img_processor.blur() + elif filter_name == 'rotate': + processed_image = img_processor.rotate() + elif filter_name == 'salt and pepper': + processed_image = img_processor.salt_n_pepper() + elif filter_name == 'segment': + processed_image = img_processor.segment() + elif filter_name == 'grayscale': + processed_image = img_processor.grayscale() + elif filter_name == 'sharpen': + processed_image = img_processor.sharpen() + elif filter_name == 'emboss': + processed_image = img_processor.emboss() + elif filter_name == 'invert colors': + processed_image = img_processor.invert_colors() + elif filter_name == 'oil painting': + processed_image = img_processor.oil_painting() + elif filter_name == 'cartoonize': + processed_image = img_processor.cartoonize() + else: + processed_image = None + + # check if the filter was applied successfully + if processed_image is not None: + # save and send the processed image + processed_image_path = img_processor.save_image(processed_image, suffix=f'_{filter_name.replace(" ", "_")}') + with open(processed_image_path, 'rb') as photo_file: + bot.send_photo(message.chat.id, photo_file) + else: + bot.reply_to(message, f"Error applying {filter_name} filter: Result is None.") + + # remove the image path from the dict + del user_images[message.chat.id] + else: + bot.reply_to(message, "Please send an image first.") + except Exception as e: + bot.reply_to(message, f"Error processing image: {e}") + +# Start polling +bot.polling() diff --git a/polybot/img_proc.py b/polybot/img_proc.py index 137ca70d..df2f79a0 100644 --- a/polybot/img_proc.py +++ b/polybot/img_proc.py @@ -1,67 +1,193 @@ -from pathlib import Path -from matplotlib.image import imread, imsave - - -def rgb2gray(rgb): - r, g, b = rgb[:, :, 0], rgb[:, :, 1], rgb[:, :, 2] - gray = 0.2989 * r + 0.5870 * g + 0.1140 * b - return gray - +import os +import cv2 +import numpy as np class Img: - - def __init__(self, path): - """ - Do not change the constructor implementation - """ - self.path = Path(path) - self.data = rgb2gray(imread(path)).tolist() - - def save_img(self): - """ - Do not change the below implementation - """ - new_path = self.path.with_name(self.path.stem + '_filtered' + self.path.suffix) - imsave(new_path, self.data, cmap='gray') - return new_path + def __init__(self, image_path): + self.image_path = image_path + self.image_data = self.load_image() + + def load_image(self): + try: + image = cv2.imread(self.image_path) + if image is not None: + return image + else: + raise FileNotFoundError(" Unable to load image.") + except Exception as e: + print(f"Error loading image: {e}") + return None + + def save_image(self, image_data, suffix='_filtered'): + try: + directory = 'images' + if not os.path.exists(directory): + os.makedirs(directory) + + file_path = os.path.join(directory, f"{os.path.basename(self.image_path).split('.')[0]}{suffix}.jpg") + cv2.imwrite(file_path, image_data) + print(f"Image saved successfully: {file_path}") + return file_path + except Exception as e: + print(f"Error saving image: {e}") + return None def blur(self, blur_level=16): - - height = len(self.data) - width = len(self.data[0]) - filter_sum = blur_level ** 2 - - result = [] - for i in range(height - blur_level + 1): - row_result = [] - for j in range(width - blur_level + 1): - sub_matrix = [row[j:j + blur_level] for row in self.data[i:i + blur_level]] - average = sum(sum(sub_row) for sub_row in sub_matrix) // filter_sum - row_result.append(average) - result.append(row_result) - - self.data = result - - def contour(self): - for i, row in enumerate(self.data): - res = [] - for j in range(1, len(row)): - res.append(abs(row[j-1] - row[j])) - - self.data[i] = res + try: + if self.image_data is None: + raise ValueError("No image data available.") + blur_level = max(1, blur_level) + blur_level = blur_level + 1 if blur_level % 2 == 0 else blur_level + blurred_image = cv2.GaussianBlur(self.image_data, (blur_level, blur_level), 0) + return blurred_image + except Exception as e: + print(f"Error applying blur: {e}") + return None def rotate(self): - # TODO remove the `raise` below, and write your implementation - raise NotImplementedError() - - def salt_n_pepper(self): - # TODO remove the `raise` below, and write your implementation - raise NotImplementedError() - - def concat(self, other_img, direction='horizontal'): - # TODO remove the `raise` below, and write your implementation - raise NotImplementedError() - - def segment(self): - # TODO remove the `raise` below, and write your implementation - raise NotImplementedError() + try: + img = cv2.imread(self.image_path) + if img is None: + raise FileNotFoundError("Unable to load image.") + rotated_img = cv2.rotate(img, cv2.ROTATE_180) + return rotated_img + except Exception as e: + print(f"Error rotating image: {e}") + return None + + def salt_n_pepper(self, amount=0.05): + try: + if self.image_data is None: + raise ValueError("No image data available.") + noisy_image = self.image_data.copy() + mask = np.random.choice([0, 1, 2], size=noisy_image.shape[:2], p=[amount / 2, amount / 2, 1 - amount]) + noisy_image[mask == 0] = 0 + noisy_image[mask == 1] = 255 + return noisy_image + except Exception as e: + print(f"Error adding salt and pepper noise: {e}") + return None + + def concat(self, other_image_data, direction='horizontal'): + try: + if self.image_data is None or other_image_data is None: + raise ValueError("Image data is missing.") + + if direction not in ['horizontal', 'vertical']: + raise ValueError("Invalid direction. Please use 'horizontal' or 'vertical'.") + + if direction == 'horizontal': + concatenated_img = np.concatenate((self.image_data, other_image_data), axis=1) + else: + concatenated_img = np.concatenate((self.image_data, other_image_data), axis=0) + + return concatenated_img + except Exception as e: + print(f"Error concatenating images: {e}") + return None + + def segment(self, num_clusters=100): + try: + if self.image_data is None: + raise ValueError("No image data available.") + + image_rgb = cv2.cvtColor(self.image_data, cv2.COLOR_BGR2RGB) + pixels = image_rgb.reshape((-1, 3)) + pixels = np.float32(pixels) + + criteria = (cv2.TERM_CRITERIA_EPS + cv2.TERM_CRITERIA_MAX_ITER, 100, 0.2) + _, labels, centers = cv2.kmeans(pixels, num_clusters, None, criteria, 10, cv2.KMEANS_RANDOM_CENTERS) + + centers = np.uint8(centers) + segmented_image = centers[labels.flatten()] + segmented_image = segmented_image.reshape(image_rgb.shape) + + # Convert segmented image back to BGR format + segmented_image_bgr = cv2.cvtColor(segmented_image, cv2.COLOR_RGB2BGR) + + # Darken the segmented image + segmented_image_bgr = segmented_image_bgr * 0.5 # Reduce brightness by 50% + + return segmented_image_bgr + except Exception as e: + print(f"Error segmenting image: {e}") + return None + + + + def grayscale(self): + try: + if self.image_data is None: + raise ValueError("No image data available.") + gray_image = cv2.cvtColor(self.image_data, cv2.COLOR_BGR2GRAY) + return cv2.cvtColor(gray_image, cv2.COLOR_GRAY2BGR) + except Exception as e: + print(f"Error converting image to grayscale: {e}") + return None + + def sharpen(self): + try: + if self.image_data is None: + raise ValueError("No image data available.") + kernel = np.array([[-1, -1, -1], [-1, 9, -1], [-1, -1, -1]]) + sharpened_image = cv2.filter2D(self.image_data, -1, kernel) + return sharpened_image + except Exception as e: + print(f"Error applying sharpen filter: {e}") + return None + + def emboss(self): + try: + if self.image_data is None: + raise ValueError("No image data available.") + kernel = np.array([[0, -1, -1], [1, 0, -1], [1, 1, 0]]) + embossed_image = cv2.filter2D(self.image_data, -1, kernel) + return embossed_image + except Exception as e: + print(f"Error applying emboss filter: {e}") + return None + + def invert_colors(self): + try: + if self.image_data is None: + raise ValueError("No image data available.") + inverted_image = cv2.bitwise_not(self.image_data) + return inverted_image + except Exception as e: + print(f"Error inverting colors: {e}") + return None + + def oil_painting(self, size=7, dynRatio=0.2): + try: + if self.image_data is None: + raise ValueError("No image data available.") + + # Convert image to grayscale + gray_image = cv2.cvtColor(self.image_data, cv2.COLOR_BGR2GRAY) + + # Apply median blur to create a smoother image + blurred_image = cv2.medianBlur(gray_image, size) + + # Apply adaptive threshold to create a binary image + _, mask = cv2.threshold(blurred_image, 0, 255, cv2.THRESH_BINARY_INV + cv2.THRESH_OTSU) + + # Create an output image using bitwise and operation with original image + oil_painting_effect = cv2.bitwise_and(self.image_data, self.image_data, mask=mask) + + return oil_painting_effect + except Exception as e: + print(f"Error applying oil painting effect: {e}") + return None + def cartoonize(self): + try: + if self.image_data is None: + raise ValueError("No image data available.") + gray_image = cv2.cvtColor(self.image_data, cv2.COLOR_BGR2GRAY) + blurred_image = cv2.medianBlur(gray_image, 7) + edges = cv2.adaptiveThreshold(blurred_image, 255, cv2.ADAPTIVE_THRESH_MEAN_C, cv2.THRESH_BINARY, 9, 9) + color = cv2.bilateralFilter(self.image_data, 9, 300, 300) + cartoon_image = cv2.bitwise_and(color, color, mask=edges) + return cartoon_image + except Exception as e: + print(f"Error cartoonizing image: {e}") + return None \ No newline at end of file diff --git a/polybot/requirements.txt b/polybot/requirements.txt index dbab2768..4b13bb11 100644 --- a/polybot/requirements.txt +++ b/polybot/requirements.txt @@ -2,4 +2,10 @@ pyTelegramBotAPI>=4.12.0 loguru>=0.7.0 requests>=2.31.0 flask>=2.3.2 -matplotlib \ No newline at end of file +matplotlib>=3.7.5 +pylint>=3.2.3 +opencv-python>=4.5.3.56 +telebot==0.0.4 +python-dotenv>=0.19.2 +pytest>=6.2.4 +unittest2>=1.1.0 diff --git a/pr-testing.Jenkinsfile b/pr-testing.Jenkinsfile new file mode 100644 index 00000000..9c78e48a --- /dev/null +++ b/pr-testing.Jenkinsfile @@ -0,0 +1,32 @@ + +pipeline { + + + agent any + + stages { + stage('Unittest'){ + steps{ + sh 'echo "testing"' + } + } + stage('Lint') { + steps { + sh ''' + cd polybot + pip install -r requirements.txt + python3 -m pylint *.py + ''' + } + } + + stage('Functional test') { + steps { + sh 'echo "testing "' + } + } + + + + } +} diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 00000000..4b13bb11 --- /dev/null +++ b/requirements.txt @@ -0,0 +1,11 @@ +pyTelegramBotAPI>=4.12.0 +loguru>=0.7.0 +requests>=2.31.0 +flask>=2.3.2 +matplotlib>=3.7.5 +pylint>=3.2.3 +opencv-python>=4.5.3.56 +telebot==0.0.4 +python-dotenv>=0.19.2 +pytest>=6.2.4 +unittest2>=1.1.0 diff --git a/tests/test_sample.py b/tests/test_sample.py new file mode 100644 index 00000000..a0bbfe47 --- /dev/null +++ b/tests/test_sample.py @@ -0,0 +1,8 @@ +import unittest + +class TestSample(unittest.TestCase): + def test_example(self): + self.assertEqual(1, 1) + +if __name__ == '__main__': + unittest.main() \ No newline at end of file