app.py

from __future__ import annotations

import logging
import os
from pathlib import Path
from typing import List, Optional

import gradio as gr
from dotenv import load_dotenv

from db import (
    configure_database,
    create_score,
    ensure_user,
    get_global_top,
    get_image_top,
    get_user_recent,
    init_db,
    normalize_username,
    scores_to_rows,
    session_scope,
    validate_username,
)
from model import ClipScorer, ImageEntry, load_image_entries

load_dotenv()

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("app")

DATABASE_URL = os.getenv("DATABASE_URL")
if not DATABASE_URL:
    raise RuntimeError("DATABASE_URL ist nicht gesetzt. Bitte in den Space-Secrets hinterlegen.")

configure_database(DATABASE_URL)
init_db()

IMAGE_ENTRIES: List[ImageEntry] = []
SCORER: Optional[ClipScorer] = None
EMBEDDING_ERROR: Optional[str] = None

try:
    IMAGE_ENTRIES = load_image_entries(Path("images.csv"))
except Exception as exc:  # noqa: BLE001
    EMBEDDING_ERROR = f"images.csv konnte nicht geladen werden: {exc}"
    logger.exception("Fehler beim Laden der images.csv", exc_info=exc)

if EMBEDDING_ERROR is None and IMAGE_ENTRIES:
    try:
        SCORER = ClipScorer()
        SCORER.load_precomputed_embeddings(IMAGE_ENTRIES)
    except Exception as exc:  # noqa: BLE001
        EMBEDDING_ERROR = (
            "Embeddings konnten nicht geladen werden. Bitte precompute_embeddings.py ausführen."
            f"\nFehler: {exc}"
        )
        logger.exception("Fehler beim Laden der Embeddings", exc_info=exc)

APP_READY = EMBEDDING_ERROR is None

HELP_TEXT = (
    "Beschreibe das angezeigte Bild in 3 bis 500 Zeichen. "
    "Die KI vergleicht deine Beschreibung mit der (theoretisch) perfekten Prompt für das angezeigte Bild. "
    "Der Score reicht von 0 (gar nicht passend) bis 1000 (perfekte Übereinstimmung)."
)

LEADERBOARD_HEADERS = [
    "Platz",
    "Benutzername",
    "Bild-ID",
    "Score",
    "Ähnlichkeit",
    "Text",
    "Zeitstempel",
]


def fetch_global_rows() -> List[List[object]]:
    with session_scope() as session:
        scores = get_global_top(session)
        return scores_to_rows(scores, include_rank=True)


def fetch_image_rows(image_id: str) -> List[List[object]]:
    if not image_id:
        return []
    with session_scope() as session:
        scores = get_image_top(session, image_id)
        return scores_to_rows(scores, include_rank=True)


def fetch_user_rows(username: str) -> List[List[object]]:
    if not username:
        return []
    canonical = normalize_username(username)
    with session_scope() as session:
        scores = get_user_recent(session, canonical)
        return scores_to_rows(scores, include_rank=True)


def handle_score(username: str, text: str, image_index: int | None):
    if not APP_READY or SCORER is None or not IMAGE_ENTRIES:
        raise gr.Error(
            "Embeddings sind nicht verfügbar. Bitte vor dem Start precompute_embeddings.py ausführen."
        )

    username_clean = (username or "").strip()
    if not validate_username(username_clean):
        raise gr.Error("Ungültiger Benutzername. Erlaubt sind 3-20 Zeichen aus A-Z, a-z, 0-9, _.-")

    text_clean = (text or "").strip()
    if len(text_clean) < 3:
        raise gr.Error("Bitte gib mindestens 3 Zeichen Text ein.")
    if len(text_clean) > 500:
        raise gr.Error("Der Beschreibungstext darf höchstens 500 Zeichen enthalten.")

    if image_index is None:
        image_index = 0
    if image_index < 0 or image_index >= len(IMAGE_ENTRIES):
        image_index = 0
    entry = IMAGE_ENTRIES[image_index]

    similarity, score = SCORER.score_text_for_image(text_clean, entry.image_id)

    with session_scope() as session:
        user = ensure_user(session, username_clean)
        create_score(
            session,
            user=user,
            image_id=entry.image_id,
            score_value=score,
            similarity=similarity,
            text=text_clean,
        )

    global_rows = fetch_global_rows()
    image_rows = fetch_image_rows(entry.image_id)
    user_rows = fetch_user_rows(username_clean)

    gr.Info("Score gespeichert!")

    return (
        gr.update(value=score),
        gr.update(value=round(similarity, 4)),
        gr.update(value=global_rows),
        gr.update(value=image_rows),
        gr.update(value=user_rows),
    )


