Dispatcharr/apps/fuse/api_views.py

import json
from urllib.parse import unquote

from django.db.models import OuterRef, Subquery, Value
from django.db.models.functions import Coalesce
from django.http import JsonResponse, FileResponse, Http404
from django.conf import settings
import os
from django.urls import reverse
from rest_framework import status, viewsets
from rest_framework.permissions import AllowAny, IsAuthenticated
from rest_framework.response import Response
from rest_framework.views import APIView

from apps.accounts.permissions import IsAdmin, IsStandardUser
from apps.vod.models import (
    VODCategory,
    Movie,
    Series,
    Episode,
    M3UMovieRelation,
    M3USeriesRelation,
    M3UEpisodeRelation,
)
from core.models import CoreSettings, FUSE_SETTINGS_KEY
from .serializers import (
    FuseEntrySerializer,
    FuseSettingsSerializer,
    FUSE_SETTINGS_DEFAULTS,
)
from .utils import is_fuse_client_request
from .client_presence import (
    touch_fuse_client_presence,
    list_fuse_clients,
    force_remove_fuse_client,
)
from .host_auth import (
    require_valid_fuse_host_token,
    has_registered_fuse_host_tokens,
    create_fuse_pairing_token,
    register_fuse_host_with_pairing_token,
)


def _track_fuse_presence_or_forbidden(request, endpoint: str):
    tracking = touch_fuse_client_presence(request=request, endpoint=endpoint)
    if tracking.get("blocked"):
        return Response(
            {
                "detail": (
                    "This FUSE client has been disconnected by an administrator. "
                    "Restart the mount service or wait for the block window to expire."
                )
            },
            status=status.HTTP_403_FORBIDDEN,
        )
    return None


def _require_fuse_token_or_authenticated(
    request,
    *,
    strict: bool,
):
    user = getattr(request, "user", None)
    if user and getattr(user, "is_authenticated", False):
        return None

    if not strict:
        user_agent = str(request.META.get("HTTP_USER_AGENT", "")).strip()
        if not is_fuse_client_request(
            client_user_agent=user_agent,
            request_meta=request.META,
        ):
            return None

    # Backward compatibility: only enforce token auth after at least one host
    # has been paired. Fresh installs can still bootstrap pairing.
    if not has_registered_fuse_host_tokens():
        return None

    ok, detail, _metadata = require_valid_fuse_host_token(request)
    if not ok:
        return Response(
            {"detail": detail},
            status=status.HTTP_401_UNAUTHORIZED,
        )
    return None


def _get_request_ip(request) -> str:
    forwarded_for = request.META.get("HTTP_X_FORWARDED_FOR", "")
    if forwarded_for:
        parts = [part.strip() for part in str(forwarded_for).split(",") if part.strip()]
        if parts:
            return parts[0]
    return str(request.META.get("REMOTE_ADDR", "")).strip() or "unknown"


def _select_best_relation(relations):
    """
    Pick the highest priority active relation.
    """
    if relations is None:
        return None
    try:
        iterable = list(relations.all()) if hasattr(relations, "all") else list(relations)
    except TypeError:
        iterable = []
    if not iterable:
        return None
    return sorted(
        iterable,
        key=lambda rel: (-getattr(rel.m3u_account, "priority", 0), rel.id),
    )[0]


def _build_logo_cache_url(request, logo_id):
    """Build absolute URL for cached VOD logo content."""
    if not logo_id:
        return None
    return request.build_absolute_uri(reverse("api:vod:vodlogo-cache", args=[logo_id]))


def _truncate_utf8(text, max_bytes=255):
    if text is None:
        return ""
    raw = str(text).encode("utf-8", errors="ignore")
    if len(raw) <= max_bytes:
        return str(text)
    raw = raw[:max_bytes]
    while raw:
        try:
            return raw.decode("utf-8")
        except UnicodeDecodeError:
            raw = raw[:-1]
    return ""


def _sanitize_raw_name(name):
    cleaned = []
    for ch in str(name or ""):
        code = ord(ch)
        if ch in ("/", "\x00") or code < 32:
            cleaned.append("_")
        else:
            cleaned.append(ch)
    return "".join(cleaned).strip()


