project-kino/nextcloud-import/backend-nextcloud-variant/app/providers/internet_archive.py

from collections.abc import Callable
from pathlib import Path
from urllib.parse import quote, urlparse

import httpx

from app.providers.base import DownloadOptions, DownloadResult, MediaMetadata, ProviderError
from app.providers.mediathek import MEDIA_EXTENSIONS, MediathekProvider

ARCHIVE_HOSTS = {"archive.org", "www.archive.org"}
METADATA_BASE = "https://archive.org/metadata"
DOWNLOAD_BASE = "https://archive.org/download"


class InternetArchiveProvider:
    name = "internet_archive"

    def can_handle(self, url: str) -> bool:
        parsed = urlparse(url)
        return parsed.scheme in {"http", "https"} and (parsed.hostname or "").lower() in ARCHIVE_HOSTS and self._identifier_from_url(url) is not None

    async def probe(self, url: str) -> MediaMetadata:
        identifier = self._identifier_from_url(url)
        if not identifier:
            raise ProviderError("Internet Archive URL must contain an item identifier")
        metadata = await self._metadata(identifier)
        meta = metadata.get("metadata") or {}
        files = metadata.get("files") or []
        selected = self._select_media_file(files)
        if selected is None:
            raise ProviderError("Internet Archive item has no accepted public media file")
        title = self._string(meta.get("title")) or identifier
        description = self._string(meta.get("description"))
        return MediaMetadata(
            provider=self.name,
            title=title,
            description=description,
            external_id=identifier,
        )

    async def download(
        self,
        url: str,
        target_dir: Path,
        options: DownloadOptions,
        progress_callback: Callable[[float], None] | None = None,
    ) -> DownloadResult:
        identifier = self._identifier_from_url(url)
        if not identifier:
            raise ProviderError("Internet Archive URL must contain an item identifier")
        metadata = await self._metadata(identifier)
        selected = self._select_media_file(metadata.get("files") or [])
        if selected is None:
            raise ProviderError("Internet Archive item has no accepted public media file")
        file_name = selected["name"]
        safe_url = f"{DOWNLOAD_BASE}/{quote(identifier, safe='')}/{quote(file_name, safe='/')}"
        return await MediathekProvider().download(safe_url, target_dir, options, progress_callback=progress_callback)

    async def _metadata(self, identifier: str) -> dict:
        url = f"{METADATA_BASE}/{quote(identifier, safe='')}"
        async with httpx.AsyncClient(timeout=httpx.Timeout(15, read=20), follow_redirects=True) as client:
            response = await client.get(url)
            if response.status_code == 404:
                raise ProviderError(f"Internet Archive item not found: {identifier}")
            response.raise_for_status()
            data = response.json()
        if not isinstance(data, dict):
            raise ProviderError("Internet Archive metadata response was not an object")
        return data

    def _identifier_from_url(self, url: str) -> str | None:
        parsed = urlparse(url)
        parts = [part for part in parsed.path.split("/") if part]
        if not parts:
            return None
        if parts[0] in {"details", "download", "metadata"} and len(parts) >= 2:
            return parts[1]
        return parts[0]

    def _select_media_file(self, files: list[dict]) -> dict | None:
        candidates: list[tuple[int, dict]] = []
        for item in files:
            if not isinstance(item, dict):
                continue
            name = item.get("name")
            if not isinstance(name, str) or not name or name.endswith("/") or ".." in Path(name).parts:
                continue
            lower = name.lower()
            ext = Path(lower).suffix
            fmt = str(item.get("format") or "").lower()
            source = str(item.get("source") or "").lower()
            if ext not in MEDIA_EXTENSIONS and "mpeg4" not in fmt and "h.264" not in fmt:
                continue
            if source in {"metadata", "derivative"} and ext not in {".mp4", ".m4v", ".mp3", ".m4a", ".ogg", ".webm"}:
                continue
            size = self._int_or_zero(item.get("size"))
            # Prefer original/public mp4-ish media, then larger files as likely higher quality.
            score = 0
            if source == "original":
                score += 1000
            if ext in {".mp4", ".m4v", ".webm"}:
                score += 500
            if size:
                score += min(size // 1_000_000, 400)
            candidates.append((score, item))
        if not candidates:
            return None
        candidates.sort(key=lambda pair: pair[0], reverse=True)
        return candidates[0][1]

    def _int_or_zero(self, value: object) -> int:
        try:
            return int(value)  # type: ignore[arg-type]
        except (TypeError, ValueError):
            return 0

    def _string(self, value: object) -> str | None:
        if isinstance(value, str):
            return value.strip() or None
        if isinstance(value, list) and value and isinstance(value[0], str):
            return value[0].strip() or None
        return None