import requests from typing import List, Dict, Any class YouTubeMCP: def __init__(self): self.api_key = "AIzaSyCx593S7DDye-DdsWwh_trEH0yr1TLBDJw" # Replace with your actual API key self.base_url = "https://www.googleapis.com/youtube/v3" def fetch_courses(self, user_skills: List[Dict], skill_gaps: List[Dict], job_profile_data: Dict = None, user_query: str = "", target_role: str = "", max_results: int = 9) -> List[Dict[str, Any]]: """Fetch courses from YouTube API prioritizing user query over job profile data.""" # PRIORITY 1: Build search query starting with user query query_terms = [] # Add user query terms with highest priority if user_query: # Extract key terms from user query, focusing on career transition keywords query_lower = user_query.lower() important_words = [word for word in query_lower.split() if word not in ['i', 'want', 'to', 'the', 'a', 'an', 'and', 'or', 'but', 'give', 'best', 'for', 'this']] query_terms.extend(important_words[:4]) # Take first 4 important words # PRIORITY 2: Add target role if identified from query if target_role: target_words = target_role.split() query_terms.extend(target_words[:3]) # Add target role words # PRIORITY 3: Add critical and high priority skill gaps critical_gaps = [gap for gap in skill_gaps if gap.get('priority') == 'critical'] high_priority_gaps = [gap for gap in skill_gaps if gap.get('priority') == 'high'] # Add critical skills first for gap in critical_gaps[:2]: skill_words = gap['skill_name'].lower().split() query_terms.extend(skill_words) # Add high priority skills if space allows for gap in high_priority_gaps[:1]: if len(query_terms) < 8: # Limit total terms skill_words = gap['skill_name'].lower().split() query_terms.extend(skill_words[:2]) # PRIORITY 4: Only add job profile if no career transition detected if not target_role and job_profile_data and job_profile_data.get('job_profile_name'): if len(query_terms) < 6: # Only if we have space job_name_words = job_profile_data['job_profile_name'].lower().split() query_terms.extend(job_name_words[:2]) # PRIORITY 5: Add medium priority skills as fallback if len(query_terms) < 4: # If we still don't have enough terms medium_gaps = [gap for gap in skill_gaps if gap.get('priority') == 'medium'] for gap in medium_gaps[:2]: skill_words = gap['skill_name'].lower().split() query_terms.extend(skill_words[:1]) # Remove duplicates while preserving order seen = set() unique_query_terms = [] for term in query_terms: if term not in seen and len(term) > 2: # Filter out very short terms seen.add(term) unique_query_terms.append(term) # Create search query with educational context educational_terms = ["course", "tutorial", "training", "guide"] # Prioritize career transition terms if detected if target_role or any(word in user_query.lower() for word in ['transition', 'change', 'switch', 'become']): educational_terms = ["career change", "transition guide", "course", "training"] # Build final query query = f"{' '.join(unique_query_terms[:6])} {' '.join(educational_terms[:2])}" endpoint = f"{self.base_url}/search" params = { "part": "snippet", "q": query, "type": "video", "maxResults": max_results, "key": self.api_key, "relevanceLanguage": "en", "videoDuration": "medium", "order": "relevance" } try: print(f"YouTube search query (prioritizing user intent): {query}") # Debug log response = requests.get(endpoint, params=params) if response.status_code == 200: results = response.json().get("items", []) # Process results with user query priority for item in results: title = item.get('snippet', {}).get('title', '').lower() description = item.get('snippet', {}).get('description', '').lower() # Calculate relevance scores with user query priority # 1. User query alignment (highest weight) query_alignment_score = 0 if user_query: query_words = [word.lower() for word in user_query.split() if len(word) > 2 and word.lower() not in ['the', 'and', 'for', 'want', 'best']] for word in query_words: if word in title: query_alignment_score += 3 # Higher weight for title match elif word in description: query_alignment_score += 1 # 2. Target role alignment (second highest weight) target_role_score = 0 if target_role: target_words = target_role.lower().split() for word in target_words: if word in title: target_role_score += 2 elif word in description: target_role_score += 1 # 3. Critical skill gaps alignment critical_skills_score = 0 for gap in critical_gaps: gap_name = gap['skill_name'].lower() if gap_name in title: critical_skills_score += 2 elif gap_name in description: critical_skills_score += 1 # 4. Job profile alignment (lowest priority) job_alignment_score = 0 if not target_role and job_profile_data and job_profile_data.get('job_profile_name'): job_name_words = job_profile_data['job_profile_name'].lower().split() for word in job_name_words: if word in title: job_alignment_score += 1 elif word in description: job_alignment_score += 0.5 # Calculate total relevance score with weighted priorities total_relevance_score = ( query_alignment_score * 4 + # Highest weight target_role_score * 3 + # Second highest critical_skills_score * 2 + # Third job_alignment_score * 1 # Lowest weight ) # Find addressed skill gaps with priority awareness addressed_gaps = [] gap_types = [] gap_priorities = [] # Check gaps in priority order all_gaps_by_priority = sorted(skill_gaps, key=lambda x: {'critical': 1, 'high': 2, 'medium': 3, 'low': 4}.get(x.get('priority', 'medium'), 3)) for gap in all_gaps_by_priority: gap_name = gap['skill_name'].lower() if gap_name in title or gap_name in description: addressed_gaps.append(gap_name) gap_types.append(gap.get('gap_type', 'unknown')) gap_priorities.append(gap.get('priority', 'medium')) if len(addressed_gaps) >= 3: # Limit to top 3 gaps break # Enhanced reason generation with user query priority reason_parts = [] if target_role: if target_role_score > 0: reason_parts.append(f"Perfect for transitioning to {target_role}") else: reason_parts.append(f"Recommended for {target_role} career transition") elif user_query and query_alignment_score > 0: reason_parts.append("Directly addresses your query requirements") elif job_profile_data and job_profile_data.get('job_profile_name'): reason_parts.append(f"Relevant for {job_profile_data['job_profile_name']} role") if addressed_gaps: critical_addressed = [gap for gap, priority in zip(addressed_gaps, gap_priorities) if priority == 'critical'] high_addressed = [gap for gap, priority in zip(addressed_gaps, gap_priorities) if priority == 'high'] if critical_addressed: reason_parts.append(f"Teaches critical transition skills: {', '.join(critical_addressed)}") elif high_addressed: reason_parts.append(f"Covers essential skills: {', '.join(high_addressed)}") else: reason_parts.append(f"Helps develop: {', '.join(addressed_gaps[:2])}") if not reason_parts: reason_parts.append("Course content aligns with your learning goals") # Add enhanced metadata item['addressed_gaps'] = addressed_gaps item['gap_types'] = gap_types item['gap_priorities'] = gap_priorities item['query_alignment_score'] = query_alignment_score item['target_role_score'] = target_role_score item['critical_skills_score'] = critical_skills_score item['job_alignment_score'] = job_alignment_score item['total_relevance_score'] = total_relevance_score item['reason'] = '. '.join(reason_parts) item['url'] = f"https://www.youtube.com/watch?v={item['id']['videoId']}" item['user_query_focused'] = bool(target_role or query_alignment_score > 0) # Sort by total relevance score (user query gets highest priority) results.sort(key=lambda x: x.get('total_relevance_score', 0), reverse=True) return results else: print(f"Error: Received status code {response.status_code} from YouTube API") print(f"Response: {response.text}") return [] except Exception as e: print(f"Error fetching YouTube courses: {str(e)}") return [] def main(): # Test with career transition scenario user_skills = [ { "skill_id": "6", "skill_type": "3", "skill_name": "Leadership" }, { "skill_id": "7", "skill_type": "1", "skill_name": "Management" }, { "skill_id": "9", "skill_type": "2", "skill_name": "Communication" } ] # Current job profile (should have lower priority) job_profile_data = { "job_profile_name": "data scientist", "job_profile_skills": [ { "skill_id": "20", "skill_type": "1", "skill_name": "Machine Learning" }, { "skill_id": "15", "skill_type": "1", "skill_name": "Python" } ] } # User query indicates career transition (should have highest priority) user_query = "i want transition to sales manager give best career advice for this" target_role = "sales manager" # Extracted from query # Skill gaps prioritizing transition to sales manager skill_gaps = [ { 'skill_id': "s1", 'skill_name': "Sales Management", 'current_level': 0, 'target_level': 3, 'gap_type': 'transition_skill', 'priority': 'critical' }, { 'skill_id': "s2", 'skill_name': "Customer Relationship Management", 'current_level': 0, 'target_level': 3, 'gap_type': 'transition_skill', 'priority': 'critical' }, { 'skill_id': "s3", 'skill_name': "Negotiation", 'current_level': 0, 'target_level': 3, 'gap_type': 'transition_skill', 'priority': 'high' }, { 'skill_id': "7", 'skill_name': "Management", 'current_level': 1, 'target_level': 3, 'gap_type': 'transition_upgrade', 'priority': 'high' } ] # Create an instance of YouTubeMCP and fetch courses youtube_mcp = YouTubeMCP() courses = youtube_mcp.fetch_courses(user_skills, skill_gaps, job_profile_data, user_query, target_role) # Print the results print("Fetched Courses:") print(f"Total courses found: {len(courses)}") print("-" * 70) for i, course in enumerate(courses, 1): snippet = course.get('snippet', {}) print(f"{i}. Title: {snippet.get('title')}") print(f" Channel: {snippet.get('channelTitle')}") print(f" URL: {course.get('url')}") print(f" Relevance Score: {course.get('relevance_score', 0)}") print(f" Addressed Gaps: {course.get('addressed_gaps', [])}") print(f" Job Alignment Score: {course.get('job_alignment_score', 0)}") print(f" Reason: {course.get('reason', 'N/A')}") print("-" * 70) if __name__ == "__main__": main()