nanobot/channels/telegram_media.py

@@ -7,6 +7,7 @@ import mimetypes
 from enum import Enum
 from pathlib import Path
 
+import httpx
 from loguru import logger
 from PIL import Image
 
@@ -105,6 +106,47 @@ def is_heic_format(path: str) -> bool:
     return ext in (".heic", ".heif")
 
 
+async def fetch_media(url: str, max_bytes: int) -> tuple[bytes, str]:
+    """
+    Download media from remote URL.
+
+    Args:
+        url: Remote URL to fetch
+        max_bytes: Maximum size to download
+
+    Returns:
+        Tuple of (content bytes, detected MIME type)
+
+    Raises:
+        ValueError: If download fails or exceeds size limit
+    """
+    try:
+        async with httpx.AsyncClient(timeout=10.0) as client:
+            response = await client.get(url, follow_redirects=True)
+            response.raise_for_status()
+
+            content = response.content
+
+            if len(content) > max_bytes:
+                raise ValueError(f"Media exceeds size limit: {len(content)} > {max_bytes}")
+
+            # Get MIME type from response or detect
+            mime = response.headers.get("content-type", "application/octet-stream")
+            # Strip charset if present (e.g., "image/jpeg; charset=utf-8" → "image/jpeg")
+            mime = mime.split(";")[0].strip()
+
+            # Detect from content if generic type
+            if mime == "application/octet-stream":
+                mime = detect_mime(url, content)
+
+            return content, mime
+
+    except httpx.TimeoutException as e:
+        raise ValueError(f"Download timeout: {url}") from e
+    except httpx.HTTPError as e:
+        raise ValueError(f"Download failed: {url}: {e}") from e
+
+
 def optimize_image(path: str, max_bytes: int = TELEGRAM_PHOTO_SIZE_LIMIT) -> bytes:
     """
     Optimize image to fit under size limit.

tests/test_telegram_media.py

@@ -2,6 +2,7 @@
 
 import io
 
+import pytest
 from PIL import Image
 
 
@@ -162,3 +163,42 @@ def test_optimize_image_png_preserve_alpha(tmp_path):
     # Load and verify alpha channel preserved
     img_opt = Image.open(io.BytesIO(optimized))
     assert img_opt.mode == "RGBA"
+
+
+@pytest.mark.asyncio
+async def test_fetch_media_success():
+    """Test fetching media from remote URL."""
+    from unittest.mock import AsyncMock, MagicMock, patch
+
+    from nanobot.channels.telegram_media import fetch_media
+
+    # Mock httpx response
+    mock_content = b"fake image data"
+    mock_response = MagicMock()
+    mock_response.content = mock_content
+    mock_response.headers = {"content-type": "image/jpeg"}
+    mock_response.raise_for_status = MagicMock()
+
+    with patch("httpx.AsyncClient") as mock_client:
+        mock_client.return_value.__aenter__.return_value.get = AsyncMock(return_value=mock_response)
+
+        content, mime = await fetch_media("https://example.com/image.jpg", max_bytes=10_000_000)
+
+        assert content == mock_content
+        assert mime == "image/jpeg"
+
+
+@pytest.mark.asyncio
+async def test_fetch_media_timeout():
+    """Test fetch media handles timeout."""
+    from unittest.mock import AsyncMock, patch
+
+    import httpx
+
+    from nanobot.channels.telegram_media import fetch_media
+
+    with patch("httpx.AsyncClient") as mock_client:
+        mock_client.return_value.__aenter__.return_value.get = AsyncMock(side_effect=httpx.TimeoutException("timeout"))
+
+        with pytest.raises(ValueError, match="timeout"):
+            await fetch_media("https://example.com/image.jpg", max_bytes=10_000_000)