config.py

# photomap/backend/config.py
"""
Configuration management for PhotoMap backend.
This module handles loading, saving, and managing photo album configurations.
It uses a YAML file to store album details and provides methods to manipulate albums.
"""
import logging
import math
import os
import threading
from functools import lru_cache
from pathlib import Path
from typing import Any, Literal

import yaml
from platformdirs import user_config_dir, user_data_dir
from pydantic import BaseModel, Field, field_validator, model_validator

from .cluster_eps import MIN_CLUSTER_EPS
from .encoders import (
    LEGACY_CLIP_MIN_SEARCH_SCORE,
    LEGACY_ENCODER_SPEC,
    default_encoder_spec,
    default_min_search_score,
)
from .util import atomic_write_text

logger = logging.getLogger(__name__)


def default_board_index_path(album_key: str) -> Path:
    """Index location for albums that have no image directory of their own.

    InvokeAI-board albums can't store ``photomap_index`` next to their
    images (the images belong to InvokeAI), so their .npz lives in the
    per-user data directory instead, keyed by album.
    """
    # The key lands in a filesystem path, and keys are user input — refuse
    # anything that could escape the indexes directory.
    if (
        not album_key
        or "/" in album_key
        or "\\" in album_key
        or "\x00" in album_key
        or ".." in album_key
    ):
        raise ValueError(f"Album key not usable as a directory name: {album_key!r}")
    data_dir = Path(user_data_dir("photomap", "photomap"))
    return data_dir / "indexes" / album_key / "embeddings.npz"


def repaired_umap_eps(raw: Any) -> float | None:
    """A Cluster Strength read off disk, brought inside the bounds the model
    now enforces.

    The API refuses an unusable value before it can be written, but a config
    file written before that bound existed can hold anything, and the model
    would now refuse to load it — which would take the whole config down over
    one album's number. The two ways a stored value can be out of bounds want
    different answers:

    * Non-finite, zero or negative is not a strength anyone chose; it is
      damage. ``None`` — "nobody chose one" — hands the album back to a value
      derived from its own coordinates, the same treatment an album that never
      had one gets, and a better answer than any constant.
    * Positive but under the floor *is* an intent: the tightest clustering the
      control can ask for. ``resolve_cluster_eps`` already floors it at
      :data:`MIN_CLUSTER_EPS` before clustering, so raising it here changes
      nothing about the map — it only stops the field from reporting a number
      the map is not using.

    Anything that is not a number at all is left alone for the model to
    reject in its own words; this repairs values, it does not launder types.
    Being liberal about *what counts as* a number matters, though: the model
    refuses out-of-bounds values now, so any number this hands back unrepaired
    takes the whole config file down, not just one album's Cluster Strength.
    That is why ``bool`` is not special-cased — pydantic accepts one as a
    float, so ``umap_eps: false`` reaches the model as ``0.0``, and a config
    holding one has to load.
    """
    if raw is None:
        return None
    try:
        value = float(raw)
    except (TypeError, ValueError):
        return raw
    if not math.isfinite(value) or value <= 0:
        logger.warning(
            f"Ignoring unusable stored Cluster Strength {raw!r}; "
            "deriving one from the album's coordinates instead."
        )
        return None
    if value < MIN_CLUSTER_EPS:
        logger.warning(
            f"Raising stored Cluster Strength {raw!r} to the {MIN_CLUSTER_EPS} "
            "the semantic map clusters with anyway."
        )
        return MIN_CLUSTER_EPS
    return value


