media-player-server/media_server/routes/links.py

"""Header quick links management API endpoints."""

import logging
import re
from typing import Any
from urllib.parse import urlparse

from fastapi import APIRouter, Depends, HTTPException, status
from pydantic import BaseModel, Field, field_validator

from ..auth import verify_token
from ..config import LinkConfig, settings
from ..config_manager import config_manager
from ..services.websocket_manager import ws_manager

router = APIRouter(prefix="/api/links", tags=["links"])
logger = logging.getLogger(__name__)

# Only allow MDI iconify slugs and safe `http(s)`-ish URLs through the API.
_MDI_ICON_RE = re.compile(r"^mdi:[a-z0-9][a-z0-9-]{0,63}$")
_ALLOWED_URL_SCHEMES = {"http", "https"}


def _validate_url(url: str) -> str:
    """Ensure the URL is well-formed http(s) — no ``javascript:`` etc."""
    parsed = urlparse(url)
    if parsed.scheme.lower() not in _ALLOWED_URL_SCHEMES:
        raise ValueError("URL must start with http:// or https://")
    if not parsed.netloc:
        raise ValueError("URL must include a host")
    return url


def _validate_icon(icon: str) -> str:
    """Restrict icon names to safe Material Design Icons slugs."""
    if not _MDI_ICON_RE.match(icon):
        raise ValueError("Icon must be of the form 'mdi:<lowercase-slug>'")
    return icon


def _require_links_management() -> None:
    """Authorise a links-CRUD operation. Operator flag + per-token admin scope."""
    if not settings.links_management:
        raise HTTPException(
            status_code=status.HTTP_403_FORBIDDEN,
            detail="Links management is disabled. Set links_management: true in config.yaml to enable.",
        )
    from ..auth import auth_enabled, token_has_scope, token_label_var
    if auth_enabled():
        label = token_label_var.get("unknown")
        if not token_has_scope(label, "admin"):
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail=f"Token '{label}' lacks required scope: admin",
            )


class LinkInfo(BaseModel):
    """Information about a configured link."""

    name: str
    url: str
    icon: str
    label: str
    description: str


class LinkCreateRequest(BaseModel):
    """Request model for creating or updating a link."""

    url: str = Field(..., description="URL to open", min_length=1, max_length=2048)
    icon: str = Field(default="mdi:link", description="MDI icon name (e.g., 'mdi:led-strip-variant')")
    label: str = Field(default="", description="Tooltip text", max_length=128)
    description: str = Field(default="", description="Optional description", max_length=512)

    @field_validator("url")
    @classmethod
    def _check_url(cls, v: str) -> str:
        return _validate_url(v)

    @field_validator("icon")
    @classmethod
    def _check_icon(cls, v: str) -> str:
        return _validate_icon(v)


def _validate_link_name(name: str) -> None:
    """Validate link name."""
    if not re.match(r"^[a-zA-Z0-9_]+$", name):
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Link name must contain only letters, numbers, and underscores",
        )
    if len(name) > 64:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="Link name must be 64 characters or less",
        )


@router.get("/list")
async def list_links(_: str = Depends(verify_token)) -> list[LinkInfo]:
    """List all configured links.

    Returns:
        List of configured links.
    """
    return [
        LinkInfo(
            name=name,
            url=config.url,
            icon=config.icon,
            label=config.label,
            description=config.description,
        )
        for name, config in settings.links.items()
    ]


@router.post("/create/{link_name}")
async def create_link(
    link_name: str,
    request: LinkCreateRequest,
    _: str = Depends(verify_token),
) -> dict[str, Any]:
    """Create a new link.

    Args:
        link_name: Link name (alphanumeric and underscores only).
        request: Link configuration.

    Returns:
        Success response with link name.
    """
    _require_links_management()
    _validate_link_name(link_name)

    if link_name in settings.links:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail=f"Link '{link_name}' already exists. Use PUT /api/links/update/{link_name} to update it.",
        )

    link_config = LinkConfig(**request.model_dump())

    try:
        config_manager.add_link(link_name, link_config)
    except Exception as e:
        logger.error(f"Failed to add link '{link_name}': {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to add link: {str(e)}",
        )

    await ws_manager.broadcast_links_changed()
    logger.info(f"Link '{link_name}' created successfully")
    return {"success": True, "link": link_name}


@router.put("/update/{link_name}")
async def update_link(
    link_name: str,
    request: LinkCreateRequest,
    _: str = Depends(verify_token),
) -> dict[str, Any]:
    """Update an existing link.

    Args:
        link_name: Link name.
        request: Updated link configuration.

    Returns:
        Success response with link name.
    """
    _require_links_management()
    _validate_link_name(link_name)

    if link_name not in settings.links:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Link '{link_name}' not found. Use POST /api/links/create/{link_name} to create it.",
        )

    link_config = LinkConfig(**request.model_dump())

    try:
        config_manager.update_link(link_name, link_config)
    except Exception as e:
        logger.error(f"Failed to update link '{link_name}': {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to update link: {str(e)}",
        )

    await ws_manager.broadcast_links_changed()
    logger.info(f"Link '{link_name}' updated successfully")
    return {"success": True, "link": link_name}


@router.delete("/delete/{link_name}")
async def delete_link(
    link_name: str,
    _: str = Depends(verify_token),
) -> dict[str, Any]:
    """Delete a link.

    Args:
        link_name: Link name.

    Returns:
        Success response with link name.
    """
    _require_links_management()
    _validate_link_name(link_name)

    if link_name not in settings.links:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"Link '{link_name}' not found",
        )

    try:
        config_manager.delete_link(link_name)
    except Exception as e:
        logger.error(f"Failed to delete link '{link_name}': {e}")
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"Failed to delete link: {str(e)}",
        )

    await ws_manager.broadcast_links_changed()
    logger.info(f"Link '{link_name}' deleted successfully")
    return {"success": True, "link": link_name}