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")) # Spend accumulates across the re-arm ladder: fold the current Apify # run's charge into the row total only at a boundary (terminal or # re-arm), because after a re-arm swaps apify_run_id the old run's # cost is no longer reachable from this row. total_cost = (run.cost_usd or 0.0) + plugins.run_cost_usd(remote) 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, cost_usd=total_cost, ) return serialize_talent_run(run) run = await TalentRuns.mark_succeeded( self.session, run.id, profiles_found=found_so_far, cost_usd=total_cost ) 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, cost_usd=total_cost ) 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 account(self): """Apify account money for the Find Talent header: balance, spend this cycle, and the observed $/profile. Balance/spend degrade to None when Apify is unreachable (or the token is missing) so the screen still renders; $/profile comes from our own recorded runs either way.""" try: summary = plugins.account_summary(await plugins.get_limits()) except (httpx.HTTPError, plugins.ApifyError, RuntimeError): summary = plugins.account_summary({}) total_cost, total_profiles = await TalentRuns.cost_totals(self.session) return { **summary, "total_cost_usd": total_cost, "total_profiles_found": total_profiles, "cost_per_profile_usd": ( total_cost / total_profiles if total_profiles else None ), } 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}