class Album(BaseModel):
    """Represents a photo album configuration."""

    key: str = Field(..., description="Unique album identifier")
    name: str = Field(..., description="Display name for the album")
    source_type: Literal["directory", "invokeai_board"] = Field(
        default="directory",
        description=(
            "Where the album's images come from: a local directory tree "
            "('directory', the default) or one or more InvokeAI gallery "
            "boards ('invokeai_board')."
        ),
    )
    image_paths: list[str] = Field(
        ..., min_length=1, description="List of paths containing images"
    )
    index: str = Field(..., description="Path to the embeddings index file")
    invokeai_url: str | None = Field(
        default=None,
        description="Base URL of the InvokeAI backend serving this album's boards",
    )
    invokeai_username: str | None = Field(
        default=None,
        description="Username for the InvokeAI backend (multi-user mode only)",
    )
    invokeai_password: str | None = Field(
        default=None,
        description="Password for the InvokeAI backend (multi-user mode only)",
    )
    invokeai_root: str | None = Field(
        default=None,
        description=(
            "Locally-accessible InvokeAI root directory; images resolve to "
            "<root>/outputs/images/<image_name>"
        ),
    )
    invokeai_board_ids: list[str] = Field(
        default_factory=list,
        description=(
            "InvokeAI board ids whose images make up this album. The special "
            "id 'none' is InvokeAI's Uncategorized bucket."
        ),
    )
    umap_eps: float | None = Field(
        default=None,
        ge=MIN_CLUSTER_EPS,
        allow_inf_nan=False,
        description=(
            "DBSCAN epsilon for the semantic map's clustering. None means "
            "'not chosen': the value is derived from the album's own UMAP "
            "coordinates (see backend/cluster_eps.py), which is what keeps "
            "small albums from clustering into nothing. Adjusting Cluster "
            "Strength in the UI stores a number here and the derived value "
            "is no longer consulted. A number stored here is bounded below "
            "by MIN_CLUSTER_EPS — the floor resolve_cluster_eps applies "
            "anyway — so the stored number and the number the map clusters "
            "with cannot disagree. A config written before that bound can "
            "still hold anything; see repaired_umap_eps."
        ),
    )
    description: str = Field(default="", description="Album description")
    encoder_spec: str = Field(
        # Resolved per-host: OpenCLIP ViT-L-14 on CUDA/macOS, lighter OpenAI CLIP
        # ViT-B/32 on CPU-only Linux/Windows. See encoders.default_encoder_spec.
        default_factory=default_encoder_spec,
        description=(
            "Image/text encoder spec. Format: '<backend>:<model>'. "
            "Examples: 'openai-clip:ViT-B/32' (legacy, CPU default), "
            "'open-clip:ViT-L-14/dfn2b', 'siglip:google/siglip2-large-patch16-256'. "
            "Changing this requires re-indexing the album."
        ),
    )
    # Per-album search controls. min_search_score defaults to None so a
    # validator can resolve it from the encoder at construction time — the
    # backends do not share a score scale, see ``default_min_search_score``.
    min_search_score: float | None = Field(
        default=None,
        description="Minimum similarity score below which results are filtered out.",
    )
    max_search_results: int = Field(
        default=100,
        description="Maximum number of search results to return.",
    )
    use_query_optimization: bool = Field(
        default=True,
        description=(
            "SigLIP only: ensemble text queries across modality templates. "
            "Ignored by CLIP/OpenCLIP."
        ),
    )
    min_image_dimension: int = Field(
        default=256,
        ge=1,
        description=(
            "Minimum image dimension (in pixels) for an image to be indexed. "
            "An image is kept only if both its width and height are >= this "
            "value. Filters out thumbnails and tiny images that contribute "
            "little to semantic search."
        ),
    )
    min_image_bytes: int = Field(
        default=8192,
        ge=0,
        description=(
            "Minimum file size (in bytes) for an image to be indexed; "
            "smaller files are rejected without being opened, which keeps "
            "scans fast on thumbnail-heavy or network-mounted libraries. "
            "0 disables the check. The 8192 default suits the default 256px "
            "min_image_dimension; lower both together."
        ),
    )

    @model_validator(mode="before")
    @classmethod
    def _derive_board_album_fields(cls, data: Any) -> Any:
        """Fill in `image_paths` and `index` for InvokeAI-board albums.

        Board albums have no user-chosen image directory: their media live
        under `<invokeai_root>/outputs/images` and `<invokeai_root>/outputs/videos`,
        and their index in the user data directory. Both must be derived
        *before* field validation because `image_paths` has `min_length=1`
        and `index` is required.

        The paths are recomputed on every construction rather than only when
        absent. They are derived, not user-chosen, so whatever was persisted
        must never win over the current root — and albums written before
        video support stored only the images directory, which is how those
        pick up the videos directory with no migration step. (`image_paths`
        is what gates file access and relative-path resolution, so a board
        video would otherwise be indexed but refused by `/videos/…`.)
        """
        if not isinstance(data, dict) or data.get("source_type") != "invokeai_board":
            return data
        root = data.get("invokeai_root")
        if root:
            outputs = Path(root).expanduser() / "outputs"
            data["image_paths"] = [
                str(outputs / "images"),
                str(outputs / "videos"),
            ]
        if not data.get("index") and data.get("key"):
            data["index"] = default_board_index_path(str(data["key"])).as_posix()
        return data

    @model_validator(mode="after")
    def _resolve_min_search_score(self) -> "Album":
        if self.min_search_score is None:
            self.min_search_score = default_min_search_score(self.encoder_spec)
        return self

    @model_validator(mode="after")
    def _validate_board_fields(self) -> "Album":
        if self.source_type != "invokeai_board":
            return self
        if not self.invokeai_url:
            raise ValueError("InvokeAI-board albums require invokeai_url")
        if not self.invokeai_url.startswith(("http://", "https://")):
            raise ValueError("invokeai_url must use http:// or https://")
        if not self.invokeai_root:
            raise ValueError("InvokeAI-board albums require invokeai_root")
        if not self.invokeai_board_ids:
            raise ValueError("InvokeAI-board albums require at least one board id")
        return self

    @field_validator("image_paths")
    @classmethod
    def expand_image_paths(cls, v: list[str]) -> list[str]:
        """Expand ``~`` in image paths."""
        return [str(Path(path).expanduser()) for path in v]

    @model_validator(mode="after")
    def _warn_about_missing_image_paths(self) -> "Album":
        """Warn about configured directories that are not on disk.

        Runs after the model is built rather than as a field validator so it
        can see ``source_type``: a board album's ``outputs/videos`` is derived
        from the InvokeAI root, and InvokeAI does not create that directory
        until it first writes a video. Warning about it would fire on every
        config load for a perfectly correct setup, so a board album's derived
        directories are exempt — for those, a wrong root surfaces as the
        pointed "none of the board files were found" error at index time,
        which is a better diagnosis than this warning could give.
        """
        if self.source_type == "invokeai_board":
            return self
        for path in self.image_paths:
            if not Path(path).exists():
                logger.warning(f"Image path does not exist: {path}")
        return self

    @field_validator("index")
    @classmethod
    def validate_index_path(cls, v: str) -> str:
        """Validate index path format."""
        index_path = Path(v).expanduser()
        if not index_path.suffix == ".npz":
            raise ValueError("Index file must have .npz extension")
        return index_path.as_posix()

    def to_dict(self) -> dict[str, Any]:
        """Convert album to dictionary format for YAML."""
        data = {
            "name": self.name,
            "source_type": self.source_type,
            "image_paths": self.image_paths,
            "index": self.index,
            "description": self.description,
            "encoder_spec": self.encoder_spec,
            "min_search_score": self.min_search_score,
            "max_search_results": self.max_search_results,
            "use_query_optimization": self.use_query_optimization,
            "min_image_dimension": self.min_image_dimension,
            "min_image_bytes": self.min_image_bytes,
        }
        # A derived Cluster Strength is written as an *absent* key rather
        # than an explicit null. Both mean the same thing to this codebase,
        # but versions before umap_eps became nullable parse a null into a
        # non-nullable float field and refuse to load the whole config —
        # i.e. writing null here would stop an older PhotoMapAI from
        # starting at all, on a config file the two share.
        if self.umap_eps is not None:
            data["umap_eps"] = self.umap_eps
        # Keep directory-album YAML free of irrelevant InvokeAI keys.
        if self.source_type == "invokeai_board":
            data["invokeai_url"] = self.invokeai_url
            data["invokeai_username"] = self.invokeai_username
            data["invokeai_password"] = self.invokeai_password
            data["invokeai_root"] = self.invokeai_root
            data["invokeai_board_ids"] = self.invokeai_board_ids
        return data

    @classmethod
    def from_dict(cls, key: str, data: dict[str, Any]) -> "Album":
        """Create Album from dictionary."""
        return cls(
            key=key,
            name=data.get("name", key.capitalize()),
            # Albums written before this field existed are directory albums.
            source_type=data.get("source_type", "directory"),
            image_paths=data.get("image_paths", []),
            index=data["index"],
            # Absent from the YAML means nobody chose one, which is exactly
            # what None asks the map to derive. The old 0.07 fallback here
            # was a number for *every* album regardless of size, which is
            # what left small albums with no clusters at all.
            #
            # This is the one door untrusted numbers come through — the API
            # refuses them now, but a file written before it did cannot be
            # allowed to make the whole config unloadable.
            umap_eps=repaired_umap_eps(data.get("umap_eps")),
            description=data.get("description", ""),
            # Legacy YAML albums predate the encoder_spec field; their indexes
            # were built with the original CLIP, so fall back to that to stay
            # cache-compatible. New albums get the host-resolved default
            # (encoders.default_encoder_spec) via the Album field default when
            # the frontend creates them.
            encoder_spec=data.get("encoder_spec", LEGACY_ENCODER_SPEC),
            min_search_score=data.get("min_search_score"),
            max_search_results=data.get("max_search_results", 100),
            use_query_optimization=data.get("use_query_optimization", True),
            min_image_dimension=data.get("min_image_dimension", 256),
            min_image_bytes=data.get("min_image_bytes", 8192),
            invokeai_url=data.get("invokeai_url"),
            invokeai_username=data.get("invokeai_username"),
            invokeai_password=data.get("invokeai_password"),
            invokeai_root=data.get("invokeai_root"),
            invokeai_board_ids=data.get("invokeai_board_ids", []),
        )