def _normalize_fuse_name(name, fallback):
    value = _sanitize_raw_name(name)
    if value in ("", ".", ".."):
        value = fallback
    value = _truncate_utf8(value, 255)
    if value in ("", ".", ".."):
        value = _truncate_utf8(fallback, 255) or "item"
    return value


def _name_with_suffix(name, suffix, is_dir):
    if not is_dir:
        dot_index = name.rfind(".")
        if dot_index > 0:
            base = name[:dot_index]
            ext = name[dot_index:]
        else:
            base = name
            ext = ""
    else:
        base = name
        ext = ""

    suffix_bytes = len((suffix + ext).encode("utf-8", errors="ignore"))
    max_base_bytes = max(1, 255 - suffix_bytes)
    base = _truncate_utf8(base, max_base_bytes)
    candidate = _truncate_utf8(f"{base}{suffix}{ext}", 255)
    if candidate in ("", ".", ".."):
        candidate = _truncate_utf8(f"item{suffix}{ext}", 255) or "item"
    return candidate


def _ensure_unique_fuse_name(name, used_names, is_dir):
    candidate = name
    counter = 2
    while candidate in used_names:
        candidate = _name_with_suffix(name, f" ({counter})", is_dir=is_dir)
        counter += 1
    used_names.add(candidate)
    return candidate


def _sanitize_entries_for_directory(directory_path, entries):
    """
    Sanitize names for filesystem safety and keep entry.path consistent with
    the emitted name so subsequent getattr/lookups resolve correctly.
    """
    prefix = "/" if not directory_path or directory_path == "/" else f"/{directory_path.strip('/')}"
    used_names = set()
    sanitized = []

    for idx, entry in enumerate(entries, start=1):
        item = dict(entry)
        is_dir = bool(item.get("is_dir"))
        fallback = f"{'dir' if is_dir else 'file'}-{idx}"
        normalized = _normalize_fuse_name(item.get("name"), fallback=fallback)
        unique_name = _ensure_unique_fuse_name(normalized, used_names, is_dir=is_dir)

        item["name"] = unique_name
        item["path"] = f"/{unique_name}" if prefix == "/" else f"{prefix}/{unique_name}"
        sanitized.append(item)

    return sanitized


def _build_segment_map(items, name_getter, fallback_prefix):
    """
    Deterministically map sanitized path segments to DB objects.
    Used for root directory listings and reverse lookup of incoming segments.
    """
    used_names = set()
    mapping = {}

    for item in items:
        fallback = f"{fallback_prefix}-{getattr(item, 'id', 'x')}"
        normalized = _normalize_fuse_name(name_getter(item), fallback=fallback)
        unique_name = _ensure_unique_fuse_name(normalized, used_names, is_dir=True)
        mapping[unique_name] = item

    return mapping


