257 lines
11 KiB
Python
257 lines
11 KiB
Python
import httpx
|
|
from fastapi import HTTPException
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from job.job_post.models import JobPosts
|
|
from talent import plugins
|
|
from talent.enums import OutreachStatus
|
|
from talent.matching import annotate_applications
|
|
from talent.models import TalentProfiles, TalentRuns
|
|
from talent.serializers import (
|
|
serialize_talent_profile,
|
|
serialize_talent_profile_detail,
|
|
serialize_talent_run,
|
|
)
|
|
from users.models import Users
|
|
|
|
|
|
def _search_basis(actor_input: dict) -> dict:
|
|
"""The identity of a search, ignoring pagination and batch-size knobs."""
|
|
return {k: v for k, v in (actor_input or {}).items() if k not in ("startPage", "maxItems")}
|
|
|
|
|
|
class Talent:
|
|
def __init__(self, session: AsyncSession):
|
|
self.session = session
|
|
|
|
async def _get_job(self, job_post_id):
|
|
job = await JobPosts.get_job_post_by_id(self.session, job_post_id)
|
|
if not job or job.is_deleted:
|
|
raise HTTPException(status_code=404, detail="Job post not found")
|
|
return job
|
|
|
|
async def start_run(self, job_post_id, payload, current_user):
|
|
job = await self._get_job(job_post_id)
|
|
|
|
active = await TalentRuns.latest_active_run(self.session, job_post_id)
|
|
if active:
|
|
raise HTTPException(
|
|
status_code=409,
|
|
detail="A talent search is already running for this job",
|
|
)
|
|
|
|
requested = payload.get("max_results")
|
|
max_results = min(int(requested), plugins.APIFY_MAX_RESULTS) if requested else plugins.APIFY_MAX_RESULTS
|
|
if max_results < 1:
|
|
raise HTTPException(status_code=422, detail="max_results must be at least 1")
|
|
# Floor of 10 per paid run (user asked for at least 10 results a
|
|
# search) — unless the env cap itself is set lower.
|
|
max_results = max(max_results, min(10, plugins.APIFY_MAX_RESULTS))
|
|
|
|
overrides = {
|
|
"keywords": payload.get("keywords"),
|
|
"location": payload.get("location"),
|
|
}
|
|
job_fields = {
|
|
"title": job.title,
|
|
"requirements": job.requirements,
|
|
"optional_skills": job.optional_skills,
|
|
"location": job.location,
|
|
"experience_min": job.experience_min,
|
|
"experience_max": job.experience_max,
|
|
}
|
|
actor_input = plugins.build_actor_input(
|
|
job_fields, max_results=max_results, overrides=overrides
|
|
)
|
|
|
|
# Re-running the same search continues deeper into LinkedIn's result
|
|
# pages (25 profiles each), so every run surfaces new people. A changed
|
|
# query/location/experience is a different search and restarts at page 1.
|
|
basis = _search_basis(actor_input)
|
|
prior_runs, _ = await TalentRuns.fetch_runs(self.session, job_post_id=job_post_id)
|
|
prior_pages = [
|
|
int((r.search_input or {}).get("startPage") or 1)
|
|
for r in prior_runs
|
|
if r.status == "succeeded" and _search_basis(r.search_input) == basis
|
|
]
|
|
if prior_pages:
|
|
actor_input = plugins.build_actor_input(
|
|
job_fields,
|
|
max_results=max_results,
|
|
overrides=overrides,
|
|
start_page=max(prior_pages) + 1,
|
|
)
|
|
run = await TalentRuns.insert_run(self.session, {
|
|
"job_post_id": job.id,
|
|
"requested_by": TalentRuns._as_uuid((current_user or {}).get("id")),
|
|
"status": "pending",
|
|
"actor_id": plugins.APIFY_ACTOR_ID,
|
|
"search_input": actor_input,
|
|
"max_results": max_results,
|
|
})
|
|
|
|
try:
|
|
started = await plugins.start_actor_run(actor_input)
|
|
except (httpx.HTTPError, plugins.ApifyError, RuntimeError) as exc:
|
|
# Keep the failed row for run history, then surface the vendor error.
|
|
await TalentRuns.mark_failed(self.session, run.id, str(exc))
|
|
raise HTTPException(status_code=502, detail=f"Apify run could not be started: {exc}")
|
|
|
|
run = await TalentRuns.mark_started(
|
|
self.session,
|
|
run.id,
|
|
apify_run_id=started.get("id"),
|
|
apify_dataset_id=started.get("defaultDatasetId"),
|
|
)
|
|
return serialize_talent_run(run)
|
|
|
|
async def run_status(self, run_id):
|
|
run = await TalentRuns.get_by_id(self.session, run_id)
|
|
if not run:
|
|
raise HTTPException(status_code=404, detail="Talent run not found")
|
|
|
|
# Terminal runs are immutable: no Apify call, no re-persist. This makes
|
|
# the poll endpoint idempotent and cheap once a run has settled.
|
|
if run.status in plugins.TERMINAL_STATUSES:
|
|
return serialize_talent_run(run)
|
|
|
|
if not run.apify_run_id:
|
|
# pending row whose start call never completed (crash between insert
|
|
# and mark_started): nothing to poll, mark it failed.
|
|
run = await TalentRuns.mark_failed(
|
|
self.session, run.id, "Run was never started on Apify"
|
|
)
|
|
return serialize_talent_run(run)
|
|
|
|
try:
|
|
remote = await plugins.get_run(run.apify_run_id)
|
|
except (httpx.HTTPError, plugins.ApifyError, RuntimeError) as exc:
|
|
if isinstance(exc, plugins.ApifyError) and exc.code == "record-not-found":
|
|
run = await TalentRuns.mark_failed(
|
|
self.session, run.id, "Apify run no longer exists"
|
|
)
|
|
return serialize_talent_run(run)
|
|
raise HTTPException(status_code=502, detail=f"Apify status check failed: {exc}")
|
|
|
|
status = plugins.local_status(remote.get("status"))
|
|
if status == "running":
|
|
run = await TalentRuns.mark_status(self.session, run.id, "running")
|
|
return serialize_talent_run(run)
|
|
|
|
if status == "succeeded":
|
|
dataset_id = run.apify_dataset_id or remote.get("defaultDatasetId")
|
|
try:
|
|
items = await plugins.get_dataset_items(dataset_id, limit=run.max_results)
|
|
except (httpx.HTTPError, plugins.ApifyError, RuntimeError) as exc:
|
|
raise HTTPException(status_code=502, detail=f"Apify dataset fetch failed: {exc}")
|
|
normalized = [
|
|
p
|
|
for p in (plugins.normalize_profile(i) for i in items)
|
|
if p and not plugins.is_excluded_profile(p)
|
|
]
|
|
job = await JobPosts.get_job_post_by_id(self.session, run.job_post_id)
|
|
if job:
|
|
job_fields = {
|
|
"title": job.title,
|
|
"requirements": job.requirements,
|
|
"optional_skills": job.optional_skills,
|
|
}
|
|
for profile in normalized:
|
|
profile["match_score"] = plugins.relevance_score(job_fields, profile)
|
|
count = await TalentProfiles.upsert_from_items(
|
|
self.session,
|
|
job_post_id=run.job_post_id,
|
|
run_id=run.id,
|
|
normalized_items=normalized,
|
|
)
|
|
found_so_far = (run.profiles_found or 0) + count
|
|
|
|
# Thin results: broaden and keep the same run going instead of
|
|
# settling for one lonely card. Each rung is a fresh actor run on
|
|
# the same row; the frontend just sees "running" a while longer.
|
|
if len(items) < min(10, run.max_results):
|
|
broadened = plugins.broaden_actor_input(run.search_input or {})
|
|
if broadened:
|
|
try:
|
|
started = await plugins.start_actor_run(broadened)
|
|
except (httpx.HTTPError, plugins.ApifyError, RuntimeError):
|
|
# Keep what we already found rather than failing the run.
|
|
started = None
|
|
if started:
|
|
run = await TalentRuns.mark_rearmed(
|
|
self.session,
|
|
run.id,
|
|
apify_run_id=started.get("id"),
|
|
apify_dataset_id=started.get("defaultDatasetId"),
|
|
search_input=broadened,
|
|
found_so_far=found_so_far,
|
|
)
|
|
return serialize_talent_run(run)
|
|
|
|
run = await TalentRuns.mark_succeeded(
|
|
self.session, run.id, profiles_found=found_so_far
|
|
)
|
|
return serialize_talent_run(run)
|
|
|
|
# failed / timed_out / aborted
|
|
message = remote.get("statusMessage") or f"Apify run {remote.get('status')}"
|
|
run = await TalentRuns.mark_failed(self.session, run.id, message, status=status)
|
|
return serialize_talent_run(run)
|
|
|
|
async def fetch_runs(self, job_post_id):
|
|
await self._get_job(job_post_id)
|
|
rows, total = await TalentRuns.fetch_runs(self.session, job_post_id=job_post_id)
|
|
return [serialize_talent_run(r) for r in rows], total
|
|
|
|
async def _outreach_actor_names(self, rows) -> dict:
|
|
return await Users.names_by_ids(
|
|
self.session,
|
|
[r.shortlisted_by for r in rows] + [r.contacted_by for r in rows],
|
|
)
|
|
|
|
async def fetch_profiles(self, job_post_id, search=None, top=None, skip=0):
|
|
await self._get_job(job_post_id)
|
|
rows, total = await TalentProfiles.fetch_profiles(
|
|
self.session, job_post_id=job_post_id, search=search, top=top, skip=skip
|
|
)
|
|
names = await self._outreach_actor_names(rows)
|
|
profiles = [serialize_talent_profile(r, user_names=names) for r in rows]
|
|
profiles = await annotate_applications(self.session, profiles)
|
|
return profiles, total
|
|
|
|
async def get_profile(self, profile_id):
|
|
row = await TalentProfiles.get_profile_by_id(self.session, profile_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Talent profile not found")
|
|
names = await self._outreach_actor_names([row])
|
|
data = serialize_talent_profile_detail(row, user_names=names)
|
|
await annotate_applications(self.session, [data])
|
|
return data
|
|
|
|
async def set_outreach_status(self, profile_id, payload, current_user):
|
|
parsed = OutreachStatus.parse(payload.get("outreach_status") or "")
|
|
if parsed is None:
|
|
raise HTTPException(
|
|
status_code=422,
|
|
detail=f"outreach_status must be one of {', '.join(OutreachStatus.values())}",
|
|
)
|
|
try:
|
|
row = await TalentProfiles.set_outreach_status(
|
|
self.session,
|
|
profile_id,
|
|
parsed.value,
|
|
actor=(current_user or {}).get("id"),
|
|
)
|
|
except ValueError as exc:
|
|
raise HTTPException(status_code=409, detail=str(exc))
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Talent profile not found")
|
|
names = await self._outreach_actor_names([row])
|
|
return serialize_talent_profile(row, user_names=names)
|
|
|
|
async def delete_profile(self, profile_id):
|
|
row = await TalentProfiles.soft_delete_profile(self.session, profile_id)
|
|
if not row:
|
|
raise HTTPException(status_code=404, detail="Talent profile not found")
|
|
return {"id": str(row.id), "deleted": True}
|