File size: 5,585 Bytes
4fd18a2 |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 |
"""Job search tool for finding and ranking relevant jobs - jobs.search endpoint."""
from typing import Dict, Any
from ..services import JobSearchService
class JobSearchTool:
"""Tool for searching and ranking job opportunities."""
def __init__(self):
self.job_search_service = JobSearchService()
def search(
self, user_id: str, query: str = "", location: str = "", job_type: str = ""
) -> Dict[str, Any]:
"""
Search for jobs and rank them by relevance to user profile.
This is the main jobs.search endpoint that pulls fresh job posts,
ranks them with GPU embeddings, and returns fit scores.
Args:
user_id: User identifier to get personalized results based on profile
query: Job search query/keywords (e.g., "Python developer", "Data scientist")
location: Preferred job location (e.g., "Remote", "New York", "San Francisco")
job_type: Type of job (e.g., "full-time", "part-time", "contract", "remote", "freelance")
Returns:
Dict with ranked job listings and fit scores:
{
"success": True,
"jobs": [
{
"id": "job_123",
"title": "Senior Python Developer",
"company": "TechCorp",
"location": "Remote",
"description": "Job description...",
"requirements": "Skills required...",
"salary": "$80,000 - $120,000",
"url": "https://company.com/jobs/123",
"posted_date": "2024-01-15T10:30:00",
"job_type": "full-time",
"fit_score": 85, # Percentage match (0-100)
"match_reasons": ["Skills match: Python, Django", "Location preference match"]
}
],
"total_found": 10,
"search_params": {
"query": "Python developer",
"location": "Remote",
"job_type": "full-time"
},
"user_profile": {
"skills_count": 15,
"location": "Remote"
}
}
"""
return self.job_search_service.search_jobs(user_id, query, location, job_type)
def get_job_details(self, job_id: str) -> Dict[str, Any]:
"""
Get detailed information about a specific job.
Args:
job_id: Unique job identifier
Returns:
Dict with detailed job information or error message
"""
# This would typically fetch from job cache or re-scrape
# For now, return a placeholder implementation
return {
"success": False,
"message": "Job details retrieval not implemented yet",
}
def get_search_suggestions(self, user_id: str) -> Dict[str, Any]:
"""
Get personalized job search suggestions based on user profile.
Args:
user_id: User identifier
Returns:
Dict with suggested search queries and parameters
"""
try:
from ..services import ProfileService
profile_service = ProfileService()
profile = profile_service.get_profile(user_id)
if not profile:
return {"success": False, "message": "User profile not found"}
# Generate suggestions based on skills and career goals
suggestions = []
# Skill-based suggestions
top_skills = profile.skills[:5] # Top 5 skills
for skill in top_skills:
suggestions.append(
{
"query": f"{skill} developer",
"reason": f"Based on your {skill} skills",
}
)
# Career goal-based suggestions
if "senior" in profile.career_goals.lower():
suggestions.append(
{
"query": "senior developer",
"reason": "Based on your career goals",
}
)
if "remote" in profile.career_goals.lower():
suggestions.append(
{
"location": "Remote",
"reason": "Based on your location preferences",
}
)
return {
"success": True,
"suggestions": suggestions[:10], # Limit to 10 suggestions
"profile_skills": profile.skills[:10],
}
except Exception as e:
return {"success": False, "message": f"Error getting suggestions: {str(e)}"}
def clear_job_cache(self) -> Dict[str, Any]:
"""
Clear the job search cache to force fresh results.
Returns:
Dict with operation result
"""
try:
self.job_search_service.jobs_cache = {}
self.job_search_service._save_cache()
# Also clear embedding index
self.job_search_service.embedding_service.clear_index()
return {
"success": True,
"message": "Job cache and embeddings cleared successfully",
}
except Exception as e:
return {"success": False, "message": f"Error clearing cache: {str(e)}"}
|