perf: optimize import persistence for 0.3.14

apps/jobs/services/pipeline.py

@@ -1,12 +1,21 @@
 from __future__ import annotations
 
+from dataclasses import dataclass
+from dataclasses import field as dataclass_field
 from decimal import Decimal
 from urllib.parse import urlsplit
 
 from django.db import IntegrityError, transaction
 from django.utils import timezone
 
-from apps.jobs.models import Employer, FieldProvenance, JobPosting, JobSourceAlias, JobVersion
+from apps.jobs.models import (
+    Employer,
+    FieldProvenance,
+    JobPosting,
+    JobSourceAlias,
+    JobVersion,
+    ScoreRun,
+)
 from apps.profiles.models import SearchProfile
 from apps.sources.adapters.base import FieldEvidence
 from apps.sources.adapters.registry import registry
@@ -29,17 +38,31 @@ RECRUITER_TERMS = {
 }
 
 
+@dataclass
+class PersistenceContext:
+    """Run-scoped reference cache; never shared between workers or imports."""
+
+    employers: dict[tuple[str, str], Employer | None] = dataclass_field(default_factory=dict)
+    active_profiles: list[SearchProfile] | None = None
+    latest_scores: dict[tuple[object, int], ScoreRun] = dataclass_field(default_factory=dict)
+
+
 def _confidence(value: float) -> Decimal:
     return Decimal(str(max(0.0, min(1.0, value))))
 
 
-def resolve_employer(draft: CanonicalJobDraft) -> Employer | None:
+def resolve_employer(
+    draft: CanonicalJobDraft, *, context: PersistenceContext | None = None
+) -> Employer | None:
     name = draft.employer_name.strip()
     domain = draft.employer_domain.strip()
     if not name and not domain:
         return None
     display_name = name or domain
     normalized = normalize_token(display_name)
