"""Real app logos for stacks, fetched once and then served from disk.

A stack called ``jellyfin`` should show *the Jellyfin logo*, not a generic
clapperboard. The logos come from the selfh.st icon set (~2900 self-hosted
apps), which is the same catalog Homarr, Homepage and Dashy draw on.

Everything crosses the network exactly once and on the server:

* the **catalog** (a JSON index of slugs and display names) is downloaded on
  startup and refreshed weekly into ``${DATA_DIR}/stack-icons/catalog.json``,
* a **logo** is downloaded the first time something asks for it and cached at
  ``${DATA_DIR}/stack-icons/logos/<slug>.png``, keyed by slug rather than by
  stack so ten Postgres stacks share one file.

Browsers therefore never talk to the CDN: they fetch logos from StackPilot's
own authenticated icon endpoint, like an uploaded image. After the first fetch
the whole feature works offline, and an installation with no outbound internet
degrades to the built-in glyphs rather than breaking — every entry point here
returns None instead of raising when the network is not there.

Matching a stack to a slug is deliberately conservative: an exact name, then the
name with punctuation rearranged, then the longest run of words inside it, then
the images its compose file pulls (``lscr.io/linuxserver/jellyfin:latest`` →
``jellyfin``). A stack whose name means nothing to the catalog gets no logo and
falls back to the keyword-derived glyph in the frontend.
"""
from __future__ import annotations

import asyncio
import json
import logging
import os
import re
import time
from typing import Iterable, Optional

import httpx
import yaml

from config import settings
from services import compose_service

logger = logging.getLogger("stackpilot.logos")

CATALOG_URL = "https://cdn.jsdelivr.net/gh/selfhst/icons/index.json"
LOGO_URL = "https://cdn.jsdelivr.net/gh/selfhst/icons/png/{slug}.png"

#: The catalog gains a handful of apps a week; there is nothing to gain from
#: checking more often, and a failed refresh simply keeps the previous copy.
CATALOG_TTL = 7 * 24 * 3600
CATALOG_MAX_BYTES = 8 * 1024 * 1024
LOGO_MAX_BYTES = 2 * 1024 * 1024
TIMEOUT = httpx.Timeout(15.0)

_SLUG_RE = re.compile(r"^[a-z0-9][a-z0-9-]{0,63}$")

#: Docker image names that are not what the catalog calls the app. Kept short on
#: purpose — this is for the cases the matcher genuinely cannot reach, not a
#: second catalog.
ALIASES = {
    "postgres": "postgresql",
    "pgsql": "postgresql",
    "mongo": "mongodb",
    "trilium": "trilium-notes",
    "wg-easy": "wireguard",
    "wg": "wireguard",
    "homeassistant": "home-assistant",
    "hass": "home-assistant",
    "pihole": "pi-hole",
    "openwebui": "open-webui",
    "nextcloud-aio": "nextcloud",
    "paperless": "paperless-ngx",
    "paperless-ng": "paperless-ngx",
    "code-server": "coder",
    "filebrowser": "file-browser",
    "qbit": "qbittorrent",
    "sab": "sabnzbd",
}

#: Image name components that say nothing about the app.
_IMAGE_NOISE = {
    "latest", "linuxserver", "lscr", "ghcr", "docker", "io", "com", "library",
    "hotio", "alpine", "amd64", "arm64v8", "bitnami", "official",
}


# --------------------------------------------------------------------------- #
# Paths
# --------------------------------------------------------------------------- #


def _root() -> str:
    return os.path.join(settings.DATA_DIR, "stack-icons")


def catalog_path() -> str:
    return os.path.join(_root(), "catalog.json")


def logo_dir() -> str:
    return os.path.join(_root(), "logos")


def logo_path(slug: str) -> Optional[str]:
    """Where a slug's logo is cached, or None if the slug is malformed.

    The slug reaches this from the database and from query strings, and it is
    about to become a filename.
    """
    if not _SLUG_RE.match(slug or ""):
        return None
    return os.path.join(logo_dir(), f"{slug}.png")


# --------------------------------------------------------------------------- #
# The catalog
# --------------------------------------------------------------------------- #


class Catalog:
    """Slug lookups built once per catalog file."""

    def __init__(self, entries: list[dict]):
        self.entries = entries
        self.slugs: set[str] = set()
        self.by_name: dict[str, str] = {}
        self.compact: dict[str, str] = {}
        for entry in entries:
            slug = (entry.get("Reference") or "").strip().lower()
            if not _SLUG_RE.match(slug):
                continue
            self.slugs.add(slug)
            # "AdGuard Home" → "adguard home", so a stack named that matches
            # even though the slug is hyphenated.
            name = _normalize(entry.get("Name") or "")
            self.by_name.setdefault(name, slug)
            # "pihole" → "pi-hole": people drop the punctuation the catalog keeps.
            self.compact.setdefault(slug.replace("-", ""), slug)
            self.compact.setdefault(name.replace(" ", ""), slug)

    def display_name(self, slug: str) -> str:
        for entry in self.entries:
            if (entry.get("Reference") or "").lower() == slug:
                return entry.get("Name") or slug
        return slug

    def __len__(self) -> int:
        return len(self.slugs)