class FuseBrowseView(APIView):
    """
    Read-only filesystem-style browsing for Movies and TV.
    """

    permission_classes = [AllowAny]

    def get(self, request, mode):
        unauthorized = _require_fuse_token_or_authenticated(request, strict=True)
        if unauthorized is not None:
            return unauthorized

        denied_response = _track_fuse_presence_or_forbidden(
            request=request,
            endpoint=f"browse:{mode}",
        )
        if denied_response is not None:
            return denied_response

        path = request.query_params.get("path", "/")
        path = unquote(path)
        # Normalize
        trimmed = path.strip("/")
        parts = [p for p in trimmed.split("/") if p] if trimmed else []

        if mode not in ("movies", "tv"):
            return Response({"detail": "Invalid mode"}, status=status.HTTP_400_BAD_REQUEST)

        if mode == "movies":
            return Response(self._browse_movies(parts, request))
        return Response(self._browse_tv(parts, request))

    def _browse_movies(self, parts, request):
        # Root -> list categories
        if len(parts) == 0:
            category_ids = (
                M3UMovieRelation.objects.filter(
                    m3u_account__is_active=True, category__isnull=False
                )
                .values_list("category_id", flat=True)
                .distinct()
            )
            categories = VODCategory.objects.filter(
                category_type="movie", id__in=category_ids
            ).order_by("name", "id")

            segment_map = _build_segment_map(
                categories,
                name_getter=lambda cat: cat.name,
                fallback_prefix="category",
            )
            entries = [
                {
                    "name": segment,
                    "path": f"/{segment}",
                    "is_dir": True,
                    "content_type": "category",
                    "uuid": None,
                }
                for segment in segment_map.keys()
            ]
            return {"path": "/", "entries": FuseEntrySerializer(entries, many=True).data}

        # Category -> list movies
        category_segment = parts[0]
        category = (
            VODCategory.objects.filter(
                name=category_segment, category_type="movie"
            ).first()
        )
        if not category:
            category_ids = (
                M3UMovieRelation.objects.filter(
                    m3u_account__is_active=True, category__isnull=False
                )
                .values_list("category_id", flat=True)
                .distinct()
            )
            categories = VODCategory.objects.filter(
                category_type="movie", id__in=category_ids
            ).order_by("name", "id")
            segment_map = _build_segment_map(
                categories,
                name_getter=lambda cat: cat.name,
                fallback_prefix="category",
            )
            category = segment_map.get(category_segment)
        if not category:
            return {"path": f"/{category_segment}", "entries": []}

        movies = (
            Movie.objects.filter(
                m3u_relations__category=category,
                m3u_relations__m3u_account__is_active=True,
            )
            .distinct()
            .annotate(
                best_extension=Coalesce(
                    Subquery(
                        M3UMovieRelation.objects.filter(
                            m3u_account__is_active=True,
                            category=category,
                            movie_id=OuterRef("pk"),
                        )
                        .order_by("-m3u_account__priority", "id")
                        .values("container_extension")[:1]
                    ),
                    Value("mp4"),
                )
            )
            .only("id", "uuid", "name", "year", "logo_id")
            .order_by("name")
        )

        entries = []
        for movie in movies:
            extension = getattr(movie, "best_extension", None) or "mp4"
            name = f"{movie.name} ({movie.year})" if movie.year else movie.name
            file_name = f"{name}.{extension}"
            poster_file_name = f"{name}.jpg"

            entries.append(
                {
                    "name": file_name,
                    "path": f"/{category_segment}/{file_name}",
                    "is_dir": False,
                    "content_type": "movie",
                    "uuid": movie.uuid,
                    "extension": extension,
                    "category": category.name,
                    # Report zero so clients don't prefetch/consume provider slots until a real read.
                    "size": 0,
                    # Omit stream_url to force clients to fetch it only when they actually read.
                    "stream_url": None,
                }
            )

            # Add sidecar poster art so Plex-like scanners can consume lightweight metadata assets
            # without touching upstream provider streams.
            if movie.logo_id:
                entries.append(
                    {
                        "name": poster_file_name,
                        "path": f"/{category_segment}/{poster_file_name}",
                        "is_dir": False,
                        "content_type": "image",
                        "uuid": None,
                        "extension": "jpg",
                        "category": category.name,
                        "size": 0,
                        "stream_url": _build_logo_cache_url(request, movie.logo_id),
                    }
                )

        entries = _sanitize_entries_for_directory(f"/{category_segment}", entries)
        return {
            "path": f"/{category_segment}",
            "entries": FuseEntrySerializer(entries, many=True).data,
        }

    def _browse_tv(self, parts, request):
        # Root -> list series
        if len(parts) == 0:
            series = (
                Series.objects.filter(
                    m3u_relations__m3u_account__is_active=True,
                )
                .distinct()
                .order_by("name", "id")
            )
            segment_map = _build_segment_map(
                series,
                name_getter=lambda serie: serie.name,
                fallback_prefix="series",
            )
            entries = [
                {
                    "name": segment,
                    "path": f"/{segment}",
                    "is_dir": True,
                    "content_type": "series",
                    "uuid": None,
                }
                for segment in segment_map.keys()
            ]
            return {"path": "/", "entries": FuseEntrySerializer(entries, many=True).data}

        # Series -> list seasons
        series_segment = parts[0]
        series_obj = Series.objects.filter(name=series_segment).first()
        if not series_obj:
            series = (
                Series.objects.filter(
                    m3u_relations__m3u_account__is_active=True,
                )
                .distinct()
                .order_by("name", "id")
            )
            segment_map = _build_segment_map(
                series,
                name_getter=lambda serie: serie.name,
                fallback_prefix="series",
            )
            series_obj = segment_map.get(series_segment)
        if not series_obj:
            return {"path": f"/{series_segment}", "entries": []}

        if len(parts) == 1:
            seasons = (
                Episode.objects.filter(series=series_obj)
                .exclude(season_number__isnull=True)
                .values_list("season_number", flat=True)
                .distinct()
            )
            season_numbers = sorted(set(seasons)) or [0]
            entries = []
            for num in season_numbers:
                label = f"Season {int(num):02d}"
                entries.append(
                    {
                        "name": label,
                        "path": f"/{series_segment}/{label}",
                        "is_dir": True,
                        "content_type": "season",
                        "uuid": None,
                        "season": int(num),
                    }
                )

            # Plex local-assets convention for show-level artwork.
            if series_obj.logo_id:
                entries.append(
                    {
                        "name": "poster.jpg",
                        "path": f"/{series_segment}/poster.jpg",
                        "is_dir": False,
                        "content_type": "image",
                        "uuid": None,
                        "extension": "jpg",
                        "size": 0,
                        "stream_url": _build_logo_cache_url(request, series_obj.logo_id),
                    }
                )

            entries = _sanitize_entries_for_directory(f"/{series_segment}", entries)
            return {
                "path": f"/{series_segment}",
                "entries": FuseEntrySerializer(entries, many=True).data,
            }

        # Season -> list episodes
        season_label = parts[1]
        try:
            season_number = int(season_label.replace("Season", "").strip())
        except Exception:
            season_number = None

        episodes = (
            Episode.objects.filter(
                series=series_obj,
                season_number=season_number,
            )
            .filter(m3u_relations__m3u_account__is_active=True)
            .distinct()
            .annotate(
                best_extension=Coalesce(
                    Subquery(
                        M3UEpisodeRelation.objects.filter(
                            m3u_account__is_active=True,
                            episode_id=OuterRef("pk"),
                        )
                        .order_by("-m3u_account__priority", "id")
                        .values("container_extension")[:1]
                    ),
                    Value("mp4"),
                )
            )
            .only("id", "uuid", "name", "series_id", "season_number", "episode_number")
            .order_by("episode_number")
        )

        entries = []
        for ep in episodes:
            extension = getattr(ep, "best_extension", None) or "mp4"
            ep_num = ep.episode_number or 0
            season_num = ep.season_number or 0
            name = f"S{season_num:02d}E{ep_num:02d} - {ep.name}"
            file_name = f"{name}.{extension}"
            stream_url = None
            if ep.uuid:
                stream_url = request.build_absolute_uri(
                    reverse(
                        "proxy:vod_proxy:vod_stream",
                        kwargs={"content_type": "episode", "content_id": ep.uuid},
                    )
                )
            entries.append(
                {
                    "name": file_name,
                    "path": f"/{series_segment}/{season_label}/{file_name}",
                    "is_dir": False,
                    "content_type": "episode",
                    "uuid": ep.uuid,
                    "extension": extension,
                    "season": season_num,
                    "episode_number": ep_num,
                    "size": 0,
                    "stream_url": None,
                }
            )

        entries = _sanitize_entries_for_directory(f"/{series_segment}/{season_label}", entries)
        return {
            "path": f"/{series_segment}/{season_label}",
            "entries": FuseEntrySerializer(entries, many=True).data,
        }


