#!/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())