app/services/input_resolver.py

from __future__ import annotations

import asyncio
import base64
import json
import mimetypes
from pathlib import Path
from typing import Any
from uuid import uuid4

import aiofiles
from fastapi import Request
from starlette.datastructures import FormData, UploadFile

from app.core.config import Settings
from app.core.exceptions import InputError
from app.models.media import InputMedia, MediaSource, OperationResult, ResolvedRequest
from app.services.cleanup import CleanupService, RequestWorkspace
from app.services.downloader import Downloader
from app.services.validator import MediaValidator
from app.services.ytdlp_service import YTDLPService


class InputResolver:
    """Normalizes multipart, JSON, Base64, n8n, URL, and raw-byte inputs."""

    INPUT_KEYS = {
        "url",
        "urls",
        "base64",
        "binary",
        "input",
        "inputs",
        "files",
        "file",
        "path",
        "temp_path",
    }

    def __init__(
        self,
        settings: Settings,
        cleanup: CleanupService,
        downloader: Downloader,
        ytdlp: YTDLPService,
        validator: MediaValidator,
    ) -> None:
        self.settings = settings
        self.cleanup = cleanup
        self.downloader = downloader
        self.ytdlp = ytdlp
        self.validator = validator

    async def resolve(self, request: Request, *, require_input: bool = True) -> ResolvedRequest:
        request_id = request.state.request_id
        workspace = await self.cleanup.create_workspace(request_id)
        content_type = request.headers.get("content-type", "").lower()
        is_ytdlp_request = request.url.path.startswith("/v1/ytdlp/")
        try:
            if content_type.startswith("multipart/form-data"):
                inputs, params = await self._multipart(request, workspace, is_ytdlp_request)
            elif "json" in content_type:
                inputs, params = await self._json(request, workspace, is_ytdlp_request)
            else:
                inputs, params = await self._raw(request, workspace, content_type)
        except Exception:
            await self.cleanup.complete(request_id)
            raise
        if require_input and not inputs:
            await self.cleanup.complete(request_id)
            raise InputError(
                "No media input found. Send multipart file(s), raw bytes, url, base64, or n8n binary data."
            )
        params = {**dict(request.query_params), **params}
        return ResolvedRequest(request_id=request_id, inputs=inputs, params=params)

    async def resolve_payload(
        self,
        payload: dict[str, Any],
        request_id: str,
        *,
        require_input: bool = True,
        ytdlp_options: dict[str, Any] | None = None,
    ) -> ResolvedRequest:
        """Resolve an in-process JSON payload through the same REST input pipeline.

        MCP and other non-HTTP transports use this method so URL, Base64, n8n
        binary, managed temporary paths, validation, and workspace handling stay
        centralized in one resolver.
        """
        workspace = await self.cleanup.create_workspace(request_id)
        try:
            descriptors = self._collect_descriptors(payload)
            params = self._params_from_payload(payload)
            inputs = [
                await self._resolve_descriptor(
                    descriptor,
                    workspace,
                    ytdlp_options=ytdlp_options,
                )
                for descriptor in descriptors
            ]
        except Exception:
            await self.cleanup.complete(request_id)
            raise
        if require_input and not inputs:
            await self.cleanup.complete(request_id)
            raise InputError(
                "No media input found. Provide url, base64, binary, temp_path, or inputs."
            )
        return ResolvedRequest(request_id=request_id, inputs=inputs, params=params)

    async def _multipart(
        self,
        request: Request,
        workspace: RequestWorkspace,
        is_ytdlp_request: bool,
    ) -> tuple[list[InputMedia], dict[str, Any]]:
        try:
            form: FormData = await request.form(
                max_files=100,
                max_fields=500,
                max_part_size=self.settings.max_upload_size,
            )
        except TypeError:
            form = await request.form()
        except Exception as exc:
            raise InputError("Invalid multipart form data") from exc
        inputs: list[InputMedia] = []
        fields: dict[str, Any] = {}
        for key, value in form.multi_items():
            if isinstance(value, UploadFile):
                inputs.append(await self._save_upload(value, workspace))
                continue
            if key in fields:
                current = fields[key]
                fields[key] = current + [value] if isinstance(current, list) else [current, value]
            else:
                fields[key] = value
        descriptors: list[dict[str, Any]] = []
        for key, value in fields.items():
            if key == "url":
                descriptors.extend({"url": item} for item in self._as_list(value))
            elif key == "urls":
                parsed = self._parse_field(value)
                descriptors.extend({"url": item} for item in self._as_list(parsed))
            elif key == "base64" or key.startswith("binary."):
                for item in self._as_list(value):
                    parsed = self._parse_field(item)
                    descriptors.append(parsed if isinstance(parsed, dict) else {"base64": parsed})
            elif key in {"input", "inputs", "files", "binary"}:
                parsed = self._parse_field(value)
                descriptors.extend(self._collect_descriptors(parsed, binary=key == "binary"))
        params = self._params_from_payload(fields)
        ytdlp_options = params if is_ytdlp_request else None
        for descriptor in descriptors:
            inputs.append(
                await self._resolve_descriptor(descriptor, workspace, ytdlp_options=ytdlp_options)
            )
        return inputs, params

    async def _json(
        self,
        request: Request,
        workspace: RequestWorkspace,
        is_ytdlp_request: bool,
    ) -> tuple[list[InputMedia], dict[str, Any]]:
        content_length = self._int(request.headers.get("content-length"))
        maximum_json = int(self.settings.max_upload_size * 1.5) + 1_048_576
        if content_length and content_length > maximum_json:
            raise InputError("JSON request body is too large")
        body = await request.body()
        if len(body) > maximum_json:
            raise InputError("JSON request body is too large")
        try:
            payload = await asyncio.to_thread(json.loads, body)
        except (json.JSONDecodeError, UnicodeDecodeError) as exc:
            raise InputError("Request body is not valid JSON") from exc
        if not isinstance(payload, dict):
            raise InputError("JSON request body must be an object")
        descriptors = self._collect_descriptors(payload)
        params = self._params_from_payload(payload)
        ytdlp_options = params if is_ytdlp_request else None
        inputs = [
            await self._resolve_descriptor(descriptor, workspace, ytdlp_options=ytdlp_options)
            for descriptor in descriptors
        ]
        return inputs, params

    async def _raw(
        self, request: Request, workspace: RequestWorkspace, content_type: str
    ) -> tuple[list[InputMedia], dict[str, Any]]:
        declared = self._int(request.headers.get("content-length"))
        if declared and declared > self.settings.max_upload_size:
            raise InputError("The request body exceeds MAX_UPLOAD_SIZE")
        filename = self.validator.safe_filename(
            request.headers.get("x-filename") or request.query_params.get("filename"),
            "input.bin",
        )
        path = workspace.uploads / f"{uuid4().hex}_{filename}"
        size = 0
        async with aiofiles.open(path, "wb") as output:
            async for chunk in request.stream():
                size += len(chunk)
                if size > self.settings.max_upload_size:
                    await output.close()
                    path.unlink(missing_ok=True)
                    raise InputError("The request body exceeds MAX_UPLOAD_SIZE")
                await output.write(chunk)
        if size == 0:
            path.unlink(missing_ok=True)
            return [], dict(request.query_params)
        mime_type = content_type.split(";", 1)[0] or self.validator.infer_mime(path)
        self.validator.validate_declared(filename, mime_type, size)
        return [
            InputMedia(
                source=MediaSource.OCTET_STREAM,
                filename=filename,
                mime_type=mime_type,
                temp_path=path,
                size=size,
            )
        ], dict(request.query_params)

    async def _save_upload(self, upload: UploadFile, workspace: RequestWorkspace) -> InputMedia:
        filename = self.validator.safe_filename(upload.filename, "upload.bin")
        path = workspace.uploads / f"{uuid4().hex}_{filename}"
        size = 0
        async with aiofiles.open(path, "wb") as output:
            while chunk := await upload.read(1024 * 1024):
                size += len(chunk)
                if size > self.settings.max_upload_size:
                    await output.close()
                    path.unlink(missing_ok=True)
                    raise InputError("The uploaded media exceeds MAX_UPLOAD_SIZE")
                await output.write(chunk)
        await upload.close()
        mime_type = upload.content_type or self.validator.infer_mime(path)
        self.validator.validate_declared(filename, mime_type, size)
        return InputMedia(
            source=MediaSource.MULTIPART,
            filename=filename,
            mime_type=mime_type,
            temp_path=path,
            size=size,
        )

    async def _resolve_descriptor(
        self,
        descriptor: dict[str, Any],
        workspace: RequestWorkspace,
        *,
        ytdlp_options: dict[str, Any] | None = None,
    ) -> InputMedia:
        if "temp_path" in descriptor or "path" in descriptor:
            return await self._resolve_local_path(descriptor, workspace)
        if "url" in descriptor:
            url = descriptor["url"]
            if not isinstance(url, str) or not url.strip():
                raise InputError("Media URL must be a non-empty string")
            await self.downloader.validate_url(url)
            if ytdlp_options is not None or await asyncio.to_thread(self.ytdlp.supports_url, url):
                options = ytdlp_options or {}
                mode = str(options.get("mode", "video"))
                if options.get("playlist") or options.get("include_formats"):
                    max_entries = self._int(options.get("max_entries", 100))
                    if max_entries is None:
                        raise InputError("max_entries must be an integer")
                    metadata = await self.ytdlp.extract_metadata(
                        url,
                        playlist=bool(options.get("playlist")),
                        include_formats=bool(options.get("include_formats")),
                        max_entries=max_entries,
                    )
                    result = OperationResult(metadata=metadata)
                    mode = "metadata"
                else:
                    result = await self.ytdlp.download(
                        url,
                        workspace.uploads,
                        mode=mode,
                        format_selector=options.get("format"),
                        audio_format=str(options.get("audio_format", "mp3")),
                    )
                if result.path is None:
                    if mode != "metadata":
                        raise InputError("yt-dlp did not return a media file")
                    result.path = workspace.uploads / "metadata.json"
                    encoded = json.dumps(
                        result.metadata, ensure_ascii=False, separators=(",", ":")
                    ).encode("utf-8")
                    async with aiofiles.open(result.path, "wb") as output:
                        await output.write(encoded)
                    result.filename = result.path.name
                    result.mime_type = "application/json"
                return InputMedia(
                    source=MediaSource.YTDLP,
                    filename=result.filename or result.path.name,
                    mime_type=result.mime_type or self.validator.infer_mime(result.path),
                    temp_path=result.path,
                    size=result.path.stat().st_size,
                    duration=result.metadata.get("duration"),
                    metadata=result.metadata,
                )
            path, mime_type = await self.downloader.download(
                url, workspace.uploads, descriptor.get("filename")
            )
            return InputMedia(
                source=MediaSource.JSON_URL,
                filename=self._original_name(path),
                mime_type=mime_type,
                temp_path=path,
                size=path.stat().st_size,
                metadata={"url": url},
            )
        data = descriptor.get("base64", descriptor.get("data"))
        if not isinstance(data, str) or not data:
            raise InputError("Input descriptor must contain a url or Base64 data")
        return await self._decode_base64(data, descriptor, workspace)

    async def _resolve_local_path(
        self, descriptor: dict[str, Any], workspace: RequestWorkspace
    ) -> InputMedia:
        raw_path = descriptor.get("temp_path", descriptor.get("path"))
        if not isinstance(raw_path, str) or not raw_path.strip():
            raise InputError("temp_path must be a non-empty string")
        candidate = Path(raw_path).expanduser().resolve()
        allowed_roots = (self.settings.temp_dir.resolve(), self.settings.output_dir.resolve())
        if not any(candidate == root or root in candidate.parents for root in allowed_roots):
            raise InputError("temp_path must be inside TEMP_DIR or OUTPUT_DIR")
        if not candidate.is_file():
            raise InputError("The managed temporary file does not exist")
        filename = self.validator.safe_filename(descriptor.get("filename"), candidate.name)
        mime_type = str(
            descriptor.get("mime_type")
            or descriptor.get("mimeType")
            or self.validator.infer_mime(candidate)
        )
        size = candidate.stat().st_size
        self.validator.validate_declared(filename, mime_type, size)
        destination = workspace.uploads / f"{uuid4().hex}_{filename}"
        async with (
            aiofiles.open(candidate, "rb") as source,
            aiofiles.open(destination, "wb") as output,
        ):
            while chunk := await source.read(1024 * 1024):
                await output.write(chunk)
        return InputMedia(
            source=MediaSource.LOCAL_PATH,
            filename=filename,
            mime_type=mime_type,
            temp_path=destination,
            size=size,
            metadata={"managed_source": str(candidate)},
        )

    async def _decode_base64(
        self, value: str, descriptor: dict[str, Any], workspace: RequestWorkspace
    ) -> InputMedia:
        mime_type = str(
            descriptor.get("mime_type") or descriptor.get("mimeType") or "application/octet-stream"
        )
        if value.startswith("data:"):
            try:
                header, value = value.split(",", 1)
            except ValueError as exc:
                raise InputError("Invalid Base64 data URI") from exc
            if ";base64" not in header:
                raise InputError("Only Base64 data URIs are supported")
            mime_type = header[5:].split(";", 1)[0] or mime_type
        compact = "".join(value.split())
        estimated_size = len(compact) * 3 // 4
        if estimated_size > self.settings.max_upload_size:
            raise InputError("Decoded Base64 media exceeds MAX_UPLOAD_SIZE")
        try:
            decoded = await asyncio.to_thread(base64.b64decode, compact, validate=True)
        except ValueError as exc:
            raise InputError("Media contains invalid Base64 data") from exc
        extension = mimetypes.guess_extension(mime_type) or ".bin"
        supplied_name = descriptor.get("filename") or descriptor.get("fileName")
        filename = self.validator.safe_filename(supplied_name, f"decoded{extension}")
        self.validator.validate_declared(filename, mime_type, len(decoded))
        path = workspace.uploads / f"{uuid4().hex}_{filename}"
        async with aiofiles.open(path, "wb") as output:
            await output.write(decoded)
        source = MediaSource.N8N_BINARY if "data" in descriptor else MediaSource.JSON_BASE64
        return InputMedia(
            source=source,
            filename=filename,
            mime_type=mime_type,
            temp_path=path,
            size=len(decoded),
        )

    def _collect_descriptors(self, value: Any, *, binary: bool = False) -> list[dict[str, Any]]:
        if isinstance(value, list):
            result: list[dict[str, Any]] = []
            for item in value:
                result.extend(self._collect_descriptors(item, binary=binary))
            return result
        if not isinstance(value, dict):
            if binary and isinstance(value, str):
                return [{"data": value}]
            return []
        if any(key in value for key in ("url", "base64", "temp_path", "path")):
            return [value]
        if binary and "data" in value and isinstance(value["data"], str):
            return [value]
        result = []
        if not binary:
            if "url" in value:
                result.append({"url": value["url"], "filename": value.get("filename")})
            if "base64" in value:
                result.append(value)
            for key in ("input", "inputs", "files", "urls"):
                if key in value:
                    items = value[key]
                    if key == "urls" and isinstance(items, list):
                        result.extend({"url": item} for item in items if isinstance(item, str))
                    else:
                        result.extend(self._collect_descriptors(items))
            if "binary" in value:
                result.extend(self._collect_descriptors(value["binary"], binary=True))
            for key, item in value.items():
                if key.startswith("binary."):
                    result.extend(self._collect_descriptors(item, binary=True))
        else:
            for item in value.values():
                result.extend(self._collect_descriptors(item, binary=True))
        return result

    def _params_from_payload(self, payload: dict[str, Any]) -> dict[str, Any]:
        params: dict[str, Any] = {}
        options = payload.get("options")
        if isinstance(options, str):
            options = self._parse_field(options)
        if isinstance(options, dict):
            params.update(options)
        for key, value in payload.items():
            if key not in self.INPUT_KEYS and not key.startswith("binary.") and key != "options":
                params[key] = self._parse_field(value)
        return params

    @staticmethod
    def _parse_field(value: Any) -> Any:
        if isinstance(value, list):
            return [InputResolver._parse_field(item) for item in value]
        if not isinstance(value, str):
            return value
        stripped = value.strip()
        if stripped.startswith(("{", "[")) or stripped in {"true", "false", "null"}:
            try:
                return json.loads(stripped)
            except json.JSONDecodeError:
                return value
        return value

    @staticmethod
    def _as_list(value: Any) -> list[Any]:
        return value if isinstance(value, list) else [value]

    @staticmethod
    def _int(value: Any) -> int | None:
        try:
            return int(value)
        except (TypeError, ValueError):
            return None

    @staticmethod
    def _original_name(path: Path) -> str:
        parts = path.name.split("_", 1)
        return parts[1] if len(parts) == 2 else path.name