class FuseSettingsViewSet(viewsets.ViewSet):
    """
    Store FUSE client guidance in CoreSettings as JSON.
    """

    permission_classes = [IsAuthenticated]
    serializer_class = FuseSettingsSerializer

    DEFAULTS = FUSE_SETTINGS_DEFAULTS.copy()

    def get_permissions(self):
        # Host-side scripts need anonymous read access to tuning values.
        if self.action in ("list", "retrieve"):
            return [AllowAny()]
        return [IsAuthenticated()]

    def _normalize_settings_value(self, raw_value):
        if isinstance(raw_value, dict):
            return raw_value
        if isinstance(raw_value, str):
            try:
                parsed = json.loads(raw_value)
                if isinstance(parsed, dict):
                    return parsed
            except json.JSONDecodeError:
                return {}
        return {}

    def _get_or_create(self):
        try:
            obj = CoreSettings.objects.get(key=FUSE_SETTINGS_KEY)
            data = self._normalize_settings_value(obj.value)
        except CoreSettings.DoesNotExist:
            data = self.DEFAULTS.copy()
            obj, _ = CoreSettings.objects.get_or_create(
                key=FUSE_SETTINGS_KEY,
                defaults={"name": "Fuse Settings", "value": data},
            )
        merged = self.DEFAULTS.copy()
        merged.update(data or {})
        return obj, merged

    def list(self, request):
        obj, data = self._get_or_create()
        serializer = FuseSettingsSerializer(data=data)
        serializer.is_valid(raise_exception=True)
        return Response(serializer.data)

    def retrieve(self, request, pk=None):
        return self.list(request)

    def update(self, request, pk=None):
        obj, current = self._get_or_create()
        serializer = FuseSettingsSerializer(data=request.data)
        serializer.is_valid(raise_exception=True)
        obj.value = serializer.validated_data
        obj.save()
        return Response(serializer.validated_data)


