394 lines
14 KiB
Python
394 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import csv
|
|
import io
|
|
import json
|
|
import zipfile
|
|
from datetime import date, timedelta
|
|
from decimal import Decimal
|
|
from html import escape
|
|
from typing import Any
|
|
|
|
from django.db import transaction
|
|
from django.utils import timezone
|
|
from django.utils.text import slugify
|
|
|
|
from apps.jobs.models import (
|
|
Application,
|
|
ApplicationTimelineEvent,
|
|
JobPosting,
|
|
ScoreRun,
|
|
)
|
|
from apps.jobs.services.pipeline import job_snapshot
|
|
from apps.profiles.models import SearchProfile
|
|
|
|
SNAPSHOT_VERSION = "1.0.0"
|
|
|
|
|
|
def _coerce_json_value(value: Any) -> Any:
|
|
if isinstance(value, date):
|
|
return value.isoformat()
|
|
if isinstance(value, Decimal):
|
|
return float(value)
|
|
if isinstance(value, set):
|
|
return sorted(value)
|
|
return value
|
|
|
|
|
|
def _timeline_dict_rows(application: Application) -> list[dict[str, str]]:
|
|
rows: list[dict[str, str]] = []
|
|
for event in application.timeline_events.order_by("created_at").all():
|
|
metadata = event.metadata if isinstance(event.metadata, dict) else {}
|
|
rows.append(
|
|
{
|
|
"timestamp": event.created_at.isoformat(),
|
|
"type": event.event_type,
|
|
"actor": str(getattr(event.user, "username", "")),
|
|
"from": str(metadata.get("from", "")),
|
|
"to": str(metadata.get("to", "")),
|
|
"note": str(metadata.get("note", "")),
|
|
}
|
|
)
|
|
return rows
|
|
|
|
|
|
def _build_export_filename(application: Application) -> str:
|
|
slug = slugify(application.job.normalized_title or application.job.original_title) or "vacature"
|
|
return f"application-{application.pk}-{slug[:48]}.zip"
|
|
|
|
|
|
def _source_links(job: JobPosting) -> list[dict[str, Any]]:
|
|
links: list[dict[str, Any]] = []
|
|
seen: set[str] = set()
|
|
for alias in job.source_aliases.select_related("source").all():
|
|
url = alias.canonical_url or alias.url
|
|
if not url:
|
|
continue
|
|
if url in seen:
|
|
continue
|
|
seen.add(url)
|
|
links.append(
|
|
{
|
|
"source": alias.source.name if alias.source else "",
|
|
"url": url,
|
|
"is_canonical": alias.is_canonical,
|
|
"source_domain": (alias.source.domain if alias.source else ""),
|
|
}
|
|
)
|
|
return links
|
|
|
|
|
|
def _latest_score_snapshot(job: JobPosting, user) -> dict[str, Any] | None:
|
|
profile = SearchProfile.objects.filter(user=user, is_active=True).first()
|
|
if not profile:
|
|
return None
|
|
run = (
|
|
ScoreRun.objects.filter(job=job, profile=profile)
|
|
.order_by("-created_at")
|
|
.only(
|
|
"score",
|
|
"recommendation",
|
|
"confidence",
|
|
"components",
|
|
"positives",
|
|
"concerns",
|
|
"hard_exclusions",
|
|
"evidence",
|
|
"profile_version",
|
|
)
|
|
.first()
|
|
)
|
|
if not run:
|
|
return None
|
|
return {
|
|
"profile_id": profile.pk,
|
|
"profile_name": profile.name,
|
|
"score": float(run.score),
|
|
"recommendation": run.recommendation,
|
|
"confidence": float(run.confidence),
|
|
"components": dict(run.components),
|
|
"positives": list(run.positives),
|
|
"concerns": list(run.concerns),
|
|
"hard_exclusions": list(run.hard_exclusions),
|
|
"evidence": _coerce_json_value(run.evidence),
|
|
"captured_at": run.created_at.isoformat(),
|
|
"profile_version": run.profile_version,
|
|
}
|
|
|
|
|
|
def _build_application_snapshot_payload(application: Application) -> dict[str, Any]:
|
|
now = timezone.now().isoformat()
|
|
snapshot = job_snapshot(application.job)
|
|
snapshot["description_html"] = application.job.description_html_sanitized
|
|
return {
|
|
"schema_version": SNAPSHOT_VERSION,
|
|
"title": application.job.original_title,
|
|
"frozen_at": now,
|
|
"job": snapshot,
|
|
"sources": _source_links(application.job),
|
|
"scores": _latest_score_snapshot(application.job, application.user),
|
|
"snapshot_kind": "application",
|
|
}
|
|
|
|
|
|
def _record_timeline_event(
|
|
*, application: Application, user, event_type: str, metadata: dict[str, Any] | None = None
|
|
) -> ApplicationTimelineEvent:
|
|
payload: dict[str, Any] = {}
|
|
if metadata:
|
|
payload.update({key: value for key, value in metadata.items() if value is not None})
|
|
return ApplicationTimelineEvent.objects.create(
|
|
application=application,
|
|
user=user,
|
|
event_type=event_type,
|
|
metadata=payload,
|
|
)
|
|
|
|
|
|
def _normalize_str(value: Any) -> str:
|
|
return ("" if value is None else str(value)).strip()
|
|
|
|
|
|
def apply_application_on_feedback(*, user, job: JobPosting) -> Application:
|
|
"""Create or promote an application when the user marks the job as applied."""
|
|
with transaction.atomic():
|
|
follow_up_date = timezone.localdate() + timedelta(days=7)
|
|
application, created = Application.objects.get_or_create(
|
|
user=user,
|
|
job=job,
|
|
defaults={
|
|
"status": Application.Status.APPLIED,
|
|
"applied_at": timezone.now(),
|
|
"follow_up_date": follow_up_date,
|
|
},
|
|
)
|
|
|
|
previous = {
|
|
"status": application.status,
|
|
"contact_name": application.contact_name,
|
|
"contact_email": application.contact_email,
|
|
"notes": application.notes,
|
|
"snapshot": application.snapshot,
|
|
}
|
|
snapshot_created = False
|
|
|
|
if application.status == Application.Status.PREPARING:
|
|
application.status = Application.Status.APPLIED
|
|
if not application.applied_at:
|
|
application.applied_at = application.applied_at or timezone.now()
|
|
if not application.follow_up_date:
|
|
application.follow_up_date = follow_up_date
|
|
if not application.snapshot:
|
|
application.snapshot = _build_application_snapshot_payload(application)
|
|
snapshot_created = True
|
|
|
|
updates: list[str] = []
|
|
if created:
|
|
updates.extend(["status", "applied_at", "follow_up_date", "snapshot", "updated_at"])
|
|
else:
|
|
if previous["status"] != application.status:
|
|
updates.extend(["status", "applied_at", "follow_up_date", "updated_at"])
|
|
if not previous["snapshot"]:
|
|
updates.append("snapshot")
|
|
if updates:
|
|
updates = sorted(set(updates))
|
|
application.save(update_fields=updates)
|
|
|
|
if created:
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.CREATED,
|
|
metadata={
|
|
"status": application.status,
|
|
},
|
|
)
|
|
if snapshot_created:
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.SNAPSHOT_CAPTURED,
|
|
metadata={
|
|
"version": SNAPSHOT_VERSION,
|
|
},
|
|
)
|
|
|
|
if previous["status"] != application.status:
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.STATUS_CHANGED,
|
|
metadata={
|
|
"from": previous["status"],
|
|
"to": application.status,
|
|
},
|
|
)
|
|
|
|
return application
|
|
|
|
|
|
def track_application_changes(
|
|
*, application: Application, user, previous: dict[str, Any], current: dict[str, Any]
|
|
) -> int:
|
|
"""Persist timeline events for manual application edits."""
|
|
events = 0
|
|
if previous.get("status") != current.get("status"):
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.STATUS_CHANGED,
|
|
metadata={
|
|
"from": previous.get("status", ""),
|
|
"to": current.get("status", ""),
|
|
},
|
|
)
|
|
events += 1
|
|
|
|
if _normalize_str(previous.get("notes")) != _normalize_str(current.get("notes")):
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.NOTES_UPDATED,
|
|
metadata={
|
|
"note": _normalize_str(current.get("notes", ""))[:250],
|
|
},
|
|
)
|
|
events += 1
|
|
|
|
if (
|
|
_normalize_str(previous.get("contact_name")) != _normalize_str(current.get("contact_name"))
|
|
) or (
|
|
_normalize_str(previous.get("contact_email"))
|
|
!= _normalize_str(current.get("contact_email"))
|
|
):
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.CONTACT_UPDATED,
|
|
metadata={
|
|
"contact_name": _normalize_str(current.get("contact_name")),
|
|
"contact_email": _normalize_str(current.get("contact_email")),
|
|
},
|
|
)
|
|
events += 1
|
|
return events
|
|
|
|
|
|
def change_application_status(*, application: Application, user, status: str) -> Application:
|
|
"""Persist an accessible pipeline status change and its audit event."""
|
|
|
|
valid_statuses = {value for value, _label in Application.Status.choices}
|
|
if status not in valid_statuses:
|
|
raise ValueError("Ongeldige sollicitatiestatus.")
|
|
previous = application.status
|
|
if previous == status:
|
|
return application
|
|
with transaction.atomic():
|
|
application.status = status
|
|
update_fields = ["status", "updated_at"]
|
|
if status == Application.Status.APPLIED and application.applied_at is None:
|
|
application.applied_at = timezone.now()
|
|
update_fields.append("applied_at")
|
|
application.save(update_fields=update_fields)
|
|
_record_timeline_event(
|
|
application=application,
|
|
user=user,
|
|
event_type=ApplicationTimelineEvent.EventType.STATUS_CHANGED,
|
|
metadata={"from": previous, "to": status},
|
|
)
|
|
return application
|
|
|
|
|
|
def build_print_html(application: Application) -> str:
|
|
snapshot = application.snapshot if isinstance(application.snapshot, dict) else {}
|
|
job = application.job
|
|
timeline_rows = _timeline_dict_rows(application)
|
|
safe_rows = []
|
|
for row in timeline_rows:
|
|
safe_rows.append(
|
|
f"<tr>"
|
|
f"<td>{escape(row['timestamp'])}</td>"
|
|
f"<td>{escape(row['type'])}</td>"
|
|
f"<td>{escape(row['actor'])}</td>"
|
|
f"<td>{escape(row['from'])}</td>"
|
|
f"<td>{escape(row['to'])}</td>"
|
|
f"<td>{escape(row['note'])}</td>"
|
|
"</tr>"
|
|
)
|
|
|
|
source_rows = []
|
|
for source in snapshot.get("sources", []):
|
|
source_name = escape(str(source.get("source", "")))
|
|
source_url = escape(str(source.get("url", "")))
|
|
source_rows.append(f"<li>{source_name}: {source_url}</li>")
|
|
source_section = "".join(source_rows) if source_rows else "<li>Niet beschikbaar</li>"
|
|
frozen_at = (
|
|
application.snapshot.get("frozen_at", "") if isinstance(application.snapshot, dict) else ""
|
|
)
|
|
snapshot_json = json.dumps(snapshot, indent=2, ensure_ascii=False, sort_keys=True)
|
|
|
|
return (
|
|
"<!doctype html><html><head><meta charset='utf-8'>"
|
|
"<title>Sollicitatiedossier</title><style>body{font-family:Arial,sans-serif;margin:24px}"
|
|
"table{border-collapse:collapse;width:100%}"
|
|
"th,td{border:1px solid #ddd;padding:8px;text-align:left}"
|
|
"th{background:#f2f2f2} </style></head><body>"
|
|
f"<h1>{escape(job.original_title)}</h1>"
|
|
f"<p>Vacature: {escape(job.canonical_url)}</p>"
|
|
f"<p>Status: {escape(application.get_status_display())}</p>"
|
|
f"<p>Geslaagd op: {escape(str(frozen_at))}</p>"
|
|
f"<h2>Bronnen</h2><ul>{source_section}</ul>"
|
|
"<h2>Timeline</h2><table><thead><tr><th>Tijd</th><th>Type</th>"
|
|
"<th>Actor</th><th>Van</th><th>Naar</th><th>Notitie</th></tr></thead><tbody>"
|
|
f"{''.join(safe_rows)}</tbody></table>"
|
|
f"<h2>Snapshot</h2><pre>{escape(snapshot_json)}</pre>"
|
|
"</body></html>"
|
|
)
|
|
|
|
|
|
def build_application_export(application: Application) -> tuple[bytes, str]:
|
|
payload = {
|
|
"application": {
|
|
"id": str(application.pk),
|
|
"job_id": str(application.job_id),
|
|
"status": application.status,
|
|
"contact_name": application.contact_name,
|
|
"contact_email": application.contact_email,
|
|
"contact_metadata": application.contact_metadata,
|
|
"notes": application.notes,
|
|
"snapshot_version": SNAPSHOT_VERSION,
|
|
"exported_at": timezone.now().isoformat(),
|
|
"applied_at": application.applied_at.isoformat() if application.applied_at else None,
|
|
"follow_up_date": application.follow_up_date.isoformat()
|
|
if application.follow_up_date
|
|
else None,
|
|
},
|
|
"snapshot": _coerce_json_value(application.snapshot),
|
|
"timeline": _timeline_dict_rows(application),
|
|
}
|
|
|
|
timeline_buffer = io.StringIO()
|
|
writer = csv.DictWriter(
|
|
timeline_buffer,
|
|
fieldnames=["timestamp", "type", "actor", "from", "to", "note"],
|
|
)
|
|
writer.writeheader()
|
|
for row in payload["timeline"]:
|
|
writer.writerow(row)
|
|
|
|
content_buffer = io.BytesIO()
|
|
with zipfile.ZipFile(content_buffer, "w", compression=zipfile.ZIP_DEFLATED) as zf:
|
|
zf.writestr(
|
|
"application.json",
|
|
json.dumps(_coerce_json_value(payload), indent=2, ensure_ascii=False, sort_keys=True),
|
|
)
|
|
zf.writestr("timeline.csv", timeline_buffer.getvalue())
|
|
zf.writestr("application_print.html", build_print_html(application))
|
|
|
|
return content_buffer.getvalue(), _build_export_filename(application)
|
|
|
|
|
|
def delete_application_dossier(*, application_id: int, user) -> bool:
|
|
deleted, _ = Application.objects.filter(pk=application_id, user=user).delete()
|
|
return bool(deleted)
|