""" Preference learning - learns from user behavior to improve recommendations over time. Tracks which jobs users save/apply to vs skip/hide to build a preference model. """ import uuid from datetime import datetime, timedelta, timezone from fastapi import APIRouter from sqlalchemy import func, select from app.api.deps import CurrentUser, DBSession from app.models.application import Application from app.models.job import Job, SavedJob router = APIRouter(prefix="/preferences/learned", tags=["preference-learning"]) @router.get("") async def get_learned_preferences(db: DBSession, user: CurrentUser): """ Analyze user's saved/applied jobs to infer preferences. Returns learned preferences that can be used to tune recommendations. """ # Get all jobs user has positively interacted with saved_result = await db.execute( select(Job) .join(SavedJob, SavedJob.job_id == Job.id) .where(SavedJob.user_id == user.id) ) saved_jobs = saved_result.scalars().all() applied_result = await db.execute( select(Job) .join(Application, Application.job_id == Job.id) .where(Application.user_id == user.id, Application.job_id.isnot(None)) ) applied_jobs = applied_result.scalars().all() all_positive_jobs = list(set(saved_jobs + applied_jobs)) if not all_positive_jobs: return { "status": "insufficient_data", "message": "Save or apply to more jobs to learn your preferences", "jobs_analyzed": 0, } # Analyze patterns companies = {} remote_types = {} seniority_levels = {} salary_ranges = [] skills_seen = {} employment_types = {} for job in all_positive_jobs: # Companies companies[job.company_name] = companies.get(job.company_name, 0) + 1 # Remote if job.remote_type: remote_types[job.remote_type] = remote_types.get(job.remote_type, 0) + 1 # Seniority if job.seniority_level: seniority_levels[job.seniority_level] = seniority_levels.get(job.seniority_level, 0) + 1 # Salary if job.salary_min and job.salary_max: salary_ranges.append({"min": job.salary_min, "max": job.salary_max}) # Skills for skill_list in [job.skills_required, job.skills_preferred]: if skill_list: for skill in skill_list: skills_seen[skill.lower()] = skills_seen.get(skill.lower(), 0) + 1 # Employment type if job.employment_type: employment_types[job.employment_type] = employment_types.get(job.employment_type, 0) + 1 # Compute preferences total = len(all_positive_jobs) preferred_remote = max(remote_types, key=remote_types.get) if remote_types else None preferred_seniority = max(seniority_levels, key=seniority_levels.get) if seniority_levels else None preferred_type = max(employment_types, key=employment_types.get) if employment_types else None salary_min_avg = int(sum(s["min"] for s in salary_ranges) / len(salary_ranges)) if salary_ranges else None salary_max_avg = int(sum(s["max"] for s in salary_ranges) / len(salary_ranges)) if salary_ranges else None top_skills = sorted(skills_seen.items(), key=lambda x: x[1], reverse=True)[:15] top_companies = sorted(companies.items(), key=lambda x: x[1], reverse=True)[:10] return { "status": "ready", "jobs_analyzed": total, "preferences": { "remote_type": { "preferred": preferred_remote, "distribution": remote_types, }, "seniority": { "preferred": preferred_seniority, "distribution": seniority_levels, }, "employment_type": { "preferred": preferred_type, "distribution": employment_types, }, "salary_range": { "average_min": salary_min_avg, "average_max": salary_max_avg, "data_points": len(salary_ranges), }, "top_skills": [{"skill": s[0], "frequency": s[1]} for s in top_skills], "preferred_companies": [{"company": c[0], "interactions": c[1]} for c in top_companies], }, "confidence": "high" if total >= 10 else "medium" if total >= 5 else "low", }