iLSP/ilsp/python_lsp/catalog.py

"""
pypi-server.i80.dk catalog fetcher with in-memory TTL cache.

Fetches the package list at startup and refreshes every hour.
No persistent storage needed — all data lives in memory.
"""

import asyncio
import logging
import time
from typing import Any

import aiohttp
from lsprotocol.types import CompletionItem, CompletionItemKind, MarkupContent, MarkupKind

logger = logging.getLogger(__name__)

PYPI_SIMPLE_URL = "https://pypi-server.i80.dk/simple/"
PYPI_BASE_URL = "https://pypi-server.i80.dk"
REFRESH_INTERVAL = 3600  # seconds


class PypiCatalog:
    """Thread-safe singleton catalog for pypi-server.i80.dk packages."""

    _packages: list[dict[str, Any]] = []
    _last_refresh: float = 0
    _lock = asyncio.Lock()
    _refresh_task: asyncio.Task | None = None

    @classmethod
    async def get_packages(cls) -> list[dict[str, Any]]:
        if not cls._packages or time.time() - cls._last_refresh > REFRESH_INTERVAL:
            await cls._refresh()
        return cls._packages

    @classmethod
    async def start_background_refresh(cls) -> None:
        if cls._refresh_task is None or cls._refresh_task.done():
            cls._refresh_task = asyncio.create_task(cls._refresh_loop())

    @classmethod
    async def _refresh_loop(cls) -> None:
        while True:
            await cls._refresh()
            await asyncio.sleep(REFRESH_INTERVAL)

    @classmethod
    async def _refresh(cls) -> None:
        async with cls._lock:
            try:
                packages = await cls._fetch()
                cls._packages = packages
                cls._last_refresh = time.time()
                logger.info("PyPI catalog refreshed: %d packages", len(packages))
            except Exception:
                logger.exception("Failed to refresh PyPI catalog")

    @classmethod
    async def _fetch(cls) -> list[dict[str, Any]]:
        packages = []
        async with aiohttp.ClientSession(timeout=aiohttp.ClientTimeout(total=15)) as session:
            async with session.get(PYPI_SIMPLE_URL) as resp:
                resp.raise_for_status()
                html = await resp.text()

        # Parse simple index HTML — each <a href="pkg-name/">pkg-name</a><br>
        # The pypi-server.i80.dk simple index uses relative hrefs: href="pkg-name/"
        import re
        for match in re.finditer(r'<a\s+href="[^"]*">([^<]+)</a>', html):
            pkg_name = match.group(1).strip()
            if pkg_name:
                packages.append({
                    "name": pkg_name,
                    "label": pkg_name,
                    "detail": "i80 package — pypi-server.i80.dk",
                    "sort_prefix": "0_i80_",  # sorts before standard packages
                })

        return packages

    @classmethod
    def as_completion_items(cls) -> list[CompletionItem]:
        return [
            CompletionItem(
                label=pkg["name"],
                kind=CompletionItemKind.Module,
                detail=pkg["detail"],
                sort_text=f"{pkg['sort_prefix']}{pkg['name']}",
                documentation=MarkupContent(
                    kind=MarkupKind.Markdown,
                    value=f"**{pkg['name']}**\n\nCustom i80/LRU package from `pypi-server.i80.dk`",
                ),
            )
            for pkg in cls._packages
        ]