Add asset preprocessing filter and enhance asset data

Features:
- Filter unprocessed assets from events and get_assets service
  - Videos require completed transcoding (encodedVideoPath)
  - Photos require generated thumbnails (thumbhash)
- Add photo_url field for images (preview-sized thumbnail)
- Simplify asset attribute names (remove asset_ prefix)
- Prioritize user-added descriptions over EXIF descriptions

Documentation:
- Update README with new asset fields and preprocessing note
- Update services.yaml parameter descriptions

Version: 2.1.0

Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>

README.md

@@ -4,6 +4,8 @@
 
 A Home Assistant custom integration that monitors [Immich](https://immich.app/) photo/video library albums for changes and exposes them as Home Assistant entities with event-firing capabilities.
 
+> **Tip:** For the best experience, use this integration with the [Immich Album Watcher Blueprint](https://github.com/DolgolyovAlexei/haos-blueprints/blob/main/Common/Immich%20Album%20Watcher.yaml) to easily create automations for album change notifications.
+
 ## Features
 
 - **Album Monitoring** - Watch selected Immich albums for asset additions and removals
@@ -31,7 +33,7 @@ A Home Assistant custom integration that monitors [Immich](https://immich.app/)
   - Detected people in the asset
 - **Services** - Custom service calls:
   - `immich_album_watcher.refresh` - Force immediate data refresh
-  - `immich_album_watcher.get_recent_assets` - Get recent assets from an album
+  - `immich_album_watcher.get_assets` - Get assets from an album with filtering and ordering
   - `immich_album_watcher.send_telegram_notification` - Send text, photo, video, or media group to Telegram
 - **Share Link Management** - Button entities to create and delete share links:
   - Create/delete public (unprotected) share links
@@ -60,8 +62,6 @@ A Home Assistant custom integration that monitors [Immich](https://immich.app/)
 3. Restart Home Assistant
 4. Add the integration via **Settings** → **Devices & Services** → **Add Integration**
 
-> **Tip:** For the best experience, use this integration with the [Immich Album Watcher Blueprint](https://github.com/DolgolyovAlexei/haos-blueprints/blob/main/Common/Immich%20Album%20Watcher.yaml) to easily create automations for album change notifications.
-
 ## Configuration
 
 | Option | Description | Default |
@@ -103,16 +103,59 @@ Force an immediate refresh of all album data:
 service: immich_album_watcher.refresh
 ```
 
-### Get Recent Assets
+### Get Assets
 
-Get the most recent assets from a specific album (returns response data):
+Get assets from a specific album with optional filtering and ordering (returns response data). Only returns fully processed assets (videos with completed transcoding, photos with generated thumbnails).
 
 ```yaml
-service: immich_album_watcher.get_recent_assets
+service: immich_album_watcher.get_assets
+target:
+  entity_id: sensor.album_name_asset_count
+data:
+  count: 10                    # Maximum number of assets (1-100)
+  filter: "favorite"           # Options: "none", "favorite", "rating"
+  filter_min_rating: 4         # Min rating (1-5), used when filter="rating"
+  order: "descending"          # Options: "ascending", "descending", "random"
+```
+
+**Parameters:**
+
+- `count` (optional, default: 10): Maximum number of assets to return (1-100)
+- `filter` (optional, default: "none"): Filter type
+  - `"none"`: No filtering, return all assets
+  - `"favorite"`: Return only favorite assets
+  - `"rating"`: Return assets with rating >= `filter_min_rating`
+- `filter_min_rating` (optional, default: 1): Minimum rating (1-5 stars), used when `filter="rating"`
+- `order` (optional, default: "descending"): Sort order by creation date
+  - `"ascending"`: Oldest first
+  - `"descending"`: Newest first
+  - `"random"`: Random order
+
+**Examples:**
+
+Get 5 most recent favorite assets:
+
+```yaml
+service: immich_album_watcher.get_assets
+target:
+  entity_id: sensor.album_name_asset_count
+data:
+  count: 5
+  filter: "favorite"
+  order: "descending"
+```
+
+Get 10 random assets rated 3 stars or higher:
+
+```yaml
+service: immich_album_watcher.get_assets
 target:
   entity_id: sensor.album_name_asset_count
 data:
   count: 10
+  filter: "rating"
+  filter_min_rating: 3
+  order: "random"
 ```
 
 ### Send Telegram Notification
@@ -181,7 +224,7 @@ data:
   chat_id: "-1001234567890"
   caption: |
     <b>Album Updated!</b>
-    New photos by <i>{{ trigger.event.data.added_assets[0].asset_owner }}</i>
+    New photos by <i>{{ trigger.event.data.added_assets[0].owner }}</i>
     <a href="https://immich.example.com/album">View Album</a>
   parse_mode: "HTML"  # Default, can be omitted
 ```
@@ -213,6 +256,8 @@ data:
 | `max_group_size` | Maximum media items per group (2-10). Large lists split into multiple groups. Default: 10 | No |
 | `chunk_delay` | Delay in milliseconds between sending multiple groups (0-60000). Useful for rate limiting. Default: 0 | No |
 | `wait_for_response` | Wait for Telegram to finish processing. Set to `false` for fire-and-forget (automation continues immediately). Default: `true` | No |
+| `max_asset_data_size` | Maximum asset size in bytes. Assets exceeding this limit will be skipped. Default: no limit | No |
+| `send_large_photos_as_documents` | Handle photos exceeding Telegram limits (10MB or 10000px dimension sum). If `true`, send as documents. If `false`, skip oversized photos. Default: `false` | No |
 
 The service returns a response with `success` status and `message_id` (single message), `message_ids` (media group), or `groups_sent` (number of groups when split). When `wait_for_response` is `false`, the service returns immediately with `{"success": true, "status": "queued"}` while processing continues in the background.
 
@@ -293,15 +338,22 @@ Each item in the `added_assets` list contains the following fields:
 | Field | Description |
 |-------|-------------|
 | `id` | Unique asset ID |
-| `asset_type` | Type of asset (`IMAGE` or `VIDEO`) |
-| `asset_filename` | Original filename of the asset |
-| `asset_created` | Date/time when the asset was originally created |
-| `asset_owner` | Display name of the user who owns the asset |
-| `asset_owner_id` | Unique ID of the user who owns the asset |
-| `asset_description` | Description/caption of the asset (from EXIF data) |
-| `asset_url` | Public URL to view the asset (only present if album has a shared link) |
+| `type` | Type of asset (`IMAGE` or `VIDEO`) |
+| `filename` | Original filename of the asset |
+| `created_at` | Date/time when the asset was originally created |
+| `owner` | Display name of the user who owns the asset |
+| `owner_id` | Unique ID of the user who owns the asset |
+| `description` | Description/caption of the asset (from EXIF data) |
+| `is_favorite` | Whether the asset is marked as favorite (`true` or `false`) |
+| `rating` | User rating of the asset (1-5 stars, or `null` if not rated) |
+| `url` | Public URL to view the asset (only present if album has a shared link) |
+| `download_url` | Direct download URL for the original file (if shared link exists) |
+| `playback_url` | Video playback URL (for VIDEO assets only, if shared link exists) |
+| `photo_url` | Photo preview URL (for IMAGE assets only, if shared link exists) |
 | `people` | List of people detected in this specific asset |
 
+> **Note:** Assets are only included in events and service responses when they are fully processed by Immich. For videos, this means transcoding must be complete (with `encodedVideoPath`). For photos, thumbnail generation must be complete (with `thumbhash`). This ensures that all media URLs are valid and accessible. Unprocessed assets are silently ignored until their processing completes.
+
 Example accessing asset owner in an automation:
 
 ```yaml
@@ -315,7 +367,7 @@ automation:
         data:
           title: "New Photos"
           message: >
-            {{ trigger.event.data.added_assets[0].asset_owner }} added
+            {{ trigger.event.data.added_assets[0].owner }} added
             {{ trigger.event.data.added_count }} photos to {{ trigger.event.data.album_name }}
 ```
 

custom_components/immich_album_watcher/const.py

@@ -57,15 +57,17 @@ ATTR_OLD_NAME: Final = "old_name"
 ATTR_NEW_NAME: Final = "new_name"
 ATTR_OLD_SHARED: Final = "old_shared"
 ATTR_NEW_SHARED: Final = "new_shared"
-ATTR_ASSET_TYPE: Final = "asset_type"
-ATTR_ASSET_FILENAME: Final = "asset_filename"
-ATTR_ASSET_CREATED: Final = "asset_created"
-ATTR_ASSET_OWNER: Final = "asset_owner"
-ATTR_ASSET_OWNER_ID: Final = "asset_owner_id"
-ATTR_ASSET_URL: Final = "asset_url"
-ATTR_ASSET_DOWNLOAD_URL: Final = "asset_download_url"
-ATTR_ASSET_PLAYBACK_URL: Final = "asset_playback_url"
-ATTR_ASSET_DESCRIPTION: Final = "asset_description"
+ATTR_ASSET_TYPE: Final = "type"
+ATTR_ASSET_FILENAME: Final = "filename"
+ATTR_ASSET_CREATED: Final = "created_at"
+ATTR_ASSET_OWNER: Final = "owner"
+ATTR_ASSET_OWNER_ID: Final = "owner_id"
+ATTR_ASSET_URL: Final = "url"
+ATTR_ASSET_DOWNLOAD_URL: Final = "download_url"
+ATTR_ASSET_PLAYBACK_URL: Final = "playback_url"
+ATTR_ASSET_DESCRIPTION: Final = "description"
+ATTR_ASSET_IS_FAVORITE: Final = "is_favorite"
+ATTR_ASSET_RATING: Final = "rating"
 
 # Asset types
 ASSET_TYPE_IMAGE: Final = "IMAGE"
@@ -76,5 +78,5 @@ PLATFORMS: Final = ["sensor", "binary_sensor", "camera", "text", "button"]
 
 # Services
 SERVICE_REFRESH: Final = "refresh"
-SERVICE_GET_RECENT_ASSETS: Final = "get_recent_assets"
+SERVICE_GET_ASSETS: Final = "get_assets"
 SERVICE_SEND_TELEGRAM_NOTIFICATION: Final = "send_telegram_notification"

custom_components/immich_album_watcher/coordinator.py

@@ -28,11 +28,13 @@ from .const import (
     ATTR_ASSET_DESCRIPTION,
     ATTR_ASSET_DOWNLOAD_URL,
     ATTR_ASSET_FILENAME,
+    ATTR_ASSET_IS_FAVORITE,
     ATTR_ASSET_OWNER,
     ATTR_ASSET_OWNER_ID,
+    ATTR_ASSET_PLAYBACK_URL,
+    ATTR_ASSET_RATING,
     ATTR_ASSET_TYPE,
     ATTR_ASSET_URL,
-    ATTR_ASSET_PLAYBACK_URL,
     ATTR_CHANGE_TYPE,
     ATTR_HUB_NAME,
     ATTR_PEOPLE,
@@ -43,6 +45,7 @@ from .const import (
     ATTR_OLD_SHARED,
     ATTR_NEW_SHARED,
     ATTR_SHARED,
+    ATTR_THUMBNAIL_URL,
     DOMAIN,
     EVENT_ALBUM_CHANGED,
     EVENT_ASSETS_ADDED,
@@ -115,6 +118,9 @@ class AssetInfo:
     owner_name: str = ""
     description: str = ""
     people: list[str] = field(default_factory=list)
+    is_favorite: bool = False
+    rating: int | None = None
+    is_processed: bool = True  # Whether asset is fully processed by Immich
 
     @classmethod
     def from_api_response(
@@ -130,23 +136,74 @@ class AssetInfo:
         if users_cache and owner_id:
             owner_name = users_cache.get(owner_id, "")
 
-        # Get description from exifInfo if available
-        description = ""
+        # Get description - prioritize user-added description over EXIF description
+        description = data.get("description", "") or ""
         exif_info = data.get("exifInfo")
-        if exif_info:
+        if not description and exif_info:
+            # Fall back to EXIF description if no user description
             description = exif_info.get("description", "") or ""
 
+        # Get favorites and rating
+        is_favorite = data.get("isFavorite", False)
+        rating = exif_info.get("rating") if exif_info else None
+
+        # Check if asset is fully processed by Immich
+        asset_type = data.get("type", ASSET_TYPE_IMAGE)
+        is_processed = cls._check_processing_status(data, asset_type)
+
         return cls(
             id=data["id"],
-            type=data.get("type", ASSET_TYPE_IMAGE),
+            type=asset_type,
             filename=data.get("originalFileName", ""),
             created_at=data.get("fileCreatedAt", ""),
             owner_id=owner_id,
             owner_name=owner_name,
             description=description,
             people=people,
+            is_favorite=is_favorite,
+            rating=rating,
+            is_processed=is_processed,
         )
 
+    @staticmethod
+    def _check_processing_status(data: dict[str, Any], asset_type: str) -> bool:
+        """Check if asset has been fully processed by Immich.
+
+        For photos: Check if thumbnails/previews have been generated
+        For videos: Check if video transcoding is complete
+
+        Args:
+            data: Asset data from API response
+            asset_type: Asset type (IMAGE or VIDEO)
+
+        Returns:
+            True if asset is fully processed, False otherwise
+        """
+        if asset_type == ASSET_TYPE_VIDEO:
+            # For videos, check if transcoding is complete
+            # Video is processed if it has an encoded video path or if isOffline is False
+            is_offline = data.get("isOffline", False)
+            if is_offline:
+                return False
+
+            # Check if video has been transcoded (has encoded video path)
+            # Immich uses "encodedVideoPath" or similar field when transcoding is done
+            has_encoded_video = bool(data.get("encodedVideoPath"))
+            return has_encoded_video
+
+        else:  # ASSET_TYPE_IMAGE
+            # For photos, check if thumbnails have been generated
+            # Photos are processed if they have thumbnail/preview paths
+            is_offline = data.get("isOffline", False)
+            if is_offline:
+                return False
+
+            # Check if thumbnails exist
+            has_thumbhash = bool(data.get("thumbhash"))
+            has_thumbnail = has_thumbhash  # If thumbhash exists, thumbnails should exist
+
+            return has_thumbnail
+
 
 @dataclass
 class AlbumData:
@@ -303,33 +360,52 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
                     self._album_name,
                 )
 
-    async def async_get_recent_assets(self, count: int = 10) -> list[dict[str, Any]]:
-        """Get recent assets from the album."""
+    async def async_get_assets(
+        self,
+        count: int = 10,
+        filter: str = "none",
+        filter_min_rating: int = 1,
+        order: str = "descending",
+    ) -> list[dict[str, Any]]:
+        """Get assets from the album with optional filtering and ordering.
+
+        Args:
+            count: Maximum number of assets to return (1-100)
+            filter: Filter type - 'none', 'favorite', or 'rating'
+            filter_min_rating: Minimum rating for assets (1-5), used when filter='rating'
+            order: Sort order - 'ascending', 'descending', or 'random'
+
+        Returns:
+            List of asset data dictionaries
+        """
         if self.data is None:
             return []
 
-        # Sort assets by created_at descending
-        sorted_assets = sorted(
-            self.data.assets.values(),
-            key=lambda a: a.created_at,
-            reverse=True,
-        )[:count]
+        # Start with all processed assets only
+        assets = [a for a in self.data.assets.values() if a.is_processed]
 
+        # Apply filtering
+        if filter == "favorite":
+            assets = [a for a in assets if a.is_favorite]
+        elif filter == "rating":
+            assets = [a for a in assets if a.rating is not None and a.rating >= filter_min_rating]
+
+        # Apply ordering
+        if order == "random":
+            import random
+            random.shuffle(assets)
+        elif order == "ascending":
+            assets = sorted(assets, key=lambda a: a.created_at, reverse=False)
+        else:  # descending (default)
+            assets = sorted(assets, key=lambda a: a.created_at, reverse=True)
+
+        # Limit count
+        assets = assets[:count]
+
+        # Build result with all available asset data (matching event data)
         result = []
-        for asset in sorted_assets:
-            asset_data = {
-                "id": asset.id,
-                "type": asset.type,
-                "filename": asset.filename,
-                "created_at": asset.created_at,
-                "description": asset.description,
-                "people": asset.people,
-                "thumbnail_url": f"{self._url}/api/assets/{asset.id}/thumbnail",
-            }
-            if asset.type == ASSET_TYPE_VIDEO:
-                video_url = self._get_asset_video_url(asset.id)
-                if video_url:
-                    asset_data["video_url"] = video_url
+        for asset in assets:
+            asset_data = self._build_asset_detail(asset, include_thumbnail=True)
             result.append(asset_data)
         return result
 
@@ -501,6 +577,68 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
             return f"{self._url}/api/assets/{asset_id}/video/playback?key={non_expired[0].key}"
         return None
 
+    def _get_asset_photo_url(self, asset_id: str) -> str | None:
+        """Get the preview-sized thumbnail URL for a photo asset."""
+        accessible_links = self._get_accessible_links()
+        if accessible_links:
+            return f"{self._url}/api/assets/{asset_id}/thumbnail?size=preview&key={accessible_links[0].key}"
+        non_expired = [link for link in self._shared_links if not link.is_expired]
+        if non_expired:
+            return f"{self._url}/api/assets/{asset_id}/thumbnail?size=preview&key={non_expired[0].key}"
+        return None
+
+    def _build_asset_detail(
+        self, asset: AssetInfo, include_thumbnail: bool = True
+    ) -> dict[str, Any]:
+        """Build asset detail dictionary with all available data.
+
+        Args:
+            asset: AssetInfo object
+            include_thumbnail: If True, include thumbnail_url
+
+        Returns:
+            Dictionary with asset details (using ATTR_* constants for consistency)
+        """
+        # Base asset data
+        asset_detail = {
+            "id": asset.id,
+            ATTR_ASSET_TYPE: asset.type,
+            ATTR_ASSET_FILENAME: asset.filename,
+            ATTR_ASSET_CREATED: asset.created_at,
+            ATTR_ASSET_OWNER: asset.owner_name,
+            ATTR_ASSET_OWNER_ID: asset.owner_id,
+            ATTR_ASSET_DESCRIPTION: asset.description,
+            ATTR_PEOPLE: asset.people,
+            ATTR_ASSET_IS_FAVORITE: asset.is_favorite,
+            ATTR_ASSET_RATING: asset.rating,
+        }
+
+        # Add thumbnail URL if requested
+        if include_thumbnail:
+            asset_detail[ATTR_THUMBNAIL_URL] = f"{self._url}/api/assets/{asset.id}/thumbnail"
+
+        # Add public viewer URL (web page)
+        asset_url = self._get_asset_public_url(asset.id)
+        if asset_url:
+            asset_detail[ATTR_ASSET_URL] = asset_url
+
+        # Add download URL (direct media file)
+        asset_download_url = self._get_asset_download_url(asset.id)
+        if asset_download_url:
+            asset_detail[ATTR_ASSET_DOWNLOAD_URL] = asset_download_url
+
+        # Add type-specific URLs
+        if asset.type == ASSET_TYPE_VIDEO:
+            asset_video_url = self._get_asset_video_url(asset.id)
+            if asset_video_url:
+                asset_detail[ATTR_ASSET_PLAYBACK_URL] = asset_video_url
+        elif asset.type == ASSET_TYPE_IMAGE:
+            asset_photo_url = self._get_asset_photo_url(asset.id)
+            if asset_photo_url:
+                asset_detail["photo_url"] = asset_photo_url  # TODO: Add ATTR_ASSET_PHOTO_URL constant
+
+        return asset_detail
+
     async def _async_update_data(self) -> AlbumData | None:
         """Fetch data from Immich API."""
         if self._session is None:
@@ -620,34 +758,37 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
         added_ids = new_state.asset_ids - old_state.asset_ids
         removed_ids = old_state.asset_ids - new_state.asset_ids
 
+        # Only include fully processed assets in added_assets
+        added_assets = [
+            new_state.assets[aid]
+            for aid in added_ids
+            if aid in new_state.assets and new_state.assets[aid].is_processed
+        ]
+
         # Detect metadata changes
         name_changed = old_state.name != new_state.name
         sharing_changed = old_state.shared != new_state.shared
 
         # Return None only if nothing changed at all
-        if not added_ids and not removed_ids and not name_changed and not sharing_changed:
+        if not added_assets and not removed_ids and not name_changed and not sharing_changed:
             return None
 
-        # Determine primary change type
+        # Determine primary change type (use added_assets not added_ids)
         change_type = "changed"
-        if name_changed and not added_ids and not removed_ids and not sharing_changed:
+        if name_changed and not added_assets and not removed_ids and not sharing_changed:
             change_type = "album_renamed"
-        elif sharing_changed and not added_ids and not removed_ids and not name_changed:
+        elif sharing_changed and not added_assets and not removed_ids and not name_changed:
             change_type = "album_sharing_changed"
-        elif added_ids and not removed_ids and not name_changed and not sharing_changed:
+        elif added_assets and not removed_ids and not name_changed and not sharing_changed:
             change_type = "assets_added"
-        elif removed_ids and not added_ids and not name_changed and not sharing_changed:
+        elif removed_ids and not added_assets and not name_changed and not sharing_changed:
             change_type = "assets_removed"
 
-        added_assets = [
-            new_state.assets[aid] for aid in added_ids if aid in new_state.assets
-        ]
-
         return AlbumChange(
             album_id=new_state.id,
             album_name=new_state.name,
             change_type=change_type,
-            added_count=len(added_ids),
+            added_count=len(added_assets),  # Count only processed assets
             removed_count=len(removed_ids),
             added_assets=added_assets,
             removed_asset_ids=list(removed_ids),
@@ -661,26 +802,10 @@ class ImmichAlbumWatcherCoordinator(DataUpdateCoordinator[AlbumData | None]):
         """Fire Home Assistant events for album changes."""
         added_assets_detail = []
         for asset in change.added_assets:
-            asset_detail = {
-                "id": asset.id,
-                ATTR_ASSET_TYPE: asset.type,
-                ATTR_ASSET_FILENAME: asset.filename,
-                ATTR_ASSET_CREATED: asset.created_at,
-                ATTR_ASSET_OWNER: asset.owner_name,
-                ATTR_ASSET_OWNER_ID: asset.owner_id,
-                ATTR_ASSET_DESCRIPTION: asset.description,
-                ATTR_PEOPLE: asset.people,
-            }
-            asset_url = self._get_asset_public_url(asset.id)
-            if asset_url:
-                asset_detail[ATTR_ASSET_URL] = asset_url
-            asset_download_url = self._get_asset_download_url(asset.id)
-            if asset_download_url:
-                asset_detail[ATTR_ASSET_DOWNLOAD_URL] = asset_download_url
-            if asset.type == ASSET_TYPE_VIDEO:
-                asset_video_url = self._get_asset_video_url(asset.id)
-                if asset_video_url:
-                    asset_detail[ATTR_ASSET_PLAYBACK_URL] = asset_video_url
+            # Only include fully processed assets
+            if not asset.is_processed:
+                continue
+            asset_detail = self._build_asset_detail(asset, include_thumbnail=False)
             added_assets_detail.append(asset_detail)
 
         event_data = {

custom_components/immich_album_watcher/manifest.json

@@ -8,5 +8,5 @@
     "iot_class": "cloud_polling",
     "issue_tracker": "https://github.com/DolgolyovAlexei/haos-hacs-immich-album-watcher/issues",
     "requirements": [],
-    "version": "2.0.0"
+    "version": "2.1.0"
 }

custom_components/immich_album_watcher/sensor.py

@@ -40,7 +40,7 @@ from .const import (
     CONF_HUB_NAME,
     CONF_TELEGRAM_BOT_TOKEN,
     DOMAIN,
-    SERVICE_GET_RECENT_ASSETS,
+    SERVICE_GET_ASSETS,
     SERVICE_REFRESH,
     SERVICE_SEND_TELEGRAM_NOTIFICATION,
 )
@@ -48,6 +48,10 @@ from .coordinator import AlbumData, ImmichAlbumWatcherCoordinator
 
 _LOGGER = logging.getLogger(__name__)
 
+# Telegram photo limits
+TELEGRAM_MAX_PHOTO_SIZE = 10 * 1024 * 1024  # 10 MB - Telegram's max photo size
+TELEGRAM_MAX_DIMENSION_SUM = 10000  # Maximum sum of width + height in pixels
+
 
 async def async_setup_entry(
     hass: HomeAssistant,
@@ -88,13 +92,18 @@ async def async_setup_entry(
     )
 
     platform.async_register_entity_service(
-        SERVICE_GET_RECENT_ASSETS,
+        SERVICE_GET_ASSETS,
         {
             vol.Optional("count", default=10): vol.All(
                 vol.Coerce(int), vol.Range(min=1, max=100)
             ),
+            vol.Optional("filter", default="none"): vol.In(["none", "favorite", "rating"]),
+            vol.Optional("filter_min_rating", default=1): vol.All(
+                vol.Coerce(int), vol.Range(min=1, max=5)
+            ),
+            vol.Optional("order", default="descending"): vol.In(["ascending", "descending", "random"]),
         },
-        "async_get_recent_assets",
+        "async_get_assets",
         supports_response=SupportsResponse.ONLY,
     )
 
@@ -171,9 +180,20 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         """Refresh data for this album."""
         await self.coordinator.async_refresh_now()
 
-    async def async_get_recent_assets(self, count: int = 10) -> ServiceResponse:
-        """Get recent assets for this album."""
-        assets = await self.coordinator.async_get_recent_assets(count)
+    async def async_get_assets(
+        self,
+        count: int = 10,
+        filter: str = "none",
+        filter_min_rating: int = 1,
+        order: str = "descending",
+    ) -> ServiceResponse:
+        """Get assets for this album with optional filtering and ordering."""
+        assets = await self.coordinator.async_get_assets(
+            count=count,
+            filter=filter,
+            filter_min_rating=filter_min_rating,
+            order=order,
+        )
         return {"assets": assets}
 
     async def async_send_telegram_notification(
@@ -188,6 +208,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         max_group_size: int = 10,
         chunk_delay: int = 0,
         wait_for_response: bool = True,
+        max_asset_data_size: int | None = None,
+        send_large_photos_as_documents: bool = False,
     ) -> ServiceResponse:
         """Send notification to Telegram.
 
@@ -216,6 +238,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                     parse_mode=parse_mode,
                     max_group_size=max_group_size,
                     chunk_delay=chunk_delay,
+                    max_asset_data_size=max_asset_data_size,
+                    send_large_photos_as_documents=send_large_photos_as_documents,
                 )
             )
             return {"success": True, "status": "queued", "message": "Notification queued for background processing"}
@@ -231,6 +255,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
             parse_mode=parse_mode,
             max_group_size=max_group_size,
             chunk_delay=chunk_delay,
+            max_asset_data_size=max_asset_data_size,
+            send_large_photos_as_documents=send_large_photos_as_documents,
         )
 
     async def _execute_telegram_notification(
@@ -244,6 +270,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         parse_mode: str = "HTML",
         max_group_size: int = 10,
         chunk_delay: int = 0,
+        max_asset_data_size: int | None = None,
+        send_large_photos_as_documents: bool = False,
     ) -> ServiceResponse:
         """Execute the Telegram notification (internal method)."""
         import json
@@ -270,18 +298,20 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         # Handle single photo
         if len(urls) == 1 and urls[0].get("type", "photo") == "photo":
             return await self._send_telegram_photo(
-                session, token, chat_id, urls[0].get("url"), caption, reply_to_message_id, parse_mode
+                session, token, chat_id, urls[0].get("url"), caption, reply_to_message_id, parse_mode,
+                max_asset_data_size, send_large_photos_as_documents
             )
 
         # Handle single video
         if len(urls) == 1 and urls[0].get("type") == "video":
             return await self._send_telegram_video(
-                session, token, chat_id, urls[0].get("url"), caption, reply_to_message_id, parse_mode
+                session, token, chat_id, urls[0].get("url"), caption, reply_to_message_id, parse_mode, max_asset_data_size
             )
 
         # Handle multiple items - send as media group(s)
         return await self._send_telegram_media_group(
-            session, token, chat_id, urls, caption, reply_to_message_id, max_group_size, chunk_delay, parse_mode
+            session, token, chat_id, urls, caption, reply_to_message_id, max_group_size, chunk_delay, parse_mode,
+            max_asset_data_size, send_large_photos_as_documents
         )
 
     async def _send_telegram_message(
@@ -332,6 +362,104 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
             _LOGGER.error("Telegram message send failed: %s", err)
             return {"success": False, "error": str(err)}
 
+    def _log_telegram_error(
+        self,
+        error_code: int | None,
+        description: str,
+        data: bytes | None = None,
+        media_type: str = "photo",
+    ) -> None:
+        """Log detailed Telegram API error with diagnostics.
+
+        Args:
+            error_code: Telegram error code
+            description: Error description from Telegram
+            data: Media data bytes (optional, for size diagnostics)
+            media_type: Type of media (photo/video)
+        """
+        error_msg = f"Telegram API error ({error_code}): {description}"
+
+        # Add diagnostic information based on error type
+        if data:
+            error_msg += f" | Media size: {len(data)} bytes ({len(data) / (1024 * 1024):.2f} MB)"
+
+            # Check dimensions for photos
+            if media_type == "photo":
+                try:
+                    from PIL import Image
+                    import io
+
+                    img = Image.open(io.BytesIO(data))
+                    width, height = img.size
+                    dimension_sum = width + height
+                    error_msg += f" | Dimensions: {width}x{height} (sum={dimension_sum})"
+
+                    # Highlight limit violations
+                    if len(data) > TELEGRAM_MAX_PHOTO_SIZE:
+                        error_msg += f" | EXCEEDS size limit ({TELEGRAM_MAX_PHOTO_SIZE / (1024 * 1024):.0f} MB)"
+                    if dimension_sum > TELEGRAM_MAX_DIMENSION_SUM:
+                        error_msg += f" | EXCEEDS dimension limit ({TELEGRAM_MAX_DIMENSION_SUM})"
+                except Exception:
+                    pass
+
+        # Provide suggestions based on error description
+        suggestions = []
+        if "dimension" in description.lower() or "PHOTO_INVALID_DIMENSIONS" in description:
+            suggestions.append("Photo dimensions too large - consider setting send_large_photos_as_documents=true")
+        elif "too large" in description.lower() or error_code == 413:
+            suggestions.append("File size too large - consider setting send_large_photos_as_documents=true or max_asset_data_size to skip large files")
+        elif "entity too large" in description.lower():
+            suggestions.append("Request entity too large - reduce max_group_size or set max_asset_data_size")
+
+        if suggestions:
+            error_msg += f" | Suggestions: {'; '.join(suggestions)}"
+
+        _LOGGER.error(error_msg)
+
+    def _check_telegram_photo_limits(
+        self,
+        data: bytes,
+    ) -> tuple[bool, str | None, int | None, int | None]:
+        """Check if photo data exceeds Telegram photo limits.
+
+        Telegram limits for photos:
+        - Max file size: 10 MB
+        - Max dimension sum: ~10,000 pixels (width + height)
+
+        Returns:
+            Tuple of (exceeds_limits, reason, width, height)
+            - exceeds_limits: True if photo exceeds limits
+            - reason: Human-readable reason (None if within limits)
+            - width: Image width in pixels (None if PIL not available)
+            - height: Image height in pixels (None if PIL not available)
+        """
+        # Check file size
+        if len(data) > TELEGRAM_MAX_PHOTO_SIZE:
+            return True, f"size {len(data)} bytes exceeds {TELEGRAM_MAX_PHOTO_SIZE} bytes limit", None, None
+
+        # Try to check dimensions using PIL
+        try:
+            from PIL import Image
+            import io
+
+            img = Image.open(io.BytesIO(data))
+            width, height = img.size
+            dimension_sum = width + height
+
+            if dimension_sum > TELEGRAM_MAX_DIMENSION_SUM:
+                return True, f"dimensions {width}x{height} (sum={dimension_sum}) exceed {TELEGRAM_MAX_DIMENSION_SUM} limit", width, height
+
+            return False, None, width, height
+        except ImportError:
+            # PIL not available, can't check dimensions
+            _LOGGER.debug("PIL not available, skipping dimension check")
+            return False, None, None, None
+        except Exception as e:
+            # Failed to check dimensions
+            _LOGGER.debug("Failed to check photo dimensions: %s", e)
+            return False, None, None, None
+
+
     async def _send_telegram_photo(
         self,
         session: Any,
@@ -341,6 +469,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         caption: str | None = None,
         reply_to_message_id: int | None = None,
         parse_mode: str = "HTML",
+        max_asset_data_size: int | None = None,
+        send_large_photos_as_documents: bool = False,
     ) -> ServiceResponse:
         """Send a single photo to Telegram."""
         import aiohttp
@@ -361,6 +491,37 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                 data = await resp.read()
                 _LOGGER.debug("Downloaded photo: %d bytes", len(data))
 
+            # Check if photo exceeds max size limit (user-defined limit)
+            if max_asset_data_size is not None and len(data) > max_asset_data_size:
+                _LOGGER.warning(
+                    "Photo size (%d bytes) exceeds max_asset_data_size limit (%d bytes), skipping",
+                    len(data), max_asset_data_size
+                )
+                return {
+                    "success": False,
+                    "error": f"Photo size ({len(data)} bytes) exceeds max_asset_data_size limit ({max_asset_data_size} bytes)",
+                    "skipped": True,
+                }
+
+            # Check if photo exceeds Telegram's photo limits
+            exceeds_limits, reason, width, height = self._check_telegram_photo_limits(data)
+            if exceeds_limits:
+                if send_large_photos_as_documents:
+                    # Send as document instead
+                    _LOGGER.info("Photo %s, sending as document", reason)
+                    return await self._send_telegram_document(
+                        session, token, chat_id, data, "photo.jpg",
+                        caption, reply_to_message_id, parse_mode
+                    )
+                else:
+                    # Skip oversized photo
+                    _LOGGER.warning("Photo %s, skipping (set send_large_photos_as_documents=true to send as document)", reason)
+                    return {
+                        "success": False,
+                        "error": f"Photo {reason}",
+                        "skipped": True,
+                    }
+
             # Build multipart form
             form = FormData()
             form.add_field("chat_id", chat_id)
@@ -386,7 +547,13 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                         "message_id": result.get("result", {}).get("message_id"),
                     }
                 else:
-                    _LOGGER.error("Telegram API error: %s", result)
+                    # Log detailed error with diagnostics
+                    self._log_telegram_error(
+                        error_code=result.get("error_code"),
+                        description=result.get("description", "Unknown Telegram error"),
+                        data=data,
+                        media_type="photo",
+                    )
                     return {
                         "success": False,
                         "error": result.get("description", "Unknown Telegram error"),
@@ -405,6 +572,7 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         caption: str | None = None,
         reply_to_message_id: int | None = None,
         parse_mode: str = "HTML",
+        max_asset_data_size: int | None = None,
     ) -> ServiceResponse:
         """Send a single video to Telegram."""
         import aiohttp
@@ -425,6 +593,18 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                 data = await resp.read()
                 _LOGGER.debug("Downloaded video: %d bytes", len(data))
 
+            # Check if video exceeds max size limit
+            if max_asset_data_size is not None and len(data) > max_asset_data_size:
+                _LOGGER.warning(
+                    "Video size (%d bytes) exceeds max_asset_data_size limit (%d bytes), skipping",
+                    len(data), max_asset_data_size
+                )
+                return {
+                    "success": False,
+                    "error": f"Video size ({len(data)} bytes) exceeds max_asset_data_size limit ({max_asset_data_size} bytes)",
+                    "skipped": True,
+                }
+
             # Build multipart form
             form = FormData()
             form.add_field("chat_id", chat_id)
@@ -450,7 +630,13 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                         "message_id": result.get("result", {}).get("message_id"),
                     }
                 else:
-                    _LOGGER.error("Telegram API error: %s", result)
+                    # Log detailed error with diagnostics
+                    self._log_telegram_error(
+                        error_code=result.get("error_code"),
+                        description=result.get("description", "Unknown Telegram error"),
+                        data=data,
+                        media_type="video",
+                    )
                     return {
                         "success": False,
                         "error": result.get("description", "Unknown Telegram error"),
@@ -460,6 +646,63 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
             _LOGGER.error("Telegram video upload failed: %s", err)
             return {"success": False, "error": str(err)}
 
+    async def _send_telegram_document(
+        self,
+        session: Any,
+        token: str,
+        chat_id: str,
+        data: bytes,
+        filename: str = "photo.jpg",
+        caption: str | None = None,
+        reply_to_message_id: int | None = None,
+        parse_mode: str = "HTML",
+    ) -> ServiceResponse:
+        """Send a photo as a document to Telegram (for oversized photos)."""
+        import aiohttp
+        from aiohttp import FormData
+
+        try:
+            # Build multipart form
+            form = FormData()
+            form.add_field("chat_id", chat_id)
+            form.add_field("document", data, filename=filename, content_type="image/jpeg")
+            form.add_field("parse_mode", parse_mode)
+
+            if caption:
+                form.add_field("caption", caption)
+
+            if reply_to_message_id:
+                form.add_field("reply_to_message_id", str(reply_to_message_id))
+
+            # Send to Telegram
+            telegram_url = f"https://api.telegram.org/bot{token}/sendDocument"
+
+            _LOGGER.debug("Uploading oversized photo as document to Telegram (%d bytes)", len(data))
+            async with session.post(telegram_url, data=form) as response:
+                result = await response.json()
+                _LOGGER.debug("Telegram API response: status=%d, ok=%s", response.status, result.get("ok"))
+                if response.status == 200 and result.get("ok"):
+                    return {
+                        "success": True,
+                        "message_id": result.get("result", {}).get("message_id"),
+                    }
+                else:
+                    # Log detailed error with diagnostics
+                    self._log_telegram_error(
+                        error_code=result.get("error_code"),
+                        description=result.get("description", "Unknown Telegram error"),
+                        data=data,
+                        media_type="document",
+                    )
+                    return {
+                        "success": False,
+                        "error": result.get("description", "Unknown Telegram error"),
+                        "error_code": result.get("error_code"),
+                    }
+        except aiohttp.ClientError as err:
+            _LOGGER.error("Telegram document upload failed: %s", err)
+            return {"success": False, "error": str(err)}
+
     async def _send_telegram_media_group(
         self,
         session: Any,
@@ -471,6 +714,8 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
         max_group_size: int = 10,
         chunk_delay: int = 0,
         parse_mode: str = "HTML",
+        max_asset_data_size: int | None = None,
+        send_large_photos_as_documents: bool = False,
     ) -> ServiceResponse:
         """Send media URLs to Telegram as media group(s).
 
@@ -511,12 +756,13 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                 if media_type == "photo":
                     _LOGGER.debug("Sending chunk %d/%d as single photo", chunk_idx + 1, len(chunks))
                     result = await self._send_telegram_photo(
-                        session, token, chat_id, url, chunk_caption, chunk_reply_to, parse_mode
+                        session, token, chat_id, url, chunk_caption, chunk_reply_to, parse_mode,
+                        max_asset_data_size, send_large_photos_as_documents
                     )
                 else:  # video
                     _LOGGER.debug("Sending chunk %d/%d as single video", chunk_idx + 1, len(chunks))
                     result = await self._send_telegram_video(
-                        session, token, chat_id, url, chunk_caption, chunk_reply_to, parse_mode
+                        session, token, chat_id, url, chunk_caption, chunk_reply_to, parse_mode, max_asset_data_size
                     )
 
                 if not result.get("success"):
@@ -529,7 +775,10 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
             _LOGGER.debug("Sending chunk %d/%d as media group (%d items)", chunk_idx + 1, len(chunks), len(chunk))
 
             # Download all media files for this chunk
-            media_files: list[tuple[str, bytes, str]] = []
+            media_files: list[tuple[str, bytes, str]] = []  # (type, data, filename)
+            oversized_photos: list[tuple[bytes, str | None]] = []  # For send_large_photos_as_documents=true
+            skipped_count = 0
+
             for i, item in enumerate(chunk):
                 url = item.get("url")
                 media_type = item.get("type", "photo")
@@ -555,72 +804,139 @@ class ImmichAlbumBaseSensor(CoordinatorEntity[ImmichAlbumWatcherCoordinator], Se
                                 "error": f"Failed to download media {chunk_idx * max_group_size + i}: HTTP {resp.status}",
                             }
                         data = await resp.read()
+                        _LOGGER.debug("Downloaded media %d: %d bytes", chunk_idx * max_group_size + i, len(data))
+
+                        # Check if media exceeds max_asset_data_size limit (user-defined limit for skipping)
+                        if max_asset_data_size is not None and len(data) > max_asset_data_size:
+                            _LOGGER.warning(
+                                "Media %d size (%d bytes) exceeds max_asset_data_size limit (%d bytes), skipping",
+                                chunk_idx * max_group_size + i, len(data), max_asset_data_size
+                            )
+                            skipped_count += 1
+                            continue
+
+                        # For photos, check Telegram limits
+                        if media_type == "photo":
+                            exceeds_limits, reason, width, height = self._check_telegram_photo_limits(data)
+                            if exceeds_limits:
+                                if send_large_photos_as_documents:
+                                    # Separate this photo to send as document later
+                                    # Caption only on first item of first chunk
+                                    photo_caption = caption if chunk_idx == 0 and i == 0 and len(media_files) == 0 else None
+                                    oversized_photos.append((data, photo_caption))
+                                    _LOGGER.info("Photo %d %s, will send as document", i, reason)
+                                    continue
+                                else:
+                                    # Skip oversized photo
+                                    _LOGGER.warning("Photo %d %s, skipping (set send_large_photos_as_documents=true to send as document)", i, reason)
+                                    skipped_count += 1
+                                    continue
+
                         ext = "jpg" if media_type == "photo" else "mp4"
                         filename = f"media_{chunk_idx * max_group_size + i}.{ext}"
                         media_files.append((media_type, data, filename))
-                        _LOGGER.debug("Downloaded media %d: %d bytes", chunk_idx * max_group_size + i, len(data))
                 except aiohttp.ClientError as err:
                     return {
                         "success": False,
                         "error": f"Failed to download media {chunk_idx * max_group_size + i}: {err}",
                     }
 
-            # Build multipart form
-            form = FormData()
-            form.add_field("chat_id", chat_id)
+            # Skip this chunk if all files were filtered out
+            if not media_files and not oversized_photos:
+                _LOGGER.info("Chunk %d/%d: all %d media items skipped",
+                            chunk_idx + 1, len(chunks), len(chunk))
+                continue
 
-            # Only use reply_to_message_id for the first chunk
-            if chunk_idx == 0 and reply_to_message_id:
-                form.add_field("reply_to_message_id", str(reply_to_message_id))
+            # Send media group if we have normal-sized files
+            if media_files:
+                # Build multipart form
+                form = FormData()
+                form.add_field("chat_id", chat_id)
 
-            # Build media JSON with attach:// references
-            media_json = []
-            for i, (media_type, data, filename) in enumerate(media_files):
-                attach_name = f"file{i}"
-                media_item: dict[str, Any] = {
-                    "type": media_type,
-                    "media": f"attach://{attach_name}",
-                }
-                # Only add caption to the first item of the first chunk
-                if chunk_idx == 0 and i == 0 and caption:
-                    media_item["caption"] = caption
-                    media_item["parse_mode"] = parse_mode
-                media_json.append(media_item)
+                # Only use reply_to_message_id for the first chunk
+                if chunk_idx == 0 and reply_to_message_id:
+                    form.add_field("reply_to_message_id", str(reply_to_message_id))
 
-                content_type = "image/jpeg" if media_type == "photo" else "video/mp4"
-                form.add_field(attach_name, data, filename=filename, content_type=content_type)
+                # Build media JSON with attach:// references
+                media_json = []
+                for i, (media_type, data, filename) in enumerate(media_files):
+                    attach_name = f"file{i}"
+                    media_item: dict[str, Any] = {
+                        "type": media_type,
+                        "media": f"attach://{attach_name}",
+                    }
+                    # Only add caption to the first item of the first chunk (if no oversized photos with caption)
+                    if chunk_idx == 0 and i == 0 and caption and not oversized_photos:
+                        media_item["caption"] = caption
+                        media_item["parse_mode"] = parse_mode
+                    media_json.append(media_item)
 
-            form.add_field("media", json.dumps(media_json))
+                    content_type = "image/jpeg" if media_type == "photo" else "video/mp4"
+                    form.add_field(attach_name, data, filename=filename, content_type=content_type)
 
-            # Send to Telegram
-            telegram_url = f"https://api.telegram.org/bot{token}/sendMediaGroup"
+                form.add_field("media", json.dumps(media_json))
 
-            try:
-                _LOGGER.debug("Uploading media group chunk %d/%d (%d files) to Telegram",
-                             chunk_idx + 1, len(chunks), len(media_files))
-                async with session.post(telegram_url, data=form) as response:
-                    result = await response.json()
-                    _LOGGER.debug("Telegram API response: status=%d, ok=%s", response.status, result.get("ok"))
-                    if response.status == 200 and result.get("ok"):
-                        chunk_message_ids = [
-                            msg.get("message_id") for msg in result.get("result", [])
-                        ]
-                        all_message_ids.extend(chunk_message_ids)
-                    else:
-                        _LOGGER.error("Telegram API error for chunk %d: %s", chunk_idx + 1, result)
-                        return {
-                            "success": False,
-                            "error": result.get("description", "Unknown Telegram error"),
-                            "error_code": result.get("error_code"),
-                            "failed_at_chunk": chunk_idx + 1,
-                        }
-            except aiohttp.ClientError as err:
-                _LOGGER.error("Telegram upload failed for chunk %d: %s", chunk_idx + 1, err)
-                return {
-                    "success": False,
-                    "error": str(err),
-                    "failed_at_chunk": chunk_idx + 1,
-                }
+                # Send to Telegram
+                telegram_url = f"https://api.telegram.org/bot{token}/sendMediaGroup"
+
+                try:
+                    _LOGGER.debug("Uploading media group chunk %d/%d (%d files) to Telegram",
+                                 chunk_idx + 1, len(chunks), len(media_files))
+                    async with session.post(telegram_url, data=form) as response:
+                        result = await response.json()
+                        _LOGGER.debug("Telegram API response: status=%d, ok=%s", response.status, result.get("ok"))
+                        if response.status == 200 and result.get("ok"):
+                            chunk_message_ids = [
+                                msg.get("message_id") for msg in result.get("result", [])
+                            ]
+                            all_message_ids.extend(chunk_message_ids)
+                        else:
+                            # Log detailed error for media group with total size info
+                            total_size = sum(len(d) for _, d, _ in media_files)
+                            _LOGGER.error(
+                                "Telegram API error for chunk %d/%d: %s | Media count: %d | Total size: %d bytes (%.2f MB)",
+                                chunk_idx + 1, len(chunks),
+                                result.get("description", "Unknown Telegram error"),
+                                len(media_files),
+                                total_size,
+                                total_size / (1024 * 1024)
+                            )
+                            # Log detailed diagnostics for the first photo in the group
+                            for media_type, data, _ in media_files:
+                                if media_type == "photo":
+                                    self._log_telegram_error(
+                                        error_code=result.get("error_code"),
+                                        description=result.get("description", "Unknown Telegram error"),
+                                        data=data,
+                                        media_type="photo",
+                                    )
+                                    break  # Only log details for first photo
+                            return {
+                                "success": False,
+                                "error": result.get("description", "Unknown Telegram error"),
+                                "error_code": result.get("error_code"),
+                                "failed_at_chunk": chunk_idx + 1,
+                            }
+                except aiohttp.ClientError as err:
+                    _LOGGER.error("Telegram upload failed for chunk %d: %s", chunk_idx + 1, err)
+                    return {
+                        "success": False,
+                        "error": str(err),
+                        "failed_at_chunk": chunk_idx + 1,
+                    }
+
+            # Send oversized photos as documents
+            for i, (data, photo_caption) in enumerate(oversized_photos):
+                _LOGGER.debug("Sending oversized photo %d/%d as document", i + 1, len(oversized_photos))
+                result = await self._send_telegram_document(
+                    session, token, chat_id, data, f"photo_{i}.jpg",
+                    photo_caption, None, parse_mode
+                )
+                if result.get("success"):
+                    all_message_ids.append(result.get("message_id"))
+                else:
+                    _LOGGER.error("Failed to send oversized photo as document: %s", result.get("error"))
+                    # Continue with other photos even if one fails
 
         return {
             "success": True,

custom_components/immich_album_watcher/services.yaml

@@ -6,9 +6,9 @@ refresh:
       integration: immich_album_watcher
       domain: sensor
 
-get_recent_assets:
-  name: Get Recent Assets
-  description: Get the most recent assets from the targeted album.
+get_assets:
+  name: Get Assets
+  description: Get assets from the targeted album with optional filtering and ordering.
   target:
     entity:
       integration: immich_album_watcher
@@ -16,7 +16,7 @@ get_recent_assets:
   fields:
     count:
       name: Count
-      description: Number of recent assets to return (1-100).
+      description: Maximum number of assets to return (1-100).
       required: false
       default: 10
       selector:
@@ -24,6 +24,44 @@ get_recent_assets:
           min: 1
           max: 100
           mode: slider
+    filter:
+      name: Filter
+      description: Filter assets by type (none, favorite, or rating-based).
+      required: false
+      default: "none"
+      selector:
+        select:
+          options:
+            - label: "None (no filtering)"
+              value: "none"
+            - label: "Favorites only"
+              value: "favorite"
+            - label: "By minimum rating"
+              value: "rating"
+    filter_min_rating:
+      name: Minimum Rating
+      description: Minimum rating for assets (1-5). Only used when filter is set to 'rating'.
+      required: false
+      default: 1
+      selector:
+        number:
+          min: 1
+          max: 5
+          mode: slider
+    order:
+      name: Order
+      description: Sort order for assets by creation date.
+      required: false
+      default: "descending"
+      selector:
+        select:
+          options:
+            - label: "Ascending (oldest first)"
+              value: "ascending"
+            - label: "Descending (newest first)"
+              value: "descending"
+            - label: "Random"
+              value: "random"
 
 send_telegram_notification:
   name: Send Telegram Notification
@@ -116,3 +154,21 @@ send_telegram_notification:
       default: true
       selector:
         boolean:
+    max_asset_data_size:
+      name: Max Asset Data Size
+      description: Maximum asset size in bytes. Assets exceeding this limit will be skipped. Leave empty for no limit.
+      required: false
+      selector:
+        number:
+          min: 1
+          max: 52428800
+          step: 1048576
+          unit_of_measurement: "bytes"
+          mode: box
+    send_large_photos_as_documents:
+      name: Send Large Photos As Documents
+      description: How to handle photos exceeding Telegram's limits (10MB or 10000px dimension sum). If true, send as documents. If false, skip oversized photos.
+      required: false
+      default: false
+      selector:
+        boolean:

custom_components/immich_album_watcher/translations/en.json

@@ -133,13 +133,25 @@
             "name": "Refresh",
             "description": "Force an immediate refresh of album data from Immich."
         },
-        "get_recent_assets": {
-            "name": "Get Recent Assets",
-            "description": "Get the most recent assets from the targeted album.",
+        "get_assets": {
+            "name": "Get Assets",
+            "description": "Get assets from the targeted album with optional filtering and ordering.",
             "fields": {
                 "count": {
                     "name": "Count",
-                    "description": "Number of recent assets to return (1-100)."
+                    "description": "Maximum number of assets to return (1-100)."
+                },
+                "filter": {
+                    "name": "Filter",
+                    "description": "Filter assets by type (none, favorite, or rating-based)."
+                },
+                "filter_min_rating": {
+                    "name": "Minimum Rating",
+                    "description": "Minimum rating for assets (1-5). Only used when filter is set to 'rating'."
+                },
+                "order": {
+                    "name": "Order",
+                    "description": "Sort order for assets by creation date."
                 }
             }
         },
@@ -186,6 +198,14 @@
                 "wait_for_response": {
                     "name": "Wait For Response",
                     "description": "Wait for Telegram to finish processing before returning. Set to false for fire-and-forget (automation continues immediately)."
+                },
+                "max_asset_data_size": {
+                    "name": "Max Asset Data Size",
+                    "description": "Maximum asset size in bytes. Assets exceeding this limit will be skipped. Leave empty for no limit."
+                },
+                "send_large_photos_as_documents": {
+                    "name": "Send Large Photos As Documents",
+                    "description": "How to handle photos exceeding Telegram's limits (10MB or 10000px dimension sum). If true, send as documents. If false, downsize to fit limits."
                 }
             }
         }

custom_components/immich_album_watcher/translations/ru.json

@@ -133,13 +133,25 @@
             "name": "Обновить",
             "description": "Принудительно обновить данные альбома из Immich."
         },
-        "get_recent_assets": {
-            "name": "Получить последние файлы",
-            "description": "Получить последние файлы из выбранного альбома.",
+        "get_assets": {
+            "name": "Получить файлы",
+            "description": "Получить файлы из выбранного альбома с возможностью фильтрации и сортировки.",
             "fields": {
                 "count": {
                     "name": "Количество",
-                    "description": "Количество возвращаемых файлов (1-100)."
+                    "description": "Максимальное количество возвращаемых файлов (1-100)."
+                },
+                "filter": {
+                    "name": "Фильтр",
+                    "description": "Фильтровать файлы по типу (none - без фильтра, favorite - только избранные, rating - по рейтингу)."
+                },
+                "filter_min_rating": {
+                    "name": "Минимальный рейтинг",
+                    "description": "Минимальный рейтинг для файлов (1-5). Используется только при filter='rating'."
+                },
+                "order": {
+                    "name": "Порядок",
+                    "description": "Порядок сортировки файлов по дате создания."
                 }
             }
         },
@@ -186,6 +198,14 @@
                 "wait_for_response": {
                     "name": "Ждать ответа",
                     "description": "Ждать завершения отправки в Telegram перед возвратом. Установите false для фоновой отправки (автоматизация продолжается немедленно)."
+                },
+                "max_asset_data_size": {
+                    "name": "Макс. размер ресурса",
+                    "description": "Максимальный размер ресурса в байтах. Ресурсы, превышающие этот лимит, будут пропущены. Оставьте пустым для отсутствия ограничения."
+                },
+                "send_large_photos_as_documents": {
+                    "name": "Большие фото как документы",
+                    "description": "Как обрабатывать фото, превышающие лимиты Telegram (10МБ или сумма размеров 10000пкс). Если true, отправлять как документы. Если false, уменьшать для соответствия лимитам."
                 }
             }
         }