# Bumped when a load-time migration is added below. 1.1.0 re-resolves the
# score floor of albums that never chose one — see ``_migrate_score_floors``.
CONFIG_VERSION = "1.1.0"


def _version_tuple(version: str) -> tuple[int, ...]:
    """``"1.10.2"`` -> ``(1, 10, 2)``, for ordering config versions.

    Only ever called with a stamp read straight out of the YAML, before the
    :class:`Config` model has had a chance to reject it. Anything that is not
    ``x.y.z`` sorts as newer than every version this build knows, so a stamp
    it cannot read is never mistaken for an un-migrated old file — belt and
    braces, since ``Config.validate_version`` refuses to build the model at
    all a few lines later and the load fails outright.
    """
    try:
        return tuple(int(part) for part in version.split("."))
    except ValueError:
        return (999,)


# Albums already announced as migrated, so the notice below is printed once
# per album per process. The migration itself is re-applied on every load
# until a save persists it (see ``load_config``), and repeating the same
# three lines at every reload would read as a change happening over and over.
_announced_score_floor_migrations: set[tuple[str, float]] = set()


def _migrate_score_floors(albums: dict[str, "Album"]) -> None:
    """Re-resolve score floors left at the old blanket CLIP default.

    Before :func:`default_min_search_score` learned that OpenCLIP scores a
    tenth of a point below OpenAI CLIP, every non-SigLIP album was given
    ``0.2`` at creation — a floor above the entire match band of the encoder
    PhotoMapAI recommends, so those albums answer most searches with nothing
    at all. The value is stored per album, so fixing the default alone would
    only help albums created after the upgrade.

    Touches only the exact 0.2, and only where the album's encoder now
    resolves to something else. That is a heuristic, not a proof of
    authorship: nothing records *who* chose a floor, so a user who typed 0.2
    into an OpenCLIP album before upgrading is moved off it along with every
    album that merely inherited it. The log line below is what tells them,
    and the value is one edit away in the search dialog. Any other floor —
    including 0.2 on the encoder whose default it still is — is left alone.

    Runs only for a config written before this build (see ``CONFIG_VERSION``),
    so 0.2 remains a value the user can choose afterwards.
    """
    for key, album in albums.items():
        resolved = default_min_search_score(album.encoder_spec)
        if album.min_search_score == LEGACY_CLIP_MIN_SEARCH_SCORE != resolved:
            album.min_search_score = resolved
            if (key, resolved) not in _announced_score_floor_migrations:
                _announced_score_floor_migrations.add((key, resolved))
                logger.info(
                    "Album %r: search score floor %.3f -> %.3f, the measured "
                    "default for %s (searches at the old floor returned almost "
                    "nothing). Adjust it in the search dialog if you preferred "
                    "the old value.",
                    key,
                    LEGACY_CLIP_MIN_SEARCH_SCORE,
                    resolved,
                    album.encoder_spec,
                )


