feat: add storage tier change events script

operator-tools/README.md

@@ -230,7 +230,7 @@ uv run operator-tools/migrate_catalog.py clone sentinel-2-l2a-staging sentinel-2
 
 **Documentation:** See [README_MIGRATIONS.md](./README_MIGRATIONS.md) for the full safe migration procedure, CLI reference, and instructions for writing new migrations.
 
-### 4. `submit_test_workflow_wh.py` - HTTP Webhook Submission
+### 4. `submit_test_workflow_wh.py` - HTTP Webhook Submission (single item)
 
 Submits a single test STAC item via HTTP webhook endpoint.
 
@@ -253,9 +253,100 @@ Edit the script to change:
 
 - `source_url`: STAC item URL to process
 - `collection`: Target collection name
-- `action`: Processing action (e.g., `convert-v1-s2-hp`, or `convert-v1-s2-hp`)
+- `action`: Processing action (e.g., `convert-v1-s2-hp`)
 
-### 5. `submit_stac_items_notebook.ipynb` - Interactive STAC Search & Submit
+### 5. `submit_test_workflow_wh_list.py` - HTTP Webhook Submission (hardcoded list)
+
+Submits a hardcoded list of items via HTTP webhook, one workflow per item, with a 1s delay between submissions.
+
+**Use case:** Batch-reprocessing a known fixed set of items (list is edited directly in the script)
+
+**Usage:**
+
+```bash
+uv run operator-tools/submit_test_workflow_wh_list.py
+```
+
+**Configuration:** Edit the `products` list and `payload` fields inside the script.
+
+### 6. `submit_storage_tier_workflows.py` - Batch Storage Tier Change via Argo
+
+Queries STAC in 24h windows over a date range and submits one `batch-change-storage-tier` webhook payload per window, triggering the `eopf-storage-tier-batch-job` WorkflowTemplate via Argo Events. Each workflow fans out over all items in that window in parallel (up to `--parallelism` pods at a time). This is the preferred approach for large date ranges — one Argo Workflow per day keeps job history clean and avoids submitting hundreds of individual workflows.
+
+**Use case:** Change the S3 storage class (e.g., to `STANDARD_IA`) for thousands of items over a multi-month date range.
+
+**Prerequisites:**
+
+- Pipeline webhook service running (port-forward, see above)
+- `requests` and `pystac-client` Python packages (already in project deps)
+- Argo Events sensor `eopf-explorer-storage-tier` deployed in the cluster (see `platform-deploy` repo)
+
+**Usage:**
+
+```bash
+# Dry run — logs what would be submitted without sending any requests
+python operator-tools/submit_storage_tier_workflows.py \
+    --start-date 2024-01-01 \
+    --end-date 2024-06-01 \
+    --collection sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA \
+    --dry-run
+
+# Live run (port-forward must be active)
+python operator-tools/submit_storage_tier_workflows.py \
+    --start-date 2024-01-01 \
+    --end-date 2024-06-01 \
+    --collection sentinel-2-l2a-staging \
+    --storage-class STANDARD_IA \
+    --webhook-url http://localhost:12000/samples
+
+# Registered-window backlog — window on when items were *registered*
+# (properties.created) rather than sensed, to drain a bulk-conversion backlog
+# to STANDARD. Uses a CQL2 filter instead of the datetime= range.
+python operator-tools/submit_storage_tier_workflows.py \
+    --date-field created \
+    --start-date 2025-11-01 \
+    --end-date 2026-04-08 \
+    --collection sentinel-2-l2a \
+    --storage-class STANDARD \
+    --process-all-assets \
+    --dry-run
+```
+
+**All options:**
+
+| Option | Default | Description |
+|--------|---------|-------------|
+| `--start-date` | required | Start of date range (`YYYY-MM-DD`) |
+| `--end-date` | required | End of date range (`YYYY-MM-DD`) |
+| `--collection` | required | STAC collection ID |
+| `--date-field` | `datetime` | Item date to window on: sensing `datetime`, registration `created`, or last-modified `updated`. Non-default fields use a CQL2 `between` filter |
+| `--storage-class` | `STANDARD` | Target S3 storage class |
+| `--stac-api-url` | prod API URL | STAC API endpoint to query |
+| `--s3-endpoint` | OVH Cloud | S3 endpoint passed to Argo workflows |
+| `--pipeline-image-version` | `v1.6.1` | Docker image tag for Argo jobs |
+| `--process-all-assets` | false | Process all assets (not just reflectance) |
+| `--webhook-url` | `localhost:12000/samples` | Webhook endpoint |
+| `--delay` | `1.0` | Seconds between window submissions |
+| `--dry-run` | false | Log payloads without sending |
+
+**Payload format** (one per 24h window):
+```json
+{
+  "action": "batch-change-storage-tier",
+  "item_ids": ["S2A_MSIL2A_...", "S2B_MSIL2A_..."],
+  "collection": "sentinel-2-l2a-staging",
+  "storage_class": "STANDARD_IA",
+  "stac_api_url": "https://...",
+  "s3_endpoint": "https://...",
+  "pipeline_image_version": "v1.6.1",
+  "process_all_assets": "false"
+}
+```
+
+**Cluster-side:** The `eopf-explorer-storage-tier` Argo Events Sensor (in `platform-deploy`) listens for `action: "batch-change-storage-tier"` and triggers `eopf-storage-tier-batch-job`, which fans out over `item_ids` — running change-tier + STAC metadata update for each item in parallel. Empty windows (no items found) are skipped without submitting a webhook.
+
+### 7. `submit_stac_items_notebook.ipynb` - Interactive STAC Search & Submit
 
 Jupyter notebook for searching and batch submitting STAC items.
 
