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 improved two-step 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) -> str: """ Step 1: Identify the field/domain based on user query and department Args: user_query: User's business goal or challenge department_name: Target department name Returns: Identified field/domain """ 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} Based on the user's goal and department, identify the specific professional field or domain this relates to. Common fields include: - Web Development (Frontend, Backend, Full-stack) - Mobile App Development - Data Science & Analytics - Machine Learning & AI - DevOps & Cloud Computing - Cybersecurity - Database Management - Software Architecture - Quality Assurance & Testing - Project Management - Digital Marketing - Sales & Customer Relations - Human Resources - Finance & Accounting - Operations Management - Product Management - UI/UX Design - Business Analysis - Research & Development Instructions: 1. Analyze the user's goal carefully in context of their department 2. Identify the most specific field/domain that matches their need 3. Consider both technical and business domains 4. Return only the field name (e.g., "Web Development", "Data Science") 5. Be specific but not overly narrow Field/Domain: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, department_name=department_name ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Clean and extract field name field_name = self._clean_field_name(response.content) logger.info(f"Identified field/domain: {field_name}") return field_name except Exception as e: logger.error(f"Error in field identification: {e}") return "General" def filter_relevant_organizational_skills(self, identified_field: str, all_skills_data: List[Dict]) -> List[Dict]: """ Step 2: Filter organizational skills that are relevant to the identified field Args: identified_field: The field/domain identified in step 1 all_skills_data: Complete list of organizational skills Returns: List of field-relevant organizational skills """ # 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 ]) prompt_template = PromptTemplate( input_variables=["identified_field", "skills_text"], template=""" You are an expert skills analyst specializing in matching skills to professional fields. Identified Field/Domain: {identified_field} Below is a list of skills available in the organization: {skills_text} Your task is to identify which skills from this list are RELEVANT to the "{identified_field}" field. Instructions: 1. Review each skill carefully 2. Determine if the skill is directly or indirectly relevant to the identified field 3. Include both technical and soft skills that are valuable in this field 4. Be inclusive rather than restrictive - if a skill could be useful, include it 5. Consider transferable skills and foundational abilities 6. Return ONLY the skill names that are relevant, one per line Format your response as: - Skill Name 1 - Skill Name 2 - Skill Name 3 ... Relevant Skills: """ ) try: formatted_prompt = prompt_template.format( identified_field=identified_field, skills_text=skills_text ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Parse relevant skill names from response relevant_skill_names = self._parse_skills_from_response(response.content) # Match back to original skill data relevant_skills = [] for skill_name in relevant_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 logger.info(f"Filtered {len(relevant_skills)} relevant organizational skills from {len(all_skills_data)} total skills") return relevant_skills except Exception as e: logger.error(f"Error in organizational skill filtering: {e}") return [] def suggest_external_skills(self, user_query: str, identified_field: str, existing_skills: List[str], needed_count: int) -> List[str]: """ Suggest external skills to fill gaps, now with field context Args: user_query: User's business goal identified_field: The identified field/domain existing_skills: Skills already identified needed_count: Number of additional skills needed Returns: List of external skill suggestions """ existing_skills_text = ", ".join(existing_skills) if existing_skills else "None" prompt_template = PromptTemplate( input_variables=["user_query", "identified_field", "existing_skills", "needed_count"], template=""" You are an expert training consultant specializing in the "{identified_field}" field. User Goal: {user_query} Field: {identified_field} Already Covered Skills: {existing_skills} Additional Skills Needed: {needed_count} Suggest {needed_count} additional external skills that are specifically valuable in the "{identified_field}" field but are not covered in the existing skills list. Instructions: 1. Focus on skills that are trending or essential in the {identified_field} field 2. Include both emerging technologies and foundational skills that might be missing 3. Consider industry best practices and future-oriented competencies for this field 4. Ensure skills complement the existing skills 5. Avoid duplicating existing skills 6. Be specific to the {identified_field} domain Return exactly {needed_count} skills in the following format: - External Skill 1 - External Skill 2 - External Skill 3 ... External Skills for {identified_field}: """ ) try: formatted_prompt = prompt_template.format( user_query=user_query, identified_field=identified_field, existing_skills=existing_skills_text, needed_count=needed_count ) response = self.llm.invoke([HumanMessage(content=formatted_prompt)]) # Extract skills from response skills = self._parse_skills_from_response(response.content) logger.info(f"Suggested {len(skills)} external skills for {identified_field}") return skills[:needed_count] # Ensure we don't exceed needed count except Exception as e: logger.error(f"Error in external skill suggestion: {e}") return [] def _clean_field_name(self, field_text: str) -> str: """Clean and normalize field name""" # Extract first line if multiple lines field_text = field_text.strip().split('\n')[0] # Remove common prefixes/suffixes field_text = field_text.replace("Field/Domain:", "").replace("Field:", "").replace("Domain:", "") # Remove quotes and extra spaces 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() # Remove bullet points, numbers, and other prefixes line = re.sub(r'^[-•*\d+\.\)]\s*', '', line) line = line.strip() if line and len(line) > 2: # Valid skill names should be more than 2 characters skills.append(line) return skills def _skills_match(self, skill1: str, skill2: str) -> bool: """Check if two skill names match with reasonable similarity""" # Normalize both skills 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() # Exact match if skill1_norm == skill2_norm: return True # High similarity match similarity = SequenceMatcher(None, skill1_norm, skill2_norm).ratio() if similarity > 0.8: return True # Substring match 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""" 📊 TRAINING NEEDS ANALYSIS REPORT ═══════════════════════════════════════════════════════════════ Organization: {analysis_result.admin_name} Department: {analysis_result.department_name} Analysis Date: {current_date} Field Identified: {analysis_result.identified_field} 🎯 EXECUTIVE SUMMARY ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ User Goal: "{analysis_result.user_query}" AI-Identified Field: {analysis_result.identified_field} Case Applied: {analysis_result.case_applied} Key Findings: • {len(org_skills)} organizational skills aligned with business goal • {len(ext_skills)} external skills identified for capability enhancement • {org_percentage:.1f}% utilization of existing organizational assets • Ready-to-implement training pathway identified 📈 STRATEGIC ANALYSIS ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Field Identification & Relevance: ├─ User Intent: {analysis_result.user_query} ├─ AI Analysis: {analysis_result.identified_field} field identified ├─ Organizational Alignment: {len(org_skills)} skills available internally └─ Strategic Fit: {"High" if len(org_skills) > len(ext_skills) else "Medium"} - Organization has {"strong" if len(org_skills) > 5 else "moderate"} infrastructure for this field 🟢 IMMEDIATE PRIORITIES (Organizational Skills) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {self._format_skill_section(org_skills + dept_skills, "organizational")} Advantages: → Training infrastructure already exists → Lower implementation cost → Can start immediately → High ROI - leverages existing investment 🟡 GROWTH OPPORTUNITIES (External Skills) ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ {self._format_skill_section(ext_skills, "external")} Investment Requirements: → External training partners/resources needed → 3-6 month implementation timeline → Strategic value: Future-proofing and competitive advantage 💼 BUSINESS IMPACT ASSESSMENT ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Resource Optimization: • {org_percentage:.1f}% skill-match rate with existing organizational capabilities • Immediate deployment possible for {len(org_skills + dept_skills)} skills • Cost efficiency through internal resource utilization Capability Gap Closure: • Current State: {analysis_result.user_query} • Target State: Comprehensive {analysis_result.identified_field.lower()} proficiency • Gap: {analysis_result.total_skills_recommended} targeted skills for structured development 🛣️ IMPLEMENTATION ROADMAP ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Phase 1 (0-3 months): Organizational Skills ├─ Priority: {"High" if dept_skills else "Medium"} ({len(org_skills + dept_skills)} skills) ├─ Resources: Internal training programs, existing courses ├─ Skills: {", ".join([s.skill_name for s in (org_skills + dept_skills)[:5]])}{"..." if len(org_skills + dept_skills) > 5 else ""} └─ Expected Outcome: Enhanced foundation in organizational tech stack Phase 2 (3-6 months): External Skills ├─ Priority: Low ({len(ext_skills)} skills) ├─ Resources: External training partners, certifications ├─ Skills: {", ".join([s.skill_name for s in ext_skills[:3]])}{"..." if len(ext_skills) > 3 else ""} └─ Expected Outcome: Advanced capabilities and competitive edge 📋 RECOMMENDED ADMIN ACTIONS ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ Immediate (This Week): 1. Review existing courses for: {", ".join([s.skill_name for s in (org_skills + dept_skills)[:3]])} 2. Assign learning paths using organizational training resources 3. Schedule progress check-ins for skill development tracking Short-term (1-3 Months): 1. Monitor progress on organizational skills 2. Research external training partners for: {", ".join([s.skill_name for s in ext_skills[:3]])} 3. Plan budget allocation for Phase 2 external training Long-term (3-6 Months): 1. Implement external training for advanced skills 2. Measure skill improvement impact on department performance 3. Scale successful approaches to other team members ✅ DECISION SUPPORT SUMMARY ━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━ RECOMMENDED APPROACH: Phased Implementation Phase 1 Benefits: • {((len(org_skills + dept_skills)) / analysis_result.total_skills_recommended * 100):.0f}% of skills can start immediately • Uses existing organizational assets • Low implementation risk • Quick wins for team morale Phase 2 Benefits: • {(len(ext_skills) / analysis_result.total_skills_recommended * 100):.0f}% of skills provide competitive advantage • Future-proofs team capabilities • Aligns with industry trends • Long-term strategic value Analysis Confidence: {"High" if analysis_result.success else "Low"} Resource Utilization Score: {org_percentage:.1f}% Implementation Complexity: {"Low" if len(org_skills) > len(ext_skills) else "Medium"} ═══════════════════════════════════════════════════════════════ Report Generated: {current_date} Analysis Engine: TNA v2.0 (Improved Field-Based Approach) ═══════════════════════════════════════════════════════════════ """ 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 improved two-step approach and integrated presentation""" 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("Improved Skill Analysis System with Presentation initialized") def analyze_skills(self, request_data: Dict) -> SkillAnalysisResponse: """ Main method to analyze skills using improved two-step approach Args: request_data: Dictionary containing all necessary data Returns: SkillAnalysisResponse with recommendations """ try: # Parse request request = self._parse_request(request_data) # Step 1: Identify field/domain using LLM identified_field = self.skill_engine.identify_field_domain( request.user_query, request.department_name or "General" ) # Step 2: Filter organizational skills relevant to the field relevant_org_skills = self.skill_engine.filter_relevant_organizational_skills( identified_field, 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, identified_field, relevant_org_skills) else: return self._handle_case_2_improved(request, identified_field, 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, identified_field: str, relevant_org_skills: List[Dict]) -> SkillAnalysisResponse: """ Handle Case 1 with improved approach: Department found AND employee count > 3 """ logger.info("Applying Case 1 (Improved): Department found with sufficient employees") 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 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 and available in organization" )) used_skill_ids.add(org_skill['skill_id']) # PRIORITY 3 (LOW): External skills if needed 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, identified_field, 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 {identified_field} skill 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"Analysis for {request.department_name} department in {identified_field} field. 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 (Improved): Department found with sufficient employees", identified_field=identified_field ) def _handle_case_2_improved(self, request: SkillAnalysisRequest, identified_field: str, relevant_org_skills: List[Dict]) -> SkillAnalysisResponse: """ Handle Case 2 with improved approach: Department not found OR employee count ≤ 3 """ logger.info("Applying Case 2 (Improved): Department not found or insufficient employees") recommendations = [] used_skill_ids = set() # PRIORITY 1 (MEDIUM): Field-relevant organizational skills 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 and available in organization" )) used_skill_ids.add(org_skill['skill_id']) # PRIORITY 2 (LOW): External skills if needed 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, identified_field, 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 {identified_field} skill to fill capability gap" )) return SkillAnalysisResponse( recommended_skills=recommendations[:self.min_skills_target], analysis_summary=f"General analysis for {identified_field} field. 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 (Improved): 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 Integrated TNA System with both presentation formats""" # Initialize system skill_system = SkillAnalysisSystem() # Sample request data (based on your template) sample_request = { "user_query": "Want to improve my tech 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"] } ] } } print("="*80) print("INTEGRATED TNA SYSTEM - DUAL FORMAT PRESENTATION") print("="*80) # Option 1: Get both formats together print("\n" + "="*50) print("OPTION 1: BOTH FORMATS TOGETHER") print("="*50) both_formats = skill_system.process_with_both_formats(json.dumps(sample_request)) print("\n[RAW ANALYSIS FORMAT]:") print("-" * 40) print(json.dumps(both_formats['raw_analysis'], indent=2)) print("\n[STRATEGIC PRESENTATION FORMAT]:") print("-" * 40) print(both_formats['strategic_presentation']) # Option 2: Get raw format only (original method) print("\n" + "="*50) print("OPTION 2: RAW FORMAT ONLY") print("="*50) raw_only = skill_system.process_json_request(json.dumps(sample_request)) print(json.dumps(raw_only, indent=2)) # Option 3: Save both formats to files print("\n" + "="*50) print("OPTION 3: SAVE TO FILES") print("="*50) file_paths = skill_system.save_analysis_report(json.dumps(sample_request), "demo_tna") print(f"Files saved: {file_paths}") if __name__ == "__main__": main()