Spaces:
Runtime error
Runtime error
File size: 3,214 Bytes
680fa2b | 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 | from typing import List, Optional
from fastapi import APIRouter, Depends, Query
from sqlalchemy.orm import Session
from backend.app.core.database import get_db
from backend.app.repositories.users import UserRepository
from backend.app.schemas.auth import WorkerSearchResult
router = APIRouter(prefix="/workers", tags=["workers"])
def calculate_ai_score(
rating: float,
completion_rate: int,
distance: float,
online: bool,
verified: bool
) -> int:
"""
Simulated AI ranking score calculation.
Formula: rating * 7 + completion_rate * 0.25 + online (25 pts) + verified (15 pts) + distance (max 25 - distance * 3)
"""
online_score = 25 if online else 0
verified_score = 15 if verified else 0
distance_score = max(0.0, 25.0 - float(distance) * 3.0)
score = (float(rating) * 7.0) + (int(completion_rate) * 0.25) + online_score + verified_score + distance_score
return round(score)
@router.get("", response_model=List[WorkerSearchResult])
def search_workers(
skill: Optional[str] = Query("All", description="Filter by worker skill type"),
radius: Optional[float] = Query(5.0, description="Max search radius in kilometers"),
online: Optional[str] = Query("yes", description="Filter: 'yes' for online only, 'all' for all"),
q: Optional[str] = Query("", alias="query", description="Search by name or skill query text"),
db: Session = Depends(get_db)
):
"""
Search and rank construction workers using the location-aware AI ranking algorithm.
"""
user_repo = UserRepository(db)
# Resolve online boolean filter
online_only = (online == "yes")
# Query database workers matching filters
workers = user_repo.get_all_workers(
skill=skill,
radius=radius,
online_only=online_only
)
results = []
text_filter = q.strip().lower()
for user in workers:
prof = user.worker_profile
if not prof:
continue
# Text query filtering (by name or skill)
if text_filter and text_filter not in f"{user.name} {prof.skill}".lower():
continue
# Compute dynamic match score
score = calculate_ai_score(
rating=prof.rating,
completion_rate=prof.completion_rate,
distance=prof.distance,
online=prof.online,
verified=prof.verified
)
results.append(
WorkerSearchResult(
id=user.id,
name=user.name,
phone=user.phone,
city=user.city,
skill=prof.skill,
rate=prof.rate,
rating=prof.rating,
distance=prof.distance,
online=prof.online,
verified=prof.verified,
completed_jobs=prof.completed_jobs,
completion_rate=prof.completion_rate,
map_x=prof.map_x,
map_y=prof.map_y,
ai_score=score
)
)
# Sort results in descending order of AI Match Score
results.sort(key=lambda w: w.ai_score, reverse=True)
return results
|