mindnet/scripts/reset_qdrant.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
"""
Name:        scripts/reset_qdrant.py
Version:     v1.1.1 (2025-09-05)
Kurzbeschreibung:
  Sicheres Zurücksetzen der Qdrant-Collections für EIN Projektpräfix. Das Skript
  ermittelt zunächst die tatsächlich betroffenen Collections und zeigt eine
  Übersicht an. Anschließend wird der User um Bestätigung gebeten (interaktive
  Abfrage), bevor eine der beiden Aktionen ausgeführt wird:

    - wipe:     Löscht die Collections komplett und legt sie gemäß Projekt-Defaults neu an
    - truncate: Löscht nur alle Points (Inhalte), behält die Collection-Settings

Aufruf (aus Projekt-Root, im venv):
  python3 -m scripts.reset_qdrant --mode wipe [--prefix PREFIX] [--yes] [--dry-run]
  python3 -m scripts.reset_qdrant --mode truncate [--prefix PREFIX] [--yes] [--dry-run]

Parameter:
  --mode       wipe | truncate  (Pflicht)
  --prefix     Collection-Prefix (Default: env COLLECTION_PREFIX oder 'mindnet')
  --yes        Keine Rückfrage, direkt ausführen (CI/CD geeignet)
  --dry-run    Nur anzeigen, was passieren würde; nichts ändern

Umgebungsvariablen (optional):
  QDRANT_URL, QDRANT_API_KEY, COLLECTION_PREFIX, VECTOR_DIM (Default 384)

Sicherheitsmerkmale:
  - Betrachtet ausschließlich Collections des angegebenen Präfixes.
  - Listet vor Ausführung die tatsächlich existierenden Ziel-Collections auf.
  - Fragt interaktiv nach Bestätigung (es sei denn --yes ist gesetzt).

Exitcodes:
  0 = OK, 1 = abgebrochen/keine Aktion, 2 = Verbindungs-/Konfigurationsfehler

Changelog:
  v1.1.0: Interaktive Bestätigung, --yes/--dry-run hinzugefügt, Preview der betroffenen Collections.
  v1.0.0: Wipe/Truncate ohne Bestätigungsabfrage.
"""
from __future__ import annotations
import argparse
import os
import sys
from typing import List

from qdrant_client import QdrantClient
from qdrant_client.http import models as rest

from app.core.qdrant import QdrantConfig, get_client, ensure_collections, collection_names


def resolve_existing_collections(client: QdrantClient, prefix: str) -> List[str]:
    """Ermittelt die *existierenden* Collections für das übergebene Präfix.
    Es werden NUR die projektdefinierten Collections betrachtet (notes/chunks/edges).
    """
    notes, chunks, edges = collection_names(prefix)
    candidates = [notes, chunks, edges]
    existing = [c for c in candidates if client.collection_exists(c)]
    return existing


def confirm_or_abort(action: str, collections: List[str], nonexisting: List[str], assume_yes: bool) -> bool:
    print("\n=== Reset-Vorschau ===")
    print(f"Aktion:    {action}")
    if collections:
        print("Betroffen (existieren):")
        for c in collections:
            print(f"  - {c}")
    else:
        print("Betroffen (existieren):   — (keine)")

    if nonexisting:
        print("Nicht vorhanden (werden bei wipe ggf. neu angelegt):")
        for c in nonexisting:
            print(f"  - {c}")

    if assume_yes:
        print("\n--yes gesetzt → führe ohne Rückfrage aus.")
        return True

    try:
        ans = input("\nFortfahren? (yes/no): ").strip().lower()
    except EOFError:
        return False
    return ans in ("y", "yes")


def delete_all_points(client: QdrantClient, collections: List[str]) -> None:
    """Löscht *alle* Points in den angegebenen Collections.
    API-Kompatibilität:
      - neuere qdrant-client:   client.delete_points(...)
      - ältere qdrant-client:   client.delete(...)
    """
    match_all = rest.Filter(must=[])
    for col in collections:
        try:
            if hasattr(client, "delete_points"):
                client.delete_points(collection_name=col, points_selector=match_all, wait=True)
            else:
                client.delete(collection_name=col, points_selector=match_all, wait=True)
        except Exception as e:
            print(f"Fehler beim Löschen der Points in {col}: {e}", file=sys.stderr)
            raise


def wipe_collections(client: QdrantClient, all_col_names: List[str], existing: List[str]) -> None:
    # Lösche nur Collections, die wirklich existieren; die anderen werden anschließend neu angelegt
    for col in existing:
        client.delete_collection(col)
    # ensure_collections legt alle benötigten Collections (notes/chunks/edges) neu an


def main():
    ap = argparse.ArgumentParser(description="Wipe oder truncate mindnet-Collections in Qdrant (mit Bestätigung).")
    ap.add_argument("--mode", choices=["wipe", "truncate"], required=True,
                    help="wipe = Collections löschen & neu anlegen; truncate = nur Inhalte löschen")
    ap.add_argument("--prefix", help="Collection-Prefix (Default: env COLLECTION_PREFIX oder 'mindnet')")
    ap.add_argument("--yes", action="store_true", help="Ohne Rückfrage ausführen (CI/CD)")
    ap.add_argument("--dry-run", action="store_true", help="Nur anzeigen, was passieren würde; nichts ändern")
    args = ap.parse_args()

    # Qdrant-Konfiguration
    try:
        cfg = QdrantConfig.from_env()
    except Exception as e:
        print(f"Konfigurationsfehler: {e}", file=sys.stderr)
        sys.exit(2)

    if args.prefix:
        cfg.prefix = args.prefix

    # Client
    try:
        client = get_client(cfg)
    except Exception as e:
        print(f"Verbindungsfehler zu Qdrant: {e}", file=sys.stderr)
        sys.exit(2)

    # Ziel-Collections: existierende & nicht existierende (per Namenskonvention)
    notes, chunks, edges = collection_names(cfg.prefix)
    all_col_names = [notes, chunks, edges]
    existing = resolve_existing_collections(client, cfg.prefix)
    nonexisting = [c for c in all_col_names if c not in existing]

    # Preview & Bestätigung
    if not confirm_or_abort(args.mode, existing, nonexisting, args.yes):
        print("Abgebrochen – keine Änderungen vorgenommen.")
        sys.exit(1)

    if args.dry_run:
        print("Dry-Run – keine Änderungen vorgenommen.")
        sys.exit(0)

    # Ausführen
    if args.mode == "wipe":
        wipe_collections(client, all_col_names, existing)
        ensure_collections(client, cfg.prefix, cfg.dim)
        print(f"Wiped & recreated (Prefix={cfg.prefix}): {all_col_names}")
    else:
        if not existing:
            print("Keine existierenden Collections zum Truncaten gefunden. Beende ohne Aktion.")
            sys.exit(0)
        delete_all_points(client, existing)
        print(f"Truncated (deleted all points in): {existing}")


if __name__ == "__main__":
    main()