scripts/create_index.py

#!/usr/bin/env python3
"""
Bootstrap script: create the Pinecone integrated embedding index.

This script pins the embedding model and vector dimension EXPLICITLY in code so
the index is reproducible from configuration, not console clicks.  Model and
dimension come from Settings (app.core.config) — the single source of truth
shared with the running application.

Safety contract
---------------
- IDEMPOTENT by default: if the index already exists, the script exits with a
  clear message and does NOT modify or overwrite it.
- Destructive recreate is behind an explicit double opt-in (--recreate AND
  --confirm-recreate) and prints a loud warning before proceeding.

Usage
-----
  # Create index (safe — skips if already exists)
  python scripts/create_index.py

  # Specify a different index name or cloud region
  python scripts/create_index.py --index-name my-index --cloud aws --region us-east-1

  # Recreate (DESTRUCTIVE — deletes the existing index first)
  python scripts/create_index.py --recreate --confirm-recreate

Requirements
------------
  PINECONE_API_KEY, PINECONE_INDEX_NAME (or --index-name), PINECONE_HOST
  (PINECONE_HOST is only needed at runtime, not for index creation).
  Set via environment variables or backend/.env.
"""

from __future__ import annotations

import argparse
import sys
import time
from pathlib import Path

# ---------------------------------------------------------------------------
# Path setup — allow importing from backend/app/
# ---------------------------------------------------------------------------
_REPO_ROOT = Path(__file__).resolve().parent.parent
_BACKEND_DIR = _REPO_ROOT / "backend"
sys.path.insert(0, str(_BACKEND_DIR))

try:
    from dotenv import load_dotenv
    load_dotenv(_BACKEND_DIR / ".env", override=False)
except ImportError:
    pass

from app.core.config import get_settings  # noqa: E402


# ---------------------------------------------------------------------------
# Constants — sourced from Settings; documented here for readability
# ---------------------------------------------------------------------------
# Cloud/region defaults.  These match the serverless config assumed during
# initial setup; override via CLI flags if your index is in a different region.
_DEFAULT_CLOUD = "aws"
_DEFAULT_REGION = "us-east-1"


def _describe_existing(pc: object, index_name: str) -> bool:
    """Return True if the index already exists, False otherwise."""
    try:
        existing = [idx.name for idx in pc.list_indexes().indexes]  # type: ignore[attr-defined]
        return index_name in existing
    except Exception as exc:  # noqa: BLE001
        print(f"  [WARN] Could not list indexes: {exc}", file=sys.stderr)
        return False


def _create(pc: object, index_name: str, settings: object, cloud: str, region: str) -> None:
    """Issue the create_index_for_model call with explicit model + dimension."""
    # Settings fields are the single source of truth — do not inline magic numbers here.
    model: str = settings.PINECONE_EMBED_MODEL          # type: ignore[attr-defined]
    dimension: int = settings.PINECONE_EMBED_DIMENSION  # type: ignore[attr-defined]
    text_field: str = settings.PINECONE_TEXT_FIELD      # type: ignore[attr-defined]

    print(f"  Creating index '{index_name}' …")
    print(f"    model     = {model}")
    print(f"    dimension = {dimension}")
    print(f"    metric    = cosine")
    print(f"    text_field= {text_field}  (field_map key)")
    print(f"    cloud     = {cloud}  region = {region}")

    pc.create_index_for_model(  # type: ignore[attr-defined]
        name=index_name,
        cloud=cloud,
        region=region,
        embed={
            "model": model,
            "field_map": {"text": text_field},
            "metric": "cosine",
            "dimension": dimension,
        },
    )
    print("  Waiting for index to become ready …")
    for attempt in range(30):
        try:
            desc = pc.describe_index(index_name)  # type: ignore[attr-defined]
            ready = getattr(getattr(desc, "status", None), "ready", False)
            if ready:
                print(f"  Index '{index_name}' is ready.")
                return
        except Exception:  # noqa: BLE001
            pass
        time.sleep(5)
        print(f"  … still waiting ({(attempt + 1) * 5}s elapsed)")
    print("  [WARN] Timed out waiting for index to become ready. Check the Pinecone console.")


def _delete(pc: object, index_name: str) -> None:
    print(f"  Deleting index '{index_name}' …", flush=True)
    pc.delete_index(index_name)  # type: ignore[attr-defined]
    # Brief pause to let the control-plane propagate the deletion.
    time.sleep(5)
    print(f"  Index '{index_name}' deleted.")


def main() -> None:
    parser = argparse.ArgumentParser(
        description="Create the Pinecone integrated embedding index (idempotent by default).",
    )
    parser.add_argument(
        "--index-name",
        default=None,
        help="Pinecone index name.  Defaults to PINECONE_INDEX_NAME from settings.",
    )
    parser.add_argument(
        "--cloud",
        default=_DEFAULT_CLOUD,
        help=f"Serverless cloud provider (default: {_DEFAULT_CLOUD}).",
    )
    parser.add_argument(
        "--region",
        default=_DEFAULT_REGION,
        help=f"Serverless cloud region (default: {_DEFAULT_REGION}).",
    )
    parser.add_argument(
        "--recreate",
        action="store_true",
        default=False,
        help="DELETE the existing index and recreate it.  DESTRUCTIVE — also requires --confirm-recreate.",
    )
    parser.add_argument(
        "--confirm-recreate",
        action="store_true",
        default=False,
        help="Second opt-in required for --recreate.  Both flags must be present.",
    )
    args = parser.parse_args()

    settings = get_settings()
    index_name: str = args.index_name or settings.PINECONE_INDEX_NAME

    from pinecone import Pinecone  # noqa: PLC0415
    pc = Pinecone(api_key=settings.PINECONE_API_KEY)

    exists = _describe_existing(pc, index_name)

    # ------------------------------------------------------------------
    # Normal (non-destructive) path
    # ------------------------------------------------------------------
    if not args.recreate:
        if exists:
            desc = pc.describe_index(index_name)
            live_model = getattr(getattr(desc, "embed", None), "model", "unknown")
            live_dim = getattr(getattr(desc, "embed", None), "dimension", "unknown")
            print(
                f"Index '{index_name}' already exists — skipping creation.\n"
                f"  live model={live_model}  dimension={live_dim}\n"
                f"  Expected: model={settings.PINECONE_EMBED_MODEL}  "
                f"dimension={settings.PINECONE_EMBED_DIMENSION}\n"
                "To recreate it, run with --recreate --confirm-recreate (DESTRUCTIVE)."
            )
        else:
            _create(pc, index_name, settings, args.cloud, args.region)
        return

    # ------------------------------------------------------------------
    # Destructive recreate — requires BOTH flags
    # ------------------------------------------------------------------
    if not args.confirm_recreate:
        print(
            "ERROR: --recreate requires --confirm-recreate as a second opt-in.\n"
            "Both flags must be present to prevent accidental data loss.",
            file=sys.stderr,
        )
        sys.exit(1)

    print(
        f"\n*** WARNING: --recreate will DELETE '{index_name}' and all its data. ***\n"
        "This is irreversible.  Proceeding in 5 seconds …"
    )
    time.sleep(5)

    if exists:
        _delete(pc, index_name)
    else:
        print(f"  Index '{index_name}' does not exist — skipping delete step.")

    _create(pc, index_name, settings, args.cloud, args.region)


if __name__ == "__main__":
    main()