_catalog: Optional[Catalog] = None
_catalog_mtime: float = 0.0


def load_catalog() -> Optional[Catalog]:
    """The cached catalog, re-read only when the file on disk changed."""
    global _catalog, _catalog_mtime
    path = catalog_path()
    try:
        mtime = os.path.getmtime(path)
    except OSError:
        return _catalog  # never downloaded, or removed under us
    if _catalog is not None and mtime == _catalog_mtime:
        return _catalog
    try:
        with open(path, "r", encoding="utf-8") as fh:
            entries = json.load(fh)
    except (OSError, json.JSONDecodeError) as exc:
        logger.warning("Icon catalog is unreadable (%s); ignoring it", exc)
        return _catalog
    if not isinstance(entries, list):
        return _catalog
    _catalog = Catalog(entries)
    _catalog_mtime = mtime
    logger.info("Loaded %d app logos from the icon catalog", len(_catalog))
    return _catalog


def catalog_age() -> Optional[float]:
    try:
        return time.time() - os.path.getmtime(catalog_path())
    except OSError:
        return None


async def refresh_catalog(force: bool = False) -> bool:
    """Download the catalog unless the copy on disk is still fresh."""
    age = catalog_age()
    if not force and age is not None and age < CATALOG_TTL:
        return False
    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=TIMEOUT) as client:
            response = await client.get(CATALOG_URL)
            response.raise_for_status()
            if len(response.content) > CATALOG_MAX_BYTES:
                raise ValueError("catalog is implausibly large")
            entries = response.json()
        if not isinstance(entries, list) or not entries:
            raise ValueError("catalog is not a non-empty list")
    except (httpx.HTTPError, ValueError, json.JSONDecodeError) as exc:
        # No internet is a normal state for a self-hosted box. Say so once and
        # carry on with the built-in glyphs.
        logger.info("Could not refresh the app icon catalog: %s", exc)
        return False
    os.makedirs(_root(), exist_ok=True)
    tmp = catalog_path() + ".tmp"
    with open(tmp, "w", encoding="utf-8") as fh:
        json.dump(entries, fh)
    os.replace(tmp, catalog_path())
    load_catalog()
    return True


async def catalog_loop() -> None:
    """Keep the catalog fresh for as long as the app runs."""
    while True:
        try:
            await refresh_catalog()
        except Exception as exc:  # noqa: BLE001 - a background loop may not die
            logger.warning("Icon catalog refresh failed: %s", exc)
        await asyncio.sleep(CATALOG_TTL)


# --------------------------------------------------------------------------- #
# Matching a stack to a slug
# --------------------------------------------------------------------------- #


def _normalize(text: str) -> str:
    return re.sub(r"[^a-z0-9]+", " ", text.lower()).strip()


def _lookup(catalog: Catalog, phrase: str) -> Optional[str]:
    """A slug for one normalized phrase, trying every spelling of it."""
    if not phrase:
        return None
    hyphenated = phrase.replace(" ", "-")
    squashed = phrase.replace(" ", "")
    if (alias := ALIASES.get(hyphenated)) and alias in catalog.slugs:
        return alias
    if hyphenated in catalog.slugs:
        return hyphenated
    if phrase in catalog.by_name:
        return catalog.by_name[phrase]
    if squashed in catalog.compact:
        return catalog.compact[squashed]
    return None


def match_slug(name: str, images: Iterable[str] = ()) -> Optional[str]:
    """The catalog slug a stack's name (then its images) points at."""
    catalog = load_catalog()
    if not catalog:
        return None

    phrase = _normalize(name)
    if (slug := _lookup(catalog, phrase)):
        return slug

    # The app's name inside a longer one ("my jellyfin stack", "medien plex").
    # Longest run of words first, so "home assistant" beats "home".
    tokens = phrase.split()
    for size in range(len(tokens), 0, -1):
        for start in range(len(tokens) - size + 1):
            gram = tokens[start : start + size]
            # A single short word is far more likely to be a coincidence than
            # an app ("app", "web", "db" are all slugs somewhere).
            if size == 1 and len(gram[0]) < 4:
                continue
            if (slug := _lookup(catalog, " ".join(gram))):
                return slug

    # Nothing in the name: ask what the stack actually runs.
    for image in images:
        if (slug := _lookup(catalog, _normalize(image))):
            return slug
    return None


