src/utils.py

"""
src/utils.py — Pure helper functions shared across the entire codebase.

All stateless — no I/O, no model calls, no side effects.
Import freely from any module.
"""

import hashlib
import re
from typing import Union
from urllib.parse import urlparse

import inflect
import numpy as np
from fastapi import Request

from .config import FACE_THRESHOLD_LOW

_inflect = inflect.engine()


# ════════════════════════════════════════════════════════════════════
#  VECTOR HELPERS
# ════════════════════════════════════════════════════════════════════

def to_list(v) -> list:
    """
    Convert a numpy array (or any object with .tolist()) to a plain Python list.
    Pinecone's SDK requires plain lists for vector values, not numpy arrays.
    """
    return v.tolist() if hasattr(v, "tolist") else v


# ════════════════════════════════════════════════════════════════════
#  SCORE NORMALIZATION
# ════════════════════════════════════════════════════════════════════

def face_ui_score(raw: float, n_faces: int = 1) -> float:
    """
    Map a raw Pinecone cosine score to a human-readable UI confidence (0.75–0.99).

    Why remap?  A raw cosine score of 0.36 (above threshold, genuine match) reads
    as "36% confident" to a user, which is misleading.  Remapping the valid match
    range [FACE_THRESHOLD_LOW, 1.0] onto [0.75, 0.99] communicates that anything
    shown IS a match; the percentage communicates relative quality within matches.

    Multi-face boost: +5 % per additional matched face, capped at 0.99.
    Rationale — an image containing 3 of 3 searched faces should rank above an
    image containing only 1 of 3.
    """
    lo   = FACE_THRESHOLD_LOW
    base = 0.75 + ((raw - lo) / (1.0 - lo)) * 0.24
    if n_faces > 1:
        base *= 1.0 + 0.05 * (n_faces - 1)
    return round(min(0.99, base), 4)


# ════════════════════════════════════════════════════════════════════
#  FILE / PATH HELPERS
# ════════════════════════════════════════════════════════════════════

def img_hash(image_path: str) -> str:
    """
    MD5 of the first 64 KB of a file — fast collision-resistant cache key.
    Reading the full file for every cache lookup on large images would be slow;
    the first 64 KB is almost always unique enough to distinguish images.
    """
    h = hashlib.md5()
    with open(image_path, "rb") as f:
        h.update(f.read(65536))
    return h.hexdigest()


def sanitize_filename(filename: str) -> str:
    """
    Make a filename safe for use in temp file paths.
    Replaces spaces with underscores; strips characters outside [a-zA-Z0-9._-].
    """
    return re.sub(r"[^\w.\-]", "", re.sub(r"\s+", "_", filename))


def standardize_category_name(name: str) -> str:
    """
    Normalize a user-supplied folder/category name for consistent storage:
      - Lowercase and underscore-separated
      - No special characters
      - Singularized (dogs → dog, shoes → shoe)

    Singularization ensures "Dog", "Dogs", "dogs" all map to the same folder.
    inflect.singular_noun() returns False for already-singular words,
    so `or clean` keeps the original in that case.
    """
    clean = re.sub(r"\s+", "_", name.strip().lower())
    clean = re.sub(r"[^\w]", "", clean)
    return _inflect.singular_noun(clean) or clean


# ════════════════════════════════════════════════════════════════════
#  CLOUDINARY HELPERS
# ════════════════════════════════════════════════════════════════════

def get_cloudinary_creds(env_url: str) -> dict:
    """
    Parse a Cloudinary Environment URL (cloudinary://key:secret@cloud_name)
    into separate credential components for SDK calls.
    Returns an empty dict for blank/invalid input.
    """
    if not env_url:
        return {}
    parsed = urlparse(env_url)
    return {
        "api_key":    parsed.username or "",
        "api_secret": parsed.password or "",
        "cloud_name": parsed.hostname or "",
    }


def cld_thumb_url(secure_url: str) -> str:
    """
    Inject a Cloudinary on-the-fly image transformation into a full-resolution
    URL, producing a 400×400 fill thumbnail for grid display.

    Cloudinary transformations are inserted between /upload/ and the version/path:
      .../image/upload/v123/folder/img.jpg
      → .../image/upload/w_400,h_400,c_fill,q_auto,f_auto/v123/folder/img.jpg

    Returns the original URL unchanged if transformation cannot be applied
    (e.g. non-standard URL format or already-transformed URL).
    """
    marker = "/image/upload/"
    idx    = secure_url.find(marker)
    if idx == -1:
        return secure_url
    base = secure_url[: idx + len(marker)]
    rest = secure_url[idx + len(marker):]
    if any(rest.startswith(p) for p in ("w_", "h_", "c_", "q_")):
        return secure_url   # already has a transformation block
    return f"{base}w_400,h_400,c_fill,q_auto,f_auto/{rest}"


def url_to_public_id(image_url: str) -> str:
    """
    Extract the Cloudinary public_id from a secure_url.
    Strips the version segment (vXXX) if present.
    Returns empty string on failure.

    Example:
      https://res.cloudinary.com/demo/image/upload/v1234/folder/img.jpg
      → folder/img
    """
    try:
        path  = urlparse(image_url).path
        parts = path.split("/")
        upload_idx = parts.index("upload")
        after = parts[upload_idx + 1:]
        if after and after[0].startswith("v") and after[0][1:].isdigit():
            after = after[1:]
        return "/".join(after).rsplit(".", 1)[0]
    except Exception:
        return ""


# ════════════════════════════════════════════════════════════════════
#  REQUEST HELPERS
# ════════════════════════════════════════════════════════════════════

def get_ip(request: Request) -> str:
    """
    Extract the real client IP from a FastAPI request.
    Respects the X-Forwarded-For header set by reverse proxies / CDNs.
    """
    xff = request.headers.get("X-Forwarded-For", "")
    return xff.split(",")[0].strip() if xff else getattr(request.client, "host", "unknown")


def is_default_key(key: str, default: str) -> bool:
    """Return True if `key` matches the shared demo key (guest mode)."""
    return bool(default) and key.strip() == default.strip()