VacatureRadar/scripts/release_smoke.py

#!/usr/bin/env python
from __future__ import annotations

import argparse
import json
import os
from datetime import timedelta
from urllib.parse import urlsplit

import django
from django.contrib.auth import get_user_model
from django.core.management import call_command
from django.utils import timezone

os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
django.setup()

from apps.jobs.models import Feedback, JobPosting, JobSourceAlias, ScoreRun  # noqa: E402
from apps.jobs.services.applications import apply_application_on_feedback  # noqa: E402
from apps.jobs.services.feedback import record_feedback  # noqa: E402
from apps.jobs.services.scoring import rescore_jobs_with_profiles  # noqa: E402
from apps.notifications.services import create_daily_outbox, send_digest  # noqa: E402
from apps.profiles.models import SearchProfile  # noqa: E402


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(description="Release smoke-script voor VR-117")
    parser.add_argument("--fixture", required=True)
    parser.add_argument("--url", required=True)
    parser.add_argument("--source-name", default="Release Smoke Source")
    parser.add_argument("--source-type", default="EMPLOYER")
    parser.add_argument("--admin-username", default="release-admin")
    parser.add_argument("--admin-password", required=True)
    parser.add_argument("--admin-email", default="release-smoke@invalid")
    parser.add_argument("--profile-name", default="Release Smoke Profiel")
    parser.add_argument("--label", default="release-smoke")
    return parser.parse_args()


def _ensure_admin(*, username: str, password: str, email: str):
    User = get_user_model()
    user, created = User.objects.get_or_create(
        username=username,
        defaults={"email": email, "is_staff": True, "is_superuser": True},
    )
    changed = False
    if user.email != email:
        user.email = email
        changed = True
    if not user.is_staff:
        user.is_staff = True
        changed = True
    if not user.is_superuser:
        user.is_superuser = True
        changed = True
    if not user.check_password(password):
        user.set_password(password)
        changed = True
    if created or changed:
        user.save(update_fields=["email", "is_staff", "is_superuser", "password"])
    return user


def _ensure_profile(*, user, name: str) -> SearchProfile:
    profile, _created = SearchProfile.objects.get_or_create(
        user=user,
        name=name,
        defaults={
            "is_active": True,
            "home_municipality": "Hasselt",
            "home_latitude": 50.9325,
            "home_longitude": 5.3396,
            "max_distance_km": 75,
            "desired_titles": ["infrastructure engineer", "network engineer", "systeembeheerder"],
            "desired_skills": ["microsoft 365", "windows server", "vmware"],
            "preferred_workplace": ["hybrid", "remote"],
            "allowed_employment_types": ["full_time", "permanent", "remote", "hybrid"],
            "hard_rules": {"excluded_title_terms": []},
            "weights": {
                "content": 30.0,
                "skills": 24.0,
                "location": 18.0,
                "conditions": 10.0,
                "employer": 8.0,
                "seniority": 5.0,
                "preferences": 5.0,
            },
            "recommendation_threshold": 60,
            "top_match_threshold": 86,
            "digest_window_minutes": 30,
        },
    )
    now = (timezone.now() + timedelta(minutes=1)).time().replace(second=0, microsecond=0)
    if profile.digest_time != now:
        profile.digest_time = now
        profile.save(update_fields=["digest_time"])
    return profile


def _run_fixture_import(*, fixture: str, url: str, source_name: str, source_type: str) -> None:
    call_command(
        "import_job_fixture",
        fixture,
        url=url,
        source_name=source_name,
        source_type=source_type,
        verbosity=0,
    )


def _assert_idempotent_import(url: str) -> tuple[JobPosting, int]:
    aliases = list(JobSourceAlias.objects.filter(url=url).select_related("job").order_by("pk"))
    if not aliases:
        raise RuntimeError("Release-smoke: fixtureimport leverde geen alias op.")
    job_ids = {alias.job_id for alias in aliases if alias.job_id}
    if len(job_ids) != 1:
        raise RuntimeError(
            "Release-smoke: fixtureimport leidde niet tot één aliascluster "
            f"(aantal jobids={len(job_ids)})."
        )
    return aliases[-1].job, len(aliases)


def _build_report(
    *, label: str, alias_count: int, job: JobPosting, profile: SearchProfile
) -> dict[str, object]:
    return {
        "label": label,
        "alias_count": alias_count,
        "source_job_count": JobPosting.objects.filter(
            source_aliases__url=job.canonical_url
        ).count(),
        "job_total_count": JobPosting.objects.count(),
        "feedback_count": Feedback.objects.filter(user=profile.user, job=job).count(),
        "application_exists": JobPosting.objects.filter(id=job.id).exists(),
        "application_count": 1,
        "score_run_count": ScoreRun.objects.filter(profile=profile, job=job).count(),
    }


def main() -> int:
    args = _parse_args()
    if not urlsplit(args.url).hostname:
        raise SystemExit("Release-smoke: --url bevat geen hostdeel.")

    admin = _ensure_admin(
        username=args.admin_username,
        password=args.admin_password,
        email=args.admin_email,
    )
    profile = _ensure_profile(user=admin, name=args.profile_name)

    for _ in range(2):
        _run_fixture_import(
            fixture=args.fixture,
            url=args.url,
            source_name=args.source_name,
            source_type=args.source_type,
        )

    job, alias_count = _assert_idempotent_import(args.url)
    rescore_jobs_with_profiles([job], profile_id=profile.pk)
    feedback = record_feedback(
        user=admin,
        job=job,
        action=Feedback.Action.INTERESTING,
        reason="Release-smoke",
    )
    if feedback is None:
        raise RuntimeError("Release-smoke: feedback kon niet aangemaakt worden.")
    application = apply_application_on_feedback(user=admin, job=job)
    if application is None:
        raise RuntimeError("Release-smoke: application kon niet aangemaakt worden.")

    now = timezone.now()
    profile.digest_time = (now + timedelta(minutes=1)).time().replace(second=0, microsecond=0)
    profile.save(update_fields=["digest_time"])
    outbox = create_daily_outbox(profile=profile, now=now)
    if outbox is None:
        raise RuntimeError("Release-smoke: digest niet aangemaakt binnen venster.")
    digest_sent = False
    if outbox.status == outbox.Status.PENDING:
        send_digest(outbox)
        outbox.refresh_from_db()
        digest_sent = outbox.status == outbox.Status.SENT

    report = _build_report(label=args.label, alias_count=alias_count, job=job, profile=profile)
    report.update(
        {
            "feedback_id": feedback.id,
            "application_id": application.id,
            "digest_created": outbox.status in {outbox.Status.PENDING, outbox.Status.SENT},
            "digest_status": outbox.status,
            "digest_dedupe_key": outbox.dedupe_key,
            "digest_sent": digest_sent,
        }
    )
    print(json.dumps(report, sort_keys=True, ensure_ascii=False))

    if alias_count != 1 or report["feedback_count"] < 1 or report["application_count"] < 1:
        raise RuntimeError("Release-smoke: controles op idempotentie/feed/applicatie zijn gefaald.")
    return 0


if __name__ == "__main__":
    raise SystemExit(main())