class FusePublicSettingsView(APIView):
    """
    Public read-only endpoint for host-side FUSE scripts to fetch tuning defaults.
    """

    permission_classes = [AllowAny]

    def get(self, request):
        unauthorized = _require_fuse_token_or_authenticated(request, strict=False)
        if unauthorized is not None:
            return unauthorized

        denied_response = _track_fuse_presence_or_forbidden(
            request=request,
            endpoint="settings:public",
        )
        if denied_response is not None:
            return denied_response

        try:
            obj = CoreSettings.objects.get(key=FUSE_SETTINGS_KEY)
            raw_value = obj.value
        except CoreSettings.DoesNotExist:
            raw_value = {}

        if isinstance(raw_value, str):
            try:
                raw_value = json.loads(raw_value)
            except json.JSONDecodeError:
                raw_value = {}
        if not isinstance(raw_value, dict):
            raw_value = {}

        merged = FUSE_SETTINGS_DEFAULTS.copy()
        merged.update(raw_value)
        serializer = FuseSettingsSerializer(data=merged)
        serializer.is_valid(raise_exception=True)
        return Response(serializer.validated_data)


class FuseStreamURLView(APIView):
    """
    Provide a stable stream URL for a given movie/episode UUID.
    """

    permission_classes = [AllowAny]

    def get(self, request, content_type, content_id):
        unauthorized = _require_fuse_token_or_authenticated(request, strict=True)
        if unauthorized is not None:
            return unauthorized

        denied_response = _track_fuse_presence_or_forbidden(
            request=request,
            endpoint=f"stream-url:{content_type}",
        )
        if denied_response is not None:
            return denied_response

        if content_type not in ("movie", "episode"):
            return Response({"detail": "Invalid content type"}, status=status.HTTP_400_BAD_REQUEST)

        try:
            model = Movie if content_type == "movie" else Episode
            if not model.objects.filter(uuid=content_id).exists():
                return Response({"detail": "Not found"}, status=status.HTTP_404_NOT_FOUND)
            stream_url = request.build_absolute_uri(
                reverse(
                    "proxy:vod_proxy:vod_stream",
                    kwargs={"content_type": content_type, "content_id": content_id},
                )
            )
            return JsonResponse({"stream_url": stream_url})
        except Exception:
            return Response({"detail": "Not found"}, status=status.HTTP_404_NOT_FOUND)


