feedback.py

from __future__ import annotations

import json
import uuid
from datetime import datetime, timezone
from pathlib import Path
from typing import Any

from config import FEEDBACK_LOG_PATH


FEEDBACK_CATEGORIES = [
    "Doğru ve yararlı",
    "Yanlış kaynak",
    "Eksik cevap",
    "Kendini tekrar ediyor",
    "Bağlamı kaçırdı",
    "Başka konuya geçti",
    "Çok uzun",
    "Çok kısa",
    "Anlaşılır değil",
    "Diğer",
]


REGRESSION_REVIEW_CATEGORIES = {
    "Yanlış kaynak",
    "Eksik cevap",
    "Kendini tekrar ediyor",
    "Bağlamı kaçırdı",
    "Başka konuya geçti",
    "Anlaşılır değil",
}


def save_feedback(chat_history: Any, rating: int | float, category: str, comment: str) -> str:
    user_message, assistant_message = _last_turn(chat_history)
    if not user_message and not assistant_message:
        return "Kaydedilecek bir diyalog bulunamadı."

    try:
        numeric_rating = int(float(rating))
    except Exception:
        numeric_rating = 0
    numeric_rating = max(1, min(numeric_rating, 10))
    normalized_category = category or "Diğer"

    record = {
        "feedback_id": str(uuid.uuid4()),
        "created_at": datetime.now(timezone.utc).isoformat(),
        "rating": numeric_rating,
        "category": normalized_category,
        "comment": (comment or "").strip(),
        "user_message": user_message,
        "assistant_message": assistant_message,
        "needs_regression_review": _needs_regression_review(numeric_rating, normalized_category),
    }
    _append_jsonl(FEEDBACK_LOG_PATH, record)
    return "Geri bildirim kaydedildi. Teşekkürler."


def _last_turn(chat_history: Any) -> tuple[str, str]:
    if not chat_history:
        return "", ""

    for item in reversed(chat_history):
        if isinstance(item, dict):
            role = item.get("role")
            if role == "assistant":
                assistant = str(item.get("content", "") or "").strip()
                user = _previous_user_message(chat_history, item)
                return user, assistant
        if isinstance(item, (list, tuple)) and len(item) >= 2:
            return str(item[0] or "").strip(), str(item[1] or "").strip()

    return "", ""


def _previous_user_message(chat_history: Any, assistant_item: dict) -> str:
    try:
        assistant_index = chat_history.index(assistant_item)
    except ValueError:
        assistant_index = len(chat_history)
    for item in reversed(chat_history[:assistant_index]):
        if isinstance(item, dict) and item.get("role") == "user":
            return str(item.get("content", "") or "").strip()
    return ""


def _append_jsonl(path: Path, record: dict) -> None:
    path.parent.mkdir(parents=True, exist_ok=True)
    with path.open("a", encoding="utf-8") as file:
        file.write(json.dumps(record, ensure_ascii=False) + "\n")


def _needs_regression_review(rating: int, category: str) -> bool:
    return rating <= 5 or category in REGRESSION_REVIEW_CATEGORIES