class Config(BaseModel):
    """Main configuration model."""

    config_version: str = Field(
        CONFIG_VERSION, description="Configuration format version"
    )
    albums: dict[str, Album] = Field(
        default_factory=dict, description="Album configurations"
    )
    locationiq_api_key: str | None = Field(
        default=None, description="LocationIQ API key for map services"
    )
    invokeai_url: str | None = Field(
        default=None,
        description="Base URL of a running InvokeAI backend, e.g. http://localhost:9090",
    )
    invokeai_username: str | None = Field(
        default=None,
        description="Username for authenticating against InvokeAI (future use)",
    )
    invokeai_password: str | None = Field(
        default=None,
        description="Password for authenticating against InvokeAI (future use)",
    )
    invokeai_board_id: str | None = Field(
        default=None,
        description=(
            "ID of the InvokeAI board that reference-image uploads should be "
            "placed in. None means Uncategorized."
        ),
    )
    encoder_idle_timeout_seconds: float = Field(
        default=30.0,
        description=(
            "Seconds of search inactivity before the cached encoder is moved "
            "from GPU VRAM to system RAM. Reload from RAM on the next query is "
            "fast (no disk/network fetch). Set to 0 to disable offloading."
        ),
    )

    @field_validator("encoder_idle_timeout_seconds")
    @classmethod
    def validate_encoder_idle_timeout_seconds(cls, v: float) -> float:
        if v < 0:
            raise ValueError("encoder_idle_timeout_seconds must be >= 0")
        return float(v)

    @field_validator("config_version")
    @classmethod
    def validate_version(cls, v: str) -> str:
        """Validate configuration version format."""
        try:
            # Simple version validation - should be in format x.y.z
            parts = v.split(".")
            if len(parts) != 3 or not all(part.isdigit() for part in parts):
                raise ValueError("Version must be in format x.y.z")
        except Exception as e:
            raise ValueError("Invalid version format") from e
        return v

    @model_validator(mode="after")
    def validate_albums(self) -> "Config":
        """Validate albums configuration."""
        if not self.albums:
            print("Warning: No albums configured")
        return self

    def to_dict(self) -> dict[str, Any]:
        """Convert config to dictionary format for YAML."""
        return {
            "config_version": self.config_version,
            "albums": {key: album.to_dict() for key, album in self.albums.items()},
            "locationiq_api_key": self.locationiq_api_key,
            "invokeai_url": self.invokeai_url,
            "invokeai_username": self.invokeai_username,
            "invokeai_password": self.invokeai_password,
            "invokeai_board_id": self.invokeai_board_id,
            "encoder_idle_timeout_seconds": self.encoder_idle_timeout_seconds,
        }