+    cache_key = (normalized, domain)
+    if context is not None and cache_key in context.employers:
+        return context.employers[cache_key]
     recruiter = any(term in normalized for term in RECRUITER_TERMS)
     employer, _ = Employer.objects.get_or_create(
         normalized_name=normalized,
@@ -61,6 +84,8 @@ def resolve_employer(draft: CanonicalJobDraft) -> Employer | None:
         changed.extend(["is_recruiter", "is_direct_employer"])
     if changed:
         employer.save(update_fields=[*changed, "updated_at"])
+    if context is not None:
+        context.employers[cache_key] = employer
     return employer
 
 
@@ -162,12 +187,81 @@ def _apply_draft(
     if direct and canonical_url and job.canonical_url != canonical_url:
         job.canonical_url = canonical_url
         changed.append("canonical_url")
-    if changed and "last_changed" not in changed:
+    substantive_changes = [field for field in changed if field != "last_seen"]
+    if substantive_changes and "last_changed" not in changed:
         job.last_changed = timezone.now()
         changed.append("last_changed")
     return changed
 
 
+def _sync_evidence(
+    *,
+    job: JobPosting,
+    alias: JobSourceAlias,
+    evidence_items: list[FieldEvidence],
+    parser_version: str,
+) -> None:
+    existing = {
+        (item.field_name, item.extraction_method): item
+        for item in FieldProvenance.objects.filter(job=job, source_alias=alias)
+    }
+    creates: list[FieldProvenance] = []
+    updates: list[FieldProvenance] = []
+    for evidence in evidence_items:
+        key = (evidence.field_name, evidence.method)
+        values = {
+            "confidence": _confidence(evidence.confidence),
+            "evidence_excerpt": evidence.evidence[:1000],
+            "parser_version": parser_version,
+        }
+        current = existing.get(key)
+        if current is None:
+            creates.append(
+                FieldProvenance(
+                    job=job,
+                    source_alias=alias,
+                    field_name=evidence.field_name,
+                    extraction_method=evidence.method,
+                    **values,
+                )
+            )
+            continue
+        changed = False
+        for field_name, value in values.items():
+            if getattr(current, field_name) != value:
+                setattr(current, field_name, value)
+                changed = True
+        if changed:
+            current.updated_at = timezone.now()
+            updates.append(current)
+    if creates:
+        FieldProvenance.objects.bulk_create(creates, batch_size=250)
+    if updates:
+        FieldProvenance.objects.bulk_update(
+            updates,
+            ["confidence", "evidence_excerpt", "parser_version", "updated_at"],
+            batch_size=250,
+        )
+
+
+def _copy_score(score: ScoreRun) -> ScoreRun:
+    return ScoreRun.objects.create(
+        job=score.job,
+        profile=score.profile,
+        profile_version=score.profile_version,
+        score=score.score,
+        confidence=score.confidence,
+        recommendation=score.recommendation,
+        components=score.components,
+        positives=score.positives,
+        concerns=score.concerns,
+        hard_exclusions=score.hard_exclusions,
+        evidence=score.evidence,
+        model_version=score.model_version,
+        prompt_version=score.prompt_version,
+    )
+
+
 @transaction.atomic
 def persist_draft(
     draft: CanonicalJobDraft,
@@ -176,14 +270,16 @@ def persist_draft(
     parser_key: str,
     parser_version: str,
     extraction_confidence: float,
+    context: PersistenceContext | None = None,
 ) -> tuple[JobPosting, DedupeDecision, bool]:
     source = document.source
-    employer = resolve_employer(draft)
+    employer = resolve_employer(draft, context=context)
     decision = find_existing_job(draft, source=source)
     direct = _source_is_direct(source, draft)
     if decision.resolved_direct:
         direct = True
     created = False
+    substantive_change = False
 
     if decision.job is None:
         try:
@@ -240,6 +336,7 @@ def persist_draft(
         if extraction_confidence > float(job.extraction_confidence):
             job.extraction_confidence = _confidence(extraction_confidence)
             changed.append("extraction_confidence")
+        substantive_change = any(field not in {"last_seen", "last_changed"} for field in changed)
         if changed:
             job.save(update_fields=list(dict.fromkeys([*changed, "updated_at"])))
 
@@ -284,31 +381,58 @@ def persist_draft(
             update_fields=["last_seen", "raw_document", "payload", "is_canonical", "updated_at"]
         )
 
-    for evidence in draft.evidence:
-        FieldProvenance.objects.update_or_create(
-            job=job,
-            source_alias=alias,
-            field_name=evidence.field_name,
-            extraction_method=evidence.method,
-            defaults={
-                "confidence": _confidence(evidence.confidence),
-                "evidence_excerpt": evidence.evidence[:1000],
-                "parser_version": parser_version,
-            },
-        )
-
-    JobVersion.objects.get_or_create(
+    _sync_evidence(
         job=job,
-        content_hash=job.content_hash,
-        defaults={"snapshot": job_snapshot(job), "changed_fields": []},
+        alias=alias,
+        evidence_items=draft.evidence,
+        parser_version=parser_version,
     )
-    for profile in SearchProfile.objects.filter(is_active=True):
-        score_and_save(job, profile)
+
+    if created or substantive_change:
+        JobVersion.objects.get_or_create(
+            job=job,
+            content_hash=job.content_hash,
+            defaults={"snapshot": job_snapshot(job), "changed_fields": []},
+        )
+        if context is not None:
+            if context.active_profiles is None:
+                context.active_profiles = list(SearchProfile.objects.filter(is_active=True))
+            profiles = context.active_profiles
+        else:
+            profiles = SearchProfile.objects.filter(is_active=True)
+        for profile in profiles:
+            score = score_and_save(job, profile)
+            if context is not None:
+                context.latest_scores[(job.pk, profile.pk)] = score
+    else:
+        if context is not None:
+            if context.active_profiles is None:
+                context.active_profiles = list(SearchProfile.objects.filter(is_active=True))
+            profiles = context.active_profiles
+        else:
+            profiles = SearchProfile.objects.filter(is_active=True)
+        for profile in profiles:
+            cache_key = (job.pk, profile.pk)
+            previous = context.latest_scores.get(cache_key) if context is not None else None
+            if previous is None:
+                previous = (
+                    ScoreRun.objects.filter(job=job, profile=profile)
+                    .order_by("-created_at")
+                    .first()
+                )
+            if previous is None or previous.profile_version != profile.version:
+                score = score_and_save(job, profile)
+            else:
+                score = _copy_score(previous)
+            if context is not None:
+                context.latest_scores[cache_key] = score
     return job, decision, created
 
 
 @transaction.atomic
-def process_raw_document(document: RawDocument) -> dict[str, int | str | list[str]]:
+def process_raw_document(
+    document: RawDocument, *, context: PersistenceContext | None = None
+) -> dict[str, int | str | list[str]]:
     result = registry.extract(document)
     document.parser_key = result.parser_key
     document.parser_version = result.parser_version
@@ -341,6 +465,7 @@ def process_raw_document(document: RawDocument) -> dict[str, int | str | list[st
             parser_key=result.parser_key,
             parser_version=result.parser_version,
             extraction_confidence=result.confidence,
+            context=context,
         )
         if was_created:
             created += 1