Build Gradio CLIP scoring app

.env.example

@@ -0,0 +1,2 @@
+# PostgreSQL-Verbindungszeichenfolge im Format postgresql+psycopg2://user:password@host:port/dbname
+DATABASE_URL=postgresql+psycopg2://user:password@host:5432/database

.gitignore

@@ -0,0 +1,15 @@
+__pycache__/
+*.pyc
+*.pyo
+*.pyd
+.env
+.env.local
+pytest_cache/
+.DS_Store
+.idea/
+.vscode/
+.eggs/
+*.egg-info/
+build/
+dist/
+.cache/

README.md

@@ -1,2 +1,109 @@
-# game2
-A web game
+# CLIP Score Challenge
+
+Eine produktionsnahe Gradio-App, die Benutzertexte gegen einen festen Bildpool bewertet. Die App nutzt CLIP (open-clip ViT-B-32, Pretrained "openai") für Text- und Bild-Embeddings und speichert alle Scores in einer externen PostgreSQL-Datenbank.
+
+## Features
+
+- Fester Bildpool aus `images.csv` (nur externe URLs, keine Uploads)
+- Vorberechnete Bild-Embeddings (Kosinus-Ähnlichkeit → Score 0–1000)
+- Deutschsprachige UI mit Leaderboard (global, pro Bild, eigene letzten Scores)
+- Persistenz via PostgreSQL (SQLAlchemy + psycopg2)
+- Keine Moderation, kein Rate-Limit
+- Skript zur Vorberechnung der Embeddings
+- Automatische Schema-Erstellung beim Start
+- Light-Tests (Score-Mapping + DB Roundtrip)
+
+## Projektstruktur
+
+```
+.
+├── app.py                   # Gradio UI & Callback-Logik
+├── db.py                    # Datenbank-Modelle & Hilfsfunktionen
+├── model.py                 # CLIP-Laden, Embeddings, Score-Berechnung
+├── precompute_embeddings.py # Skript zum Vorberechnen der Bild-Embeddings
+├── images.csv               # Bildpool (image_id, image_url, clip_model, embedding_path)
+├── embeddings/              # .npy-Embeddings (initial leer, Skript befüllt)
+├── tests/                   # Pytest-Tests
+├── requirements.txt
+├── runtime.txt              # Python-Version für HF Spaces (3.10)
+├── .env.example             # Beispiel für lokale Entwicklung
+└── README.md
+```
+
+## Vorbereitung
+
+1. **Python 3.10** installieren (lokal oder via venv/conda).
+2. Repository klonen und Abhängigkeiten installieren:
+   ```bash
+   pip install -r requirements.txt
+   ```
+3. `.env` anlegen (siehe `.env.example`) oder `DATABASE_URL` direkt exportieren.
+   ```bash
+   export DATABASE_URL=postgresql+psycopg2://user:pass@host:5432/dbname
+   ```
+
+## Embeddings vorrechnen
+
+Das Skript lädt jedes Bild aus `images.csv`, berechnet das CLIP-Embedding und speichert es unter `embedding_path`.
+
+```bash
+python precompute_embeddings.py --csv images.csv --model-name ViT-B-32 --pretrained openai
+```
+
+Hinweise:
+- Die Bild-URLs müssen öffentlich erreichbar sein.
+- Beim ersten Lauf wird das Modell automatisch geladen (~1x pro Space).
+- Embedding-Dateien (`.npy`) werden im `embeddings/` Ordner gespeichert. Diese Dateien **müssen** ins Repo eingecheckt oder im Space verfügbar sein.
+
+## Lokale Entwicklung
+
+1. Embeddings berechnen (siehe oben).
+2. App starten:
+   ```bash
+   python app.py
+   ```
+3. Gradio öffnet standardmäßig `http://127.0.0.1:7860`.
+
+## Tests
+
+```bash
+pytest
+```
+
+## Deployment auf Hugging Face Spaces
+
+1. **Space anlegen**
+   - Typ: *Gradio*
+   - Runtime: Python 3.10 (`runtime.txt` ist bereits enthalten)
+
+2. **Secrets setzen**
+   - Im Space `Settings` → `New secret`
+   - Schlüssel: `DATABASE_URL`
+   - Wert: PostgreSQL-Verbindungsstring (z. B. von Neon/Supabase)
+
+3. **Embeddings bereitstellen**
+   - Lokal `precompute_embeddings.py` ausführen
+   - Die erzeugten `.npy`-Dateien unter `embeddings/` committen und zum Space pushen (z. B. via `git add embeddings/*.npy`)
+   - Alternativ die Dateien manuell im Space hochladen
+
+4. **Code pushen**
+   - Repo-Inhalt in den Space pushen (oder per `Add file` hochladen)
+
+5. **Space starten**
+   - Beim Start erstellt `app.py` automatisch das DB-Schema (`users`, `scores` + Indizes)
+   - UI erscheint mit deutschem Leaderboard
+
+## Datenbankschema
+
+- `users`: speichert `username` (kanonisch, lowercase) und `display_name`
+- `scores`: enthält `username`, `canonical_username`, `image_id`, `score`, `similarity`, `text`, `created_at`
+- Mehrere Scores pro Benutzer sind erlaubt, keine Deduplication
+
+## Hinweise
+
+- Scores werden deterministisch berechnet (gleicher Text + Bild → gleicher Score)
+- Keine IP- oder personenbezogene Daten werden geloggt
+- Für produktiven Einsatz unbedingt abgesicherte Postgres-Instanz verwenden
+- Bei neuen Bildern `images.csv` erweitern, Skript erneut laufen lassen und `.npy`-Dateien committen
+
+Viel Spaß beim Scoren! 🎯