def handle_next_image(current_index: int | None):
    if not IMAGE_ENTRIES:
        raise gr.Error("Keine Bilder konfiguriert.")
    if current_index is None:
        current_index = 0
    new_index = (current_index + 1) % len(IMAGE_ENTRIES)
    entry = IMAGE_ENTRIES[new_index]
    image_rows = fetch_image_rows(entry.image_id)
    return (
        new_index,
        gr.update(value=entry.image_url),
        gr.update(value=f"**Bild-ID:** {entry.image_id}"),
        gr.update(value=entry.image_id),
        gr.update(value=image_rows),
    )


def handle_image_dropdown(image_id: str):
    rows = fetch_image_rows(image_id)
    return gr.update(value=rows)


def handle_username_change(username: str):
    if not username:
        return gr.update(value=[])
    username_clean = username.strip()
    if not validate_username(username_clean):
        gr.Warning("Benutzername ungültig. Zeige keine Ergebnisse.")
        return gr.update(value=[])
    rows = fetch_user_rows(username_clean)
    return gr.update(value=rows)


def build_interface() -> gr.Blocks:
    status_message = ""
    if EMBEDDING_ERROR:
        status_message = f"⚠️ {EMBEDDING_ERROR}"
    elif not IMAGE_ENTRIES:
        status_message = "⚠️ Keine Bilder konfiguriert."
    else:
        status_message = "✅ Bereit zum Scoren!"

    initial_index = 0 if IMAGE_ENTRIES else None
    initial_entry = IMAGE_ENTRIES[0] if IMAGE_ENTRIES else None
    global_rows = fetch_global_rows() if APP_READY else []
    image_rows = fetch_image_rows(initial_entry.image_id) if initial_entry else []

    image_choices = [entry.image_id for entry in IMAGE_ENTRIES]

    with gr.Blocks(title="KI Prompt Challenge", theme=gr.themes.Soft()) as demo:
        gr.Markdown("# KI Prompt Challenge")
        gr.Markdown(HELP_TEXT)
        gr.Markdown(status_message)

        image_state = gr.State(initial_index)

        with gr.Row():
            with gr.Column(scale=3):
                image_component = gr.Image(
                    value=initial_entry.image_url if initial_entry else None,
                    label="Bild",
                    show_download_button=False,
                )
                image_info = gr.Markdown(
                    f"**Bild-ID:** {initial_entry.image_id}" if initial_entry else "Kein Bild geladen."
                )
                next_button = gr.Button(
                    "Nächstes Bild",
                    variant="secondary",
                    interactive=bool(IMAGE_ENTRIES),
                )
            with gr.Column(scale=2):
                username_input = gr.Textbox(
                    label="Benutzername",
                    placeholder="3-20 Zeichen (A-Z, a-z, 0-9, _.-)",
                )
                text_input = gr.Textbox(
                    label="Beschreibungstext",
                    placeholder="Was siehst du auf dem Bild?",
                    lines=5,
                )
                score_button = gr.Button(
                    "Scoren",
                    variant="primary",
                    interactive=APP_READY and bool(IMAGE_ENTRIES),
                )
                score_output = gr.Number(label="Score", value=0, precision=0)
                similarity_output = gr.Number(label="Ähnlichkeit", value=0.0, precision=4)

        gr.Markdown("### Leaderboard")
        with gr.Tabs():
            with gr.Tab("Top 50"):
                global_df = gr.Dataframe(
                    headers=LEADERBOARD_HEADERS,
                    value=global_rows,
                    datatype=[
                        "number",
                        "str",
                        "str",
                        "number",
                        "number",
                        "str",
                        "str",
                    ],
                    interactive=False,
                    wrap=True,
                )
            with gr.Tab("Dieses Bild Top 50"):
                image_dropdown = gr.Dropdown(
                    choices=image_choices,
                    value=initial_entry.image_id if initial_entry else None,
                    label="Bild auswählen",
                    interactive=bool(image_choices),
                )
                image_df = gr.Dataframe(
                    headers=LEADERBOARD_HEADERS,
                    value=image_rows,
                    datatype=[
                        "number",
                        "str",
                        "str",
                        "number",
                       "number",
                        "str",
                        "str",
                    ],
                    interactive=False,
                    wrap=True,
                )
            with gr.Tab("Meine letzten 50"):
                user_df = gr.Dataframe(
                    headers=LEADERBOARD_HEADERS,
                    value=[],
                    datatype=[
                        "number",
                        "str",
                        "str",
                        "number",
                        "number",
                        "str",
                        "str",
                    ],
                    interactive=False,
                    wrap=True,
                )

        next_button.click(
            handle_next_image,
            inputs=[image_state],
            outputs=[image_state, image_component, image_info, image_dropdown, image_df],
        )

        score_button.click(
            handle_score,
            inputs=[username_input, text_input, image_state],
            outputs=[score_output, similarity_output, global_df, image_df, user_df],
        )

        image_dropdown.change(handle_image_dropdown, inputs=[image_dropdown], outputs=[image_df])
        username_input.change(handle_username_change, inputs=[username_input], outputs=[user_df])

    return demo


demo = build_interface()


if __name__ == "__main__":
    demo.launch()