import json import logging from typing import Dict, List, Optional, Tuple from dataclasses import dataclass from difflib import SequenceMatcher import re from collections import Counter from datetime import datetime from langchain_ollama import ChatOllama from langchain.prompts import PromptTemplate from langchain.schema import HumanMessage # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) @dataclass class Skill: """Data class for skill information""" skill_id: int skill_name: str skill_level: Optional[str] = None skill_count: Optional[int] = None @dataclass class Course: """Data class for course information""" course_id: str name: str short_description: str description: str skills: List[str] @dataclass class SkillAnalysisRequest: """Data class for skill analysis request""" user_query: str admin_name: str department_name: str filtered_skills: List[Skill] all_skills_data: List[Dict] all_courses_data: Dict @dataclass class SkillRecommendation: """Data class for skill recommendation""" skill_id: int skill_name: str skill_level: str priority: str source: str # 'department_existing', 'organizational_relevant', 'external' justification: str @dataclass class SkillAnalysisResponse: """Data class for skill analysis response""" recommended_skills: List[SkillRecommendation] analysis_summary: str department_coverage: float total_skills_recommended: int case_applied: str identified_field: str @dataclass class AnalysisResult: """Data class for complete analysis result""" success: bool identified_field: str case_applied: str total_skills_recommended: int department_coverage: float analysis_summary: str recommended_skills: List[SkillRecommendation] user_query: str admin_name: str department_name: str class SkillInferenceEngine: """Handles skill inference using LLM with enhanced specificity-aware approach""" def __init__(self, model_name: str = "gemma3:12b"): """Initialize the LLM model""" try: self.llm = ChatOllama(model=model_name) logger.info(f"Initialized Skill LLM with model: {model_name}") except Exception as e: logger.error(f"Failed to initialize Skill LLM: {e}") raise def identify_field_domain(self, user_query: str, department_name: str) -> dict: """ Enhanced Step 1: Identify the field/domain with specificity analysis Args: user_query: User's business goal or challenge department_name: Target department name Returns: Dictionary containing field analysis with specificity information """ prompt_template = PromptTemplate( input_variables=["user_query", "department_name"], template=""" You are an expert business analyst specializing in identifying professional fields and domains. User Goal/Challenge: {user_query} Department: {department_name} Analyze the user's goal and provide a detailed field analysis including specificity level. SPECIFICITY LEVELS: - GENERAL: Broad, vague requests like "improve tech skills", "better at development", "enhance coding" - SPECIFIC: Domain-focused like "improve frontend skills", "better at web development", "enhance mobile development" - HIGHLY_SPECIFIC: Technology-specific like "improve Angular skills", "better at React development", "enhance AWS cloud skills" ANALYSIS REQUIRED: 1. Identify the primary professional field/domain 2. Determine specificity level of the request 3. Extract any specific technologies/tools mentioned 4. Determine if this requires broad exploration or targeted learning Return your analysis in this EXACT JSON format: {{ "primary_field": "specific field name", "specificity_level": "GENERAL|SPECIFIC|HIGHLY_SPECIFIC", "key_technologies": ["tech1", "tech2"], "skill_focus": "broad|targeted", "learning_approach": "exploration|specialization|mastery" }} EXAMPLES: "improve tech skills" → {{"primary_field": "Software Development", "specificity_level": "GENERAL", "key_technologies": [], "skill_focus": "broad", "learning_approach": "exploration"}} "improve Angular skills" → {{"primary_field": "Frontend Development", "specificity_level": "HIGHLY_SPECIFIC", "key_technologies": ["Angular", "TypeScript"], "skill_focus": "targeted", "learning_approach": "mastery"}} "improve web development" → {{"primary_field": "Web Development", "specificity_level": "SPECIFIC", "key_technologies": ["HTML", "CSS", "JavaScript"], "skill_focus": "broad", "learning_approach": "specialization"}} Field Analysis: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, department_name=department_name ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Parse JSON response field_analysis = self._parse_field_analysis(response.content) logger.info(f"Field analysis: {field_analysis}") return field_analysis except Exception as e: logger.error(f"Error in field identification: {e}") return { "primary_field": "General", "specificity_level": "GENERAL", "key_technologies": [], "skill_focus": "broad", "learning_approach": "exploration" } def filter_relevant_organizational_skills(self, field_analysis: dict, all_skills_data: List[Dict]) -> List[Dict]: """ Enhanced Step 2: Filter organizational skills based on field analysis and specificity Args: field_analysis: Result from identify_field_domain with specificity info all_skills_data: Complete list of organizational skills Returns: List of field-relevant organizational skills """ primary_field = field_analysis["primary_field"] specificity_level = field_analysis["specificity_level"] key_technologies = field_analysis["key_technologies"] # Create a formatted list of skills for the LLM skills_text = "\n".join([ f"- {skill['skill_name']} (ID: {skill['skill_id']})" for skill in all_skills_data ]) # Different prompts based on specificity level if specificity_level == "HIGHLY_SPECIFIC": return self._filter_highly_specific_skills(primary_field, key_technologies, skills_text, all_skills_data) elif specificity_level == "SPECIFIC": return self._filter_specific_skills(primary_field, skills_text, all_skills_data) else: # GENERAL return self._filter_general_skills(primary_field, skills_text, all_skills_data) def _filter_highly_specific_skills(self, primary_field: str, key_technologies: List[str], skills_text: str, all_skills_data: List[Dict]) -> List[Dict]: """Filter skills for highly specific technology requests""" key_tech_str = ", ".join(key_technologies) prompt_template = PromptTemplate( input_variables=["primary_field", "key_technologies", "skills_text"], template=""" HIGHLY SPECIFIC SKILL FILTERING for {primary_field} Target Technologies: {key_technologies} Available Skills: {skills_text} Since this is a HIGHLY SPECIFIC request focusing on {key_technologies}, filter skills that are: 1. DIRECTLY related to {key_technologies} 2. Essential prerequisites for {key_technologies} 3. Advanced concepts that build upon {key_technologies} 4. Testing, debugging, or deployment tools for {key_technologies} 5. Integration technologies that work with {key_technologies} Be VERY SELECTIVE - only include skills that have clear relevance to {key_technologies}. Exclude general programming concepts unless they're specifically crucial for {key_technologies}. Return ONLY the skill names that are relevant, one per line: - Skill Name 1 - Skill Name 2 ... Relevant Skills for {key_technologies}: """ ) try: formatted_prompt = prompt_template.format( primary_field=primary_field, key_technologies=key_tech_str, skills_text=skills_text ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) relevant_skill_names = self._parse_skills_from_response(response.content) relevant_skills = self._match_skills_to_data(relevant_skill_names, all_skills_data) logger.info(f"Filtered {len(relevant_skills)} highly specific skills for {key_tech_str}") return relevant_skills except Exception as e: logger.error(f"Error in highly specific skill filtering: {e}") return [] def _filter_specific_skills(self, primary_field: str, skills_text: str, all_skills_data: List[Dict]) -> List[Dict]: """Filter skills for specific domain requests""" prompt_template = PromptTemplate( input_variables=["primary_field", "skills_text"], template=""" SPECIFIC SKILL FILTERING for {primary_field} Available Skills: {skills_text} Since this is a SPECIFIC domain request for {primary_field}, filter skills that are: 1. Core skills essential to {primary_field} 2. Common tools and technologies used in {primary_field} 3. Methodologies and practices specific to {primary_field} 4. Soft skills that are particularly valuable in {primary_field} 5. Industry-standard technologies for {primary_field} Include both technical and soft skills that are domain-relevant. Be moderately selective - include skills that are commonly needed in {primary_field}. Return ONLY the skill names that are relevant, one per line: - Skill Name 1 - Skill Name 2 ... Relevant Skills for {primary_field}: """ ) try: formatted_prompt = prompt_template.format( primary_field=primary_field, skills_text=skills_text ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) relevant_skill_names = self._parse_skills_from_response(response.content) relevant_skills = self._match_skills_to_data(relevant_skill_names, all_skills_data) logger.info(f"Filtered {len(relevant_skills)} specific domain skills for {primary_field}") return relevant_skills except Exception as e: logger.error(f"Error in specific skill filtering: {e}") return [] def _filter_general_skills(self, primary_field: str, skills_text: str, all_skills_data: List[Dict]) -> List[Dict]: """Filter skills for general broad requests""" prompt_template = PromptTemplate( input_variables=["primary_field", "skills_text"], template=""" GENERAL SKILL FILTERING for {primary_field} Available Skills: {skills_text} Since this is a GENERAL request for improving skills in {primary_field}, filter skills that are: 1. Foundational skills that provide broad value in {primary_field} 2. Transferable skills useful across different areas of {primary_field} 3. Essential soft skills for {primary_field} professionals 4. Popular tools and technologies in {primary_field} 5. Skills that provide good career foundation in {primary_field} Be INCLUSIVE rather than restrictive - include skills that could be valuable for someone wanting to improve broadly in {primary_field}. Include both technical and soft skills that build a well-rounded foundation. Return ONLY the skill names that are relevant, one per line: - Skill Name 1 - Skill Name 2 ... Relevant Skills for {primary_field}: """ ) try: formatted_prompt = prompt_template.format( primary_field=primary_field, skills_text=skills_text ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) relevant_skill_names = self._parse_skills_from_response(response.content) relevant_skills = self._match_skills_to_data(relevant_skill_names, all_skills_data) logger.info(f"Filtered {len(relevant_skills)} general foundation skills for {primary_field}") return relevant_skills except Exception as e: logger.error(f"Error in general skill filtering: {e}") return [] def suggest_external_skills(self, user_query: str, field_analysis: dict, existing_skills: List[str], needed_count: int) -> List[str]: """ Enhanced external skill suggestion based on specificity level Args: user_query: User's business goal field_analysis: Complete field analysis with specificity info existing_skills: Skills already identified needed_count: Number of additional skills needed Returns: List of external skill suggestions """ primary_field = field_analysis["primary_field"] specificity_level = field_analysis["specificity_level"] key_technologies = field_analysis["key_technologies"] learning_approach = field_analysis["learning_approach"] existing_skills_text = ", ".join(existing_skills) if existing_skills else "None" if specificity_level == "HIGHLY_SPECIFIC": return self._suggest_highly_specific_external_skills( user_query, primary_field, key_technologies, existing_skills_text, needed_count ) elif specificity_level == "SPECIFIC": return self._suggest_specific_external_skills( user_query, primary_field, existing_skills_text, needed_count ) else: # GENERAL return self._suggest_general_external_skills( user_query, primary_field, existing_skills_text, needed_count ) def _suggest_highly_specific_external_skills(self, user_query: str, primary_field: str, key_technologies: List[str], existing_skills: str, needed_count: int) -> List[str]: """Suggest external skills for highly specific technology requests""" key_tech_str = ", ".join(key_technologies) prompt_template = PromptTemplate( input_variables=["user_query", "primary_field", "key_technologies", "existing_skills", "needed_count"], template=""" HIGHLY SPECIFIC EXTERNAL SKILLS for {key_technologies} User Goal: {user_query} Field: {primary_field} Target Technologies: {key_technologies} Already Covered: {existing_skills} Skills Needed: {needed_count} Suggest {needed_count} ADVANCED and SPECIALIZED external skills specifically for {key_technologies}. Focus on: 1. Advanced {key_technologies} concepts and patterns 2. {key_technologies} performance optimization techniques 3. {key_technologies} testing and debugging tools 4. {key_technologies} deployment and CI/CD 5. {key_technologies} ecosystem libraries and extensions 6. Advanced architectural patterns for {key_technologies} 7. {key_technologies} security best practices Avoid basic programming concepts - focus on skills that specifically enhance {key_technologies} expertise. Return exactly {needed_count} skills in this format: - External Skill 1 - External Skill 2 ... Advanced {key_technologies} Skills: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, primary_field=primary_field, key_technologies=key_tech_str, existing_skills=existing_skills, needed_count=needed_count ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) skills = self._parse_skills_from_response(response.content) logger.info(f"Suggested {len(skills)} highly specific external skills for {key_tech_str}") return skills[:needed_count] except Exception as e: logger.error(f"Error in highly specific external skill suggestion: {e}") return [] def _suggest_specific_external_skills(self, user_query: str, primary_field: str, existing_skills: str, needed_count: int) -> List[str]: """Suggest external skills for specific domain requests""" prompt_template = PromptTemplate( input_variables=["user_query", "primary_field", "existing_skills", "needed_count"], template=""" SPECIFIC DOMAIN EXTERNAL SKILLS for {primary_field} User Goal: {user_query} Field: {primary_field} Already Covered: {existing_skills} Skills Needed: {needed_count} Suggest {needed_count} external skills that are trending and valuable in the {primary_field} domain. Focus on: 1. Current industry trends in {primary_field} 2. Emerging technologies in {primary_field} 3. Professional best practices for {primary_field} 4. Tools that are becoming standard in {primary_field} 5. Skills that differentiate professionals in {primary_field} Include both technical and methodology skills relevant to {primary_field}. Return exactly {needed_count} skills in this format: - External Skill 1 - External Skill 2 ... Trending {primary_field} Skills: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, primary_field=primary_field, existing_skills=existing_skills, needed_count=needed_count ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) skills = self._parse_skills_from_response(response.content) logger.info(f"Suggested {len(skills)} specific domain external skills for {primary_field}") return skills[:needed_count] except Exception as e: logger.error(f"Error in specific external skill suggestion: {e}") return [] def _suggest_general_external_skills(self, user_query: str, primary_field: str, existing_skills: str, needed_count: int) -> List[str]: """Suggest external skills for general broad requests""" prompt_template = PromptTemplate( input_variables=["user_query", "primary_field", "existing_skills", "needed_count"], template=""" GENERAL FOUNDATIONAL EXTERNAL SKILLS for {primary_field} User Goal: {user_query} Field: {primary_field} Already Covered: {existing_skills} Skills Needed: {needed_count} Suggest {needed_count} diverse external skills that provide a strong foundation across {primary_field}. Focus on: 1. Fundamental skills every {primary_field} professional should have 2. Cross-cutting skills that apply to multiple areas of {primary_field} 3. Soft skills essential for {primary_field} success 4. Basic tools and technologies that are widely used in {primary_field} 5. Skills that provide good career versatility in {primary_field} Provide a well-rounded mix of technical, methodological, and soft skills. Return exactly {needed_count} skills in this format: - External Skill 1 - External Skill 2 ... Foundation {primary_field} Skills: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, primary_field=primary_field, existing_skills=existing_skills, needed_count=needed_count ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) skills = self._parse_skills_from_response(response.content) logger.info(f"Suggested {len(skills)} general foundation external skills for {primary_field}") return skills[:needed_count] except Exception as e: logger.error(f"Error in general external skill suggestion: {e}") return [] def _parse_field_analysis(self, response_content: str) -> dict: """Parse field analysis from LLM response""" import json import re try: # Extract JSON from response json_match = re.search(r'\{.*\}', response_content, re.DOTALL) if json_match: field_analysis = json.loads(json_match.group()) # Validate required fields required_fields = ["primary_field", "specificity_level", "key_technologies", "skill_focus", "learning_approach"] for field in required_fields: if field not in field_analysis: field_analysis[field] = self._get_default_value(field) return field_analysis else: # Fallback parsing return self._fallback_field_analysis(response_content) except Exception as e: logger.error(f"Error parsing field analysis: {e}") return { "primary_field": "General", "specificity_level": "GENERAL", "key_technologies": [], "skill_focus": "broad", "learning_approach": "exploration" } def _get_default_value(self, field_name: str): """Get default values for missing fields""" defaults = { "primary_field": "General", "specificity_level": "GENERAL", "key_technologies": [], "skill_focus": "broad", "learning_approach": "exploration" } return defaults.get(field_name, "") def _fallback_field_analysis(self, response_content: str) -> dict: """Fallback parsing when JSON extraction fails""" # Simple keyword-based analysis content_lower = response_content.lower() # Detect specificity level if any(tech in content_lower for tech in ["angular", "react", "vue", "python", "java", "aws", "docker"]): specificity = "HIGHLY_SPECIFIC" focus = "targeted" approach = "mastery" elif any(domain in content_lower for domain in ["web development", "mobile development", "data science", "frontend", "backend"]): specificity = "SPECIFIC" focus = "broad" approach = "specialization" else: specificity = "GENERAL" focus = "broad" approach = "exploration" # Extract field name (first meaningful line) field_name = self._clean_field_name(response_content) return { "primary_field": field_name, "specificity_level": specificity, "key_technologies": [], "skill_focus": focus, "learning_approach": approach } def _match_skills_to_data(self, skill_names: List[str], all_skills_data: List[Dict]) -> List[Dict]: """Match skill names back to original skill data""" relevant_skills = [] for skill_name in skill_names: for org_skill in all_skills_data: if self._skills_match(skill_name, org_skill['skill_name']): relevant_skills.append(org_skill) break return relevant_skills def _clean_field_name(self, field_text: str) -> str: """Clean and normalize field name""" field_text = field_text.strip().split('\n')[0] field_text = field_text.replace("Field/Domain:", "").replace("Field:", "").replace("Domain:", "") field_text = field_text.strip(' "\'') return field_text def _parse_skills_from_response(self, response_text: str) -> List[str]: """Parse skills from LLM response""" skills = [] lines = response_text.strip().split('\n') for line in lines: line = line.strip() line = re.sub(r'^[-•*\d+\.\)]\s*', '', line) line = line.strip() if line and len(line) > 2: skills.append(line) return skills def _skills_match(self, skill1: str, skill2: str) -> bool: """Check if two skill names match with reasonable similarity""" skill1_norm = re.sub(r'[^a-zA-Z0-9\s]', '', skill1.lower()).strip() skill2_norm = re.sub(r'[^a-zA-Z0-9\s]', '', skill2.lower()).strip() if skill1_norm == skill2_norm: return True similarity = SequenceMatcher(None, skill1_norm, skill2_norm).ratio() if similarity > 0.8: return True if skill1_norm in skill2_norm or skill2_norm in skill1_norm: return True return False class SkillMatcher: """Handles skill matching and filtering operations""" @staticmethod def filter_department_skills(filtered_skills: List[Skill], coverage_threshold: float = 0.5) -> List[Skill]: """ Filter out skills that are already well-covered in the department Args: filtered_skills: Current department skills with counts coverage_threshold: Threshold for considering a skill well-covered Returns: List of skills that need improvement """ if not filtered_skills: return [] # Calculate total employee count total_employees = max([skill.skill_count for skill in filtered_skills], default=1) # Filter skills that are under-covered under_covered_skills = [] for skill in filtered_skills: coverage_ratio = skill.skill_count / total_employees if coverage_ratio < coverage_threshold: under_covered_skills.append(skill) logger.info(f"Filtered {len(under_covered_skills)} under-covered skills from {len(filtered_skills)} total skills") return under_covered_skills class TNAPresentation: """Handles presentation of TNA analysis in multiple formats""" def __init__(self): """Initialize the presentation system""" logger.info("TNA Presentation System initialized") def generate_raw_analysis(self, analysis_result: AnalysisResult) -> Dict: """ Generate raw analysis output in JSON format Args: analysis_result: Complete analysis result Returns: Dictionary with raw analysis data """ # Convert skills to JSON format skills_json = [] for skill in analysis_result.recommended_skills: skills_json.append({ 'skill_id': skill.skill_id, 'skill_name': skill.skill_name, 'skill_level': skill.skill_level, 'priority': skill.priority, 'source': skill.source, 'justification': skill.justification }) raw_analysis = { 'success': analysis_result.success, 'identified_field': analysis_result.identified_field, 'case_applied': analysis_result.case_applied, 'total_skills_recommended': analysis_result.total_skills_recommended, 'department_coverage': round(analysis_result.department_coverage, 4), 'analysis_summary': analysis_result.analysis_summary, 'recommended_skills': skills_json, 'metadata': { 'user_query': analysis_result.user_query, 'admin_name': analysis_result.admin_name, 'department_name': analysis_result.department_name, 'analysis_timestamp': datetime.now().isoformat(), 'total_organizational_skills': len([s for s in analysis_result.recommended_skills if s.source == 'organizational_relevant']), 'total_external_skills': len([s for s in analysis_result.recommended_skills if s.source == 'external']), 'priority_breakdown': self._get_priority_breakdown(analysis_result.recommended_skills) } } logger.info("Raw analysis generated successfully") return raw_analysis def generate_strategic_presentation(self, analysis_result: AnalysisResult) -> str: """ Generate strategic executive presentation Args: analysis_result: Complete analysis result Returns: Formatted string with strategic presentation """ # Calculate metrics org_skills = [s for s in analysis_result.recommended_skills if s.source == 'organizational_relevant'] ext_skills = [s for s in analysis_result.recommended_skills if s.source == 'external'] dept_skills = [s for s in analysis_result.recommended_skills if s.source == 'department_existing'] org_percentage = (len(org_skills) / len(analysis_result.recommended_skills)) * 100 if analysis_result.recommended_skills else 0 current_date = datetime.now().strftime("%B %d, %Y") strategic_report = f""" Hello {analysis_result.admin_name}, We noticed you shared: "{analysis_result.user_query}". That’s a great growth mindset! Wanting to improve in {analysis_result.identified_field} shows real initiative. Here’s what we’ve analyzed and suggest for you: ━━━━━━━━━━━━━━━━━━━━━━━ What You Need to Grow in {analysis_result.identified_field} ━━━━━━━━━━━━━━━━━━━━━━━ To really level up your {analysis_result.identified_field.lower()} skills, we identified a combination of organizational skills you already have access to, plus some external skills that will give you an extra edge. Core Organizational Skills You Can Start With: {self._format_skill_section(org_skills + dept_skills, "organizational")} These are skills already available internally through your organization’s resources. It’s like already having some tools in your toolbox! High-Value External Skills That Unlock the Next Level: {self._format_skill_section(ext_skills, "external")} These are the more advanced or specialized areas where external training, certifications, or hands-on projects could make a big difference. ━━━━━━━━━━━━━━━━━━━━━━━ Prioritized Skill Recommendations ━━━━━━━━━━━━━━━━━━━━━━━ Based on your profile and our analysis, here’s a practical order to focus on: 1️ Most Important – Start Here: {", ".join([s.skill_name for s in (org_skills + dept_skills)[:3]])} Why? You already have access to resources for these. Quick wins, low setup time, high ROI. 2️ Secondary Focus – Build Deeper Expertise: {", ".join([s.skill_name for s in ext_skills[:3]])} These are high-impact but need external support or new learning paths. 3️ Bonus Round – Become Truly Unstoppable: {", ".join([s.skill_name for s in ext_skills[3:6]])} Once you’ve mastered the core and secondary skills, these will push your capabilities into the top professional bracket. ━━━━━━━━━━━━━━━━━━━━━━━ Your Current Capability Assessment ━━━━━━━━━━━━━━━━━━━━━━━ - Skill Match with Organization: {org_percentage:.1f}% - Existing Skills Found: {len(org_skills + dept_skills)} Good news — you already have a solid foundation! Having {len(org_skills + dept_skills)} relevant skills puts you one step ahead. ━━━━━━━━━━━━━━━━━━━━━━━ Suggested Learning Path ━━━━━━━━━━━━━━━━━━━━━━━ Phase 1 (0–3 Months): Strengthen Existing Skills - Focus on {", ".join([s.skill_name for s in (org_skills + dept_skills)[:3]])} - Use internal training programs - Regular check-ins to track progress Phase 2 (3–6 Months): Expand with External Skills - Focus on {", ".join([s.skill_name for s in ext_skills[:3]])} - Plan for external courses, certifications, or mentorship - Budget allocation and resource setup required ━━━━━━━━━━━━━━━━━━━━━━━ Final Thoughts ━━━━━━━━━━━━━━━━━━━━━━━ You’ve already taken the first step by identifying your goal. Now with this structured plan, you’re on the path to mastering {analysis_result.identified_field.lower()}. Let’s keep the momentum going! Generated on: {current_date} Analysis Confidence: {"High" if analysis_result.success else "Low"} Resource Utilization: {org_percentage:.1f}% — TNA v2.0 Insights Engine """ logger.info("Strategic presentation generated successfully") return strategic_report def _format_skill_section(self, skills: List[SkillRecommendation], section_type: str) -> str: """Format skills section for strategic presentation""" if not skills: return "No skills identified in this category." formatted_skills = [] for i, skill in enumerate(skills, 1): formatted_skills.append(f"{i:2d}. {skill.skill_name} ({skill.skill_level})") return "\n".join(formatted_skills) def _get_priority_breakdown(self, skills: List[SkillRecommendation]) -> Dict[str, int]: """Get breakdown of skills by priority""" breakdown = {'High': 0, 'Medium': 0, 'Low': 0} for skill in skills: if skill.priority in breakdown: breakdown[skill.priority] += 1 return breakdown def generate_both_formats(self, analysis_result: AnalysisResult) -> Dict: """ Generate both raw and strategic formats Args: analysis_result: Complete analysis result Returns: Dictionary containing both formats """ return { 'raw_analysis': self.generate_raw_analysis(analysis_result), 'strategic_presentation': self.generate_strategic_presentation(analysis_result) } class SkillAnalysisSystem: """Main Skill Analysis System with enhanced specificity-aware approach""" def __init__(self, model_name: str = "gemma3:12b"): """Initialize the skill analysis system""" self.skill_engine = SkillInferenceEngine(model_name) self.skill_matcher = SkillMatcher() self.presenter = TNAPresentation() self.min_skills_target = 15 logger.info("Enhanced Specificity-Aware Skill Analysis System initialized") def analyze_skills(self, request_data: Dict) -> SkillAnalysisResponse: """ Main method to analyze skills using enhanced specificity-aware approach Args: request_data: Dictionary containing all necessary data Returns: SkillAnalysisResponse with recommendations """ try: # Parse request request = self._parse_request(request_data) # Step 1: Enhanced field/domain identification with specificity analysis field_analysis = self.skill_engine.identify_field_domain( request.user_query, request.department_name or "General" ) # Step 2: Filter organizational skills based on field analysis and specificity relevant_org_skills = self.skill_engine.filter_relevant_organizational_skills( field_analysis, request.all_skills_data ) # Determine which case to apply if (request.department_name and request.department_name.lower() != "null" and len(request.filtered_skills) > 3): return self._handle_case_1_improved(request, field_analysis, relevant_org_skills) else: return self._handle_case_2_improved(request, field_analysis, relevant_org_skills) except Exception as e: logger.error(f"Error in skill analysis: {e}") return SkillAnalysisResponse( recommended_skills=[], analysis_summary=f"Error in analysis: {str(e)}", department_coverage=0.0, total_skills_recommended=0, case_applied="Error", identified_field="Unknown" ) def _handle_case_1_improved(self, request: SkillAnalysisRequest, field_analysis: dict, relevant_org_skills: List[Dict]) -> SkillAnalysisResponse: """ Handle Case 1 with enhanced specificity-aware approach: Department found AND employee count > 3 """ logger.info("Applying Case 1 (Enhanced): Department found with sufficient employees") identified_field = field_analysis["primary_field"] specificity_level = field_analysis["specificity_level"] recommendations = [] used_skill_ids = set() # PRIORITY 1 (HIGH): Under-covered department skills under_covered_dept_skills = self.skill_matcher.filter_department_skills( request.filtered_skills, coverage_threshold=0.5 ) for dept_skill in under_covered_dept_skills: if dept_skill.skill_id not in used_skill_ids: recommendations.append(SkillRecommendation( skill_id=dept_skill.skill_id, skill_name=dept_skill.skill_name, skill_level=dept_skill.skill_level or "Intermediate", priority="High", source="department_existing", justification=f"Under-covered in department: {dept_skill.skill_count} employees out of {max([s.skill_count for s in request.filtered_skills])}" )) used_skill_ids.add(dept_skill.skill_id) # PRIORITY 2 (MEDIUM): Field-relevant organizational skills (now specificity-aware) for org_skill in relevant_org_skills: if org_skill['skill_id'] not in used_skill_ids: recommendations.append(SkillRecommendation( skill_id=org_skill['skill_id'], skill_name=org_skill['skill_name'], skill_level="Intermediate", priority="Medium", source="organizational_relevant", justification=f"Relevant to {identified_field} field ({specificity_level.lower()} level) and available in organization" )) used_skill_ids.add(org_skill['skill_id']) # PRIORITY 3 (LOW): External skills if needed (now specificity-aware) if len(recommendations) < self.min_skills_target: needed_count = self.min_skills_target - len(recommendations) existing_skill_names = [rec.skill_name for rec in recommendations] external_skills = self.skill_engine.suggest_external_skills( request.user_query, field_analysis, # Pass full field analysis instead of just field name existing_skill_names, needed_count ) # Add external skills with generated IDs for i, ext_skill in enumerate(external_skills): recommendations.append(SkillRecommendation( skill_id=9000 + i, skill_name=ext_skill, skill_level="Beginner", priority="Low", source="external", justification=f"External {specificity_level.lower()} skill for {identified_field} to enhance capabilities" )) # Calculate department coverage dept_coverage = len(under_covered_dept_skills) / len(request.filtered_skills) if request.filtered_skills else 0 return SkillAnalysisResponse( recommended_skills=recommendations[:self.min_skills_target], analysis_summary=f"Enhanced analysis for {request.department_name} department in {identified_field} field ({specificity_level}). Found {len(relevant_org_skills)} relevant organizational skills.", department_coverage=dept_coverage, total_skills_recommended=len(recommendations[:self.min_skills_target]), case_applied="Case 1 (Enhanced): Department found with sufficient employees", identified_field=identified_field ) def _handle_case_2_improved(self, request: SkillAnalysisRequest, field_analysis: dict, relevant_org_skills: List[Dict]) -> SkillAnalysisResponse: """ Handle Case 2 with enhanced specificity-aware approach: Department not found OR employee count ≤ 3 """ logger.info("Applying Case 2 (Enhanced): Department not found or insufficient employees") identified_field = field_analysis["primary_field"] specificity_level = field_analysis["specificity_level"] recommendations = [] used_skill_ids = set() # PRIORITY 1 (MEDIUM): Field-relevant organizational skills (now specificity-aware) for org_skill in relevant_org_skills: if org_skill['skill_id'] not in used_skill_ids: recommendations.append(SkillRecommendation( skill_id=org_skill['skill_id'], skill_name=org_skill['skill_name'], skill_level="Intermediate", priority="Medium", source="organizational_relevant", justification=f"Relevant to {identified_field} field ({specificity_level.lower()} level) and available in organization" )) used_skill_ids.add(org_skill['skill_id']) # PRIORITY 2 (LOW): External skills if needed (now specificity-aware) if len(recommendations) < self.min_skills_target: needed_count = self.min_skills_target - len(recommendations) existing_skill_names = [rec.skill_name for rec in recommendations] external_skills = self.skill_engine.suggest_external_skills( request.user_query, field_analysis, # Pass full field analysis instead of just field name existing_skill_names, needed_count ) # Add external skills for i, ext_skill in enumerate(external_skills): recommendations.append(SkillRecommendation( skill_id=9000 + i, skill_name=ext_skill, skill_level="Beginner", priority="Low", source="external", justification=f"External {specificity_level.lower()} skill for {identified_field} to fill capability gap" )) return SkillAnalysisResponse( recommended_skills=recommendations[:self.min_skills_target], analysis_summary=f"Enhanced general analysis for {identified_field} field ({specificity_level}). Found {len(relevant_org_skills)} relevant organizational skills.", department_coverage=0.0, total_skills_recommended=len(recommendations[:self.min_skills_target]), case_applied="Case 2 (Enhanced): Department not found or insufficient employees", identified_field=identified_field ) def _parse_request(self, request_data: Dict) -> SkillAnalysisRequest: """Parse request data into SkillAnalysisRequest object""" # Parse filtered skills filtered_skills = [] if 'filtered_skills' in request_data: for skill_data in request_data['filtered_skills']: filtered_skills.append(Skill( skill_id=skill_data['skill_id'], skill_name=skill_data['skill_name'], skill_level=skill_data.get('skill_level'), skill_count=skill_data.get('skill_count') )) return SkillAnalysisRequest( user_query=request_data['user_query'], admin_name=request_data['admin_name'], department_name=request_data['department_name'], filtered_skills=filtered_skills, all_skills_data=request_data.get('all_skills_data', []), all_courses_data=request_data.get('all_courses_data', {}) ) def process_json_request(self, json_data: str) -> Dict: """ Process JSON request and return JSON response (Raw format only) Args: json_data: JSON string with request data Returns: Dictionary with response data """ try: request_data = json.loads(json_data) response = self.analyze_skills(request_data) # Convert to JSON-serializable format recommendations_json = [] for rec in response.recommended_skills: recommendations_json.append({ 'skill_id': rec.skill_id, 'skill_name': rec.skill_name, 'skill_level': rec.skill_level, 'priority': rec.priority, 'source': rec.source, 'justification': rec.justification }) return { 'success': True, 'identified_field': response.identified_field, 'recommended_skills': recommendations_json, 'analysis_summary': response.analysis_summary, 'department_coverage': round(response.department_coverage, 4), 'total_skills_recommended': response.total_skills_recommended, 'case_applied': response.case_applied } except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") return { 'success': False, 'error': f'Invalid JSON format: {e}', 'recommended_skills': [] } except Exception as e: logger.error(f"Processing error: {e}") return { 'success': False, 'error': str(e), 'recommended_skills': [] } def process_with_both_formats(self, json_data: str) -> Dict: """ Process JSON request and return BOTH raw and strategic formats Args: json_data: JSON string with request data Returns: Dictionary with both presentation formats """ try: request_data = json.loads(json_data) analysis_response = self.analyze_skills(request_data) # Convert to AnalysisResult format for presentation analysis_result = AnalysisResult( success=True, identified_field=analysis_response.identified_field, case_applied=analysis_response.case_applied, total_skills_recommended=analysis_response.total_skills_recommended, department_coverage=analysis_response.department_coverage, analysis_summary=analysis_response.analysis_summary, recommended_skills=analysis_response.recommended_skills, user_query=request_data['user_query'], admin_name=request_data['admin_name'], department_name=request_data['department_name'] ) # Generate both presentations both_formats = self.presenter.generate_both_formats(analysis_result) logger.info("Both presentation formats generated successfully") return both_formats except json.JSONDecodeError as e: logger.error(f"JSON decode error: {e}") return { 'raw_analysis': { 'success': False, 'error': f'Invalid JSON format: {e}', 'recommended_skills': [] }, 'strategic_presentation': f"Error: Invalid JSON format - {e}" } except Exception as e: logger.error(f"Processing error: {e}") return { 'raw_analysis': { 'success': False, 'error': str(e), 'recommended_skills': [] }, 'strategic_presentation': f"Error: {str(e)}" } def save_analysis_report(self, json_data: str, filename_prefix: str = "tna_analysis") -> Dict[str, str]: """ Process analysis and save both formats to files Args: json_data: JSON string with request data filename_prefix: Prefix for saved files Returns: Dictionary with file paths """ try: request_data = json.loads(json_data) analysis_response = self.analyze_skills(request_data) # Convert to AnalysisResult format analysis_result = AnalysisResult( success=True, identified_field=analysis_response.identified_field, case_applied=analysis_response.case_applied, total_skills_recommended=analysis_response.total_skills_recommended, department_coverage=analysis_response.department_coverage, analysis_summary=analysis_response.analysis_summary, recommended_skills=analysis_response.recommended_skills, user_query=request_data['user_query'], admin_name=request_data['admin_name'], department_name=request_data['department_name'] ) # Save using presenter timestamp = datetime.now().strftime("%Y%m%d_%H%M%S") # Generate both formats both_formats = self.presenter.generate_both_formats(analysis_result) # Save raw analysis as JSON raw_filename = f"{filename_prefix}_raw_{timestamp}.json" with open(raw_filename, 'w', encoding='utf-8') as f: json.dump(both_formats['raw_analysis'], f, indent=2, ensure_ascii=False) # Save strategic presentation as text strategic_filename = f"{filename_prefix}_strategic_{timestamp}.txt" with open(strategic_filename, 'w', encoding='utf-8') as f: f.write(both_formats['strategic_presentation']) logger.info(f"Analysis saved to {raw_filename} and {strategic_filename}") return { 'raw_file': raw_filename, 'strategic_file': strategic_filename, 'success': True } except Exception as e: logger.error(f"Error saving analysis: {e}") return { 'success': False, 'error': str(e) } # Example usage def main(): """Example usage of the Enhanced Specificity-Aware TNA System""" # Initialize system skill_system = SkillAnalysisSystem() # Sample request data (based on your template) sample_request_general = { "user_query": "Want to improve my Python skill ability", "admin_name": "Ascent", "department_name": "Developer", "filtered_skills": [ { "skill_id": 120, "skill_name": "Python", "skill_level": "Intermediate", "skill_count": 6 }, { "skill_id": 1442, "skill_name": "Emotional Intelligence", "skill_level": "Advanced", "skill_count": 3 } ], "all_skills_data": [ {"skill_id": 1388, "skill_name": "negotiation and closure"}, {"skill_id": 1389, "skill_name": "market analysis"}, {"skill_id": 1390, "skill_name": "product knowledge"}, {"skill_id": 121, "skill_name": "JavaScript"}, {"skill_id": 122, "skill_name": "React"}, {"skill_id": 123, "skill_name": "Node.js"}, {"skill_id": 124, "skill_name": "HTML"}, {"skill_id": 125, "skill_name": "CSS"}, {"skill_id": 126, "skill_name": "SQL"}, {"skill_id": 127, "skill_name": "Git"}, {"skill_id": 128, "skill_name": "Docker"} ], "all_courses_data": { "3": [ { "courseId": "73", "name": "Java Course", "short_description": "Java Course", "description": "", "skills": ["java", "javascript", "coding"] } ] } } sample_request_specific = { "user_query": "Want to improve my Angular skills", "admin_name": "Ascent", "department_name": "Developer", "filtered_skills": [ { "skill_id": 120, "skill_name": "Python", "skill_level": "Intermediate", "skill_count": 6 }, { "skill_id": 1442, "skill_name": "Emotional Intelligence", "skill_level": "Advanced", "skill_count": 3 } ], "all_skills_data": [ {"skill_id": 121, "skill_name": "JavaScript"}, {"skill_id": 122, "skill_name": "TypeScript"}, {"skill_id": 123, "skill_name": "Angular"}, {"skill_id": 124, "skill_name": "HTML"}, {"skill_id": 125, "skill_name": "CSS"}, {"skill_id": 126, "skill_name": "React"}, {"skill_id": 127, "skill_name": "Vue.js"}, {"skill_id": 128, "skill_name": "RxJS"} ], "all_courses_data": { "3": [ { "courseId": "73", "name": "Angular Advanced Course", "short_description": "Advanced Angular Development", "description": "", "skills": ["angular", "typescript", "rxjs"] } ] } } print("="*80) print("ENHANCED SPECIFICITY-AWARE TNA SYSTEM DEMONSTRATION") print("="*80) # Test General Query print("\n" + "="*50) print("TEST 1: GENERAL QUERY - 'improve python ability'") print("="*50) general_result = skill_system.process_json_request(json.dumps(sample_request_general)) print(f"Field Identified: {general_result['identified_field']}") print(f"Total Skills: {general_result['total_skills_recommended']}") print(f"Summary: {general_result['analysis_summary']}") print("\nTop 5 Recommended Skills:") for i, skill in enumerate(general_result['recommended_skills'], 1): print(f"{i}. {skill['skill_name']} ({skill['source']}) - {skill['justification']}") # Test Specific Query print("\n" + "="*50) print("TEST 2: HIGHLY SPECIFIC QUERY - 'improve Angular skills'") print("="*50) specific_result = skill_system.process_json_request(json.dumps(sample_request_specific)) print(f"Field Identified: {specific_result['identified_field']}") print(f"Total Skills: {specific_result['total_skills_recommended']}") print(f"Summary: {specific_result['analysis_summary']}") print("\nTop 5 Recommended Skills:") for i, skill in enumerate(specific_result['recommended_skills'], 1): print(f"{i}. {skill['skill_name']} ({skill['source']}) - {skill['justification']}") print("\n" + "="*80) print("ENHANCEMENT SUCCESS: Different queries now produce different, contextually appropriate skill recommendations!") print("="*80) ##### print("Strategic") json_data = json.dumps(sample_request_specific) result = skill_system.process_with_both_formats(json_data) strategic_report = result['strategic_presentation'] print(strategic_report) if __name__ == "__main__": main()