from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from typing import List, Dict, Optional, Any, Union import json import logging import asyncio from dataclasses import asdict from fastapi.responses import JSONResponse from fastapi.exceptions import RequestValidationError # Import the enhanced skill analysis system, course filter, online course system, and caching from tna_skill_second import SkillAnalysisSystem from tester import DomainAwareCourseFilter from online_course import OnlineCourseRecommendationSystem from course_cache import course_cache # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = FastAPI(title="TNA System API", version="1.0.0") # Initialize the enhanced skill analysis system, course filter, and online course system skill_analysis_system = SkillAnalysisSystem(model_name="gemma3:12b") course_filter_system = DomainAwareCourseFilter(model_name="gemma3:12b", batch_size=30) online_course_system = OnlineCourseRecommendationSystem(model_name="gemma3:12b") @app.exception_handler(RequestValidationError) async def validation_exception_handler(request: Request, exc: RequestValidationError): logger.error(f"Validation error: {exc}") return JSONResponse( status_code=422, content={"detail": exc.errors(), "body": exc.body} ) @app.middleware("http") async def log_requests(request: Request, call_next): if request.url.path == "/analyze-skills-courses": body = await request.body() logger.info(f"Raw request body: {body.decode()}") # Re-create request for further processing async def receive(): return {"type": "http.request", "body": body} request._receive = receive response = await call_next(request) return response # =================== 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 confidence: Optional[str] = "none" # =================== 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[Union[str, Dict[str, Any]]] class SkillsCourseRequest(BaseModel): user_query: str admin_name: str department_name: str filtered_skills: List[Dict[str, Any]] # Changed to match PHP structure 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 OnlineCourseRecommendation(BaseModel): title: str platform: str instructor: str duration: str skill_level: str description: str url: str price: str rating: str skills_covered: List[str] relevance_score: int reason: str class SkillsCourseResponse(BaseModel): skill_analysis: Dict[str, Any] courses_offline: List[CourseRecommendation] online_recommended_courses: List[OnlineCourseRecommendation] strategic_presentation: str class OnlineCourseRequest(BaseModel): domain: str field_prompt: str max_courses: Optional[int] = 10 class OnlineCourseResponse(BaseModel): success: bool domain: str field_prompt: str total_recommendations: int course_recommendations: List[OnlineCourseRecommendation] search_summary: str from department import analyze_department # =================== FULL TNA ENDPOINT MODELS =================== class DepartmentData(BaseModel): id: int name: str total_users: int filtered_skills: List[Dict[str, Any]] class FullTnaRequest(BaseModel): user_query: str admin_name: str departments: List[DepartmentData] all_skills_data: List[AllSkillData] offline_courses: List[OfflineCourse] class FullTnaResponse(SkillsCourseResponse): identified_department: str department_id: Optional[int] confidence: str async def perform_enhanced_analysis( user_query: str, admin_name: str, department_name: str, filtered_skills: List[Dict[str, Any]], total_users_in_department: int, all_skills_data: List[AllSkillData], offline_courses: List[OfflineCourse] ) -> SkillsCourseResponse: """ Reusable core logic for skills and course analysis with async optimization """ try: logger.info(f"Enhanced analysis for query: {user_query[:100]}...") # Convert filtered_skills to the format expected by tna_skill_second filtered_skills_formatted = [] if filtered_skills: for skill_group in filtered_skills: if 'skills' in skill_group: # Handle the PHP format: [{'user_id': 0, 'skills': ['Python', 'JS']}] for i, skill_name in enumerate(skill_group['skills']): filtered_skills_formatted.append({ "skill_id": 5000 + i, # Generate temp IDs for dept skills "skill_name": skill_name, "skill_level": "Intermediate", "skill_count": total_users_in_department // 2 # Estimate }) # Convert all_skills_data format all_skills_formatted = [] for skill in all_skills_data: all_skills_formatted.append({ "skill_id": skill.skill_id, "skill_name": skill.skill_name }) # Convert offline courses format for skill analysis (needs simple format) all_courses_formatted = {} if offline_courses: all_courses_formatted["courses"] = [] for course in offline_courses: # Truncate description slightly for the main skill analysis context too desc = course.description if len(desc) > 800: desc = desc[:797] + "..." all_courses_formatted["courses"].append({ "courseId": course.course_id, "name": course.course_name, "short_description": course.short_description, "description": desc, "skills": course.skills }) # Prepare request data for tna_skill_second analysis_request = { "user_query": user_query, "admin_name": admin_name, "department_name": department_name, "filtered_skills": filtered_skills_formatted, "all_skills_data": all_skills_formatted, "all_courses_data": all_courses_formatted } # Call enhanced skill analysis system (Blocking call, run in executor if needed, but keeping simple for core logic) # Using run_in_executor to avoid blocking the event loop loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, skill_analysis_system.process_json_request, json.dumps(analysis_request) ) logger.info(f"Enhanced analysis completed - Field: {result.get('identified_field', 'Unknown')}, Skills: {result.get('total_skills_recommended', 0)}") # Define async function for Offline Course Filtering async def get_offline_matches(): if not offline_courses: return [] try: # Prepare course data for filtering with AGGRESSIVE truncation for speed courses_for_filtering = [] for course in offline_courses: # Convert skills to string format if needed skills_str = "" if course.skills: if isinstance(course.skills, list): skills_str = ", ".join([str(s) for s in course.skills]) else: skills_str = str(course.skills) # Truncate description to 500 chars to speed up LLM token processing desc = course.description if len(desc) > 500: desc = desc[:497] + "..." courses_for_filtering.append({ "course_id": course.course_id, "course_name": course.course_name, "course_short_description": course.short_description, "course_description": desc, "skill": skills_str }) # Create course filtering request course_filter_request = { "domain": result.get('identified_field', 'Technology'), "field_prompt": user_query, "courses": courses_for_filtering } # Run filtering in executor course_filter_result = await loop.run_in_executor( None, course_filter_system.process_json_request, json.dumps(course_filter_request) ) if course_filter_result.get('success'): matches = course_filter_result.get('course_matches', []) logger.info(f"Course filtering completed - {len(matches)} matches found") return matches else: logger.warning(f"Course filtering failed: {course_filter_result.get('error')}") return [] except Exception as e: logger.error(f"Course filtering error: {e}") return [] # Define async function for Online Course Recommendations async def get_online_matches(): try: identified_field = result.get('identified_field', 'Technology') # Check cache first cached_online_courses = course_cache.get_cached_courses( domain=identified_field, field_prompt=user_query, course_type="online" ) if cached_online_courses: logger.info(f"Using cached online courses: {len(cached_online_courses)} courses") return cached_online_courses else: # Generate new online course recommendations # Request 7 to ensure we have at least 5 high quality ones online_course_request = { "domain": identified_field, "field_prompt": user_query, "max_courses": 7 } online_course_result = await loop.run_in_executor( None, online_course_system.process_json_request, json.dumps(online_course_request) ) if online_course_result.get('success'): recs = online_course_result.get('course_recommendations', []) # Cache the results course_cache.cache_courses( domain=identified_field, field_prompt=user_query, courses=recs, course_type="online" ) logger.info(f"Generated and cached {len(recs)} online courses") return recs else: logger.warning(f"Online course generation failed: {online_course_result.get('error')}") return [] except Exception as e: logger.error(f"Online course recommendation error: {e}") return [] # Execute both tasks concurrently course_matches, online_courses = await asyncio.gather(get_offline_matches(), get_online_matches()) # Build story-format strategic presentation dept_skills = [s for s in result.get('recommended_skills', []) if s['source'] == 'department_existing'] org_skills = [s for s in result.get('recommended_skills', []) if s['source'] == 'organizational_relevant'] ext_skills = [s for s in result.get('recommended_skills', []) if s['source'] == 'external'] total_skills = len(result.get('recommended_skills', [])) identified_field = result.get('identified_field', 'your chosen field') strategic_parts = [] # Story Opening strategic_parts.append(f"📈 **Your Journey to Master {identified_field}**") strategic_parts.append("=" * 50) strategic_parts.append("") strategic_parts.append(f"Hello {admin_name}! 👋") strategic_parts.append("") strategic_parts.append(f'We heard you say: *"{user_query}"* - and that\'s exactly the kind of growth mindset that separates good professionals from great ones!') strategic_parts.append("") strategic_parts.append(f"After analyzing your current situation in {identified_field}, we\'ve crafted a personalized skill development roadmap just for you. Here\'s your story:") strategic_parts.append("") # Department Skills Story if dept_skills and department_name.lower() != "null": strategic_parts.append("🏢 **Chapter 1: Building on Your Department Foundation**") strategic_parts.append("") strategic_parts.append(f"Good news! Your {department_name} department already has some valuable skills that align perfectly with your goals. Think of these as your existing superpowers that just need some sharpening:") strategic_parts.append("") for i, skill in enumerate(dept_skills, 1): strategic_parts.append(f" {i}. **{skill['skill_name']}** - {skill['justification']}") strategic_parts.append("") strategic_parts.append("💡 *Why start here?* You already have colleagues using these skills, which means mentorship opportunities and immediate practical application!") strategic_parts.append("") # Organization Skills Story if org_skills: chapter_num = 2 if dept_skills else 1 strategic_parts.append(f"🌟 **Chapter {chapter_num}: Expanding Your Organizational Arsenal**") strategic_parts.append("") strategic_parts.append("Your organization is a goldmine of additional skills that can accelerate your growth. These are the hidden gems available right within your company:") strategic_parts.append("") for i, skill in enumerate(org_skills, 1): strategic_parts.append(f" {i}. **{skill['skill_name']}** - {skill['justification']}") strategic_parts.append("") strategic_parts.append("🎯 *The advantage?* These skills are already valued by your organization, so developing them directly boosts your internal career prospects!") strategic_parts.append("") # External Skills Story if ext_skills: chapter_num = 3 if (dept_skills and org_skills) else (2 if (dept_skills or org_skills) else 1) strategic_parts.append(f"🚀 **Chapter {chapter_num}: The Competitive Edge - External Skills**") strategic_parts.append("") strategic_parts.append("To truly excel and stand out in the market, here are the cutting-edge skills that will give you the competitive advantage:") strategic_parts.append("") for i, skill in enumerate(ext_skills, 1): strategic_parts.append(f" {i}. **{skill['skill_name']}** - {skill['justification']}") strategic_parts.append("") strategic_parts.append("⚡ *The impact?* These skills are already valued by your organization, so developing them directly boosts your internal career prospects!") strategic_parts.append("") # Course Recommendations Story # Sort courses by match percentage sorted_courses = sorted(course_matches, key=lambda x: x.get('match_percentage', 0), reverse=True) top_offline_courses = sorted_courses[:5] # STRICTLY limit to top 5 if top_offline_courses: final_chapter_num = len([x for x in [dept_skills, org_skills, ext_skills] if x]) + 1 strategic_parts.append(f"📚 **Chapter {final_chapter_num}: Your Learning Pathway - Recommended Courses**") strategic_parts.append("") strategic_parts.append("Great news! We found some fantastic courses in your organization that perfectly align with your skill development goals:") strategic_parts.append("") for i, course in enumerate(top_offline_courses, 1): # Show top 5 courses percentage = course.get('match_percentage', 0) strategic_parts.append(f" {i}. **{course['course_name']}** ({percentage}% match)") strategic_parts.append(f" Why it's perfect: {course['reason']}") strategic_parts.append("") if len(course_matches) > 5: strategic_parts.append(f" ...and {len(course_matches) - 5} more courses available!") strategic_parts.append("") strategic_parts.append("🎓 *The advantage?* These courses are available right in your organization, meaning:") strategic_parts.append(" • No external training costs") strategic_parts.append(" • Immediate access and enrollment") strategic_parts.append(" • Tailored to your company's needs") strategic_parts.append(" • Support from internal training team") strategic_parts.append("") # Online Courses Story # Sort online courses by match percentage sorted_online_courses = sorted(online_courses, key=lambda x: x.get('match_percentage', 0), reverse=True) top_online_courses = sorted_online_courses[:5] # STRICTLY limit to top 5 if top_online_courses: final_chapter_num = len([x for x in [dept_skills, org_skills, ext_skills, top_offline_courses] if x]) + 1 strategic_parts.append(f"🌐 **Chapter {final_chapter_num}: Global Learning - Top Online Courses**") strategic_parts.append("") strategic_parts.append("To complement your organizational learning, here are some outstanding online courses from top platforms:") strategic_parts.append("") for i, course in enumerate(top_online_courses, 1): # Show top 5 online courses percentage = course.get('match_percentage', 0) platform = course.get('platform', 'Online Platform') price = course.get('price', 'Check Platform') rating = course.get('rating', 'N/A') duration = course.get('duration', 'Self-paced') strategic_parts.append(f" {i}. **{course.get('course_name', 'Course Title')}** ({percentage}% match)") strategic_parts.append(f" Platform: {platform} | Duration: {duration} | Rating: {rating}") strategic_parts.append(f" Price: {price}") strategic_parts.append(f" Why it's excellent: {course.get('recommendation_reason', 'High-quality content')}") strategic_parts.append("") if len(online_courses) > 5: strategic_parts.append(f" ...and {len(online_courses) - 5} more online courses available!") strategic_parts.append("") strategic_parts.append("🎓 *The online advantage:*") strategic_parts.append(" • Learn from world-class instructors") strategic_parts.append(" • Flexible, self-paced learning") strategic_parts.append(" • Industry-recognized certifications") strategic_parts.append(" • Latest industry trends and practices") strategic_parts.append(" • Global community of learners") strategic_parts.append("") # Success Story Ending strategic_parts.append("🎯 **Your Success Formula**") strategic_parts.append("") if dept_skills: strategic_parts.append(f"**Phase 1 (Months 1-2):** Master your department skills - {', '.join([s['skill_name'] for s in dept_skills[:3]])}") if org_skills: phase_num = 2 if dept_skills else 1 strategic_parts.append(f"**Phase {phase_num} (Months {2 if dept_skills else 1}-{4 if dept_skills else 3}):** Leverage organizational skills - {', '.join([s['skill_name'] for s in org_skills[:3]])}") if ext_skills: phase_num = 3 if (dept_skills and org_skills) else (2 if (dept_skills or org_skills) else 1) months_start = 4 if (dept_skills and org_skills) else (3 if (dept_skills or org_skills) else 1) months_end = 6 if (dept_skills and org_skills) else (5 if (dept_skills or org_skills) else 3) strategic_parts.append(f"**Phase {phase_num} (Months {months_start}-{months_end}):** Gain competitive edge - {', '.join([s['skill_name'] for s in ext_skills[:3]])}") strategic_parts.append("") strategic_parts.append("📊 **Your Complete Development Package:**") strategic_parts.append(f"• Total Skills Identified: **{total_skills}**") strategic_parts.append(f"• Department Advantage: **{len(dept_skills)} skills**") strategic_parts.append(f"• Organizational Resources: **{len(org_skills)} skills**") strategic_parts.append(f"• External Growth Opportunities: **{len(ext_skills)} skills**") if top_offline_courses: strategic_parts.append(f"• Internal Courses Available: **{len(top_offline_courses)} courses**") if top_online_courses: strategic_parts.append(f"• Top Online Courses: **{len(top_online_courses)} courses**") strategic_parts.append("") strategic_parts.append("🌟 **The Bottom Line:**") total_courses_count = len(top_offline_courses) + len(top_online_courses) if total_courses_count > 0: strategic_parts.append(f"You're not just getting a skill list - you're getting a complete learning ecosystem! With {total_skills} targeted skills, {len(top_offline_courses)} internal courses, and {len(top_online_courses)} world-class online courses, you have everything needed to become a {identified_field.lower()} powerhouse!") else: strategic_parts.append(f"You're not just improving random skills - you're building a strategic capability stack that will make you a {identified_field.lower()} powerhouse!") strategic_parts.append("") strategic_parts.append("Ready to begin this exciting journey? Let's make it happen! 🚀") strategic_parts.append("") strategic_parts.append("---") strategic_parts.append("*Generated by TNA v2.0 - Your Personalized Learning Companion*") strategic_presentation = "\n".join(strategic_parts) # Format skill analysis for response skill_analysis_dict = { "recommended_skills": result.get('recommended_skills', []), "total_skills": total_skills, "identified_field": result.get('identified_field', ''), "case_applied": result.get('case_applied', '') } # Convert course matches to the expected format offline_courses_formatted = [] for course_match in top_offline_courses: offline_courses_formatted.append({ "course_id": course_match['course_id'], "name": course_match['course_name'], "reason": course_match['reason'], "percentage": float(course_match['match_percentage']) }) # Convert online courses to the expected format online_courses_formatted = [] for online_course in top_online_courses: online_courses_formatted.append({ "title": online_course.get('course_name', ''), "platform": online_course.get('platform', ''), "instructor": online_course.get('instructor', ''), "duration": online_course.get('duration', ''), "skill_level": online_course.get('skill_level', ''), "description": online_course.get('description', ''), "url": online_course.get('url', ''), "price": online_course.get('price', ''), "rating": online_course.get('rating', ''), "skills_covered": online_course.get('skills_covered', []), "relevance_score": online_course.get('match_percentage', 0), "reason": online_course.get('recommendation_reason', '') }) return SkillsCourseResponse( skill_analysis=skill_analysis_dict, courses_offline=offline_courses_formatted, online_recommended_courses=online_courses_formatted, strategic_presentation=strategic_presentation ) except Exception as e: logger.error(f"Enhanced skills analysis failed: {e}") return SkillsCourseResponse( skill_analysis={"recommended_skills": [], "total_skills": 0}, courses_offline=[], online_recommended_courses=[], strategic_presentation=f"Unable to complete enhanced analysis at this time. Error: {str(e)}" ) @app.post("/analyze-skills-courses", response_model=SkillsCourseResponse) async def analyze_skills_courses_endpoint(request: SkillsCourseRequest): """ Enhanced skill analysis using tna_skill_second.py Flow: 1. Field identification with specificity analysis 2. Smart skill filtering by domain relevance 3. Priority-based recommendations (dept → org → external) 4. Professional strategic presentation """ return await perform_enhanced_analysis( user_query=request.user_query, admin_name=request.admin_name, department_name=request.department_name, filtered_skills=request.filtered_skills, total_users_in_department=request.total_users_in_department, all_skills_data=request.all_skills_data, offline_courses=request.offline_courses ) @app.post("/analyze-full-tna", response_model=FullTnaResponse) async def analyze_full_tna_endpoint(request: FullTnaRequest): """ Combined endpoint performing Department Identification + Skills & Course Analysis """ try: logger.info(f"Full TNA analysis for query: {request.user_query[:100]}...") # 1. Identify Department # Convert departments to simple format for analysis simple_departments = [{"id": d.id, "name": d.name} for d in request.departments] # Run blocking department analysis in executor loop = asyncio.get_running_loop() dept_result = await loop.run_in_executor( None, analyze_department, request.user_query, simple_departments ) identified_dept_name = dept_result.get('identified_department', '') identified_dept_id = dept_result.get('department_id') confidence = dept_result.get('confidence', 'none') logger.info(f"Identified Department: {identified_dept_name} (ID: {identified_dept_id})") # 2. Retrieve Data for Identified Department target_department_data = None if identified_dept_id: for d in request.departments: if d.id == identified_dept_id: target_department_data = d break # Fallback if no specific department matched or found department_name_for_analysis = identified_dept_name if identified_dept_name else "General" filtered_skills = [] total_users = 0 if target_department_data: filtered_skills = target_department_data.filtered_skills total_users = target_department_data.total_users department_name_for_analysis = target_department_data.name # Ensure name matches exact data # 3. Perform Skills & Course Analysis skills_response = await perform_enhanced_analysis( user_query=request.user_query, admin_name=request.admin_name, department_name=department_name_for_analysis, filtered_skills=filtered_skills, total_users_in_department=total_users, all_skills_data=request.all_skills_data, offline_courses=request.offline_courses ) # 4. Merge Results return FullTnaResponse( skill_analysis=skills_response.skill_analysis, courses_offline=skills_response.courses_offline, online_recommended_courses=skills_response.online_recommended_courses, strategic_presentation=skills_response.strategic_presentation, identified_department=identified_dept_name, department_id=identified_dept_id, confidence=confidence ) except Exception as e: logger.error(f"Full TNA analysis failed: {e}") return FullTnaResponse( skill_analysis={"recommended_skills": [], "total_skills": 0}, courses_offline=[], online_recommended_courses=[], strategic_presentation=f"Unable to complete full analysis at this time. Error: {str(e)}", identified_department="", department_id=None, confidence="high" # Error state but returning valid model ) @app.post("/analyze-department", response_model=DepartmentResponse) async def analyze_department_endpoint(request: DepartmentRequest): """ Analyze user goal and identify relevant department """ try: logger.info(f"Analyzing: {request.user_prompt[:100]}...") # Log first 100 chars # Convert to simple format departments = [ {"id": dept.id, "name": dept.name} for dept in request.organization_departments ] # Get analysis (this is now just one simple function call!) loop = asyncio.get_running_loop() result = await loop.run_in_executor( None, analyze_department, request.user_prompt, departments ) logger.info(f"Result: {result.get('identified_department', 'None')} " f"(confidence: {result.get('confidence', 'none')})") # Return response return DepartmentResponse(**result) except Exception as e: logger.error(f"Analysis failed: {e}") # Return empty result instead of error (matching your PHP behavior) return DepartmentResponse( identified_department="", department_id=None, confidence="none" ) 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"] } ] }