app.py

@@ -0,0 +1,319 @@
+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. "
+    "CLIP vergleicht deine Beschreibung mit dem vorab berechneten Bild-Embedding. "
+    "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="CLIP Score Challenge", theme=gr.themes.Soft()) as demo:
+        gr.Markdown("# CLIP Score 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="Aktuelles 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("Global 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()

db.py

@@ -0,0 +1,227 @@
+from __future__ import annotations
+
+import os
+import re
+from contextlib import contextmanager
+from datetime import datetime, timezone
+from typing import List, Optional, Sequence
+
+from sqlalchemy import (
+    Column,
+    DateTime,
+    Float,
+    ForeignKey,
+    Index,
+    Integer,
+    String,
+    Text,
+    create_engine,
+    func,
+    select,
+)
+from sqlalchemy.engine import Engine
+from sqlalchemy.orm import Session, declarative_base, relationship, sessionmaker
+
+Base = declarative_base()
+
+
+USERNAME_REGEX = re.compile(r"^[A-Za-z0-9_.-]{3,20}$")
+
+_engine: Optional[Engine] = None
+_SessionLocal: Optional[sessionmaker] = None
+
+
+class User(Base):
+    __tablename__ = "users"
+
+    id = Column(Integer, primary_key=True)
+    username = Column(String(20), unique=True, nullable=False, index=True)  # kanonisch (lowercase)
+    display_name = Column(String(20), nullable=False)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False)
+
+    scores = relationship("Score", back_populates="user", cascade="all, delete-orphan")
+
+
+class Score(Base):
+    __tablename__ = "scores"
+
+    id = Column(Integer, primary_key=True)
+    user_id = Column(Integer, ForeignKey("users.id", ondelete="CASCADE"), nullable=False, index=True)
+    username = Column(String(20), nullable=False)  # Originalschreibweise
+    canonical_username = Column(String(20), nullable=False, index=True)
+    image_id = Column(String(100), nullable=False, index=True)
+    score = Column(Integer, nullable=False)
+    similarity = Column(Float, nullable=False)
+    text = Column(Text, nullable=False)
+    created_at = Column(DateTime(timezone=True), server_default=func.now(), nullable=False, index=True)
+
+    user = relationship("User", back_populates="scores")
+
+    __table_args__ = (
+        Index("ix_scores_global", score.desc(), created_at.asc(), id.asc()),
+        Index("ix_scores_image", "image_id", score.desc(), created_at.asc()),
+        Index("ix_scores_user", "canonical_username", created_at.desc(), id.desc()),
+    )
+
+
+def configure_database(database_url: Optional[str] = None) -> Engine:
+    """Initialisiert Engine und SessionFactory."""
+
+    global _engine, _SessionLocal
+
+    if database_url is None:
+        database_url = os.getenv("DATABASE_URL")
+    if not database_url:
+        raise RuntimeError("DATABASE_URL ist nicht gesetzt.")
+
+    _engine = create_engine(database_url, future=True, pool_pre_ping=True)
+    _SessionLocal = sessionmaker(bind=_engine, autoflush=False, expire_on_commit=False, future=True)
+    return _engine
+
+
+def get_engine() -> Engine:
+    if _engine is None:
+        raise RuntimeError("Datenbank wurde noch nicht konfiguriert. Rufen Sie configure_database() auf.")
+    return _engine
+
+
+def init_db() -> None:
+    engine = get_engine()
+    Base.metadata.create_all(engine)
+
+
+@contextmanager
+def session_scope() -> Session:
+    if _SessionLocal is None:
+        raise RuntimeError("SessionFactory nicht initialisiert. configure_database() zuerst aufrufen.")
+    session: Session = _SessionLocal()
+    try:
+        yield session
+        session.commit()
+    except Exception:
+        session.rollback()
+        raise
+    finally:
+        session.close()
+
+
+def normalize_username(username: str) -> str:
+    return username.strip().lower()
+
+
+def validate_username(username: str) -> bool:
+    return bool(USERNAME_REGEX.match(username))
+
+
+def ensure_user(session: Session, username_input: str) -> User:
+    """Sucht oder erstellt einen Nutzer."""
+
+    normalized = normalize_username(username_input)
+    stmt = select(User).where(User.username == normalized)
+    user = session.execute(stmt).scalar_one_or_none()
+    if user:
+        return user
+
+    user = User(username=normalized, display_name=username_input.strip())
+    session.add(user)
+    session.flush()
+    return user
+
+
+def create_score(
+    session: Session,
+    *,
+    user: User,
+    image_id: str,
+    score_value: int,
+    similarity: float,
+    text: str,
+) -> Score:
+    entry = Score(
+        user_id=user.id,
+        username=user.display_name,
+        canonical_username=user.username,
+        image_id=image_id,
+        score=score_value,
+        similarity=similarity,
+        text=text,
+    )
+    session.add(entry)
+    session.flush()
+    return entry
+
+
+def _format_timestamp(value: datetime | str | None) -> str:
+    if value is None:
+        return ""
+    if isinstance(value, str):
+        return value
+    if value.tzinfo is None:
+        value = value.replace(tzinfo=timezone.utc)
+    return value.astimezone(timezone.utc).strftime("%Y-%m-%d %H:%M:%SZ")
+
+
+def scores_to_rows(scores: Sequence[Score], include_rank: bool = True) -> List[List[object]]:
+    rows: List[List[object]] = []
+    for index, score in enumerate(scores, start=1):
+        base_row = [
+            score.username,
+            score.image_id,
+            score.score,
+            round(score.similarity, 4),
+            score.text,
+            _format_timestamp(score.created_at),
+        ]
+        if include_rank:
+            rows.append([index, *base_row])
+        else:
+            rows.append(base_row)
+    return rows
+
+
+def get_global_top(session: Session, limit: int = 50) -> List[Score]:
+    stmt = (
+        select(Score)
+        .order_by(Score.score.desc(), Score.created_at.asc(), Score.id.asc())
+        .limit(limit)
+    )
+    return list(session.scalars(stmt))
+
+
+def get_image_top(session: Session, image_id: str, limit: int = 50) -> List[Score]:
+    stmt = (
+        select(Score)
+        .where(Score.image_id == image_id)
+        .order_by(Score.score.desc(), Score.created_at.asc(), Score.id.asc())
+        .limit(limit)
+    )
+    return list(session.scalars(stmt))
+
+
+def get_user_recent(session: Session, canonical_username: str, limit: int = 50) -> List[Score]:
+    stmt = (
+        select(Score)
+        .where(Score.canonical_username == canonical_username)
+        .order_by(Score.created_at.desc(), Score.id.desc())
+        .limit(limit)
+    )
+    return list(session.scalars(stmt))
+
+
+__all__ = [
+    "Base",
+    "Score",
+    "User",
+    "configure_database",
+    "create_score",
+    "ensure_user",
+    "get_engine",
+    "get_global_top",
+    "get_image_top",
+    "get_user_recent",
+    "init_db",
+    "normalize_username",
+    "scores_to_rows",
+    "session_scope",
+    "validate_username",
+]

images.csv

@@ -0,0 +1,4 @@
+image_id,image_url,clip_model,embedding_path
+cat_wiki,https://upload.wikimedia.org/wikipedia/commons/thumb/3/3a/Cat03.jpg/640px-Cat03.jpg,ViT-B-32,embeddings/cat_wiki.npy
+mountain_sunrise,https://upload.wikimedia.org/wikipedia/commons/thumb/9/99/Swiss_Mountains_%28228640584%29.jpeg/640px-Swiss_Mountains_%28228640584%29.jpeg,ViT-B-32,embeddings/mountain_sunrise.npy
+metro_station,https://upload.wikimedia.org/wikipedia/commons/thumb/5/55/Berlin_U-Bahn_station_Brandenburger_Tor_platform.jpg/640px-Berlin_U-Bahn_station_Brandenburger_Tor_platform.jpg,ViT-B-32,embeddings/metro_station.npy

model.py

@@ -0,0 +1,191 @@
+from __future__ import annotations
+
+import csv
+import logging
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Dict, Iterable, List, Optional
+
+logger = logging.getLogger(__name__)
+
+
+@dataclass(frozen=True)
+class ImageEntry:
+    """Container für Bildmetadaten und Pfade zu Embeddings."""
+
+    image_id: str
+    image_url: str
+    clip_model: str
+    embedding_path: Path
+
+
+def load_image_entries(csv_path: Path | str) -> List[ImageEntry]:
+    """Liest die Bildliste aus einer CSV-Datei."""
+
+    path = Path(csv_path)
+    if not path.exists():
+        raise FileNotFoundError(f"Die Datei {path} existiert nicht.")
+
+    entries: List[ImageEntry] = []
+    with path.open("r", encoding="utf-8") as handle:
+        reader = csv.DictReader(handle)
+        required_fields = {"image_id", "image_url", "clip_model", "embedding_path"}
+        if reader.fieldnames is None:
+            raise ValueError("images.csv besitzt keine Kopfzeile.")
+        missing_fields = required_fields.difference(reader.fieldnames)
+        if missing_fields:
+            raise ValueError(f"images.csv fehlt/fehlen folgende Felder: {sorted(missing_fields)}")
+
+        for row in reader:
+            image_id = row["image_id"].strip()
+            image_url = row["image_url"].strip()
+            clip_model = row["clip_model"].strip()
+            embedding_path = Path(row["embedding_path"].strip())
+            if not image_id or not image_url or not clip_model:
+                logger.warning("Zeile in images.csv übersprungen: %s", row)
+                continue
+            if not embedding_path.is_absolute():
+                embedding_path = (path.parent / embedding_path).resolve()
+            entries.append(
+                ImageEntry(
+                    image_id=image_id,
+                    image_url=image_url,
+                    clip_model=clip_model,
+                    embedding_path=embedding_path,
+                )
+            )
+
+    if not entries:
+        raise ValueError("images.csv enthält keine gültigen Einträge.")
+
+    return entries
+
+
+def similarity_to_score(similarity: float) -> int:
+    """Wandelt eine Kosinusähnlichkeit (-1 bis 1) in einen Score von 0 bis 1000 um."""
+
+    clipped = max(-1.0, min(1.0, similarity))
+    score = int(round(((clipped + 1.0) / 2.0) * 1000))
+    return score
+
+
+def _require_numpy():
+    try:
+        import numpy as np  # type: ignore
+    except ModuleNotFoundError as exc:  # pragma: no cover - defensive fallback
+        raise ModuleNotFoundError("numpy wird benötigt, ist aber nicht installiert.") from exc
+    return np
+
+
+def _require_torch():
+    try:
+        import torch  # type: ignore
+    except ModuleNotFoundError as exc:  # pragma: no cover - defensive fallback
+        raise ModuleNotFoundError("torch wird benötigt, ist aber nicht installiert.") from exc
+    return torch
+
+
+def _require_open_clip():
+    try:
+        import open_clip  # type: ignore
+    except ModuleNotFoundError as exc:  # pragma: no cover - defensive fallback
+        raise ModuleNotFoundError("open-clip-torch wird benötigt, ist aber nicht installiert.") from exc
+    return open_clip
+
+
+class ClipScorer:
+    """Wrapper um CLIP für Text-/Bild-Embeddings und Scores."""
+
+    def __init__(
+        self,
+        model_name: str = "ViT-B-32",
+        pretrained: str = "openai",
+        device: Optional[str] = None,
+    ) -> None:
+        self.model_name = model_name
+        self.pretrained = pretrained
+        torch = _require_torch()
+        open_clip = _require_open_clip()
+        self._torch = torch
+        self.device = device or ("cuda" if torch.cuda.is_available() else "cpu")
+        logger.info("Lade CLIP Modell %s (%s) auf %s", model_name, pretrained, self.device)
+        self.model, _, self.preprocess = open_clip.create_model_and_transforms(
+            model_name,
+            pretrained=pretrained,
+        )
+        self.model.to(self.device)
+        self.model.eval()
+        for parameter in self.model.parameters():
+            parameter.requires_grad = False
+        self.tokenizer = open_clip.get_tokenizer(model_name)
+        self._image_embeddings: Dict[str, Any] = {}
+
+    def load_precomputed_embeddings(self, entries: Iterable[ImageEntry]) -> None:
+        """Lädt Embeddings aus .npy-Dateien und speichert sie intern."""
+
+        loaded = 0
+        for entry in entries:
+            if entry.clip_model != self.model_name:
+                logger.warning(
+                    "Überspringe Bild %s: erwartet Modell %s, gefunden %s",
+                    entry.image_id,
+                    self.model_name,
+                    entry.clip_model,
+                )
+                continue
+            if not entry.embedding_path.exists():
+                raise FileNotFoundError(
+                    f"Embedding-Datei für {entry.image_id} fehlt: {entry.embedding_path}"
+                )
+            np = _require_numpy()
+            torch = self._torch
+            array = np.load(entry.embedding_path)
+            if array.ndim > 1:
+                array = array.squeeze()
+            tensor = torch.from_numpy(array).to(self.device)
+            tensor = tensor.to(dtype=torch.float32)
+            norm = torch.linalg.norm(tensor)
+            if norm == 0:
+                raise ValueError(f"Embedding für {entry.image_id} hat Norm 0.")
+            tensor = tensor / norm
+            self._image_embeddings[entry.image_id] = tensor
+            loaded += 1
+
+        if loaded == 0:
+            raise ValueError("Keine Embeddings konnten geladen werden.")
+        logger.info("%d Embeddings geladen.", loaded)
+
+    def encode_text(self, text: str) -> Any:
+        torch = self._torch
+        tokens = self.tokenizer([text])
+        tokens = tokens.to(self.device)
+        with torch.no_grad():
+            text_features = self.model.encode_text(tokens).float()
+        text_features = text_features / text_features.norm(dim=-1, keepdim=True)
+        return text_features[0]
+
+    def get_image_embedding(self, image_id: str) -> Any:
+        try:
+            return self._image_embeddings[image_id]
+        except KeyError as exc:
+            raise KeyError(f"Kein Embedding für Bild-ID {image_id} geladen.") from exc
+
+    def compute_similarity(self, text_embedding: Any, image_embedding: Any) -> float:
+        torch = self._torch
+        similarity = torch.matmul(text_embedding, image_embedding)
+        return float(similarity.item())
+
+    def score_text_for_image(self, text: str, image_id: str) -> tuple[float, int]:
+        text_embedding = self.encode_text(text)
+        image_embedding = self.get_image_embedding(image_id)
+        similarity = self.compute_similarity(text_embedding, image_embedding)
+        score = similarity_to_score(similarity)
+        return similarity, score
+
+
+__all__ = [
+    "ClipScorer",
+    "ImageEntry",
+    "load_image_entries",
+    "similarity_to_score",
+]

precompute_embeddings.py

@@ -0,0 +1,75 @@
+from __future__ import annotations
+
+import argparse
+import io
+import logging
+from pathlib import Path
+from typing import Iterable
+
+import numpy as np
+import requests
+import torch
+from PIL import Image
+
+from model import ClipScorer, ImageEntry, load_image_entries
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger("precompute_embeddings")
+
+
+def download_image(url: str) -> Image.Image:
+    response = requests.get(url, timeout=60)
+    response.raise_for_status()
+    image = Image.open(io.BytesIO(response.content)).convert("RGB")
+    return image
+
+
+def save_embedding(path: Path, embedding: torch.Tensor) -> None:
+    path.parent.mkdir(parents=True, exist_ok=True)
+    array = embedding.detach().cpu().numpy().astype(np.float32)
+    np.save(path, array)
+
+
+def compute_embeddings(entries: Iterable[ImageEntry], model_name: str = "ViT-B-32", pretrained: str = "openai") -> None:
+    scorer = ClipScorer(model_name=model_name, pretrained=pretrained)
+    processed = 0
+    for entry in entries:
+        if entry.clip_model != model_name:
+            logger.info(
+                "Überspringe %s, da clip_model %s nicht zum Modell %s passt.",
+                entry.image_id,
+                entry.clip_model,
+                model_name,
+            )
+            continue
+        if entry.embedding_path.exists():
+            logger.info("Embedding für %s existiert bereits (%s)", entry.image_id, entry.embedding_path)
+            continue
+        logger.info("Lade Bild %s", entry.image_id)
+        image = download_image(entry.image_url)
+        image_tensor = scorer.preprocess(image).unsqueeze(0).to(scorer.device)
+        with torch.no_grad():
+            features = scorer.model.encode_image(image_tensor).float()
+        features = features / features.norm(dim=-1, keepdim=True)
+        save_embedding(entry.embedding_path, features[0])
+        processed += 1
+        logger.info("Embedding für %s gespeichert (%s)", entry.image_id, entry.embedding_path)
+    logger.info("Fertig. %d Embeddings erzeugt.", processed)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser(description="Berechnet CLIP-Embeddings für die Bilder aus images.csv")
+    parser.add_argument("--csv", default="images.csv", help="Pfad zur images.csv")
+    parser.add_argument("--model-name", default="ViT-B-32", help="open-clip Modellname")
+    parser.add_argument("--pretrained", default="openai", help="Checkpoint/Pretraining-Name")
+    return parser.parse_args()
+
+
+def main() -> None:
+    args = parse_args()
+    entries = load_image_entries(Path(args.csv))
+    compute_embeddings(entries, model_name=args.model_name, pretrained=args.pretrained)
+
+
+if __name__ == "__main__":
+    main()

requirements.txt

@@ -0,0 +1,12 @@
+gradio>=4.29.0,<5.0.0
+open-clip-torch>=2.24.0
+torch>=2.1.0
+torchvision>=0.16.0
+numpy>=1.24.0
+pandas>=2.0.0
+sqlalchemy>=2.0.0
+psycopg2-binary>=2.9.0
+requests>=2.31.0
+pillow>=10.0.0
+python-dotenv>=1.0.0
+pytest>=7.4.0

runtime.txt

@@ -0,0 +1 @@
+python-3.10

tests/conftest.py

@@ -0,0 +1,6 @@
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parent.parent
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))

