384 lines
15 KiB
Python
384 lines
15 KiB
Python
"""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
|