@@ -333,6 +424,9 @@ python manage_collections.py clean test-coll --clean-s3 -y
 | `manage_collections.py` | Viewing collection statistics | `python manage_collections.py info coll-id --s3-stats` |
 | `manage_collections.py` | Batch operations on all items | `python manage_collections.py clean coll-id --clean-s3 -y` |
 | `manage_collections.py` | Collection lifecycle management | `python manage_collections.py create/delete` |
+| `submit_test_workflow_wh.py` | Testing pipeline with one known item | edit script then `uv run submit_test_workflow_wh.py` |
+| `submit_test_workflow_wh_list.py` | Reprocessing a small known fixed set | edit products list then `uv run submit_test_workflow_wh_list.py` |
+| `submit_storage_tier_workflows.py` | Changing storage tier for a large date range (one Argo batch workflow per 24h window) | `python submit_storage_tier_workflows.py --start-date ... --dry-run` |
 
 ### Benefits of This Workflow
 

operator-tools/submit_storage_tier_workflows.py

@@ -0,0 +1,182 @@
+#!/usr/bin/env python3
+"""Submit storage tier change workflows via HTTP webhook for a date range of STAC items.
+
+Queries STAC in 24h windows and POSTs one webhook payload per window,
+triggering the eopf-storage-tier-batch-job WorkflowTemplate via Argo Events.
+Each workflow fans out to per-item parallel processing within Argo.
+"""
+
+import argparse
+import logging
+import sys
+import time
+from datetime import UTC, datetime, timedelta
+from typing import cast
+
+import requests
+from pystac_client import Client
+
+logging.basicConfig(
+    level=logging.INFO,
+    format="%(asctime)s - %(name)s - %(levelname)s - %(message)s",
+)
+logger = logging.getLogger(__name__)
+
+
+def generate_time_windows(
+    start_date: datetime, end_date: datetime, window_hours: int = 24
+) -> list[tuple[str, str]]:
+    """Return list of (window_start_iso, window_end_iso) tuples covering start_date to end_date.
+
+    Each window is window_hours long; the last window is truncated to end_date.
+    """
+    windows = []
+    current = start_date
+    delta = timedelta(hours=window_hours)
+    while current < end_date:
+        window_end = min(current + delta, end_date)
+        windows.append(
+            (
+                current.isoformat().replace("+00:00", "Z"),
+                window_end.isoformat().replace("+00:00", "Z"),
+            )
+        )
+        current = window_end
+    return windows
+
+
+def query_stac_items(
+    stac_api_url: str,
+    collection: str,
+    window_start: str,
+    window_end: str,
+    date_field: str = "datetime",
+) -> list[str]:
+    """Query STAC for items whose ``date_field`` falls in the window. Returns item IDs.
+
+    ``date_field="datetime"`` uses pystac-client's native ``datetime=`` range
+    (the item's sensing time). Other fields (``created``, ``updated``) filter on
+    the registration/update timestamp via a CQL2 ``between`` filter, since those
+    properties are not reachable through the ``datetime=`` kwarg.
+    """
+    catalog = Client.open(stac_api_url)
+    if date_field == "datetime":
+        search = catalog.search(
+            collections=[collection],
+            datetime=f"{window_start}/{window_end}",
+            limit=100,
+        )
+    else:
+        search = catalog.search(
+            collections=[collection],
+            filter={
+                "op": "between",
+                "args": [{"property": date_field}, window_start, window_end],
+            },
+            filter_lang="cql2-json",
+            limit=100,
+        )
+    return [item.id for item in search.items()]
+
+
+def submit_batch(webhook_url: str, payload: dict[str, object], dry_run: bool) -> bool:
+    """POST a JSON payload to the webhook endpoint. Returns True on success."""
+    if dry_run:
+        logger.info(f"[dry-run] Would submit: {payload}")
+        return True
+    try:
+        response = requests.post(
+            webhook_url,
+            json=payload,
+            headers={"Content-Type": "application/json"},
+            timeout=30,
+        )
+        if response.status_code != 200:
+            logger.warning(
+                f"Non-200 response for window with {len(cast(list[str], payload.get('item_ids', [])))} items: "
+                f"{response.status_code} {response.text}"
+            )
+        return response.status_code == 200
+    except Exception as e:
+        logger.error(f"Error submitting batch: {e}")
+        return False
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(
+        description="Submit storage tier batch workflows via webhook for a date range of STAC items."
+    )
+    parser.add_argument("--start-date", required=True, help="Start date (YYYY-MM-DD)")
+    parser.add_argument("--end-date", required=True, help="End date (YYYY-MM-DD)")
+    parser.add_argument("--collection", required=True, help="STAC collection ID")
+    parser.add_argument(
+        "--date-field",
+        choices=["datetime", "created", "updated"],
+        default="datetime",
+        help="Item date to window on: sensing 'datetime' (default), registration "
+        "'created', or last-modified 'updated'. Non-default fields use a CQL2 filter.",
+    )
+    parser.add_argument("--storage-class", default="STANDARD", help="S3 storage class")
+    parser.add_argument("--stac-api-url", default="https://api.explorer.eopf.copernicus.eu/stac")
+    parser.add_argument("--s3-endpoint", default="https://s3.de.io.cloud.ovh.net")
+    parser.add_argument("--pipeline-image-version", default="v1.6.1")
+    parser.add_argument("--process-all-assets", action="store_true")
+    parser.add_argument("--webhook-url", default="http://localhost:12000/samples")
+    parser.add_argument(
+        "--delay", type=float, default=1.0, help="Delay between window submissions in seconds"
+    )
+    parser.add_argument("--dry-run", action="store_true")
+    args = parser.parse_args()
+
+    try:
+        start_date = datetime.fromisoformat(args.start_date).replace(tzinfo=UTC)
+        end_date = datetime.fromisoformat(args.end_date).replace(tzinfo=UTC)
+    except ValueError as e:
+        logger.error(f"Invalid date format: {e}")
+        sys.exit(1)
+
+    if end_date <= start_date:
+        logger.error("--end-date must be after --start-date")
+        sys.exit(1)
+
+    windows = generate_time_windows(start_date, end_date)
+    logger.info(f"Processing {len(windows)} 24h windows from {args.start_date} to {args.end_date}")
+
+    total_submitted = 0
+    total_failed = 0
+
+    for i, (window_start, window_end) in enumerate(windows, 1):
+        logger.info(f"[{i}/{len(windows)}] Querying window {window_start} to {window_end}")
+        item_ids = query_stac_items(
+            args.stac_api_url, args.collection, window_start, window_end, args.date_field
+        )
+        logger.info(f"  Found {len(item_ids)} items")
+
+        if not item_ids:
+            logger.info("  Skipping empty window")
+            continue
+
+        payload: dict[str, object] = {
+            "action": "batch-change-storage-tier",
+            "item_ids": item_ids,
+            "collection": args.collection,
+            "storage_class": args.storage_class,
+            "stac_api_url": args.stac_api_url,
+            "s3_endpoint": args.s3_endpoint,
+            "pipeline_image_version": args.pipeline_image_version,
+            "process_all_assets": str(args.process_all_assets).lower(),
+        }
+        success = submit_batch(args.webhook_url, payload, args.dry_run)
+        if success:
+            total_submitted += 1
+        else:
+            total_failed += 1
+
+        if i < len(windows):
+            time.sleep(args.delay)
+
+    logger.info(f"Done. Submitted: {total_submitted}, Failed: {total_failed}")
+
+
+if __name__ == "__main__":
+    main()