from fastapi import FastAPI, HTTPException, Request from pydantic import BaseModel from typing import List, Dict, Optional, Any, Union import json import logging 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 @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 """ try: logger.info(f"Enhanced analysis for query: {request.user_query[:100]}...") # Convert filtered_skills to the format expected by tna_skill_second filtered_skills_formatted = [] if request.filtered_skills: for skill_group in request.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": request.total_users_in_department // 2 # Estimate }) # Convert all_skills_data format all_skills_formatted = [] for skill in request.all_skills_data: all_skills_formatted.append({ "skill_id": skill.skill_id, "skill_name": skill.skill_name }) # Convert offline courses format all_courses_formatted = {} if request.offline_courses: all_courses_formatted["courses"] = [] for course in request.offline_courses: all_courses_formatted["courses"].append({ "courseId": course.course_id, "name": course.course_name, "short_description": course.short_description, "description": course.description, "skills": course.skills }) # Prepare request data for tna_skill_second analysis_request = { "user_query": request.user_query, "admin_name": request.admin_name, "department_name": request.department_name, "filtered_skills": filtered_skills_formatted, "all_skills_data": all_skills_formatted, "all_courses_data": all_courses_formatted } # Call enhanced skill analysis system result = 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)}") # Filter offline courses using domain-aware course filter course_matches = [] if request.offline_courses: try: # Prepare course data for filtering courses_for_filtering = [] for course in request.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) courses_for_filtering.append({ "course_id": course.course_id, "course_name": course.course_name, "course_short_description": course.short_description, "course_description": course.description, "skill": skills_str }) # Create course filtering request course_filter_request = { "domain": result.get('identified_field', 'Technology'), "field_prompt": request.user_query, "courses": courses_for_filtering } # Filter courses course_filter_result = course_filter_system.process_json_request(json.dumps(course_filter_request)) if course_filter_result.get('success'): course_matches = course_filter_result.get('course_matches', []) logger.info(f"Course filtering completed - {len(course_matches)} matches found") else: logger.warning(f"Course filtering failed: {course_filter_result.get('error')}") except Exception as e: logger.error(f"Course filtering error: {e}") course_matches = [] # Get online course recommendations with caching online_courses = [] try: identified_field = result.get('identified_field', 'Technology') # Check cache first cached_online_courses = course_cache.get_cached_courses( domain=identified_field, field_prompt=request.user_query, course_type="online" ) if cached_online_courses: online_courses = cached_online_courses logger.info(f"Using cached online courses: {len(online_courses)} courses") else: # Generate new online course recommendations online_course_request = { "domain": identified_field, "field_prompt": request.user_query, "max_courses": 5 # Limit to 5 online courses for story format } online_course_result = online_course_system.process_json_request(json.dumps(online_course_request)) if online_course_result.get('success'): online_courses = online_course_result.get('course_recommendations', []) # Cache the results course_cache.cache_courses( domain=identified_field, field_prompt=request.user_query, courses=online_courses, course_type="online" ) logger.info(f"Generated and cached {len(online_courses)} online courses") else: logger.warning(f"Online course generation failed: {online_course_result.get('error')}") except Exception as e: logger.error(f"Online course recommendation error: {e}") online_courses = [] # 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 {request.admin_name}! 👋") strategic_parts.append("") strategic_parts.append(f'We heard you say: *"{request.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 request.department_name.lower() != "null": strategic_parts.append("🏢 **Chapter 1: Building on Your Department Foundation**") strategic_parts.append("") strategic_parts.append(f"Good news! Your {request.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 are the skills that differentiate you from your peers and make you invaluable in the job market!") strategic_parts.append("") # Course Recommendations Story if course_matches: 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("") # Sort courses by match percentage sorted_courses = sorted(course_matches, key=lambda x: x.get('match_percentage', 0), reverse=True) for i, course in enumerate(sorted_courses[:5], 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 if online_courses: final_chapter_num = len([x for x in [dept_skills, org_skills, ext_skills, course_matches] 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("") # Sort online courses by match percentage sorted_online_courses = sorted(online_courses, key=lambda x: x.get('match_percentage', 0), reverse=True) for i, course in enumerate(sorted_online_courses[:3], 1): # Show top 3 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) > 3: strategic_parts.append(f" ...and {len(online_courses) - 3} 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 course_matches: strategic_parts.append(f"• Internal Courses Available: **{len(course_matches)} courses**") if online_courses: strategic_parts.append(f"• Top Online Courses: **{len(online_courses)} courses**") strategic_parts.append("") strategic_parts.append("🌟 **The Bottom Line:**") total_courses = len(course_matches) + len(online_courses) if total_courses > 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(course_matches)} internal courses, and {len(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 course_matches: 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 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-department", response_model=DepartmentResponse) async def analyze_department_endpoint(request: DepartmentRequest): """ Analyze user goal and identify relevant department Simple flow: 1. Receive user prompt and departments 2. Ask LLM to match 3. Return the result """ 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!) result = 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"] } ] } if __name__ == "__main__": import uvicorn uvicorn.run(app, host="0.0.0.0", port=8000)