from fastapi import FastAPI, HTTPException from pydantic import BaseModel from typing import List, Dict, Optional, Any import json import logging from dataclasses import asdict # Import your existing modules from tna_dep import TNASystem from tna_skill_second import SkillAnalysisSystem from tna_course_trial import CourseFieldIdentifier # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="TNA System API", version="1.0.0") # =================== DEPARTMENT ENDPOINT MODELS =================== class Department(BaseModel): id: int name: str class DepartmentRequest(BaseModel): user_prompt: str organization_departments: List[Department] class DepartmentResponse(BaseModel): identified_department: str department_id: int # =================== SKILLS & COURSES ENDPOINT MODELS =================== class FilteredSkill(BaseModel): skill_id: int skill_name: str skill_level: Optional[str] = None skill_count: Optional[int] = None class AllSkillData(BaseModel): skill_id: int skill_name: str class OfflineCourse(BaseModel): course_id: str course_name: str short_description: str description: str skills: List[str] class SkillsCourseRequest(BaseModel): user_query: str admin_name: str department_name: str filtered_skills: List[FilteredSkill] total_users_in_department: int all_skills_data: List[AllSkillData] offline_courses: List[OfflineCourse] class CourseRecommendation(BaseModel): course_id: str name: str reason: str percentage: float class SkillsCourseResponse(BaseModel): skill_analysis: Dict[str, Any] courses_offline: List[CourseRecommendation] strategic_presentation: str # Add this new field class SkillsCourseResponseWithStrategic(BaseModel): skill_analysis: Dict[str, Any] courses_offline: List[CourseRecommendation] strategic_presentation: str # New field for strategic presentation class SkillsCourseResponseOptions(BaseModel): skill_analysis: Dict[str, Any] courses_offline: List[CourseRecommendation] strategic_presentation: Optional[str] = None # Optional strategic presentation # =================== SYSTEM INITIALIZATION =================== class TNAIntegratedSystem: """Integrated TNA System combining department, skill, and course analysis""" def __init__(self): """Initialize all three systems""" self.department_system = TNASystem() self.skill_system = SkillAnalysisSystem() self.course_system = CourseFieldIdentifier() logger.info("TNA Integrated System initialized") def process_skills_and_courses(self, request: SkillsCourseRequest) -> SkillsCourseResponse: """ Process skill analysis and course recommendations with strategic presentation Args: request: SkillsCourseRequest with all necessary data Returns: SkillsCourseResponse with skill analysis, course recommendations, and strategic presentation """ try: # Step 1: Skill Analysis skill_request = { "user_query": request.user_query, "admin_name": request.admin_name, "department_name": request.department_name, "filtered_skills": [ { "skill_id": skill.skill_id, "skill_name": skill.skill_name, "skill_level": skill.skill_level, "skill_count": skill.skill_count } for skill in request.filtered_skills ], "all_skills_data": [ { "skill_id": skill.skill_id, "skill_name": skill.skill_name } for skill in request.all_skills_data ] } # Get both raw analysis and strategic presentation both_formats = self.skill_system.process_with_both_formats( json.dumps(skill_request) ) skill_result = both_formats['raw_analysis'] strategic_presentation = both_formats['strategic_presentation'] logger.info(f"Skill analysis completed: {skill_result.get('total_skills_recommended', 0)} skills") # Step 2: Course Analysis (only if skill analysis succeeded) courses_offline = [] if skill_result.get('success', False) and skill_result.get('identified_field'): course_request = { "identified_field": skill_result['identified_field'], "courses": [ { "course_id": course.course_id, "course_name": course.course_name, "course_short_description": course.short_description, "course_description": course.description, "skill": ",".join(course.skills) if course.skills else "" } for course in request.offline_courses ] } course_result = self.course_system.process_json_request( json.dumps(course_request) ) logger.info(f"Course analysis: {course_result.get('total_matching_courses', 0)} courses matched") # Generate course recommendations with reasons and percentages if course_result.get('success', False): courses_offline = self._generate_course_recommendations( course_result['matching_courses'], skill_result, course_result.get('filtering_reasoning', '') ) # Combine results - now including strategic presentation return SkillsCourseResponse( skill_analysis={ "identified_field": skill_result.get('identified_field', 'Unknown'), "total_skills_recommended": skill_result.get('total_skills_recommended', 0), "recommended_skills": skill_result.get('recommended_skills', []), "analysis_summary": skill_result.get('analysis_summary', ''), "case_applied": skill_result.get('case_applied', ''), "department_coverage": skill_result.get('department_coverage', 0.0), "success": skill_result.get('success', False) }, courses_offline=courses_offline, strategic_presentation=strategic_presentation # Add this line ) except Exception as e: logger.error(f"Error in skills and courses processing: {e}") raise HTTPException(status_code=500, detail=f"Skills and courses processing failed: {str(e)}") def _generate_course_recommendations(self, matching_courses: List[Dict], skill_result: Dict, reasoning: str) -> List[CourseRecommendation]: """ Generate course recommendations with reasons and percentages Args: matching_courses: List of courses that matched the field skill_result: Skill analysis result reasoning: LLM reasoning for course selection Returns: List of CourseRecommendation objects """ recommendations = [] identified_field = skill_result.get('identified_field', 'Unknown') recommended_skills = skill_result.get('recommended_skills', []) # Calculate skill coverage for each course for i, course in enumerate(matching_courses): # Calculate percentage based on position and field relevance # Higher percentage for courses that appear first (more relevant) base_percentage = max(95 - (i * 5), 60) # Start at 95%, decrease by 5% each position, min 60% # Adjust percentage based on skill matches skill_match_bonus = self._calculate_skill_match_bonus(course, recommended_skills) final_percentage = min(base_percentage + skill_match_bonus, 98) # Generate reason reason = self._generate_course_reason(course, identified_field, skill_match_bonus > 0) recommendations.append(CourseRecommendation( course_id=course['course_id'], name=course['course_name'], reason=reason, percentage=round(final_percentage, 1) )) # Sort by percentage (highest first) recommendations.sort(key=lambda x: x.percentage, reverse=True) return recommendations def _calculate_skill_match_bonus(self, course: Dict, recommended_skills: List[Dict]) -> float: """Calculate bonus percentage based on skill matches""" if not recommended_skills: return 0 course_skills = course.get('skill', '').lower().split(',') course_skills = [skill.strip() for skill in course_skills if skill.strip()] if not course_skills: return 0 matches = 0 for skill_rec in recommended_skills: skill_name = skill_rec.get('skill_name', '').lower() for course_skill in course_skills: if skill_name in course_skill or course_skill in skill_name: matches += 1 break # Bonus: 2% per skill match, max 10% return min(matches * 2, 10) def _generate_course_reason(self, course: Dict, identified_field: str, has_skill_match: bool) -> str: """Generate explanation for why this course was recommended""" course_name = course.get('course_name', 'Unknown Course') if has_skill_match: return f"Directly relevant to {identified_field} field with matching skill requirements. Course '{course_name}' covers essential technologies and concepts needed for your goal." else: return f"Relevant to {identified_field} field based on course content and description. Course '{course_name}' provides foundational knowledge that supports your learning objectives." # Initialize the integrated system tna_system = TNAIntegratedSystem() # =================== API ENDPOINTS =================== @app.post("/analyze-department", response_model=DepartmentResponse) async def analyze_department(request: DepartmentRequest): """ Department Analysis endpoint - FIRST CALL Identifies the most relevant department based on user prompt """ try: logger.info(f"Processing department analysis for: {request.user_prompt}") # Prepare department request department_request = { "user_prompt": request.user_prompt, "organization_departments": [ {"id": dept.id, "name": dept.name} for dept in request.organization_departments ] } # Process department analysis department_result = tna_system.department_system.process_json_request( json.dumps(department_request) ) logger.info(f"Department analysis result: {department_result}") # Extract results if department_result.get('success', False) and department_result.get('matched', False): identified_department = department_result.get('department_name', 'Unknown') department_id = department_result.get('department_id', 0) else: # If no match found, return the inferred department with ID 0 identified_department = department_result.get('inferred_department', 'Unknown') department_id = 0 result = DepartmentResponse( identified_department=identified_department, department_id=department_id ) logger.info(f"Department analysis completed: {identified_department} (ID: {department_id})") return result except Exception as e: logger.error(f"Department analysis failed: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.post("/analyze-skills-courses", response_model=SkillsCourseResponse) async def analyze_skills_courses(request: SkillsCourseRequest): """ Skills and Courses Analysis endpoint - SECOND CALL Processes skill analysis and course recommendations based on department context """ try: logger.info(f"Processing skills and courses analysis for: {request.user_query}") result = tna_system.process_skills_and_courses(request) logger.info("Skills and courses analysis completed successfully") return result except Exception as e: logger.error(f"Skills and courses analysis failed: {e}") raise HTTPException(status_code=500, detail=str(e)) @app.get("/health") async def health_check(): """Health check endpoint""" return {"status": "healthy", "message": "TNA System API is running"} @app.get("/") async def root(): """Root endpoint with API information""" return { "message": "TNA System API", "version": "1.0.0", "flow": "Two-step process: 1) Department Analysis → 2) Skills & Courses Analysis", "endpoints": { "analyze-department": "POST /analyze-department - FIRST CALL: Department identification", "analyze-skills-courses": "POST /analyze-skills-courses - SECOND CALL: Skills analysis and course recommendations", "health": "GET /health - Health check", "docs": "GET /docs - API documentation" } } # =================== EXAMPLE REQUESTS FOR TESTING =================== example_department_request = { "user_prompt": "Want to improve my teams sales ability", "organization_departments": [ {"id": 160, "name": "Operations"}, {"id": 161, "name": "HR"}, {"id": 162, "name": "Finance"}, {"id": 170, "name": "Product Management"}, {"id": 171, "name": "sales"}, {"id": 172, "name": "Developer"} ] } example_skills_courses_request = { "user_query": "Want to improve my technical ability", "admin_name": "Ascent", "department_name": "Developer", "filtered_skills": [ { "skill_id": 120, "skill_name": "Python", "skill_level": "Intermediate", "skill_count": 3 } ], "total_users_in_department": 5, "all_skills_data": [ {"skill_id": 1375, "skill_name": "communication skill"}, {"skill_id": 1376, "skill_name": "technical skills"}, {"skill_id": 121, "skill_name": "JavaScript"}, {"skill_id": 122, "skill_name": "Python"}, {"skill_id": 123, "skill_name": "Angular"} ], "offline_courses": [ { "course_id": "32", "course_name": "General Knowledge", "short_description": "General Knowledge", "description": "", "skills": [] }, { "course_id": "43", "course_name": "Python Programming for Complete Beginners", "short_description": "Python Programming for Complete Beginners", "description": "Learn Python from scratch", "skills": ["python", "programming"] }, { "course_id": "63", "course_name": "Angular Basic", "short_description": "Angular framework course", "description": "Complete Angular course", "skills": ["angular", "typescript", "javascript"] } ] } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)