import json from datetime import datetime, timedelta import pytz import os import os import cv2 import subprocess import re import tiktoken import base64 def extract_json_from_string(text): """ Extracts JSON content from a string and converts it into a dictionary. """ match = re.search(r'\{.*\}', text, re.DOTALL) if match: json_content = match.group(0) try: return json.loads(json_content) except json.JSONDecodeError as e: print(f"Error decoding JSON: {e}") return None else: print("No valid JSON found in the string.") return None 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 to_camel_case(snake_str): """Convert snake_case string to camelCase.""" components = snake_str.split('_') return components[0] + ''.join(x.title() for x in components[1:]) def convert_keys_to_camel_case(data): """Recursively convert all keys in a dictionary to camelCase.""" if isinstance(data, dict): return {to_camel_case(k): convert_keys_to_camel_case(v) for k, v in data.items()} elif isinstance(data, list): return [convert_keys_to_camel_case(item) for item in data] else: return data def classify_technical_skills(data): """ Classify technical skills into strengths and weaknesses based on their ratings. - Rating <= 2.5: Weakness - Rating >= 3.0: Strength """ strengths = [] weaknesses = [] for skill in data['technicalSkills']: skill_name = skill['skill'] try: rating = float(skill['rating']) # Convert rating to a float if rating <= 2.5: weaknesses.append(skill_name) elif rating >= 3.0: strengths.append(skill_name) except ValueError: print(f"Invalid rating for skill '{skill_name}': {skill['rating']}") return strengths, weaknesses def markdown_to_json(markdown_str): # Try finding JSON with ```json code block first if "```json" in markdown_str: json_start = markdown_str.index("```json") + len("```json") json_end = markdown_str.rindex("```") json_content = markdown_str[json_start:json_end].strip() # Fallback: Handle case with just ``` (without 'json') elif "```" in markdown_str: json_start = markdown_str.index("```") + len("```") json_end = markdown_str.rindex("```") json_content = markdown_str[json_start:json_end].strip() else: raise ValueError("Invalid format: Missing code block markers") # Clean up JSON string json_content = json_content.replace("\\n", "").replace("\\", "") # Try to parse the cleaned string as JSON try: json_obj = json.loads(json_content) except json.JSONDecodeError as e: raise ValueError(f"Error parsing JSON: {e}") return json_obj def validate_input(first_name, email, job_title, technologies, experience): if not first_name or not first_name.strip(): return False if not email or '@' not in email: return False if not job_title or not job_title.strip(): return False if not isinstance(technologies, list) or len(technologies) == 0: return False if not isinstance(experience, (int, float)) or experience <= 0: return False return True def extract_json(input_text): """ Extract the JSON block from the provided text. """ try: start_index = input_text.find("{") end_index = input_text.rfind("}") + 1 if start_index != -1 and end_index != -1: json_data = input_text[start_index:end_index] return json_data.strip() # Strip extra spaces or newlines else: raise ValueError("No JSON data found in the input text.") except Exception as e: raise ValueError(f"Error extracting JSON: {e}") def trim_prompt(user_input, max_tokens=4000): """ Trims the input prompt to ensure it does not exceed max_tokens (default: 6000). Args: user_input (str): The user's text input. max_tokens (int): Maximum number of tokens allowed. Returns: str: The trimmed text prompt. """ # Load tokenizer (defaulting to GPT-4 encoding) enc = tiktoken.get_encoding("cl100k_base") # Tokenize the input text tokens = enc.encode(user_input) # Trim the tokens if they exceed max_tokens if len(tokens) > max_tokens: tokens = tokens[:max_tokens] # Keep only first max_tokens # Decode back to text trimmed_text = enc.decode(tokens) return trimmed_text def subtract_minutes_from_epoch(epoch_time: int, minutes: int) -> int: print(f"Original epoch time: {epoch_time}") print(f"Original local time: {datetime.fromtimestamp(epoch_time)}") print(f"Minutes to subtract: {minutes}") # Subtract the specified number of minutes from the epoch time subracted_time= epoch_time - (minutes * 60) print(f"Subtratcted epoch Time : {subracted_time}") print(f"Subtratcted local Time : {datetime.fromtimestamp(subracted_time)}") return epoch_time - (minutes * 60) def is_upcoming(input_time, input_date): try: # Combine the date and time into a single datetime object input_datetime = datetime.strptime(f"{input_date} {input_time}", "%Y-%m-%d %H:%M") # Get the current datetime current_datetime = datetime.now() # Compare the input datetime with the current datetime return input_datetime >= current_datetime except ValueError as e: print(f"Invalid input format: {e}") return False def convert_timestamp_to_datetime(unix_timestamp, timezone_str): # Correct: get UTC datetime utc_time = datetime.utcfromtimestamp(unix_timestamp) # Convert to local timezone local_tz = pytz.timezone(timezone_str) local_time = pytz.utc.localize(utc_time).astimezone(local_tz) input_time = local_time.strftime("%H:%M") input_date = local_time.strftime("%Y-%m-%d") return input_time, input_date def mp4_to_wav(input_mp4, output_wav): try: # Ensure input file exists if not os.path.exists(input_mp4): raise FileNotFoundError(f"Input file not found: {input_mp4}") # Convert MP4 to WAV using FFmpeg command = [ "ffmpeg", "-y", "-i", input_mp4, "-acodec", "pcm_s16le", "-ar", "16000", "-ac", "1", output_wav ] subprocess.run(command, check=True) # Ensure output file was created if not os.path.exists(output_wav): raise FileNotFoundError(f"Conversion failed: {output_wav} was not created") print(f"Conversion successful: {output_wav}") return output_wav except subprocess.CalledProcessError as e: raise RuntimeError(f"FFmpeg error: {e}") except Exception as e: raise RuntimeError(f"Unexpected error: {e}") def capture_screenshot(video_path, output_image_path): # Open the video file cap = cv2.VideoCapture(video_path) if not cap.isOpened(): raise FileNotFoundError(f"Could not open video file: {video_path}") # Get total frame count and FPS total_frames = int(cap.get(cv2.CAP_PROP_FRAME_COUNT)) fps = int(cap.get(cv2.CAP_PROP_FPS)) # Calculate midpoint frame mid_frame = total_frames // 2 # Set video to the middle frame cap.set(cv2.CAP_PROP_POS_FRAMES, mid_frame) # Read the frame ret, frame = cap.read() if not ret: raise RuntimeError("Failed to capture frame at midpoint") # Save the frame as an image cv2.imwrite(output_image_path, frame) print(f"Screenshot saved: {output_image_path}") # Release resources cap.release()