131 lines
4.5 KiB
Python
131 lines
4.5 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import os
|
|
from collections import Counter
|
|
from pathlib import Path
|
|
|
|
import django
|
|
|
|
os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings")
|
|
django.setup()
|
|
|
|
from apps.profiles.models import SearchProfile # noqa: E402
|
|
|
|
|
|
def _load_profile(profile_id: int | None, username: str | None) -> SearchProfile:
|
|
if profile_id:
|
|
return SearchProfile.objects.get(pk=profile_id)
|
|
if username:
|
|
profile = SearchProfile.objects.filter(user__username=username, is_active=True).first()
|
|
if profile is None:
|
|
raise ValueError(f"Geen actief profiel gevonden voor gebruiker {username}")
|
|
return profile
|
|
raise ValueError("Kies --profile-id of --user.")
|
|
|
|
|
|
def _collect_feedback_churn(profile: SearchProfile) -> dict:
|
|
from apps.jobs.models import Feedback
|
|
|
|
all_feedback = Feedback.objects.filter(profile=profile).order_by("created_at")
|
|
status = Counter()
|
|
features = Counter()
|
|
sample_count = 0
|
|
false_negatives = []
|
|
|
|
for feedback in all_feedback:
|
|
learning = (
|
|
feedback.metadata.get("learning") if isinstance(feedback.metadata, dict) else None
|
|
)
|
|
if not isinstance(learning, dict):
|
|
continue
|
|
reason_code = learning.get("reason_code", "")
|
|
status[learning.get("status", "")] += 1
|
|
if feature := learning.get("feature"):
|
|
features[feature] += 1
|
|
sample_count += 1
|
|
if learning.get("status") == "queued" and reason_code.startswith("non_learning_"):
|
|
false_negatives.append(
|
|
{
|
|
"job_id": str(feedback.job_id),
|
|
"reason_code": reason_code,
|
|
}
|
|
)
|
|
|
|
return {
|
|
"samples": dict(status),
|
|
"features": dict(features),
|
|
"total_feedback": all_feedback.count(),
|
|
"sample_count": sample_count,
|
|
"queued_non_learning_signals": false_negatives[:25],
|
|
}
|
|
|
|
|
|
def _collect_weight_churn(profile: SearchProfile) -> list[dict[str, float]]:
|
|
revisions = list(profile.revisions.order_by("version"))
|
|
if len(revisions) < 2:
|
|
return []
|
|
|
|
base = revisions[0].snapshot
|
|
latest = revisions[-1].snapshot
|
|
base_weights = base.get("weights", {})
|
|
latest_weights = latest.get("weights", {})
|
|
churn = []
|
|
for feature in sorted(set(base_weights) | set(latest_weights)):
|
|
before = float(base_weights.get(feature, 0.0))
|
|
after = float(latest_weights.get(feature, 0.0))
|
|
if before == after:
|
|
continue
|
|
churn.append({"feature": feature, "delta": round(after - before, 3)})
|
|
return sorted(churn, key=lambda item: abs(item["delta"]), reverse=True)
|
|
|
|
|
|
def main(argv: list[str] | None = None) -> int:
|
|
parser = argparse.ArgumentParser()
|
|
parser.add_argument("--profile-id", type=int, default=None)
|
|
parser.add_argument("--user", default=None)
|
|
parser.add_argument("--top", type=int, default=5)
|
|
args = parser.parse_args(argv)
|
|
|
|
if not args.profile_id and not args.user:
|
|
raise SystemExit("Gebruik --profile-id of --user.")
|
|
|
|
profile = _load_profile(args.profile_id, args.user)
|
|
feedback_report = _collect_feedback_churn(profile)
|
|
churn = _collect_weight_churn(profile)
|
|
report = {
|
|
"profile_id": profile.pk,
|
|
"profile_name": profile.name,
|
|
"learning_enabled": profile.learning_enabled,
|
|
"feedback": feedback_report,
|
|
"weight_churn": churn[: args.top],
|
|
}
|
|
output = Path("feedback_learning_report.txt")
|
|
output.write_text(
|
|
"\n".join(
|
|
[
|
|
f"Feedbacklearningrapport voor {profile.name} (id={profile.pk})",
|
|
f"Learning actief: {'ja' if profile.learning_enabled else 'nee'}",
|
|
f"Feedbacktotalen: {feedback_report['total_feedback']}",
|
|
f"Statuscounts: {feedback_report['samples']}",
|
|
f"Featurecounts: {feedback_report['features']}",
|
|
"Top-churn:",
|
|
]
|
|
+ [f" - {item['feature']}: {item['delta']:+.3f}" for item in report["weight_churn"]]
|
|
+ [f"\nMogelijke non-learn false negatives (max {args.top}):"]
|
|
+ [
|
|
f" - {item['job_id']} ({item['reason_code']})"
|
|
for item in feedback_report["queued_non_learning_signals"][: args.top]
|
|
]
|
|
)
|
|
+ "\n",
|
|
encoding="utf-8",
|
|
)
|
|
print(output.read_text(encoding="utf-8"))
|
|
return 0
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|