class ConfigManager:
    """Manages PhotoMap configuration file with Pydantic validation."""

    def __init__(self, config_path: Path | None = None):
        """Initialize configuration manager.

        Args:
            config_path: Optional custom path to config file. If None, uses platform default.
        """
        self.config_path = config_path or self._get_default_config_path()
        self._config: Config | None = None
        # Re-entrant so a mutator can call load_config() and save_config()
        # inside the same critical section. Without this lock, two concurrent
        # API requests can clobber each other's writes when both read-modify-
        # write the YAML config.
        self._lock = threading.RLock()

    def _get_default_config_path(self) -> Path:
        """Get platform-specific default configuration file path."""
        # Try environment variable first
        if "PHOTOMAP_CONFIG" in os.environ:
            return Path(os.environ["PHOTOMAP_CONFIG"])

        # Use platformdirs for cross-platform config directory
        config_dir = Path(user_config_dir("photomap", "photomap"))
        config_dir.mkdir(parents=True, exist_ok=True)
        return config_dir / "config.yaml"

    def get_locationiq_api_key(self) -> str | None:
        """Get the LocationIQ API key."""
        if os.environ.get(
            "PHOTOMAP_ALBUM_LOCKED"
        ):  # In locked-down environments, do not leak the locationIQ key
            return
        config = self.load_config()
        return config.locationiq_api_key

    def set_locationiq_api_key(self, api_key: str | None) -> None:
        """Set the LocationIQ API key.

        Args:
            api_key: API key string or None to remove
        """
        if os.environ.get(
            "PHOTOMAP_ALBUM_LOCKED"
        ):  # In locked-down environments, do not allow changing the locationIQ key
            raise PermissionError("Album configuration is locked.")
        with self._lock:
            config = self.load_config()
            # Strip whitespace and treat empty strings as None
            config.locationiq_api_key = (
                api_key.strip() if api_key and api_key.strip() else None
            )
            self._config = config
            self.save_config()
            # Clear cache after saving to ensure fresh reads
            self._config = None

    def get_invokeai_settings(self) -> dict[str, str | None]:
        """Return the configured InvokeAI connection settings."""
        config = self.load_config()
        return {
            "url": config.invokeai_url,
            "username": config.invokeai_username,
            "password": config.invokeai_password,
            "board_id": config.invokeai_board_id,
        }

    def set_invokeai_settings(
        self,
        url: str | None = None,
        username: str | None = None,
        password: str | None = None,
        board_id: str | None = None,
    ) -> None:
        """Update the InvokeAI connection settings.

        Empty strings are normalized to None so that the fields can be cleared.
        """

        def _clean(value: str | None) -> str | None:
            if value is None:
                return None
            value = value.strip()
            return value or None

        with self._lock:
            config = self.load_config()
            config.invokeai_url = _clean(url)
            config.invokeai_username = _clean(username)
            config.invokeai_password = _clean(password)
            config.invokeai_board_id = _clean(board_id)
            self._config = config
            self.save_config()
            self._config = None

    def load_config(self) -> Config:
        """Load configuration from YAML file."""
        with self._lock:
            if self._config is None:
                if not self.config_path.exists():
                    self._config = Config(
                        config_version=CONFIG_VERSION,
                        albums={},
                        locationiq_api_key=None,
                    )
                else:
                    try:
                        with open(self.config_path) as f:
                            config_data = yaml.safe_load(f)

                        # Convert album dictionaries to Album objects
                        albums = {}
                        for key, album_data in config_data.get("albums", {}).items():
                            albums[key] = Album.from_dict(key, album_data)

                        extra: dict[str, Any] = {}
                        if "encoder_idle_timeout_seconds" in config_data:
                            extra["encoder_idle_timeout_seconds"] = config_data[
                                "encoder_idle_timeout_seconds"
                            ]
                        # Migrations run once, against a config older than
                        # this build. Re-running them on every load would make
                        # the migrated-away value permanently unsettable — a
                        # user typing 0.2 back into an OpenCLIP album would
                        # watch the next read undo it.
                        stored_version = config_data.get("config_version", "1.0.0")
                        if _version_tuple(stored_version) < _version_tuple(
                            CONFIG_VERSION
                        ):
                            _migrate_score_floors(albums)
                            stored_version = CONFIG_VERSION
                        # A config stamped *newer* than this build keeps its
                        # stamp: rewriting it as 1.1.0 would tell the build
                        # that wrote it that its own migrations had run.
                        self._config = Config(
                            config_version=stored_version,
                            albums=albums,
                            locationiq_api_key=config_data.get("locationiq_api_key"),
                            invokeai_url=config_data.get("invokeai_url"),
                            invokeai_username=config_data.get("invokeai_username"),
                            invokeai_password=config_data.get("invokeai_password"),
                            invokeai_board_id=config_data.get("invokeai_board_id"),
                            **extra,
                        )
                        # Deliberately not saved here. Loading a config is a
                        # read, and a rewrite drops comments and any key this
                        # build does not know — an unprompted one, on every
                        # user's first start after upgrading, is not a trade
                        # worth making. The new stamp rides along with the
                        # next save the user actually causes, and until then
                        # the migration is simply re-applied in memory.

                    except Exception as e:
                        raise RuntimeError(
                            f"Failed to load configuration from {self.config_path}: {e}"
                        ) from e

            return self._config

    def save_config(self):
        """Save current configuration to file."""
        with self._lock:
            if self._config is None:
                raise RuntimeError("No configuration loaded")

            try:
                payload = yaml.safe_dump(
                    self._config.to_dict(), default_flow_style=False, indent=2
                )
                atomic_write_text(self.config_path, payload)
            except Exception as e:
                raise RuntimeError(
                    f"Failed to save configuration to {self.config_path}: {e}"
                ) from e

    def get_albums(self) -> dict[str, Album]:
        """Get all albums."""
        config = self.load_config()
        return config.albums.copy()

    def get_album(self, key: str) -> Album | None:
        """Get a specific album by key."""
        albums = self.get_albums()
        return albums.get(key)

    def add_album(self, album: Album) -> bool:
        """Add a new album.

        Args:
            album: Album object to add

        Returns:
            True if added successfully, False if key already exists
        """
        with self._lock:
            config = self.load_config()

            if album.key in config.albums:
                return False

            config.albums[album.key] = album
            self._config = config
            try:
                self.save_config()
            finally:
                # Invalidated even when the save raised. The cached config now
                # holds a change that is not on disk, and leaving it in place
                # lets a later, unrelated save write out an edit the caller
                # was told had failed — a stored Cluster Strength that came
                # back from the dead when the album was next renamed, for one.
                self._config = None  # Clear cache to ensure fresh reads
            return True

    def update_album(self, album: Album) -> bool:
        """Update an existing album.

        Args:
            album: Album object with updated data

        Returns:
            True if updated successfully, False if key doesn't exist
        """
        with self._lock:
            config = self.load_config()

            if album.key not in config.albums:
                return False

            config.albums[album.key] = album
            self._config = config
            try:
                self.save_config()
            finally:
                # Invalidated even when the save raised. The cached config now
                # holds a change that is not on disk, and leaving it in place
                # lets a later, unrelated save write out an edit the caller
                # was told had failed — a stored Cluster Strength that came
                # back from the dead when the album was next renamed, for one.
                self._config = None  # Clear cache to ensure fresh reads
            return True

    def delete_album(self, key: str) -> bool:
        """Delete an album by key.

        Args:
            key: Album key to delete

        Returns:
            True if deleted successfully, False if key doesn't exist
        """
        with self._lock:
            config = self.load_config()

            if key not in config.albums:
                return False

            del config.albums[key]
            self._config = config
            try:
                self.save_config()
            finally:
                # Invalidated even when the save raised. The cached config now
                # holds a change that is not on disk, and leaving it in place
                # lets a later, unrelated save write out an edit the caller
                # was told had failed — a stored Cluster Strength that came
                # back from the dead when the album was next renamed, for one.
                self._config = None  # Clear cache to ensure fresh reads
            return True

    def get_photo_albums_dict(self) -> dict[str, str]:
        """Get albums in the old PHOTO_ALBUMS format for backward compatibility.

        Returns:
            Dictionary mapping album keys to their first image path
        """
        albums = self.get_albums()
        return {
            key: album.image_paths[0] if album.image_paths else ""
            for key, album in albums.items()
        }

    def find_image_in_album(self, album_key: str, relative_path: str) -> Path | None:
        """Find the full path of an image in any of the album's image paths.

        Args:
            album_key: Album key
            relative_path: Relative path to the image

        Returns:
            Full path to the image if found, None otherwise
        """
        album = self.get_album(album_key)
        if not album:
            return None

        for image_path in album.image_paths:
            full_path = Path(image_path) / relative_path
            if full_path.exists():
                return full_path

        # If not found, return path from first directory (for error handling)
        if album.image_paths:
            return Path(album.image_paths[0]) / relative_path
        return None

    def get_relative_path(self, full_path: str, album_key: str) -> str | None:
        """Get relative path of image within album's image paths.

        Args:
            full_path: Full path to the image
            album_key: Album key

        Returns:
            Relative path if found, None otherwise
        """
        album = self.get_album(album_key)
        if not album:
            return None

        fp = Path(full_path)
        for image_path in [Path(x).resolve() for x in album.image_paths]:
            try:
                return fp.relative_to(image_path).as_posix()
            except ValueError:
                continue

        # If not found in any path, return the filename
        return fp.name

    def validate_config(self) -> bool:
        """Validate the current configuration.

        Returns:
            True if configuration is valid
        """
        try:
            self.load_config()
            return True
        except Exception as e:
            print(f"Configuration validation failed: {e}")
            return False

    def has_albums(self) -> bool:
        """Check if any albums are configured.

        Returns:
            True if at least one album exists, False otherwise
        """
        config = self.load_config()
        return len(config.albums) > 0

    def is_first_run(self) -> bool:
        """Check if this is the first run (no config file and no albums).

        Returns:
            True if this appears to be the first run
        """
        return not self.config_path.exists() or not self.has_albums()

    def reload_config(self) -> Config:
        """Force reload configuration from file, clearing any cached data.

        Returns:
            Freshly loaded Config object
        """
        self._config = None  # Clear the cache
        return self.load_config()  # This will now re-read from file