class FuseConnectedClientsView(APIView):
    """
    List and force-remove host machines connected via the Dispatcharr FUSE client.
    """

    def get_permissions(self):
        if self.request.method.upper() == "POST":
            return [IsAdmin()]
        return [IsStandardUser()]

    def get(self, request):
        include_inactive = str(request.query_params.get("include_inactive", "")).strip().lower()
        payload = list_fuse_clients(
            include_inactive=include_inactive in {"1", "true", "yes", "on"}
        )
        return Response(payload)

    def post(self, request):
        client_id = str(request.data.get("client_id", "")).strip()
        if not client_id:
            return Response(
                {"detail": "client_id is required"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        block_seconds = request.data.get("block_seconds")
        try:
            block_seconds = int(block_seconds) if block_seconds is not None else None
        except (TypeError, ValueError):
            return Response(
                {"detail": "block_seconds must be an integer"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        removed_by = getattr(request.user, "username", "") if getattr(request, "user", None) else ""
        try:
            result = force_remove_fuse_client(
                client_id=client_id,
                block_seconds=block_seconds,
                removed_by=removed_by,
            )
        except ValueError:
            return Response(
                {"detail": "Invalid client_id"},
                status=status.HTTP_400_BAD_REQUEST,
            )
        except Exception:
            return Response(
                {"detail": "Unable to force-remove this FUSE client right now"},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )

        return Response(
            {
                "success": True,
                "message": "FUSE client force-removed",
                **result,
            }
        )


class FusePairingTokenView(APIView):
    """
    Create a short-lived one-time pairing token for Linux host install scripts.
    """

    permission_classes = [IsStandardUser]

    def post(self, request):
        ttl_seconds = request.data.get("ttl_seconds")
        try:
            ttl_seconds = int(ttl_seconds) if ttl_seconds is not None else None
        except (TypeError, ValueError):
            return Response(
                {"detail": "ttl_seconds must be an integer"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        created_by = getattr(request.user, "username", "") if getattr(request, "user", None) else ""
        try:
            payload = create_fuse_pairing_token(
                created_by=created_by,
                ttl_seconds=ttl_seconds,
            )
        except RuntimeError:
            return Response(
                {"detail": "Pairing service unavailable right now"},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )

        return Response(payload, status=status.HTTP_201_CREATED)


class FuseRegisterHostView(APIView):
    """
    Exchange a one-time pairing token for a host-scoped FUSE auth token.
    """

    permission_classes = [AllowAny]

    def post(self, request):
        pairing_token = str(request.data.get("pairing_token", "")).strip()
        hostname = str(request.data.get("hostname", "")).strip() or "unknown"
        client_id_hint = str(request.data.get("client_id", "")).strip()
        mountpoint_hint = str(request.data.get("mountpoint", "")).strip()
        request_ip = _get_request_ip(request)

        if not pairing_token:
            return Response(
                {"detail": "pairing_token is required"},
                status=status.HTTP_400_BAD_REQUEST,
            )

        try:
            payload = register_fuse_host_with_pairing_token(
                pairing_token=pairing_token,
                hostname=hostname,
                request_ip=request_ip,
                client_id_hint=client_id_hint,
                mountpoint_hint=mountpoint_hint,
            )
        except ValueError as exc:
            return Response(
                {"detail": str(exc)},
                status=status.HTTP_400_BAD_REQUEST,
            )
        except RuntimeError:
            return Response(
                {"detail": "Pairing service unavailable right now"},
                status=status.HTTP_503_SERVICE_UNAVAILABLE,
            )

        return Response(payload, status=status.HTTP_201_CREATED)


class FuseClientDownloadView(APIView):
    """
    Serve host-side FUSE setup scripts from the local server.
    """

    permission_classes = [AllowAny]

    SCRIPT_MAP = {
        "python": ("fuse_client.py", "fuse_client.py", "text/x-python"),
        "linux": ("mount_linux.sh", "mount_linux.sh", "text/x-shellscript"),
        "linux-systemd": ("install_systemd_mounts.sh", "install_systemd_mounts.sh", "text/x-shellscript"),
    }

    TARGET_ALIASES = {
        "py": "python",
        "systemd": "linux-systemd",
    }

    def get(self, request, target="python"):
        normalized_target = (target or "python").strip().lower()
        normalized_target = self.TARGET_ALIASES.get(normalized_target, normalized_target)

        if normalized_target not in self.SCRIPT_MAP:
            return Response(
                {
                    "detail": (
                        "Invalid target. Supported values: "
                        + ", ".join(sorted(self.SCRIPT_MAP.keys()))
                    )
                },
                status=status.HTTP_400_BAD_REQUEST,
            )

        file_name, download_name, content_type = self.SCRIPT_MAP[normalized_target]
        script_path = os.path.join(settings.BASE_DIR, "apps", "fuse", "fuse_client", file_name)
        if not os.path.exists(script_path):
            raise Http404(f"Script not found: {file_name}")

        return FileResponse(
            open(script_path, "rb"),
            as_attachment=True,
            filename=download_name,
            content_type=content_type,
        )