"""Talent sourcing tables: Apify actor runs and the LinkedIn profiles they find. `talent_runs` is one row per paid actor run (vendor-id trio mirrors the Buffer columns on job_posts). `talent_profiles` is deduped per job by normalized LinkedIn URL across re-runs; `raw` keeps the full dataset item verbatim because actor output fields vary between actors and versions. """ import uuid from datetime import datetime, timezone from sqlalchemy import DateTime, JSON, UniqueConstraint, func from sqlalchemy.ext.asyncio import AsyncSession from sqlmodel import Field, SQLModel, select from talent.enums import ALLOWED_OUTREACH_TRANSITIONS, OutreachStatus def _now() -> datetime: return datetime.now(timezone.utc) # Local run lifecycle. `pending` exists only between row insert and the Apify # start call succeeding; everything after start is driven by Apify's status. TERMINAL_RUN_STATUSES = ("succeeded", "failed", "timed_out", "aborted") # Profile fields refreshed when a later run re-finds the same person. Kept at # module level: an underscore-prefixed class attribute on a SQLModel becomes a # Pydantic ModelPrivateAttr, which is not iterable. `is_deleted` and the # outreach_* columns are deliberately absent — a dismissed profile stays # dismissed, and a re-run must not reset a recruiter's shortlist/contact state. MUTABLE_PROFILE_FIELDS = ( "public_id", "full_name", "headline", "location", "current_title", "current_company", "avatar_url", "summary", "skills", "match_score", "raw", ) class TalentRuns(SQLModel, table=True): __tablename__ = "talent_runs" id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) job_post_id: uuid.UUID = Field(index=True, foreign_key="job_posts.id") requested_by: uuid.UUID | None = Field(default=None, foreign_key="users.id") status: str = Field(default="pending") actor_id: str = Field(default="") search_input: dict = Field(default_factory=dict, sa_type=JSON) max_results: int = Field(default=0) apify_run_id: str | None = Field(default=None) apify_dataset_id: str | None = Field(default=None) apify_error: str | None = Field(default=None) profiles_found: int = Field(default=0) # Accumulated Apify spend (usageTotalUsd) across the row's whole re-arm # ladder. server_default is load-bearing: the column arrives as an ALTER # on a populated table (018_talent_run_cost.sql). cost_usd: float = Field(default=0, sa_column_kwargs={"server_default": "0"}) started_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) finished_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) is_deleted: bool = Field(default=False) @staticmethod def _as_uuid(record_id) -> uuid.UUID | None: if record_id in (None, ""): return None try: return uuid.UUID(str(record_id)) except ValueError: return None @classmethod async def get_by_id(cls, session: AsyncSession, record_id): uid = cls._as_uuid(record_id) if uid is None: return None statement = select(cls).where(cls.id == uid, cls.is_deleted == False) # noqa: E712 result = await session.execute(statement) return result.scalars().first() @classmethod async def fetch_runs(cls, session: AsyncSession, *, job_post_id): jid = cls._as_uuid(job_post_id) if jid is None: return [], 0 statement = select(cls).where( cls.job_post_id == jid, cls.is_deleted == False # noqa: E712 ) count_statement = select(func.count()).select_from(statement.subquery()) total = (await session.execute(count_statement)).scalar_one() statement = statement.order_by(cls.created_at.desc()) result = await session.execute(statement) return list(result.scalars().all()), total @classmethod async def latest_active_run(cls, session: AsyncSession, job_post_id): jid = cls._as_uuid(job_post_id) if jid is None: return None statement = ( select(cls) .where( cls.job_post_id == jid, cls.is_deleted == False, # noqa: E712 cls.status.not_in(TERMINAL_RUN_STATUSES), ) .order_by(cls.created_at.desc()) ) result = await session.execute(statement) return result.scalars().first() @classmethod async def insert_run(cls, session: AsyncSession, fields: dict): row = cls(**fields) session.add(row) await session.commit() return await cls.get_by_id(session, row.id) @classmethod async def _update(cls, session: AsyncSession, record_id, fields: dict): row = await cls.get_by_id(session, record_id) if not row: return None for key, value in fields.items(): setattr(row, key, value) row.updated_at = _now() session.add(row) await session.commit() await session.refresh(row) return row @classmethod async def mark_started(cls, session: AsyncSession, record_id, *, apify_run_id, apify_dataset_id): return await cls._update(session, record_id, { "status": "running", "apify_run_id": apify_run_id, "apify_dataset_id": apify_dataset_id, "started_at": _now(), }) @classmethod async def mark_rearmed( cls, session: AsyncSession, record_id, *, apify_run_id, apify_dataset_id, search_input: dict, found_so_far: int, cost_usd: float | None = None, ): """Point the SAME run row at a broadened follow-up actor run. Status stays "running" so the frontend keeps polling and the 409 active-run guard keeps holding; profiles_found and cost_usd accumulate across the ladder's batches (the caller folds the finished rung's usageTotalUsd in BEFORE apify_run_id swaps — after the swap the old run's cost is no longer reachable from this row). """ fields: dict = { "status": "running", "apify_run_id": apify_run_id, "apify_dataset_id": apify_dataset_id, "search_input": search_input, "profiles_found": found_so_far, } if cost_usd is not None: fields["cost_usd"] = cost_usd return await cls._update(session, record_id, fields) @classmethod async def mark_status(cls, session: AsyncSession, record_id, status: str): fields: dict = {"status": status} if status in TERMINAL_RUN_STATUSES: fields["finished_at"] = _now() return await cls._update(session, record_id, fields) @classmethod async def mark_failed( cls, session: AsyncSession, record_id, error: str, *, status: str = "failed", cost_usd: float | None = None, ): fields: dict = { "status": status, "apify_error": (error or "")[:2000], "finished_at": _now(), } if cost_usd is not None: fields["cost_usd"] = cost_usd return await cls._update(session, record_id, fields) @classmethod async def mark_succeeded( cls, session: AsyncSession, record_id, *, profiles_found: int, cost_usd: float | None = None, ): fields: dict = { "status": "succeeded", "profiles_found": profiles_found, "apify_error": None, "finished_at": _now(), } if cost_usd is not None: fields["cost_usd"] = cost_usd return await cls._update(session, record_id, fields) @classmethod async def cost_totals(cls, session: AsyncSession) -> tuple[float, int]: """(total cost, total profiles) over succeeded runs with recorded cost. Rows from before 018_talent_run_cost.sql carry cost_usd = 0 and are excluded so they cannot drag the $/profile average toward zero. """ statement = select( func.coalesce(func.sum(cls.cost_usd), 0.0), func.coalesce(func.sum(cls.profiles_found), 0), ).where( cls.status == "succeeded", cls.is_deleted == False, # noqa: E712 cls.cost_usd > 0, ) total_cost, total_profiles = (await session.execute(statement)).one() return float(total_cost or 0.0), int(total_profiles or 0) class TalentProfiles(SQLModel, table=True): __tablename__ = "talent_profiles" __table_args__ = ( UniqueConstraint("job_post_id", "linkedin_url", name="uq_talent_profiles_job_url"), ) id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True) job_post_id: uuid.UUID = Field(index=True, foreign_key="job_posts.id") run_id: uuid.UUID = Field(foreign_key="talent_runs.id") last_run_id: uuid.UUID | None = Field(default=None) linkedin_url: str public_id: str | None = Field(default=None) full_name: str | None = Field(default=None) headline: str | None = Field(default=None) location: str | None = Field(default=None) current_title: str | None = Field(default=None) current_company: str | None = Field(default=None) avatar_url: str | None = Field(default=None) summary: str | None = Field(default=None) skills: list = Field(default_factory=list, sa_type=JSON) match_score: int | None = Field(default=None) raw: dict = Field(default_factory=dict, sa_type=JSON) # Manual outreach funnel (OutreachStatus): sourced -> shortlisted -> # contacted. server_default is load-bearing: the column arrives as an ALTER # on a populated table (017_talent_outreach.sql). outreach_status: str = Field(default="sourced", sa_column_kwargs={"server_default": "sourced"}) shortlisted_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) shortlisted_by: uuid.UUID | None = Field(default=None, foreign_key="users.id") contacted_at: datetime | None = Field(default=None, sa_type=DateTime(timezone=True)) contacted_by: uuid.UUID | None = Field(default=None, foreign_key="users.id") first_seen_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) last_seen_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) created_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) updated_at: datetime = Field(default_factory=_now, sa_type=DateTime(timezone=True)) is_deleted: bool = Field(default=False) @classmethod async def fetch_profiles( cls, session: AsyncSession, *, job_post_id, search=None, top=None, skip=0 ): jid = TalentRuns._as_uuid(job_post_id) if jid is None: return [], 0 statement = select(cls).where( cls.job_post_id == jid, cls.is_deleted == False # noqa: E712 ) if search: pattern = f"%{search}%" statement = statement.where( cls.full_name.ilike(pattern) | cls.headline.ilike(pattern) | cls.current_company.ilike(pattern) ) count_statement = select(func.count()).select_from(statement.subquery()) total = (await session.execute(count_statement)).scalar_one() statement = statement.order_by( cls.match_score.desc().nulls_last(), cls.last_seen_at.desc(), cls.created_at.desc(), ) if skip: statement = statement.offset(skip) if top: statement = statement.limit(top) result = await session.execute(statement) return list(result.scalars().all()), total @classmethod async def upsert_from_items( cls, session: AsyncSession, *, job_post_id, run_id, normalized_items: list[dict] ) -> int: """Insert new profiles, refresh re-found ones. One commit for the batch. Dedupe key is (job_post_id, linkedin_url); dismissed rows are refreshed too but keep is_deleted=True so a re-run cannot resurrect them. """ jid = TalentRuns._as_uuid(job_post_id) rid = TalentRuns._as_uuid(run_id) persisted = 0 for item in normalized_items: url = item.get("linkedin_url") if not url: continue statement = select(cls).where( cls.job_post_id == jid, cls.linkedin_url == url ) existing = (await session.execute(statement)).scalars().first() if existing: for key in MUTABLE_PROFILE_FIELDS: if item.get(key) is not None: setattr(existing, key, item[key]) existing.last_run_id = rid existing.last_seen_at = _now() existing.updated_at = _now() session.add(existing) else: session.add(cls( job_post_id=jid, run_id=rid, last_run_id=rid, linkedin_url=url, **{key: item.get(key) for key in MUTABLE_PROFILE_FIELDS}, )) persisted += 1 await session.commit() return persisted @classmethod async def get_profile_by_id(cls, session: AsyncSession, record_id): uid = TalentRuns._as_uuid(record_id) if uid is None: return None statement = select(cls).where(cls.id == uid, cls.is_deleted == False) # noqa: E712 return (await session.execute(statement)).scalars().first() @classmethod async def set_outreach_status(cls, session: AsyncSession, record_id, status: str, *, actor=None): """Idempotent outreach-funnel writer. Raises ValueError on a disallowed transition; the stamp pair for a stage is set on entry and cleared on undo, so shortlist stamps survive contacted and its undo. """ row = await cls.get_profile_by_id(session, record_id) if not row: return None previous = row.outreach_status if previous == status: return row if status not in ALLOWED_OUTREACH_TRANSITIONS.get(previous, set()): raise ValueError(f"cannot move a {previous} profile to {status}") actor_id = TalentRuns._as_uuid(actor) if status == OutreachStatus.SHORTLISTED.value: if previous == OutreachStatus.CONTACTED.value: row.contacted_at = None row.contacted_by = None else: row.shortlisted_at = _now() row.shortlisted_by = actor_id elif status == OutreachStatus.CONTACTED.value: row.contacted_at = _now() row.contacted_by = actor_id else: # back to sourced row.shortlisted_at = None row.shortlisted_by = None row.outreach_status = status row.updated_at = _now() session.add(row) await session.commit() return row @classmethod async def soft_delete_profile(cls, session: AsyncSession, record_id): row = await cls.get_profile_by_id(session, record_id) if not row: return None row.is_deleted = True row.updated_at = _now() session.add(row) await session.commit() return row import job.job_post.models as _job_post_models # noqa: E402, F401 import users.models as _users_models # noqa: E402, F401