tests/test_db.py

@@ -0,0 +1,38 @@
+import db
+
+
+def test_db_roundtrip(tmp_path):
+    database_path = tmp_path / "test.db"
+    database_url = f"sqlite+pysqlite:///{database_path}"
+    db.configure_database(database_url)
+    db.init_db()
+
+    with db.session_scope() as session:
+        user = db.ensure_user(session, "TestUser")
+        db.create_score(
+            session,
+            user=user,
+            image_id="image_1",
+            score_value=777,
+            similarity=0.42,
+            text="Ein Testeintrag",
+        )
+        user_id = user.id
+
+    with db.session_scope() as session:
+        same_user = db.ensure_user(session, "testuser")
+        assert same_user.id == user_id
+        global_scores = db.get_global_top(session)
+        assert len(global_scores) == 1
+        assert global_scores[0].score == 777
+        image_scores = db.get_image_top(session, "image_1")
+        assert len(image_scores) == 1
+        user_scores = db.get_user_recent(session, db.normalize_username("TESTUSER"))
+        assert len(user_scores) == 1
+        rows = db.scores_to_rows(global_scores)
+        assert rows[0][0] == 1  # Rang
+        assert rows[0][1] == "TestUser"
+        assert rows[0][2] == "image_1"
+        assert rows[0][3] == 777
+        assert rows[0][4] == round(0.42, 4)
+        assert "Ein Testeintrag" in rows[0][5]

tests/test_model.py

@@ -0,0 +1,14 @@
+from model import similarity_to_score
+
+
+def test_similarity_to_score_boundaries():
+    assert similarity_to_score(-1.0) == 0
+    assert similarity_to_score(1.0) == 1000
+    assert similarity_to_score(0.0) == 500
+
+
+def test_similarity_to_score_clipping_and_rounding():
+    assert similarity_to_score(1.2) == 1000
+    assert similarity_to_score(-1.3) == 0
+    assert similarity_to_score(0.1234) == 562
+    assert similarity_to_score(-0.3333) == 333