# Convenience functions for creating albums
def create_album(
    key: str,
    name: str,
    image_paths: list[str] | None,
    index: str | None,
    umap_eps: float | None = None,
    description: str = "",
    encoder_spec: str | None = None,
    min_search_score: float | None = None,
    max_search_results: int | None = None,
    use_query_optimization: bool | None = None,
    min_image_dimension: int | None = None,
    min_image_bytes: int | None = None,
    source_type: str = "directory",
    invokeai_url: str | None = None,
    invokeai_username: str | None = None,
    invokeai_password: str | None = None,
    invokeai_root: str | None = None,
    invokeai_board_ids: list[str] | None = None,
) -> Album:
    """Create a new Album instance with validation.

    Each optional parameter falls through to the Album default when omitted,
    so existing callers don't have to thread every field. ``image_paths``
    and ``index`` may be None for InvokeAI-board albums — the Album model
    derives them from ``invokeai_root`` and the album key.
    """
    image_paths = [str(Path(x).expanduser().resolve()) for x in image_paths or []]
    fields: dict[str, object] = {
        "key": key,
        "name": name,
        "source_type": source_type,
        "image_paths": image_paths,
        "umap_eps": umap_eps,
        "description": description,
        "invokeai_url": invokeai_url,
        "invokeai_username": invokeai_username,
        "invokeai_password": invokeai_password,
        "invokeai_root": invokeai_root,
        "invokeai_board_ids": invokeai_board_ids or [],
    }
    if index is not None:
        fields["index"] = str(Path(index).expanduser().resolve())
    if encoder_spec is not None:
        fields["encoder_spec"] = encoder_spec
    if min_search_score is not None:
        fields["min_search_score"] = min_search_score
    if max_search_results is not None:
        fields["max_search_results"] = max_search_results
    if use_query_optimization is not None:
        fields["use_query_optimization"] = use_query_optimization
    if min_image_dimension is not None:
        fields["min_image_dimension"] = min_image_dimension
    if min_image_bytes is not None:
        fields["min_image_bytes"] = min_image_bytes
    return Album(**fields)


@lru_cache(maxsize=1)
def get_config_manager(config_path: Path | None = None) -> ConfigManager:
    """Get a singleton instance of ConfigManager."""
    return ConfigManager(config_path=config_path)