Merge pull request #69 from ESA-APEx/ogc_result_api

OGC API: Integration of result fetching

Author: JanssenBrm

app/platforms/implementations/ogc_api_process.py

@@ -1,5 +1,8 @@
+import json
 import re
 from typing import List
+
+import requests
 from app.auth import exchange_token, get_current_user_claims
 from fastapi import Response
 from loguru import logger
@@ -27,8 +30,10 @@
     "https://schemas.stacspec.org/v1.0.0/collection-spec/json-schema/collection.json"
 )
 
-GEOJSON_FEATURECOLLECTION_SCHEMA = "https://schemas.opengis.net/ogcapi/" \
+GEOJSON_FEATURECOLLECTION_SCHEMA = (
+    "https://schemas.opengis.net/ogcapi/"
     "features/part1/1.0/openapi/schemas/featureCollectionGeoJSON.yaml"
+)
 
 
 @register_platform(ProcessTypeEnum.OGC_API_PROCESS)
@@ -292,6 +297,206 @@ def _map_ogcapi_status(self, ogcapi_status: StatusCode) -> ProcessingStatusEnum:
             logger.warning(f"Mapping of unknown OGC API status: {ogcapi_status}")
             return ProcessingStatusEnum.UNKNOWN
 
+    def _extract_download_link_from_asset(self, asset: dict) -> str | None:
+        """
+        Extracts the download link from an asset dictionary. Checks if the
+        `href` field is present and contains an HTTPS URL. If this is not
+        the case, look for an alternative link in `alternate`
+        field. If no valid link is found, return None.
+
+        Args:
+            asset (dict): The asset dictionary.
+
+        Returns:
+            str | None: The download link if available, otherwise None.
+        """
+        refs = [
+            asset.get("href"),
+            asset.get("alternate", {}).get("https", {}).get("href"),
+        ]
+        for href in refs:
+            if href and href.startswith("https://"):
+                return href
+        return None
+
+    def _generate_signed_url(self, href: str, user_token: str) -> str:
+        """
+        Generate a signed URL for the given href using the provided user token.
+        This is a placeholder implementation and should be replaced with
+        actual logic to generate signed URLs.
+
+        Args:
+            href (str): The original href.
+            user_token (str): The user token to be used for signing.
+        """
+        # TODO - Add implementation
+        logger.debug(f"Generating signed URL for href: {href} with user token.")
+        response = requests.get(
+            href,
+            headers={"Authorization": f"Bearer {user_token}"},
+            allow_redirects=False,
+        )
+        signed_url = response.headers["location"]
+        logger.debug(f"Signed URL: {signed_url}")
+        return signed_url
+
+    def _update_assets_hrefs(self, assets: dict, user_token: str) -> dict:
+        """
+        Update the hrefs of the assets to be HTTPS URLs. If the current
+        href is an S3 URL, the code will look into `alternate` links to
+        find an HTTPS URL. If no HTTPS URL is found, the original href
+        will be kept.
+        """
+        updated_assets = {}
+        for asset_name, asset in assets.items():
+            updated_asset = asset.copy()
+            href = self._extract_download_link_from_asset(asset)
+            if not href:
+                logger.warning(
+                    "No valid HTTPS download link found for asset "
+                    f"'{asset_name}'. Keeping original href. "
+                    "Skipping asset..."
+                )
+            else:
+                href = self._generate_signed_url(href, user_token)
+                updated_asset["href"] = href
+                updated_assets[asset_name] = updated_asset
+
+        return updated_assets
+
+    def _build_collection_from_features(
+        self,
+        features: list,
+        assets: dict,
+        result_name: str,
+        user_token: str,
+        details: ServiceDetails,
+        internal_job_id: str,
+    ) -> Collection:
+        """
+        Build a STAC Collection from a list of GeoJSON features and their
+        aggregated assets. The spatial extent is derived from the feature
+        bounding boxes and the temporal extent from the feature datetime
+        properties.
+
+        Args:
+            features: GeoJSON feature list.
+            assets: Aggregated asset dict collected from those features.
+            result_name: Identifier used as the collection ID.
+            user_token: Token used to sign asset hrefs.
+            details: Service details containing namespace and application information.
+            internal_job_id: Internal job identifier.
+
+        Returns:
+            A STAC Collection.
+        """
+        # Spatial extent — union of per-feature bboxes
+        min_x, min_y = float("inf"), float("inf")
+        max_x, max_y = float("-inf"), float("-inf")
+        found_bbox = False
+        for feature in features:
+            bbox = feature.get("bbox")
+            if isinstance(bbox, (list, tuple)) and len(bbox) >= 4:
+                min_x = min(min_x, float(bbox[0]))
+                min_y = min(min_y, float(bbox[1]))
+                max_x = max(max_x, float(bbox[2]))
+                max_y = max(max_y, float(bbox[3]))
+                found_bbox = True
+        spatial_bbox: tuple[float, float, float, float] = (
+            (min_x, min_y, max_x, max_y) if found_bbox else (-180.0, -90.0, 180.0, 90.0)
+        )
+
+        # Temporal extent — min/max of all datetime-like properties
+        datetimes: list[str] = []
+        for feature in features:
+            props = feature.get("properties") or {}
+            for dt_key in ("datetime", "start_datetime", "end_datetime"):
+                dt_val = props.get(dt_key)
+                if isinstance(dt_val, str):
+                    datetimes.append(dt_val)
+        temporal_interval: list[list] = (
+            [[min(datetimes), max(datetimes)]] if datetimes else [[None, None]]
+        )
+
+        updated_assets = self._update_assets_hrefs(assets, user_token)
+
+        logger.debug(
+            f"Building STAC Collection '{result_name}' from {len(features)} feature(s) "
+            f"with {len(updated_assets)} asset(s)."
+        )
+
+        return Collection(
+            id=f"{details.namespace}-{internal_job_id}",
+            stac_version=STAC_VERSION,
+            title=f"Results for {details.application}",
+            description=(
+                f"OGC API process result items for job '{internal_job_id}' "
+                f"of application '{details.application}'."
+            ),
+            type="Collection",
+            license="proprietary",
+            links=Links([]),
+            extent=Extent(
+                spatial=SpatialExtent(bbox=[spatial_bbox]),
+                temporal=TimeInterval(interval=temporal_interval),
+            ),
+            assets=updated_assets,
+        )
+
+    def _extract_assets_from_feature_collection(
+        self,
+        feature_collection: dict,
+        *,
+        result_name: str,
+        user_token: str,
+        details: ServiceDetails,
+        internal_job_id: str,
+    ) -> Collection:
+        assets: dict = {}
+        features = feature_collection.get("features", [])
+        logger.debug(f"Feature collection: {json.dumps(feature_collection, indent=2)}")
+        for feature in features:
+            feature_assets = feature.get("assets")
+            if isinstance(feature_assets, dict):
+                assets.update(feature_assets)
+                continue
+
+            # Some providers expose assets through an item link
+            # instead of inlining them in the feature.
+            for link in feature.get("links", []):
+                if "collection" == link.get("rel") and link.get("href"):
+                    collection_link: str = link.get("href")
+                    logger.debug(
+                        f"GeoJSON FeatureCollection results: '{result_name}' "
+                        f"points to a valid collection URL: {collection_link}"
+                    )
+
+                    response: HTTPXResponse = http_get(
+                        collection_link,
+                        follow_redirects=True,
+                        headers={"Authorization": f"Bearer {user_token}"},
+                    )
+                    response.raise_for_status()
+                    collection_data = response.json()
+                    collection_data["assets"] = self._update_assets_hrefs(
+                        collection_data.get("assets", {}), user_token
+                    )
+                    collection = Collection.model_validate(collection_data)
+                    logger.debug(
+                        f"Extracted collection '{collection.id}' "
+                        f"with assets: {list((collection.assets or {}).keys())}"
+                    )
+                    return collection
+
+        return self._build_collection_from_features(
+            features,
+            assets,
+            result_name,
+            user_token,
+            details,
+            internal_job_id
+        )
+
     async def get_job_status(
         self, user_token: str, job_id: str, details: ServiceDetails
     ) -> ProcessingStatusEnum:
@@ -352,17 +557,18 @@ async def get_job_results(
                 and qualified_value.var_schema.actual_instance
             ):
                 schema_reference = qualified_value.var_schema.actual_instance
+                media_type = getattr(qualified_value, "media_type", None)
                 logger.debug(
                     f"Processing result\n* Name: '{result_name}'\n"
-                    "* media type: {qualified_value.media_type}\n"
-                    "* Python type: {type(qualified_value.value)}\n"
-                    "* schema {qualified_value.var_schema}..."
+                    f"* media type: {media_type}\n"
+                    f"* Python type: {type(qualified_value.value)}\n"
+                    f"* schema {qualified_value.var_schema}..."
                 )
 
                 if not isinstance(schema_reference, str):
                     logger.warning(
                         f"Processing result name: '{result_name}' can not be processed, "
-                        "schema of type {type(schema_reference)} not recognized"
+                        f"schema of type {type(schema_reference)} not recognized"
                     )
                     continue
 
@@ -375,29 +581,20 @@ async def get_job_results(
                     logger.success(
                         f"GeoJSON FeatureCollection found in results: '{result_name}'"
                     )
-                    feature_collection = qualified_value.value.oneof_schema_2_validator or {}
-                    for feature in feature_collection.get("features", []):
-                        for link in feature.get("links", []):
-                            if "collection" == link.get("rel") and link.get("href"):
-                                collection_link: str = link.get("href")
-                                logger.success(
-                                    f"GeoJSON FeatureCollection results: '{result_name}' "
-                                    "points to a valid collection URL: {collection_link}"
-                                )
-
-                                response: HTTPXResponse = http_get(
-                                    collection_link,
-                                    follow_redirects=True,
-                                    headers={
-                                        "Authorization": f"Bearer {exchanged_token}"
-                                    },
-                                )
-                                response.raise_for_status()
-                                return Collection.model_validate(response.json())
+                    feature_collection = (
+                        qualified_value.value.oneof_schema_2_validator or {}
+                    )
+                    return self._extract_assets_from_feature_collection(
+                        feature_collection,
+                        result_name=result_name,
+                        user_token=exchanged_token or user_token,
+                        details=details,
+                        internal_job_id=internal_job_id,
+                    )
                 else:
                     logger.warning(
                         f"Processing result: '{result_name}' can not be processed, "
-                        "schema {schema_reference} not yet managed"
+                        f"schema {schema_reference} not yet managed"
                     )
 
         # result not found, send back an empty collection
@@ -417,6 +614,7 @@ async def get_job_results(
                 spatial=SpatialExtent(bbox=[(-180.0, -90.0, 180.0, 90.0)]),
                 temporal=TimeInterval(interval=[[None, None]]),
             ),
+            assets={},
         )
 
     async def get_service_parameters(