import cv2 import numpy as np import pyautogui import threading import time import os import sounddevice as sd from selenium import webdriver from selenium.webdriver.chrome.service import Service from webdriver_manager.chrome import ChromeDriverManager from selenium.webdriver.common.by import By from datetime import datetime from azure_connection import upload_to_azure_with_signed_url, upload_img_to_azure_with_signed_url import base64 from ext_api_func import * from selenium.webdriver.support.ui import WebDriverWait from selenium.webdriver.support import expected_conditions as EC from selenium.webdriver.chrome.options import Options from json_evaluation import json_report from screen_record import record_screen, stop_video_recording from record_audio import start_audio_recording, stop_audio_recording, merge_audio_video # from merge_audio import merge_audio_video import re import logging # Global variable to control the recording process is_recording = False output = None logger = logging.getLogger("pm2_logger") logger.setLevel(logging.INFO) handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - %(levelname)s - %(message)s') handler.setFormatter(formatter) logger.addHandler(handler) # Define the codec fourcc = cv2.VideoWriter_fourcc(*"mp4v") # Global variables to track recording status audio_thread = None video_thread = None def extract_value_in_parentheses(input_string): """Extract the value inside parentheses from a string.""" match = re.search(r'\((\d+)\)', input_string) if match: return int(match.group(1)) # Return the number inside parentheses as an integer else: return None # Return None if no value is found inside parentheses def remove_file(file_path): try: if os.path.exists(file_path): os.remove(file_path) print(f"File removed: {file_path}") else: print(f"File not found, skipping: {file_path}") except Exception as e: print(f"Error occurred while removing {file_path}: {e}") def file_to_base64(file_path): try: # Open the file in binary mode with open(file_path, "rb") as file: # Read the file's content and encode it to Base64 base64_string = base64.b64encode(file.read()).decode('utf-8') return base64_string except FileNotFoundError: return "File not found. Please provide a valid file path." except Exception as e: return f"An error occurred: {str(e)}" def take_screenshot(uniqueid): try: output_path = f"{uniqueid}.png" # Capture the screenshot screenshot = pyautogui.screenshot() # Save the screenshot to the specified file screenshot.save(output_path) print(f"Screenshot saved at {output_path}") return output_path except Exception as e: print(f"Error taking screenshot: {e}") return None def stop_recording(audio_file,video_file,output_file): """ Stops both the screen and audio recording. """ global audio_thread, video_thread # Stop audio recording stop_audio_recording(audio_file) stop_video_recording() audio_thread.join() # Wait for the audio thread to finish # Stop video recording (we'll stop the screen recording once the function is called) video_thread.join() # Wait for the video thread to finish # Merge audio and video files merge_audio_video(video_file, audio_file, output_file) print(f"Final MP4 video saved as {output_file}") # def start_recording(audio_file,video_file): # """ # Starts both the screen and audio recording. # """ # global audio_thread, video_thread # # Start audio recording in a separate thread # audio_thread = threading.Thread(target=start_audio_recording, args=(audio_file,)) # audio_thread.start() # print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>") # # Start screen recording in a separate thread # video_thread = threading.Thread(target=record_screen, args=(video_file,)) # video_thread.start() # print("Recording started. Call stop_recording() to stop.") # Function to recheck the roster text periodically def recheck_roster_text(driver, roster_xpath,uniqueId): while True: time.sleep(60) # Wait for 5 minutes print("Rechecking roster text...") try: roster_text = driver.find_element(By.XPATH, roster_xpath).text print(f"Rechecked roster text: {roster_text}") participants =extract_value_in_parentheses(roster_text) # Check for "In this meeting (1)" if roster_text == "In this meeting (1)": print("No member in the meeting.") leave_button_xpath = '//*[@id="hangup-button"]/button' driver.find_element(By.XPATH, leave_button_xpath).click() print("'Leave' button clicked") break elif participants >= 3 : print("Taking screenshot") screenshot_path = take_screenshot(uniqueId) except Exception as e: print(f"Error while rechecking roster text: {e}") break # def upload_chunk_in_realtime(frame, frame_count): # chunk_file_name = f"chunk_{frame_count}.mp4" # cv2.imwrite(chunk_file_name, frame) # blob_name = f"chunks/{chunk_file_name}" # return upload_to_azure_with_signed_url(chunk_file_name, blob_name) # def start_recording(video_file_name): # global is_recording, output, screen_width, screen_height # # Get the screen size # screen_width, screen_height = pyautogui.size() # # Create a VideoWriter object to save the video in MP4 format # output = cv2.VideoWriter(video_file_name, fourcc, 20.0, (screen_width, screen_height)) # is_recording = True # frame_count = 0 # while is_recording: # # Take a screenshot # screenshot = pyautogui.screenshot() # # Convert the screenshot to a numpy array # frame = np.array(screenshot) # # Convert RGB to BGR (OpenCV uses BGR) # frame = cv2.cvtColor(frame, cv2.COLOR_RGB2BGR) # # Write the frame to the video # output.write(frame) # # Upload the current frame chunk # blob_url = upload_chunk_in_realtime(frame, frame_count) # if blob_url: # print(f"Uploaded chunk {frame_count}: {blob_url}") # frame_count += 1 # # Release the VideoWriter # output.release() # cv2.destroyAllWindows() # def stop_recording(): # global is_recording # is_recording = False # def run_meeting_recording(meeting_link, input_time, input_date, uniqueId, node_url,interviewId, max_retries=3, delay=5): # print("Waiting for the scheduled time...") # output_audio_file = uniqueId + ".wav" # output_video_file = uniqueId + ".avi" # output_merge_file = uniqueId + ".mp4" # print("Output file assigned:", output_audio_file) # # Wait until current time and date match the input # while True: # current_time = datetime.now().strftime("%H:%M") # current_date = datetime.now().strftime("%Y-%m-%d") # if input_time == current_time and input_date == current_date: # break # time.sleep(1) # print("Started recording session.") # # Set up ChromeDriver using Service with headless options # try: # options = Options() # options.add_argument("--no-sandbox") # Necessary for certain Linux environments # options.add_argument("--disable-dev-shm-usage") # Fixes some Docker issues # options.add_argument("--disable-gpu") # Disables GPU hardware acceleration # service = Service(ChromeDriverManager().install()) # driver = webdriver.Chrome(service=service, options=options) # except Exception as e: # print(f"Error initializing ChromeDriver: {e}") # retry_count = 0 # while retry_count < max_retries: # try: # print(f"Attempt {retry_count + 1} of {max_retries}") # try: # # Navigate to the Teams URL # driver.get(meeting_link) # print("Navigated to Teams") # time.sleep(60) # Wait for the page to load # print("Enter name") # # Automatically enter the name 'bot' into the input field # # name_input_xpath = '/html/body/div[1]/div/div/div/div[5]/div/div[1]/div/div/div/div/div[2]/div[1]/div/div/div[2]/div[2]/span/input' # name_input_xpath = '//*[@id="app"]/div/div/div/div[4]/div/div[1]/div/div/div/div/div[2]/div[1]/div/div/div[2]/div[2]/span/input' # # Wait for the input element to be visible and interactable # # WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.XPATH, name_input_xpath))) # print("Name input field is ready") # driver.find_element(By.XPATH, name_input_xpath).send_keys('bot') # print('Entered name: bot') # # Automatically click the Join button # time.sleep(5) # join_button_xpath = '//*[@id="prejoin-join-button"]' # driver.find_element(By.XPATH, join_button_xpath).click() # print('Join button clicked') # # # Start recording in a separate thread # # recording_thread = threading.Thread(target=start_recording, args=(output_audio_file,output_video_file,)) # # recording_thread.start() # global audio_thread, video_thread # # Start audio recording in a separate thread # audio_thread = threading.Thread(target=start_audio_recording, args=(output_audio_file,)) # audio_thread.start() # print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>") # # # Start screen recording in a separate thread # # video_thread = threading.Thread(target=record_screen, args=(output_video_file,)) # # video_thread.start() # print("Recording started. Call stop_recording() to stop.") # # Wait for the meeting interface to fully load # time.sleep(10) # # Click on the "People" button # print("Clicking the 'People' button...") # people_button_xpath = '//*[@id="roster-button"]' # driver.find_element(By.XPATH, people_button_xpath).click() # print("'People' button clicked") # # Extract and print the value at the specified XPath # print("Extracting value from roster title section...") # roster_xpath = '//*[@id="roster-title-section-2"]/span' # WebDriverWait(driver, 20).until(EC.presence_of_element_located((By.XPATH, roster_xpath))) # roster_text = driver.find_element(By.XPATH, roster_xpath).text # print("-------------------------------------------------") # print(f"Value in roster title section: {roster_text}") # # Wait for the next 10 minutes # print("Waiting for 10 minutes...") # # time.sleep(600) # 600 seconds = 10 minutes # print("Started rechecking roster text...") # # Start a thread to recheck roster text every 5 minutes # recheck_roster_text(driver,roster_xpath,uniqueId) # finally: # # Stop the recording # # stop_recording(output_audio_file,output_video_file,output_merge_file) # stop_recording(output_audio_file) # # Stop the browser after a delay # time.sleep(10) # driver.quit() # print("Browser closed.") # # Upload the final video to Azure # screenshot_path = uniqueId + ".png" # # blob_url = upload_to_azure_with_signed_url(output_merge_file, output_merge_file) # blob_url = "https://interviewbotstorage.blob.core.windows.net/interviewbot/" + uniqueId + ".mp4" # screenshot_url = upload_img_to_azure_with_signed_url(screenshot_path,screenshot_path) # json_data = json_report(output_audio_file) # json_data['snapshot'] = screenshot_url # print("json report done", json_data) # response = send_interview_report(node_url, interviewId, blob_url, json_data) # print("Node Api response:", response) # remove_file(output_audio_file) # remove_file(output_merge_file) # remove_file(output_video_file) # remove_file(screenshot_path) # print(blob_url) # print(screenshot_url) # except Exception as e: # print(f"Error during meeting recording: {e}") # retry_count += 1 # time.sleep(delay) # if retry_count == max_retries: # driver.quit() # print("Max retries reached. Exiting.") def run_meeting_recording(meeting_link, input_time, input_date, uniqueId, node_url,interviewId, max_retries=3, delay=5): print("Waiting for the scheduled time...") output_audio_file = uniqueId + ".wav" output_video_file = uniqueId + ".avi" output_merge_file = uniqueId + ".mp4" print("Output file assigned:", output_audio_file) # Wait until current time and date match the input while True: current_time = datetime.now().strftime("%H:%M") current_date = datetime.now().strftime("%Y-%m-%d") if input_time == current_time and input_date == current_date: break time.sleep(1) print("Started recording session.") # Set up ChromeDriver using Service with headless options try: options = Options() options.add_argument("--no-sandbox") # Necessary for certain Linux environments options.add_argument("--disable-dev-shm-usage") # Fixes some Docker issues options.add_argument("--disable-gpu") # Disables GPU hardware acceleration service = Service(ChromeDriverManager().install()) driver = webdriver.Chrome(service=service, options=options) except Exception as e: print(f"Error initializing ChromeDriver: {e}") try: # Navigate to the Teams URL driver.get(meeting_link) print("Navigated to Teams") time.sleep(60) # Wait for the page to load print("Enter name") # Automatically enter the name 'bot' into the input field # name_input_xpath = '/html/body/div[1]/div/div/div/div[5]/div/div[1]/div/div/div/div/div[2]/div[1]/div/div/div[2]/div[2]/span/input' name_input_xpath = '//*[@id="app"]/div/div/div/div[4]/div/div[1]/div/div/div/div/div[2]/div[1]/div/div/div[2]/div[2]/span/input' # Wait for the input element to be visible and interactable # WebDriverWait(driver, 30).until(EC.element_to_be_clickable((By.XPATH, name_input_xpath))) print("Name input field is ready") driver.find_element(By.XPATH, name_input_xpath).send_keys('bot') print('Entered name: bot') # Automatically click the Join button time.sleep(5) join_button_xpath = '//*[@id="prejoin-join-button"]' driver.find_element(By.XPATH, join_button_xpath).click() print('Join button clicked') # # Start recording in a separate thread # recording_thread = threading.Thread(target=start_recording, args=(output_audio_file,output_video_file,)) # recording_thread.start() global audio_thread, video_thread # Start audio recording in a separate thread audio_thread = threading.Thread(target=start_audio_recording, args=(output_audio_file,)) audio_thread.start() print(">>>>>>>>>>>>>>>>>>>>>>>>>>>>>") # Start screen recording in a separate thread video_thread = threading.Thread(target=record_screen, args=(output_video_file,)) video_thread.start() print("Recording started. Call stop_recording() to stop.") # Wait for the meeting interface to fully load time.sleep(10) # Click on the "People" button print("Clicking the 'People' button...") people_button_xpath = '//*[@id="roster-button"]' driver.find_element(By.XPATH, people_button_xpath).click() print("'People' button clicked") # Extract and print the value at the specified XPath print("Extracting value from roster title section...") roster_xpath = '//*[@id="roster-title-section-2"]/span' WebDriverWait(driver, 20).until(EC.presence_of_element_located((By.XPATH, roster_xpath))) roster_text = driver.find_element(By.XPATH, roster_xpath).text print("-------------------------------------------------") print(f"Value in roster title section: {roster_text}") # Wait for the next 10 minutes print("Waiting for 10 minutes...") time.sleep(600) # 600 seconds = 10 minutes print("Started rechecking roster text...") # Start a thread to recheck roster text every 5 minutes recheck_roster_text(driver,roster_xpath,uniqueId) finally: # Stop the recording stop_recording(output_audio_file,output_video_file,output_merge_file) # stop_recording(output_audio_file) # Stop the browser after a delay time.sleep(10) driver.quit() print("Browser closed.") # Upload the final video to Azure screenshot_path = uniqueId + ".png" if os.path.exists(screenshot_path): print("Screenshot exists") else: screenshot_path = "not_found.png" blob_url = upload_to_azure_with_signed_url(output_video_file, output_video_file) blob_url = "https://interviewbotstorage.blob.core.windows.net/interviewbot/" + uniqueId + ".mp4" screenshot_url = upload_img_to_azure_with_signed_url(screenshot_path,screenshot_path) logger.info("started json report") json_data = json_report(output_audio_file) json_data['snapshot'] = screenshot_url print("json report done", json_data) # Log the JSON data logger.info("Logging JSON data:") logger.info(json_data) response = send_interview_report(node_url, interviewId, blob_url, json_data) logger.info("Logging response data:") logger.info(response) print("Node Api response:", response) remove_file(output_audio_file) remove_file(output_merge_file) remove_file(output_video_file) remove_file(screenshot_path) print(blob_url) print(screenshot_url)