def images_for(stack_id: str) -> list[str]:
    """Image names a stack's compose file pulls, most specific part first.

    ``lscr.io/linuxserver/jellyfin:latest`` contributes ``jellyfin``: the tag,
    the registry and the vendor namespace say nothing about which app it is.
    """
    directory = compose_service.stack_dir(stack_id)
    compose_file = compose_service.find_compose_file(directory)
    if not compose_file:
        return []
    try:
        with open(compose_file, "r", encoding="utf-8", errors="replace") as fh:
            data = yaml.safe_load(fh) or {}
    except (OSError, yaml.YAMLError):
        return []
    out: list[str] = []
    for spec in (data.get("services") or {}).values():
        if not isinstance(spec, dict):
            continue
        image = spec.get("image")
        if not isinstance(image, str) or not image:
            continue
        # Strip the tag/digest, then take the last path segment.
        base = image.split("@")[0].rsplit(":", 1)[0]
        candidate = base.rstrip("/").split("/")[-1]
        if candidate and candidate not in _IMAGE_NOISE and candidate not in out:
            out.append(candidate)
    return out


#: Per-stack results, keyed by what they were computed from. Matching is pure
#: string work, but it reads the compose file, and the stacks list runs it for
#: every row on every poll.
_resolved: dict[str, tuple[tuple, Optional[str]]] = {}


def _signature(stack_id: str, name: str) -> tuple:
    try:
        mtime = os.path.getmtime(
            compose_service.find_compose_file(compose_service.stack_dir(stack_id)) or ""
        )
    except OSError:
        mtime = 0.0
    return (name, mtime, _catalog_mtime)


def auto_slug(stack_id: str, name: str) -> Optional[str]:
    """The logo a stack gets with nothing configured, or None for no match.

    Cached against the stack's name and its compose file's mtime, so a rename or
    an edited compose re-matches and everything else is a dictionary hit.
    """
    signature = _signature(stack_id, name)
    cached = _resolved.get(stack_id)
    if cached and cached[0] == signature:
        return cached[1]
    slug = match_slug(name, images_for(stack_id))
    _resolved[stack_id] = (signature, slug)
    return slug


def forget(stack_id: str) -> None:
    """Drop a stack's memoized match (it was deleted, or renamed by clone)."""
    _resolved.pop(stack_id, None)


# --------------------------------------------------------------------------- #
# The logo files
# --------------------------------------------------------------------------- #

#: Slugs currently being downloaded, so N rows asking at once fetch once.
_inflight: dict[str, asyncio.Task] = {}


async def ensure_logo(slug: str) -> Optional[str]:
    """Path to a slug's cached logo, downloading it the first time."""
    path = logo_path(slug)
    if not path:
        return None
    if os.path.isfile(path):
        return path
    catalog = load_catalog()
    if catalog and slug not in catalog.slugs:
        return None

    if (task := _inflight.get(slug)) is None:
        task = asyncio.create_task(_download(slug, path))
        _inflight[slug] = task
        task.add_done_callback(lambda _t, s=slug: _inflight.pop(s, None))
    try:
        return await asyncio.shield(task)
    except asyncio.CancelledError:
        # The *caller* went away (client disconnected); the download itself is
        # shielded and still finishes for whoever asks next.
        raise
    except Exception:  # noqa: BLE001 - a missing logo is not an error
        return None


async def _download(slug: str, path: str) -> Optional[str]:
    try:
        async with httpx.AsyncClient(follow_redirects=True, timeout=TIMEOUT) as client:
            response = await client.get(LOGO_URL.format(slug=slug))
            response.raise_for_status()
            data = response.content
    except httpx.HTTPError as exc:
        logger.info("Could not fetch the logo for '%s': %s", slug, exc)
        return None
    if not data.startswith(b"\x89PNG\r\n\x1a\n") or len(data) > LOGO_MAX_BYTES:
        logger.info("Ignoring the logo for '%s': not a plausible PNG", slug)
        return None
    os.makedirs(logo_dir(), exist_ok=True)
    tmp = f"{path}.{os.getpid()}.tmp"
    try:
        with open(tmp, "wb") as fh:
            fh.write(data)
        os.replace(tmp, path)
    except OSError as exc:
        logger.warning("Could not cache the logo for '%s': %s", slug, exc)
        return None
    return path


def search(query: str, limit: int = 60) -> list[dict]:
    """Catalog entries matching a search term, best match first."""
    catalog = load_catalog()
    if not catalog:
        return []
    needle = _normalize(query)
    results: list[tuple[int, str, dict]] = []
    for entry in catalog.entries:
        slug = (entry.get("Reference") or "").lower()
        name = entry.get("Name") or slug
        if not _SLUG_RE.match(slug):
            continue
        haystack = _normalize(name)
        if not needle:
            rank = 2
        elif haystack == needle or slug == needle.replace(" ", "-"):
            rank = 0
        elif haystack.startswith(needle) or slug.startswith(needle.replace(" ", "-")):
            rank = 1
        elif needle in haystack or needle.replace(" ", "-") in slug:
            rank = 2
        else:
            continue
        results.append((rank, haystack, {"slug": slug, "name": name}))
    results.sort(key=lambda row: (row[0], row[1]))
    return [row[2] for row in results[:limit]]