Web Harvesting API

Create and manage web harvest tasks (screenshots, scraping). Supports async polling and bulk operations. Requires SCRAPE_TOKEN.

Bases: BaseAPIClient

API client for Silo web harvesting operations.

This class provides methods for creating and managing automated web content harvesting tasks using the Authentic8 Silo platform. The harvester can collect various types of web content including assets, visual captures, and videos.

The harvesting API allows you to: - Create asset collection tasks (files, documents, media) - Generate visual captures (screenshots, PDFs, MHTML) - Record video captures of web interactions - Monitor task progress and status - Download completed harvest results - Manage task lifecycle and cleanup

Note

The ext API has no command to list or enumerate harvest tasks. To retrieve task IDs after the fact, extract logs of type HARVEST using :class:~silo_sdk.logging.extraction_api.LogExtractionAPI and parse the entries for task IDs, status, and output file references. The recommended pattern is to store task IDs locally at creation time via :meth:create_harvest_task.

Supported harvest types: - Asset: Download files/pages via wget. Limited to 6 egress locations (Singapore, Dubai, Frankfurt, Sao Paulo, Johannesburg, New York City). - Single: Single-URL fetch available from all egress locations; uses wget_params. - Visual: Capture screenshots, generate PDFs, save MHTML archives (all egress). - Video: Record video of web page interactions and navigation (all egress).

Attributes:

Name	Type	Description
VALID_TASK_TYPES	set[str]	Set of valid task type strings: {"asset", "visual", "video", "single"}
VALID_VISUAL_PARAMS	set[str]	Set of valid visual task parameter keys including: output_pdf: Generate PDF output (default: screenshot PNG) output_native_pdf: Generate a native (non-screenshot) PDF conform_page_to_paper: Fit page content to paper size output_mhtml: Generate MHTML archive output output_image: Generate screenshot image (default format) scale: Rendering scale factor (0.1-2.0) landscape: Landscape orientation for PDF (boolean) paper: Paper size for PDF (e.g., "A4", "Letter") max_scroll: Maximum scroll depth for full-page capture page_strategy: Page load strategy ("eager", "complete") wait_nav: Navigation wait condition (string) chrome_stealth: Enable stealth mode to reduce bot detection xff: Forward X-Forwarded-For header (boolean) translation_only: Only apply translation, skip other processing cookies_out: Export session cookies after capture
VALID_ASSET_PARAMS	set[str]	Set of valid asset task parameter keys (wget-style options). Sent as wget_params in the wire format. Note: return-cookies is a top-level task_params field, not a wget flag.
VALID_VIDEO_PARAMS	set[str]	Set of valid video task parameter keys (yt-dlp-style options). Sent as vid_params in the wire format. Key corrections: use referer (not referrer) and merge-output-format (not merge-outputformat).

Example

from silo_sdk import HarvesterAPI, load_config config = load_config() harvester = HarvesterAPI(config)

Create an asset harvesting task

task_id = harvester.create_harvest_task( ... task_type="asset", ... urls=["https://example.com/document.pdf"], ... egress_info={"name": "New York, NY"}, ... dest_path="/downloads/", ... dest_name="document.pdf", ... task_params={"recursive": True, "level": 2} ... )

Create task with specific egress categories

task_id = harvester.create_harvest_task( ... task_type="visual", ... urls=["https://example.com/page"], ... egress_info={ ... "name": "Singapore", ... "categories": { ... "connectivity": "datacenter", ... "availability": None, ... "protocol": "tor" ... } ... }, ... task_params={"output_pdf": True} ... )

Monitor task until completion

result = harvester.wait_for_completion(task_id)

Download the result

harvester.download_task_result(task_id, "/local/path/document.pdf")

Source code in silo_sdk/harvesting/harvester_api.py

class HarvesterAPI(BaseAPIClient):
    """API client for Silo web harvesting operations.

    This class provides methods for creating and managing automated web content
    harvesting tasks using the Authentic8 Silo platform. The harvester can collect
    various types of web content including assets, visual captures, and videos.

    The harvesting API allows you to:
    - Create asset collection tasks (files, documents, media)
    - Generate visual captures (screenshots, PDFs, MHTML)
    - Record video captures of web interactions
    - Monitor task progress and status
    - Download completed harvest results
    - Manage task lifecycle and cleanup

    Note:
        The ext API has no command to list or enumerate harvest tasks. To
        retrieve task IDs after the fact, extract logs of type ``HARVEST``
        using :class:`~silo_sdk.logging.extraction_api.LogExtractionAPI`
        and parse the entries for task IDs, status, and output file
        references. The recommended pattern is to store task IDs locally
        at creation time via :meth:`create_harvest_task`.

    Supported harvest types:
    - Asset: Download files/pages via wget. Limited to 6 egress locations (Singapore,
      Dubai, Frankfurt, Sao Paulo, Johannesburg, New York City).
    - Single: Single-URL fetch available from all egress locations; uses wget_params.
    - Visual: Capture screenshots, generate PDFs, save MHTML archives (all egress).
    - Video: Record video of web page interactions and navigation (all egress).

    Attributes:
        VALID_TASK_TYPES: Set of valid task type strings: ``{"asset", "visual", "video", "single"}``
        VALID_VISUAL_PARAMS: Set of valid visual task parameter keys including:

            - ``output_pdf``: Generate PDF output (default: screenshot PNG)
            - ``output_native_pdf``: Generate a native (non-screenshot) PDF
            - ``conform_page_to_paper``: Fit page content to paper size
            - ``output_mhtml``: Generate MHTML archive output
            - ``output_image``: Generate screenshot image (default format)
            - ``scale``: Rendering scale factor (0.1-2.0)
            - ``landscape``: Landscape orientation for PDF (boolean)
            - ``paper``: Paper size for PDF (e.g., ``"A4"``, ``"Letter"``)
            - ``max_scroll``: Maximum scroll depth for full-page capture
            - ``page_strategy``: Page load strategy (``"eager"``, ``"complete"``)
            - ``wait_nav``: Navigation wait condition (string)
            - ``chrome_stealth``: Enable stealth mode to reduce bot detection
            - ``xff``: Forward X-Forwarded-For header (boolean)
            - ``translation_only``: Only apply translation, skip other processing
            - ``cookies_out``: Export session cookies after capture

        VALID_ASSET_PARAMS: Set of valid asset task parameter keys (wget-style options).
            Sent as ``wget_params`` in the wire format. Note: ``return-cookies`` is a
            top-level ``task_params`` field, not a wget flag.
        VALID_VIDEO_PARAMS: Set of valid video task parameter keys (yt-dlp-style options).
            Sent as ``vid_params`` in the wire format. Key corrections: use ``referer``
            (not ``referrer``) and ``merge-output-format`` (not ``merge-outputformat``).

    Example:
        >>> from silo_sdk import HarvesterAPI, load_config
        >>> config = load_config()
        >>> harvester = HarvesterAPI(config)
        >>>
        >>> # Create an asset harvesting task
        >>> task_id = harvester.create_harvest_task(
        ...     task_type="asset",
        ...     urls=["https://example.com/document.pdf"],
        ...     egress_info={"name": "New York, NY"},
        ...     dest_path="/downloads/",
        ...     dest_name="document.pdf",
        ...     task_params={"recursive": True, "level": 2}
        ... )
        >>>
        >>> # Create task with specific egress categories
        >>> task_id = harvester.create_harvest_task(
        ...     task_type="visual",
        ...     urls=["https://example.com/page"],
        ...     egress_info={
        ...         "name": "Singapore",
        ...         "categories": {
        ...             "connectivity": "datacenter",
        ...             "availability": None,
        ...             "protocol": "tor"
        ...         }
        ...     },
        ...     task_params={"output_pdf": True}
        ... )
        >>>
        >>> # Monitor task until completion
        >>> result = harvester.wait_for_completion(task_id)
        >>>
        >>> # Download the result
        >>> harvester.download_task_result(task_id, "/local/path/document.pdf")
    """

    # Valid parameters for different harvest types
    VALID_ASSET_PARAMS: set[str] = {
        "timeout",
        "limit-rate",
        "wait",
        "random-wait",
        "quota",
        "user",
        "password",
        "local-encoding",
        "remote-encoding",
        "http-user",
        "http-password",
        "no-cookies",
        "keep-session-cookies",
        "header",
        "max-redirect",
        "save-headers",
        "user-agent",
        "secure-protocol",
        "https-only",
        "no-check-certificate",
        "ftp-user",
        "ftp-password",
        "recursive",
        "level",
        "convert-links",
        "accept",
        "reject",
        "accept-regex",
        "reject-regex",
        "domains",
        "exclude-domains",
        "post-data",
        "execute",
        "span-hosts",
        "no-parent",
        "no-directories",
        "page-requisites",
        "adjust-extension",
    }

    VALID_VISUAL_PARAMS: set[str] = {
        "scale",
        "paper",
        "landscape",
        "cookies_out",
        "output_pdf",
        "output_mhtml",
        "output_image",
        "output_native_pdf",
        "conform_page_to_paper",
        "max_scroll",
        "page_strategy",
        "notes",
        "suppress_pdf_cover",
        "id_egress",
        "emulate",
        "is_mobile",
        "has_touch",
        "user_agent",
        "viewport_width",
        "timezone",
        "accept_language",
        "slice_avoid_top",
        "slice_avoid_bottom",
        "scroll_delay",
        "final_delay",
        "wait_nav",
        "translate_original_lang",
        "translate_target_lang",
        "translation_only",
        "ignore_cert_error",
        "chrome_stealth",
        "xff",
    }

    VALID_VIDEO_PARAMS: set[str] = {
        "no-check-certificate",
        "write-description",
        "write-info-json",
        "write-all-thumbnails",
        "write-auto-sub",
        "no-call-home",
        "write-sub",
        "all-subs",
        "list-formats",
        "audio",
        "format",
        "recode-video",
        "remux-video",
        "user-agent",
        "username",
        "password",
        "referer",
        "add-header",
        "all-formats",
        "merge-output-format",
        "cookies",
        "filename",
        "transcribe",
        "transcribe-lang",
        "transcribe-text-only",
        "transcription-service",
        "translate-target-lang",
        "file-id",
        "full_path",
    }

    VALID_TASK_TYPES: set[str] = {"asset", "visual", "video", "single"}

    TASK_PARAMS_TOP_LEVEL_KEYS: set[str] = {
        "return-cookies",
        "cookies",
        "all_files",
    }

    @cached_property
    def _valid_params_map(self) -> dict[str, set[str]]:
        """Get map of task types to valid parameters."""
        return {
            "asset": self.VALID_ASSET_PARAMS,
            "visual": self.VALID_VISUAL_PARAMS,
            "video": self.VALID_VIDEO_PARAMS,
            "single": self.VALID_ASSET_PARAMS,
        }

    @classmethod
    def _top_level_keys_for_type(cls, task_type: str) -> set[str]:
        """Return the top-level task_params keys for the given task type.

        For video tasks, ``cookies`` goes in ``vid_params`` (yt-dlp list format),
        not at the top level. Visual tasks also exclude ``cookies`` since it is
        not a supported parameter for that task type. Asset/single use the full set.
        """
        if task_type in {"video", "visual"}:
            return {"return-cookies", "all_files"}
        return cls.TASK_PARAMS_TOP_LEVEL_KEYS

    @staticmethod
    def _normalize_task_params(
        task_params: dict[str, Any] | None,
    ) -> dict[str, Any] | None:
        """Normalize Python-style aliases to wire-format keys."""
        if task_params and "return_cookies" in task_params:
            if "return-cookies" in task_params:
                raise ValidationError(
                    "Cannot specify both 'return_cookies' and 'return-cookies'; "
                    "use one or the other"
                )
            task_params = dict(task_params)
            task_params["return-cookies"] = task_params.pop("return_cookies")
        return task_params

    def __init__(self, config: dict[str, Any]) -> None:
        """Initialize the harvester API client.

        Args:
            config: Configuration dictionary containing API settings

        Raises:
            ConfigurationError: If required configuration is missing or invalid
        """
        super().__init__(config)

        # Comprehensive configuration validation
        validation_errors = validate_harvester_config(config)
        if validation_errors:
            raise ConfigurationError(
                f"Harvester configuration validation failed: {'; '.join(validation_errors)}"
            )

        # Basic required configuration (legacy support)
        validate_config_for_api(config, ["SCRAPE_TOKEN", "BUCKET_ID"])

        self.auth_token = config["SCRAPE_TOKEN"]
        self.dest_bucket_id = config["BUCKET_ID"]

        # Initialize task status cache if enabled
        self._status_cache_enabled, self._status_cache_ttl, self._status_cache = (
            init_status_cache(config)
        )

        # Initialize file API for downloading results
        self.file_api = FileAPI(config)
        self._ssl_context = self.create_ssl_context()

        self.logger.info("Initialized HarvesterAPI client with validated configuration")

    def _require_task_id(self, task_id: str) -> None:
        if not task_id or not isinstance(task_id, str):
            raise ValidationError("task_id must be a non-empty string")

    def __enter__(self) -> HarvesterAPI:
        """Enter context manager for resource management.

        Returns:
            Self for use in with statement
        """
        self.logger.debug("Entering HarvesterAPI context manager")
        return self

    def __exit__(
        self,
        exc_type: type[BaseException] | None,
        exc_val: BaseException | None,
        exc_tb: TracebackType | None,
    ) -> None:
        """Exit context manager and perform cleanup.

        Args:
            exc_type: Exception type if an exception occurred
            exc_val: Exception value if an exception occurred
            exc_tb: Exception traceback if an exception occurred
        """
        # Suppress unused parameter warnings - these are required by the context manager protocol
        _ = exc_type, exc_val, exc_tb
        try:
            self.cleanup_resources()
        except Exception as e:
            self.logger.warning("Error during context manager cleanup: %s", e)
        finally:
            self.logger.debug("Exited HarvesterAPI context manager")

    @api_tag(MethodType.UTILITY)
    def cleanup_resources(self) -> None:
        """Clean up resources and perform housekeeping tasks.

        This method should be called when done with the HarvesterAPI instance
        to properly clean up resources, clear caches, and close connections.
        """
        try:
            # Clear status cache
            if self._status_cache:
                cache_size = len(self._status_cache)
                self._status_cache.clear()
                if cache_size > 0:
                    self.logger.info("Cleared %d cached task statuses", cache_size)

            # Close HTTP session if it exists
            if hasattr(self, "session") and self.session:
                self.session.close()
                self.logger.debug("Closed HTTP session")

            # Clean up file API resources
            if hasattr(self, "file_api") and hasattr(
                self.file_api, "cleanup_resources"
            ):
                self.file_api.cleanup_resources()

        except Exception as e:
            self.logger.error("Error during resource cleanup: %s", e)

    @api_tag(MethodType.API_COMMAND, api_command="create_harvest_task")
    def create_harvest_task(
        self,
        task_type: str,
        urls: list[str],
        egress_info: dict[str, Any],
        dest_path: str = "/",
        dest_name: str | None = None,
        dest_bucket_id: str | None = None,
        task_params: dict[str, Any] | None = None,
        run_after: int | str | None = None,
    ) -> str:
        """Create a new harvest task for web content collection.

        Creates a harvesting task that will collect content from the specified URL
        using the given parameters and store the results in the designated location.

        Args:
            task_type: Type of harvest task. Must be one of ``"asset"``,
                ``"visual"``, ``"video"``, or ``"single"``.

                - ``"visual"`` — Full-page capture. Output format controlled by ``task_params``:

                  - Screenshot (default): no params, or ``{"output_image": True}``
                  - PDF: ``{"output_pdf": True}``
                  - MHTML: ``{"output_mhtml": True}``

                  **IMPORTANT:** ``"screenshot"``, ``"pdf"``, and ``"mhtml"`` are **not**
                  valid ``task_type`` values. Always use ``"visual"`` with ``task_params``.

                - ``"asset"`` — Recursive asset download (HTML, JS, CSS, images).
                  Control depth/filters via ``task_params`` (wget-style options).
                - ``"video"`` — Video recording of page interaction (youtube-dl-style options).

            urls: List containing exactly one URL string to harvest.
            egress_info: Egress location dict with required ``"name"`` key
                (e.g., ``{"name": "New York, NY"}``). Optional ``"categories"``
                key accepts ``"connectivity"``, ``"availability"``, ``"protocol"``.
                For non-``"asset"`` task types, ``"connectivity"``/``"protocol"``
                are validated against what is actually available at ``"name"``
                (see :mod:`silo_sdk.harvesting.egress_validation`). Does not
                apply to ``"asset"`` tasks.
            dest_path: Destination path in storage (defaults to ``"/"``)
            dest_name: Name for the result file (auto-generated if not provided)
            dest_bucket_id: Destination bucket ID (uses default if not provided)
            task_params: Task-specific output parameters. Most keys are
                converted to a list of ``{"name": key}`` or
                ``{"name": key, "value": val}`` dicts sent under the
                appropriate wire key (``vis_params``, ``wget_params``,
                ``vid_params``). The following keys are instead injected
                directly at the top level of the wire ``task_params`` object:

                - ``return-cookies`` — send back cookies from the session
                - ``cookies`` — Netscape-format cookie file content (string)
                - ``all_files`` — include all fetched files alongside the main
                  content. Valid for ``"visual"``, ``"single"``, and ``"video"``
                  tasks. Raises ``ValidationError`` for ``"asset"`` tasks.


                Key visual params: ``output_pdf``, ``output_mhtml``,
                ``output_image``, ``scale``, ``landscape``, ``paper``.
                See class ``Attributes`` for full param sets per task type.
            run_after: Optional schedule time — Unix epoch integer
                (e.g., ``1707849000``) or ISO 8601 UTC string
                (e.g., ``"2026-02-13T18:30:00Z"``). Task enters ``"wait"``
                status until the specified time.

        Note:
            Wire format parameter mappings:

            - ``task_type`` → ``task_params.request_type``
            - ``urls`` → ``task_params.urls``
            - ``task_params`` (Python dict) → list of ``{"name": key}`` or
              ``{"name": key, "value": val}`` dicts sent as:

              - ``task_params.vis_params`` for ``"visual"`` tasks
              - ``task_params.wget_params`` for ``"asset"`` tasks
              - ``task_params.vid_params`` for ``"video"`` tasks

            - ``egress_info``, ``dest_bucket_id``, ``dest_path``, ``dest_name``,
              ``dest_auth_token`` remain at the top-level command dict
            - ``run_after`` is injected at **top-level command dict** alongside
              ``egress_info``, **NOT** inside ``task_params``

            Wire format structure for visual task::

                {
                    "command": "create_harvest_task",
                    "egress_info": {"name": "New York, NY"},
                    "dest_bucket_id": "bucket_abc123",
                    "dest_path": "/",
                    "dest_name": "capture.pdf",
                    "dest_auth_token": "<file_token>",
                    "task_params": {
                        "request_type": "visual",
                        "urls": ["https://example.com"],
                        "vis_params": [
                            {"name": "output_pdf"},
                            {"name": "paper", "value": "A4"}
                        ]
                    },
                    "run_after": 1707849000
                }

        Returns:
            Task ID string for monitoring and managing the harvest task

        Raises:
            HarvesterAPIError: If task creation fails
            ValidationError: If parameters are invalid

        Example:
            >>> # Create asset harvest task
            >>> task_id = api.create_harvest_task(
            ...     task_type="asset",
            ...     urls=["https://example.com/files/"],
            ...     egress_info={
            ...         "name": "New York City",
            ...         "categories": {
            ...             "connectivity": "datacenter",
            ...             "availability": "private",
            ...             "protocol": "direct"
            ...         }
            ...     },
            ...     dest_path="/downloads/",
            ...     dest_name="harvested_files.zip",
            ...     task_params={
            ...         "recursive": True,
            ...         "level": 2,
            ...         "accept": "*.pdf,*.doc,*.docx"
            ...     }
            ... )
            >>>
            >>> # Create visual PDF capture
            >>> task_id = api.create_harvest_task(
            ...     task_type="visual",  # ← NOT "pdf"
            ...     urls=["https://example.com/report"],
            ...     egress_info={"name": "Singapore"},
            ...     dest_name="report_capture.pdf",
            ...     task_params={
            ...         "output_pdf": True,  # ← controls output format
            ...         "paper": "A4",
            ...         "landscape": False
            ...     }
            ... )
            >>>
            >>> # Create visual MHTML capture
            >>> task_id = api.create_harvest_task(
            ...     task_type="visual",  # ← NOT "mhtml"
            ...     urls=["https://example.com/page"],
            ...     egress_info={"name": "New York, NY"},
            ...     task_params={"output_mhtml": True}  # ← MHTML format
            ... )
        """
        task_params = self._normalize_task_params(task_params)
        top_level_keys = self._top_level_keys_for_type(task_type)

        # Validate all input parameters
        validate_task_creation_params(
            task_type,
            urls,
            egress_info,
            dest_path,
            task_params,
            self.VALID_TASK_TYPES,
            self._valid_params_map,
            top_level_keys=top_level_keys,
        )
        if run_after is not None:
            validate_run_after(run_after)

        # Prepare normalized data for API
        normalized_egress_info = normalize_egress_info(egress_info)
        bucket_id = dest_bucket_id or self.dest_bucket_id
        final_dest_name = generate_dest_name_if_needed(dest_name, task_type)

        # Build and execute the task creation request
        payload = build_task_creation_payload(
            task_type,
            urls,
            normalized_egress_info,
            bucket_id,
            dest_path,
            final_dest_name,
            task_params or {},
            self.file_api.auth_token,
            run_after=run_after,
            top_level_keys=top_level_keys,
        )

        return self._execute_task_creation(
            task_type, urls, normalized_egress_info, payload
        )

    def _execute_task_creation(
        self,
        task_type: str,
        urls: list[str],
        normalized_egress_info: dict[str, Any],
        payload: list[dict[str, Any]],
    ) -> str:
        """Execute the task creation API call.

        Args:
            task_type: Type of harvest task
            urls: List of URLs being harvested
            normalized_egress_info: Normalized egress information
            payload: API payload for task creation

        Returns:
            Task ID of the created task

        Raises:
            HarvesterAPIError: If task creation fails
        """
        try:
            self.logger.debug(
                "Creating %s harvest task for %d URLs with egress location: %s",
                task_type,
                len(urls),
                normalized_egress_info.get("name", "unknown"),
            )

            response = self._make_api_request("POST", "api/", self.auth_token, payload)
            result = self._extract_api_result(response)

            if isinstance(result, dict) and "task_id" in result:
                task_id: str = str(result["task_id"])
                self.logger.info("Created harvest task: %s", task_id)
                return task_id

            raise HarvesterAPIError(f"No task_id in response: {result}")

        except Exception as e:
            if isinstance(e, HarvesterAPIError):
                raise
            raise HarvesterAPIError(f"Failed to create harvest task: {e}") from e

    @api_tag(MethodType.API_COMMAND, api_command="find_harvest_task")
    def find_harvest_task(
        self,
        task_id: str,
        use_cache: bool = True,
        finished: bool | None = None,
    ) -> dict[str, Any]:
        """Retrieve the status and details of a specific harvest task.

        Gets comprehensive information about a harvest task including its current
        status, progress, results, and any error information. Supports caching
        to reduce API calls for frequently polled tasks.

        Args:
            task_id: The ID of the task to retrieve
            use_cache: Whether to use cached results (default: True)
            finished: If True, return only completed tasks. If False, return only
                in-progress tasks. If omitted, return regardless of completion state.

        Returns:
            Dictionary containing task details including:
            - task_id: Task identifier
            - status: Current task status (wait, claimed, running, done, error)
            - progress: Task progress percentage (if available)
            - created_ts: Task creation timestamp
            - started_ts: Task start timestamp (if started)
            - completed_ts: Task completion timestamp (if completed)
            - result_file_id: File ID of the result (if completed successfully)
            - last_error: Error message (if status is error)
            - task_params: Original task parameters

        Raises:
            HarvesterAPIError: If task retrieval fails
            ValidationError: If task_id is invalid

        Example:
            >>> task_info = api.find_harvest_task("task_123")
            >>> print(f"Status: {task_info['status']}")
            >>> if task_info['status'] == 'done':
            ...     print(f"Result file: {task_info['result_file_id']}")
            >>> elif task_info['status'] == 'error':
            ...     print(f"Error: {task_info['last_error']}")
        """
        self._require_task_id(task_id)

        # Check cache if enabled and requested
        if use_cache and self._status_cache_enabled and self._status_cache is not None:
            cached_result = get_cached_task_status(
                self._status_cache, task_id, self._status_cache_ttl
            )
            if cached_result is not None:
                self.logger.debug("Retrieved task status from cache: %s", task_id)
                return cached_result

        command: dict[str, Any] = {"command": "find_harvest_task", "task_id": task_id}
        if finished is not None:
            command["finished"] = finished
        payload = [command]

        try:
            self.logger.debug("Finding harvest task: %s", task_id)
            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            # Debug logging for response structure (only in debug mode)
            self.logger.debug("Processing response for task: %s", task_id)

            # Handle different response formats more robustly
            try:
                result = self._extract_api_result(response)
            except Exception as extract_error:
                # Fallback to direct response processing if extraction fails
                self.logger.debug(
                    "Standard extraction failed, trying direct response processing: %s",
                    extract_error,
                )
                if isinstance(response, list) and len(response) > 1:
                    result = response[1]
                else:
                    raise HarvesterAPIError(
                        f"Unexpected response format: {response}"
                    ) from extract_error

            # Handle different result formats
            if isinstance(result, list) and len(result) > 0:
                task_details = result[0]
            elif isinstance(result, dict):
                task_details = result
            elif isinstance(result, str):
                # Handle string responses that might contain error information
                if "task produced no output" in result or "TypeError" in result:
                    # Preserve the original server error message without additional wrapping
                    raise HarvesterAPIError(
                        f"Task {task_id} failed with server error: {result}"
                    )
                else:
                    raise HarvesterAPIError(
                        f"Unexpected string response for task_id: {task_id} - {result}"
                    )
            else:
                raise HarvesterAPIError(
                    f"No task details found for task_id: {task_id} - "
                    f"got result type: {type(result)}"
                )

            # Ensure task_details is a dictionary
            if not isinstance(task_details, dict):
                # Handle string task details that might contain error information
                if isinstance(task_details, str) and (
                    "task produced no output" in task_details
                    or "TypeError" in task_details
                ):
                    raise HarvesterAPIError(
                        f"Task {task_id} failed with server error: {task_details}"
                    )
                else:
                    raise HarvesterAPIError(
                        f"Task details must be a dictionary, "
                        f"got {type(task_details)}: {task_details}"
                    )

            self.logger.debug("Retrieved task details for: %s", task_id)

            # Cache the result if caching is enabled
            if self._status_cache_enabled:
                cache_task_status(
                    self._status_cache, task_id, task_details, self._status_cache_ttl
                )

            return task_details

        except Exception as e:
            if isinstance(e, HarvesterAPIError):
                raise
            raise HarvesterAPIError(f"Failed to find harvest task: {e}") from e

    @api_tag(MethodType.API_COMMAND, api_command="delete_harvest_task")
    def delete_harvest_task(self, task_id: str) -> bool:
        """Delete a specific harvest task.

        Permanently removes a harvest task and its associated data. This action
        cannot be undone. If the task is still running, it will be cancelled.

        Args:
            task_id: The ID of the task to delete

        Returns:
            True if deletion was successful, False otherwise

        Raises:
            HarvesterAPIError: If task deletion fails
            ValidationError: If task_id is invalid

        Example:
            >>> success = api.delete_harvest_task("task_123")
            >>> if success:
            ...     print("Task deleted successfully")
        """
        self._require_task_id(task_id)

        payload = [{"command": "delete_harvest_task", "task_id": task_id}]

        try:
            self.logger.debug("Deleting harvest task: %s", task_id)
            response = self._make_api_request("POST", "api/", self.auth_token, payload)

            result = self._extract_api_result(response)
            if isinstance(result, dict) and result.get("deleted"):
                self.logger.info("Successfully deleted harvest task: %s", task_id)
                return True

            self.logger.warning("Task deletion may have failed: %s", result)
            return False

        except Exception as e:
            if isinstance(e, HarvesterAPIError):
                raise
            raise HarvesterAPIError(f"Failed to delete harvest task: {e}") from e

    @api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task"])
    def wait_for_completion(
        self,
        task_id: str,
        max_wait_time: int = 1800,  # 30 minutes
        check_interval: int = 10,
        progress_callback: Callable[..., Any] | None = None,
    ) -> dict[str, Any]:
        """Wait for a harvest task to complete, with optional progress monitoring.

        Monitors a harvest task until it completes successfully, encounters an error,
        or the maximum wait time is exceeded. Provides optional progress callbacks
        for real-time monitoring.

        Args:
            task_id: The ID of the task to monitor
            max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
            check_interval: Time between status checks in seconds (default: 10)
            progress_callback: Optional function to call with progress updates

        Returns:
            Final task status dictionary when completed

        Raises:
            HarvesterAPIError: If task fails or times out
            ValidationError: If parameters are invalid

        Example:
            >>> def progress_handler(task_info):
            ...     print(f"Status: {task_info['status']}")
            ...     if 'progress' in task_info:
            ...         print(f"Progress: {task_info['progress']}%")
            >>>
            >>> result = api.wait_for_completion(
            ...     task_id="task_123",
            ...     max_wait_time=600,  # 10 minutes
            ...     progress_callback=progress_handler
            ... )
        """
        self._require_task_id(task_id)

        if not isinstance(max_wait_time, int) or max_wait_time <= 0:
            raise ValidationError("max_wait_time must be a positive integer")

        if not isinstance(check_interval, int) or check_interval <= 0:
            raise ValidationError("check_interval must be a positive integer")

        start_time = time.time()
        checks_performed = 0

        self.logger.info(
            "Waiting for task completion: %s (max %d seconds)", task_id, max_wait_time
        )

        try:
            while True:
                # Check if we've exceeded the maximum wait time
                elapsed_time = time.time() - start_time
                if elapsed_time > max_wait_time:
                    raise HarvesterAPIError(
                        f"Task {task_id} did not complete within {max_wait_time} seconds"
                    )

                # Get current task status
                try:
                    task_status = self.find_harvest_task(task_id)
                    if not isinstance(task_status, dict):
                        raise HarvesterAPIError(
                            f"Invalid task status format: {type(task_status)}"
                        )

                    status = task_status.get("status")
                    checks_performed += 1

                    # Call progress callback if provided
                    if progress_callback is not None:
                        try:
                            progress_callback(task_status)
                        except Exception as e:
                            self.logger.warning("Progress callback failed: %s", e)

                    # Check terminal states
                    if status == "done":
                        self.logger.info(
                            "Task %s completed successfully after %d checks (%.1f seconds)",
                            task_id,
                            checks_performed,
                            elapsed_time,
                        )
                        return task_status

                    elif status == "error":
                        error_msg = task_status.get("last_error", "Unknown error")
                        raise HarvesterAPIError(
                            f"Task {task_id} failed with server error: {error_msg}"
                        )

                except HarvesterAPIError as e:
                    # Re-check the original error message format
                    error_str = str(e)
                    if (
                        "task produced no output" in error_str
                        and "TypeError" in error_str
                    ):
                        # This indicates a response parsing issue from find_harvest_task
                        self.logger.error(
                            "Task status retrieval failed due to response format issue: %s",
                            e,
                        )
                        # Don't re-wrap the error message - just re-raise the original
                        raise
                    else:
                        raise

                if status in ["wait", "claimed", "running"]:
                    # Task is still in progress
                    self.logger.debug(
                        "Task %s status: %s (check %d, %.1f seconds elapsed)",
                        task_id,
                        status,
                        checks_performed,
                        elapsed_time,
                    )
                    time.sleep(check_interval)

                elif status not in ["done", "error"]:
                    raise HarvesterAPIError(f"Unexpected task status: {status}")

        except KeyboardInterrupt as exc:
            self.logger.info("Task monitoring interrupted by user")
            raise HarvesterAPIError("Task monitoring was interrupted") from exc

    @api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task"])
    def wait_for_completion_with_progress(
        self,
        task_id: str,
        task_type: str,
        max_wait_time: int = 1800,  # 30 minutes
        check_interval: int = 10,
        progress_callback: Callable[..., Any] | None = None,
    ) -> TaskProgress:
        """Wait for a harvest task to complete with enhanced progress tracking.

        Monitors a harvest task with detailed progress metrics including timing,
        status transitions, and estimated completion times.

        Args:
            task_id: The ID of the task to monitor
            task_type: Type of harvest task
            max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
            check_interval: Time between status checks in seconds (default: 10)
            progress_callback: Optional function to call with TaskProgress updates

        Returns:
            TaskProgress object with detailed metrics when completed

        Raises:
            HarvesterAPIError: If task fails or times out
            ValidationError: If parameters are invalid

        Example:
            >>> def progress_handler(progress: TaskProgress):
            ...     print(f"Status: {progress.status} ({progress.progress_percentage}%)")
            ...     if progress.estimated_completion:
            ...         eta = datetime.fromtimestamp(progress.estimated_completion)
            ...         print(f"ETA: {eta}")
            >>>
            >>> progress = api.wait_for_completion_with_progress(
            ...     task_id="task_123",
            ...     task_type="asset",
            ...     progress_callback=progress_handler
            ... )
            >>> print(f"Completed in {progress.get_duration():.1f} seconds")
        """
        self._require_task_id(task_id)

        if not isinstance(max_wait_time, int) or max_wait_time <= 0:
            raise ValidationError("max_wait_time must be a positive integer")

        if not isinstance(check_interval, int) or check_interval <= 0:
            raise ValidationError("check_interval must be a positive integer")

        # Initialize progress tracker
        progress = TaskProgress(task_id, task_type)

        start_time = time.time()
        checks_performed = 0

        self.logger.info(
            "Waiting for task completion with progress tracking: %s (max %d seconds)",
            task_id,
            max_wait_time,
        )

        try:
            while True:
                # Check if we've exceeded the maximum wait time
                elapsed_time = time.time() - start_time
                if elapsed_time > max_wait_time:
                    progress.set_error(f"Task timeout after {max_wait_time} seconds")
                    raise HarvesterAPIError(
                        f"Task {task_id} did not complete within {max_wait_time} seconds"
                    )

                # Get current task status
                task_status = self.find_harvest_task(task_id)
                status: str = task_status.get("status", "unknown")
                checks_performed += 1

                # Update progress tracker
                progress.update_status(status, task_status.get("progress"))
                progress.retry_count = checks_performed - 1

                # Call progress callback if provided
                if progress_callback is not None:
                    try:
                        progress_callback(progress)
                    except Exception as e:
                        self.logger.warning("Progress callback failed: %s", e)

                # Check terminal states
                if status == "done":
                    self.logger.info(
                        "Task %s completed successfully after %d checks (%.1f seconds)",
                        task_id,
                        checks_performed,
                        elapsed_time,
                    )
                    return progress

                elif status == "error":
                    error_msg = task_status.get("last_error", "Unknown error")
                    progress.set_error(error_msg)
                    raise HarvesterAPIError(
                        f"Task {task_id} failed with server error: {error_msg}"
                    )

                elif status in ["wait", "claimed", "running"]:
                    # Task is still in progress
                    self.logger.debug(
                        "Task %s status: %s (check %d, %.1f seconds elapsed)",
                        task_id,
                        status,
                        checks_performed,
                        elapsed_time,
                    )
                    time.sleep(check_interval)

                else:
                    progress.set_error(f"Unexpected task status: {status}")
                    raise HarvesterAPIError(f"Unexpected task status: {status}")

        except KeyboardInterrupt as exc:
            progress.set_error("Task monitoring interrupted by user")
            self.logger.info("Task monitoring interrupted by user")
            raise HarvesterAPIError("Task monitoring was interrupted") from exc

    @api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task", "getfile"])
    def download_task_result(
        self,
        task_id: str,
        output_path: str,
        verify_completion: bool = True,
    ) -> str:
        """Download the result of a completed harvest task.

        Downloads the harvested content from a completed task to the local filesystem.
        Optionally verifies that the task is completed before attempting download.

        Args:
            task_id: The ID of the completed task
            output_path: Local path where to save the downloaded file
            verify_completion: Whether to verify task completion before download

        Returns:
            Path of the downloaded file

        Raises:
            HarvesterAPIError: If download fails or task is not completed
            ValidationError: If parameters are invalid

        Example:
            >>> downloaded_file = api.download_task_result(
            ...     task_id="task_123",
            ...     output_path="/local/downloads/harvest_result.zip"
            ... )
            >>> print(f"Downloaded to: {downloaded_file}")
        """
        self._require_task_id(task_id)

        if not output_path:
            raise ValidationError("output_path must be provided")

        # Verify task completion if requested
        if verify_completion:
            task_status = self.find_harvest_task(task_id)
            if task_status.get("status") != "done":
                raise HarvesterAPIError(
                    f"Task {task_id} is not completed. "
                    f"Current status: {task_status.get('status')}"
                )

            result_file_id = task_status.get("result_file_id")
            if not result_file_id:
                raise HarvesterAPIError(f"No result file found for task {task_id}")
        else:
            # Get task details to find result file ID
            task_status = self.find_harvest_task(task_id)
            result_file_id = task_status.get("result_file_id")
            if not result_file_id:
                raise HarvesterAPIError(f"No result file found for task {task_id}")

        try:
            self.logger.debug("Downloading result for task: %s", task_id)
            downloaded_path = self.file_api.download_file(result_file_id, output_path)

            self.logger.info(
                "Successfully downloaded task result: %s -> %s",
                task_id,
                downloaded_path,
            )
            return downloaded_path

        except FileAPIError as e:
            raise HarvesterAPIError(f"Failed to download task result: {e}") from e

    @api_tag(MethodType.UTILITY)
    @classmethod
    def get_valid_params(cls, task_type: str) -> set[str]:
        """Get valid parameters for a specific task type.

        Args:
            task_type: Type of harvest task ("asset", "visual", or "video")

        Returns:
            Set of valid parameter names for the task type

        Raises:
            ValidationError: If task_type is invalid

        Example:
            >>> asset_params = HarvesterAPI.get_valid_params("asset")
            >>> print("Valid asset parameters:")
            >>> for param in sorted(asset_params):
            ...     print(f"  - {param}")
        """
        if task_type in {"asset", "single"}:
            return cls.VALID_ASSET_PARAMS.copy()
        elif task_type == "visual":
            return cls.VALID_VISUAL_PARAMS.copy()
        elif task_type == "video":
            return cls.VALID_VIDEO_PARAMS.copy()
        else:
            raise ValidationError(
                f"Invalid task_type: {task_type}. "
                f"Valid types are: {', '.join(sorted(cls.VALID_TASK_TYPES))}"
            )

    @api_tag(MethodType.UTILITY)
    @classmethod
    def get_valid_task_types(cls) -> set[str]:
        """Get all valid task types.

        Returns:
            Set of valid task type names

        Example:
            >>> task_types = HarvesterAPI.get_valid_task_types()
            >>> print("Supported task types:")
            >>> for task_type in sorted(task_types):
            ...     print(f"  - {task_type}")
        """
        return cls.VALID_TASK_TYPES.copy()

    # Async Methods (require httpx)

    @api_tag(
        MethodType.ASYNC_VARIANT,
        api_command="create_harvest_task",
        sync_of="create_harvest_task",
    )
    async def create_harvest_task_async(
        self,
        task_type: str,
        urls: list[str],
        egress_info: dict[str, Any],
        dest_path: str = "/",
        dest_name: str | None = None,
        dest_bucket_id: str | None = None,
        task_params: dict[str, Any] | None = None,
        max_retries: int = 5,
        backoff_factor: int = 2,
        run_after: int | str | None = None,
    ) -> str:
        """Asynchronously create a new harvest task for web content collection.

        This async version allows for non-blocking task creation and can be used
        in concurrent workflows for creating multiple tasks simultaneously.

        Args:
            task_type: Type of harvest task ("asset", "single", "visual", or "video")
            urls: Array containing exactly one URL string to harvest
            egress_info: Egress location information containing:
                - name: Egress location name (required, e.g., "New York, NY")
                - categories: Optional object with connectivity, availability, protocol
                  Each category field can be a string value or null. For
                  non-"asset" task types, connectivity/protocol are validated
                  against what is actually available at name.
            dest_path: Destination path in storage (defaults to root "/")
            dest_name: Name for the result file (auto-generated if not provided)
            dest_bucket_id: Destination bucket ID (uses default if not provided)
            task_params: Dictionary of task-specific parameters
            max_retries: Maximum number of retries for explicit HTTP 429
                rate-limit responses (default: 5). Ambiguous transport failures
                are not retried because task creation is non-idempotent.
            backoff_factor: Positive integer exponential backoff multiplier for
                rate-limit retries (default: 2)
            run_after: Optional schedule time — Unix epoch integer
                (e.g., 1707849000) or ISO 8601 UTC string
                (e.g., "2026-02-13T18:30:00Z"). When provided the task
                enters "wait" status until the specified time.

        Returns:
            Task ID string for monitoring and managing the harvest task

        Raises:
            HarvesterAPIError: If task creation fails
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def create_tasks():
            ...     harvester = HarvesterAPI(config)
            ...     task_id = await harvester.create_harvest_task_async(
            ...         task_type="asset",
            ...         urls=["https://example.com/files/"],
            ...         egress_info={"name": "New York, NY"},
            ...         dest_path="/downloads/"
            ...     )
            ...     return task_id
            >>> task_id = asyncio.run(create_tasks())
        """
        validate_positive_int(max_retries, "max_retries")
        validate_positive_int(backoff_factor, "backoff_factor")

        task_params = self._normalize_task_params(task_params)
        top_level_keys = self._top_level_keys_for_type(task_type)
        validate_task_creation_params(
            task_type,
            urls,
            egress_info,
            dest_path,
            task_params,
            self.VALID_TASK_TYPES,
            self._valid_params_map,
            top_level_keys=top_level_keys,
        )

        if run_after is not None:
            validate_run_after(run_after)

        normalized_egress_info = normalize_egress_info(egress_info)
        bucket_id = dest_bucket_id or self.dest_bucket_id
        final_dest_name = generate_dest_name_if_needed(dest_name, task_type)

        payload_commands = build_task_creation_payload(
            task_type,
            urls,
            normalized_egress_info,
            bucket_id,
            dest_path,
            final_dest_name,
            task_params or {},
            self.file_api.auth_token,
            run_after=run_after,
            top_level_keys=top_level_keys,
        )

        # Build API payload
        api_url = f"{self.base_url.rstrip('/')}/api/"
        payload = [{"command": "setauth", "data": self.auth_token}] + payload_commands

        async with httpx.AsyncClient(verify=self._ssl_context) as client:
            for attempt in range(max_retries):
                try:
                    response = await client.post(
                        api_url,
                        json=payload,
                        headers={
                            "Accept": "application/json",
                            "Content-Type": "application/json",
                        },
                        timeout=httpx.Timeout(30),
                    )
                    if response.status_code == 200:
                        task_response = response.json()
                        if (
                            isinstance(task_response, list)
                            and len(task_response) > 1
                            and isinstance(task_response[1], dict)
                        ):
                            result_inner = task_response[1].get("result")
                            result_task_id: str | None = (
                                result_inner.get("task_id")
                                if isinstance(result_inner, dict)
                                else None
                            )
                            if result_task_id:
                                self.logger.info(
                                    "Created async harvest task: %s", result_task_id
                                )
                                return result_task_id

                        raise HarvesterAPIError(
                            f"Unexpected response format: {task_response}"
                        )

                    elif response.status_code == 429:
                        if attempt == max_retries - 1:
                            break
                        try:
                            retry_after = int(
                                response.headers.get(
                                    "Retry-After", backoff_factor**attempt
                                )
                            )
                        except ValueError:
                            retry_after = int(backoff_factor**attempt)
                        self.logger.warning(
                            "Rate limit hit, retrying after %s seconds...", retry_after
                        )
                        await asyncio.sleep(retry_after)
                    else:
                        raise HarvesterAPIError(
                            f"API request failed with status code {response.status_code}"
                        )

                except httpx.HTTPError as e:
                    raise HarvesterAPIError(
                        "Harvest task creation failed after the request may have "
                        "reached the server; the task may have been created. "
                        "The request was not retried to avoid creating a duplicate: "
                        f"{e}"
                    ) from e
                except HarvesterAPIError:
                    raise
                except Exception as e:
                    raise HarvesterAPIError(
                        f"Failed to create async harvest task: {e}"
                    ) from e

        raise HarvesterAPIError("Rate-limit retries exhausted; task was not created")

    @api_tag(
        MethodType.ASYNC_VARIANT,
        api_command="find_harvest_task",
        sync_of="wait_for_completion",
    )
    async def wait_for_completion_async(
        self,
        task_id: str,
        max_wait_time: int = 1800,  # 30 minutes
        check_interval: int = 10,
        progress_callback: Callable[..., Any] | None = None,
    ) -> dict[str, Any]:
        """Asynchronously wait for a harvest task to complete.

        This async version allows for non-blocking task monitoring and can be used
        to monitor multiple tasks concurrently.

        Args:
            task_id: The ID of the task to monitor
            max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
            check_interval: Time between status checks in seconds (default: 10)
            progress_callback: Optional function to call with progress updates

        Returns:
            Final task status dictionary when completed. Shape matches the
            synchronous :meth:`wait_for_completion`: it is the raw task
            dictionary as returned by the server (``status``, ``progress``,
            ``task_params``, ``worker_id``, and any other server-reported
            fields), merged with the normalized convenience fields produced
            by :func:`~silo_sdk.utils.harvester_utils.parse_task_response`
            (``task_metadata``, and normalized ``task_id``,
            ``result_file_id``, ``last_error``, ``run_after``, ``created_ts``,
            ``started_ts``, ``completed_ts``). No server-reported fields are
            dropped.

        Raises:
            HarvesterAPIError: If task fails or times out
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def monitor_task():
            ...     harvester = HarvesterAPI(config)
            ...     def progress_handler(task_info):
            ...         print(f"Status: {task_info['status']}")
            ...
            ...     result = await harvester.wait_for_completion_async(
            ...         task_id="task_123",
            ...         progress_callback=progress_handler
            ...     )
            ...     return result
            >>> result = asyncio.run(monitor_task())
        """
        self._require_task_id(task_id)

        if not isinstance(max_wait_time, int) or max_wait_time <= 0:
            raise ValidationError("max_wait_time must be a positive integer")

        if not isinstance(check_interval, int) or check_interval <= 0:
            raise ValidationError("check_interval must be a positive integer")

        max_retries = max_wait_time // check_interval
        api_url = f"{self.base_url.rstrip('/')}/api/"

        self.logger.info("Starting async status check loop for task ID: %s", task_id)
        retries = 0
        last_response = None
        start_time = time.time()

        try:
            async with httpx.AsyncClient(verify=self._ssl_context) as client:
                while retries < max_retries:
                    # Bound the loop by wall-clock elapsed time in addition to
                    # retry count. Retry count alone assumes each iteration
                    # costs ~check_interval seconds, but a slow API response
                    # (up to the 30s per-request timeout below) isn't counted
                    # against max_wait_time, which lets the real wait exceed
                    # it. This check only ever exits *earlier* than the retry
                    # count would (never later), so it doesn't change
                    # behavior when checks are fast.
                    if time.time() - start_time > max_wait_time:
                        self.logger.warning(
                            "Wall-clock max_wait_time (%s seconds) exceeded for "
                            "task %s before the retry budget was exhausted.",
                            max_wait_time,
                            task_id,
                        )
                        break

                    self.logger.debug(
                        f"Checking status of task {task_id} "
                        f"(attempt {retries + 1}/{max_retries})"
                    )

                    payload = [
                        {"command": "setauth", "data": self.auth_token},
                        {"command": "find_harvest_task", "task_id": task_id},
                    ]

                    try:
                        response = await client.post(
                            api_url,
                            json=payload,
                            headers={
                                "Accept": "application/json",
                                "Content-Type": "application/json",
                            },
                            timeout=httpx.Timeout(30),
                        )
                        response.raise_for_status()
                        last_response = response.json()

                        if (
                            last_response
                            and isinstance(last_response, list)
                            and len(last_response) > 1
                            and isinstance(last_response[1], dict)
                        ):
                            task_result = last_response[1].get("result", [])
                        else:
                            self.logger.warning(
                                f"Invalid response format for task {task_id}: "
                                f"{last_response}"
                            )
                            raise HarvesterAPIError(
                                f"Invalid response format for task {task_id}"
                            )

                        if task_result:
                            task_status = task_result[0].get("status", "")
                            self.logger.info(f"Task {task_id} status: {task_status}")

                            # Call progress callback if provided
                            if progress_callback is not None:
                                try:
                                    progress_callback(task_result[0])
                                except Exception as e:
                                    self.logger.warning(
                                        "Progress callback failed: %s", e
                                    )

                            if task_status == "done":
                                self.logger.info(
                                    f"Task {task_id} completed successfully."
                                )
                                task_info = parse_task_response(last_response)
                                if not task_info:
                                    raise HarvesterAPIError(
                                        f"Failed to parse task response for {task_id}"
                                    )
                                # Align this method's return shape with the
                                # synchronous wait_for_completion(), which
                                # returns the full raw server task dict (see
                                # find_harvest_task/task_result[0] above).
                                # parse_task_response() only surfaces 9
                                # hardcoded keys, silently dropping fields
                                # like progress/task_params/worker_id — merge
                                # the raw dict as the base and layer the
                                # normalized fields on top so nothing the
                                # server sent is lost.
                                full_task_info: dict[str, Any] = {
                                    **task_result[0],
                                    **task_info,
                                }
                                return full_task_info
                            elif task_status == "error":
                                error_msg = task_result[0].get(
                                    "last_error", "Unknown error"
                                )
                                raise HarvesterAPIError(
                                    f"Task {task_id} failed with server error: {error_msg}"
                                )
                            elif task_status == "wait":
                                run_after = task_result[0].get("run_after")
                                if run_after:
                                    run_after_time = datetime.fromtimestamp(
                                        run_after, tz=UTC
                                    )
                                    self.logger.info(
                                        f"Task {task_id} is in 'wait' status, "
                                        f"scheduled to run at: {run_after_time} UTC."
                                    )
                                else:
                                    self.logger.info(
                                        f"Task {task_id} is in 'wait' status, "
                                        f"but no scheduled time is available."
                                    )
                            elif task_status in ["claimed", "running"]:
                                self.logger.info(
                                    f"Task {task_id} is in progress... "
                                    f"Status: {task_status}"
                                )
                            else:
                                self.logger.warning(
                                    f"Unexpected task status for task {task_id}: "
                                    f"{task_status}"
                                )
                        else:
                            self.logger.warning(
                                f"No task result found for task {task_id}"
                            )

                    except httpx.HTTPError as e:
                        self.logger.error(
                            f"Error: Request failed for task {task_id} with exception {e}."
                        )
                        raise HarvesterAPIError(
                            f"Request failed for task {task_id}: {e}"
                        ) from e

                    await asyncio.sleep(check_interval)
                    retries += 1

            # Either the retry budget or the wall-clock max_wait_time was
            # exhausted without the task reaching a terminal state.
            self.logger.error(
                f"Timed out waiting for task {task_id}. "
                f"Task may not have completed successfully."
            )
            raise HarvesterAPIError(
                f"Task {task_id} did not complete within {max_wait_time} seconds"
            )

        except HarvesterAPIError:
            raise
        except Exception as e:
            raise HarvesterAPIError(f"Failed to monitor async task: {e}") from e

    @api_tag(
        MethodType.ASYNC_VARIANT,
        api_command="create_harvest_task",
        sync_of="create_harvest_task",
    )
    async def process_harvest_workflow_async(
        self,
        task_type: str,
        urls: list[str],
        egress_info: dict[str, Any],
        output_dir: str,
        dest_path: str = "/",
        dest_name: str | None = None,
        task_params: dict[str, Any] | None = None,
        use_chunked_download: bool = False,
        max_wait_time: int = 1800,
        check_interval: int = 20,
    ) -> dict[str, Any] | None:
        """Process a complete harvest workflow asynchronously.

        This high-level async method orchestrates the entire harvest workflow:
        1. Creates a harvest task
        2. Monitors its status until completion
        3. Downloads the result file
        4. Returns comprehensive task information

        Args:
            task_type: Type of harvest task ("asset", "visual", or "video")
            urls: Array containing exactly one URL string to harvest
            egress_info: Egress location information containing:
                - name: Egress location name (required, e.g., "New York, NY")
                - categories: Optional object with connectivity, availability, protocol
                  Each category field can be a string value or null. For
                  non-"asset" task types, connectivity/protocol are validated
                  against what is actually available at name.
            output_dir: Directory to save downloaded results
            dest_path: Destination path in storage (default: "/")
            dest_name: Name for result file (auto-generated if not provided)
            task_params: Dictionary of task-specific parameters
            use_chunked_download: Whether to use chunked download (default: False)
            max_wait_time: Maximum time to wait for completion (default: 30 minutes)
            check_interval: Time between status checks (default: 20 seconds)

        Returns:
            Dictionary with task information and results, None if failed

        Raises:
            HarvesterAPIError: If workflow fails
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def harvest_site():
            ...     harvester = HarvesterAPI(config)
            ...     result = await harvester.process_harvest_workflow_async(
            ...         task_type="asset",
            ...         urls=["https://example.com/documents/"],
            ...         egress_info={"name": "New York, NY"},
            ...         output_dir="./downloads",
            ...         task_params={"recursive": True, "level": 2}
            ...     )
            ...     return result
            >>> result = asyncio.run(harvest_site())
        """
        # Validate egress_info with task-type-aware validation
        validate_egress_for_task_type(task_type, egress_info)

        # Create output directory
        output_path = Path(output_dir)
        output_path.mkdir(parents=True, exist_ok=True)

        self.logger.info(
            "Processing %s task for URL: %s", task_type, urls[0] if urls else "<empty>"
        )

        try:
            # Step 1: Create harvest task
            task_id = await self.create_harvest_task_async(
                task_type=task_type,
                urls=urls,
                egress_info=egress_info,
                dest_path=dest_path,
                dest_name=dest_name,
                task_params=task_params,
            )

            self.logger.info("Started %s task with ID: %s", task_type, task_id)

            # Step 2: Monitor task status
            task_info = await self.wait_for_completion_async(
                task_id=task_id,
                max_wait_time=max_wait_time,
                check_interval=check_interval,
            )

            if not task_info:
                self.logger.error("No status response received for task %s", task_id)
                return None

            result_file_id = task_info.get("result_file_id")
            if not result_file_id:
                self.logger.error("Missing result file ID for task %s", task_id)
                return task_info

            # Step 3: Download result file
            output_file_path = output_path / f"harvest-{task_id}.zip"

            try:
                self.logger.info("Downloading result for task %s", task_id)

                if use_chunked_download:
                    downloaded_file = await self.file_api.download_file_chunked_async(
                        file_id=result_file_id, output_path=str(output_file_path)
                    )
                else:
                    downloaded_file = (
                        await self.file_api.download_file_with_retry_async(
                            file_id=result_file_id, output_path=str(output_file_path)
                        )
                    )

                if downloaded_file:
                    self.logger.info("Downloaded file saved to: %s", downloaded_file)
                    task_info["downloaded_file"] = str(downloaded_file)
                    task_info["download_success"] = True
                else:
                    self.logger.error(
                        "Failed to download result file for task %s", task_id
                    )
                    task_info["download_success"] = False

            except Exception as e:
                self.logger.error("Error downloading file for task %s: %s", task_id, e)
                task_info["download_success"] = False
                task_info["download_error"] = str(e)

            return task_info

        except Exception as e:
            if isinstance(e, (HarvesterAPIError, ValidationError)):
                raise
            raise HarvesterAPIError(
                f"Failed to process async harvest workflow: {e}"
            ) from e

    @api_tag(
        MethodType.ASYNC_VARIANT,
        api_command="create_harvest_task",
        sync_of="create_harvest_task",
    )
    async def bulk_create_and_monitor_async(
        self,
        tasks: list[dict[str, Any]],
        output_dir: str,
        max_concurrent: int = 5,
        egress_info: dict[str, Any] | None = None,
    ) -> list[dict[str, Any] | None]:
        """Create and monitor multiple harvest tasks concurrently.
        This method processes multiple harvest tasks in parallel with configurable
        concurrency limits to efficiently handle bulk harvesting operations.

        Args:
            tasks: List of task dictionaries, each containing:
                - task_type: Type of harvest task
                - urls: List containing exactly one URL string to harvest
                  (canonical key — matches :meth:`create_harvest_task`).
                  A single ``url`` string key is also accepted as a
                  convenience alias and is wrapped into a one-item list.
                - egress_info: Optional per-task egress location dict. When
                  provided, it overrides the bulk-level ``egress_info`` for
                  this task only — this is how you mix asset tasks (which
                  need one of the 6 datacenter locations) with non-asset
                  tasks in the same call.
                - params: Task-specific parameters (dict format)
                - dest_path: Optional destination path (default: "/")
                - dest_name: Optional destination name
            output_dir: Directory to save downloaded results
            max_concurrent: Positive maximum number of concurrent tasks (default: 5)
            egress_info: Default egress location information used for any
                task that doesn't specify its own ``egress_info``
                (default: {"name": "New York, NY"}).
                Note: asset tasks require locations from the 6-location datacenter
                list (e.g., "New York City", not "New York, NY"). If mixing asset
                and non-asset tasks, set ``egress_info`` per-task (see above) or
                use separate calls.

        Returns:
            List of task results (same order as input), None for failed tasks

        Raises:
            HarvesterAPIError: If bulk processing fails
            ValidationError: If parameters are invalid

        Example:
            >>> import asyncio
            >>> async def bulk_harvest():
            ...     harvester = HarvesterAPI(config)
            ...     tasks = [
            ...         {
            ...             "task_type": "asset",
            ...             "urls": ["https://site1.com"],
            ...             "egress_info": {"name": "New York City"},
            ...             "params": {"recursive": True}
            ...         },
            ...         {
            ...             "task_type": "visual",
            ...             "urls": ["https://site2.com"],
            ...             "params": {"output_pdf": True}
            ...         }
            ...     ]
            ...     results = await harvester.bulk_create_and_monitor_async(
            ...         tasks=tasks,
            ...         output_dir="./downloads",
            ...         max_concurrent=3
            ...     )
            ...     return results
            >>> results = asyncio.run(bulk_harvest())
        """
        validate_positive_int(max_concurrent, "max_concurrent")

        # Set default egress_info if not provided
        if egress_info is None:
            egress_info = {"name": "New York, NY"}

        # Create semaphore to limit concurrent tasks
        semaphore = asyncio.Semaphore(max_concurrent)

        async def process_single_task(
            task_config: dict[str, Any],
        ) -> dict[str, Any] | None:
            """Process a single task with semaphore control."""
            async with semaphore:
                try:
                    # Get task_params from either 'params' or 'task_params' key
                    task_params = task_config.get("task_params") or task_config.get(
                        "params", {}
                    )

                    # Accept both 'urls' (list — canonical, matches
                    # create_harvest_task/create_harvest_task_async) and a
                    # single 'url' string as a convenience alias, so callers
                    # don't hit a KeyError from a docstring/param mismatch.
                    task_urls = task_config.get("urls")
                    if task_urls is None:
                        single_url = task_config.get("url")
                        task_urls = [single_url] if single_url is not None else None
                    if not task_urls:
                        raise ValidationError(
                            "Each task dict must include a non-empty 'urls' "
                            "list (or a single 'url' string)"
                        )

                    # Honor this task's own egress_info when provided,
                    # falling back to the bulk-level default. Previously the
                    # outer-scope egress_info was always used, making it
                    # impossible to mix asset tasks (which require one of
                    # the 6 datacenter locations) with non-asset tasks in a
                    # single bulk call, despite the docstring advertising
                    # per-task egress as the way to do so.
                    task_egress_info = task_config.get("egress_info", egress_info)

                    return await self.process_harvest_workflow_async(
                        task_type=task_config["task_type"],
                        urls=task_urls,
                        egress_info=task_egress_info,
                        output_dir=output_dir,
                        dest_path=task_config.get("dest_path", "/"),
                        dest_name=task_config.get("dest_name"),
                        task_params=(
                            task_params if isinstance(task_params, dict) else None
                        ),
                    )
                except Exception as e:
                    self.logger.error("Task failed: %s", e)
                    return None

        # Process all tasks concurrently
        self.logger.info(
            "Processing %d tasks with max concurrency: %d", len(tasks), max_concurrent
        )

        try:
            results = await asyncio.gather(
                *[process_single_task(task) for task in tasks]
            )

            processed_results: list[dict[str, Any] | None] = [
                r if isinstance(r, dict) else None for r in results
            ]

            successful_tasks = sum(1 for r in processed_results if r is not None)
            self.logger.info(
                f"Completed bulk processing: {successful_tasks}/{len(tasks)} tasks successful"
            )

            return processed_results

        except Exception as e:
            if isinstance(e, (HarvesterAPIError, ValidationError)):
                raise
            raise HarvesterAPIError(f"Failed to process bulk async harvest: {e}") from e

__init__

__init__(config: dict[str, Any]) -> None

Initialize the harvester API client.

Parameters:

Name	Type	Description	Default
config	dict[str, Any]	Configuration dictionary containing API settings	required

Raises:

Type	Description
ConfigurationError	If required configuration is missing or invalid

Source code in silo_sdk/harvesting/harvester_api.py

def __init__(self, config: dict[str, Any]) -> None:
    """Initialize the harvester API client.

    Args:
        config: Configuration dictionary containing API settings

    Raises:
        ConfigurationError: If required configuration is missing or invalid
    """
    super().__init__(config)

    # Comprehensive configuration validation
    validation_errors = validate_harvester_config(config)
    if validation_errors:
        raise ConfigurationError(
            f"Harvester configuration validation failed: {'; '.join(validation_errors)}"
        )

    # Basic required configuration (legacy support)
    validate_config_for_api(config, ["SCRAPE_TOKEN", "BUCKET_ID"])

    self.auth_token = config["SCRAPE_TOKEN"]
    self.dest_bucket_id = config["BUCKET_ID"]

    # Initialize task status cache if enabled
    self._status_cache_enabled, self._status_cache_ttl, self._status_cache = (
        init_status_cache(config)
    )

    # Initialize file API for downloading results
    self.file_api = FileAPI(config)
    self._ssl_context = self.create_ssl_context()

    self.logger.info("Initialized HarvesterAPI client with validated configuration")

__enter__

__enter__() -> HarvesterAPI

Enter context manager for resource management.

Returns:

Type	Description
HarvesterAPI	Self for use in with statement

Source code in silo_sdk/harvesting/harvester_api.py

def __enter__(self) -> HarvesterAPI:
    """Enter context manager for resource management.

    Returns:
        Self for use in with statement
    """
    self.logger.debug("Entering HarvesterAPI context manager")
    return self

__exit__

__exit__(exc_type: type[BaseException] | None, exc_val: BaseException | None, exc_tb: TracebackType | None) -> None

Exit context manager and perform cleanup.

Parameters:

Name	Type	Description	Default
exc_type	type[BaseException] | None	Exception type if an exception occurred	required
exc_val	BaseException | None	Exception value if an exception occurred	required
exc_tb	TracebackType | None	Exception traceback if an exception occurred	required

Source code in silo_sdk/harvesting/harvester_api.py

def __exit__(
    self,
    exc_type: type[BaseException] | None,
    exc_val: BaseException | None,
    exc_tb: TracebackType | None,
) -> None:
    """Exit context manager and perform cleanup.

    Args:
        exc_type: Exception type if an exception occurred
        exc_val: Exception value if an exception occurred
        exc_tb: Exception traceback if an exception occurred
    """
    # Suppress unused parameter warnings - these are required by the context manager protocol
    _ = exc_type, exc_val, exc_tb
    try:
        self.cleanup_resources()
    except Exception as e:
        self.logger.warning("Error during context manager cleanup: %s", e)
    finally:
        self.logger.debug("Exited HarvesterAPI context manager")

cleanup_resources

cleanup_resources() -> None

Clean up resources and perform housekeeping tasks.

This method should be called when done with the HarvesterAPI instance to properly clean up resources, clear caches, and close connections.

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.UTILITY)
def cleanup_resources(self) -> None:
    """Clean up resources and perform housekeeping tasks.

    This method should be called when done with the HarvesterAPI instance
    to properly clean up resources, clear caches, and close connections.
    """
    try:
        # Clear status cache
        if self._status_cache:
            cache_size = len(self._status_cache)
            self._status_cache.clear()
            if cache_size > 0:
                self.logger.info("Cleared %d cached task statuses", cache_size)

        # Close HTTP session if it exists
        if hasattr(self, "session") and self.session:
            self.session.close()
            self.logger.debug("Closed HTTP session")

        # Clean up file API resources
        if hasattr(self, "file_api") and hasattr(
            self.file_api, "cleanup_resources"
        ):
            self.file_api.cleanup_resources()

    except Exception as e:
        self.logger.error("Error during resource cleanup: %s", e)

create_harvest_task

create_harvest_task(task_type: str, urls: list[str], egress_info: dict[str, Any], dest_path: str = '/', dest_name: str | None = None, dest_bucket_id: str | None = None, task_params: dict[str, Any] | None = None, run_after: int | str | None = None) -> str

Create a new harvest task for web content collection.

Creates a harvesting task that will collect content from the specified URL using the given parameters and store the results in the designated location.

Parameters:

Name	Type	Description	Default
task_type	str	Type of harvest task. Must be one of "asset", "visual", "video", or "single". "visual" — Full-page capture. Output format controlled by task_params: Screenshot (default): no params, or {"output_image": True} PDF: {"output_pdf": True} MHTML: {"output_mhtml": True} IMPORTANT: "screenshot", "pdf", and "mhtml" are not valid task_type values. Always use "visual" with task_params. "asset" — Recursive asset download (HTML, JS, CSS, images). Control depth/filters via task_params (wget-style options). "video" — Video recording of page interaction (youtube-dl-style options).	required
urls	list[str]	List containing exactly one URL string to harvest.	required
egress_info	dict[str, Any]	Egress location dict with required "name" key (e.g., {"name": "New York, NY"}). Optional "categories" key accepts "connectivity", "availability", "protocol". For non-"asset" task types, "connectivity"/"protocol" are validated against what is actually available at "name" (see :mod:silo_sdk.harvesting.egress_validation). Does not apply to "asset" tasks.	required
dest_path	str	Destination path in storage (defaults to "/")	'/'
dest_name	str | None	Name for the result file (auto-generated if not provided)	None
dest_bucket_id	str | None	Destination bucket ID (uses default if not provided)	None
task_params	dict[str, Any] | None	Task-specific output parameters. Most keys are converted to a list of {"name": key} or {"name": key, "value": val} dicts sent under the appropriate wire key (vis_params, wget_params, vid_params). The following keys are instead injected directly at the top level of the wire task_params object: return-cookies — send back cookies from the session cookies — Netscape-format cookie file content (string) all_files — include all fetched files alongside the main content. Valid for "visual", "single", and "video" tasks. Raises ValidationError for "asset" tasks. Key visual params: output_pdf, output_mhtml, output_image, scale, landscape, paper. See class Attributes for full param sets per task type.	None
run_after	int | str | None	Optional schedule time — Unix epoch integer (e.g., 1707849000) or ISO 8601 UTC string (e.g., "2026-02-13T18:30:00Z"). Task enters "wait" status until the specified time.	None

Note

Wire format parameter mappings:

• task_type → task_params.request_type
• urls → task_params.urls

• task_params (Python dict) → list of {"name": key} or {"name": key, "value": val} dicts sent as:

• task_params.vis_params for "visual" tasks

• task_params.wget_params for "asset" tasks

• task_params.vid_params for "video" tasks

• egress_info, dest_bucket_id, dest_path, dest_name, dest_auth_token remain at the top-level command dict

• run_after is injected at top-level command dict alongside egress_info, NOT inside task_params

Wire format structure for visual task::

{
    "command": "create_harvest_task",
    "egress_info": {"name": "New York, NY"},
    "dest_bucket_id": "bucket_abc123",
    "dest_path": "/",
    "dest_name": "capture.pdf",
    "dest_auth_token": "<file_token>",
    "task_params": {
        "request_type": "visual",
        "urls": ["https://example.com"],
        "vis_params": [
            {"name": "output_pdf"},
            {"name": "paper", "value": "A4"}
        ]
    },
    "run_after": 1707849000
}

Returns:

Type	Description
str	Task ID string for monitoring and managing the harvest task

Raises:

Type	Description
HarvesterAPIError	If task creation fails
ValidationError	If parameters are invalid

Example

Create asset harvest task

task_id = api.create_harvest_task( ... task_type="asset", ... urls=["https://example.com/files/"], ... egress_info={ ... "name": "New York City", ... "categories": { ... "connectivity": "datacenter", ... "availability": "private", ... "protocol": "direct" ... } ... }, ... dest_path="/downloads/", ... dest_name="harvested_files.zip", ... task_params={ ... "recursive": True, ... "level": 2, ... "accept": ".pdf,.doc,*.docx" ... } ... )

Create visual PDF capture

task_id = api.create_harvest_task( ... task_type="visual", # ← NOT "pdf" ... urls=["https://example.com/report"], ... egress_info={"name": "Singapore"}, ... dest_name="report_capture.pdf", ... task_params={ ... "output_pdf": True, # ← controls output format ... "paper": "A4", ... "landscape": False ... } ... )

Create visual MHTML capture

task_id = api.create_harvest_task( ... task_type="visual", # ← NOT "mhtml" ... urls=["https://example.com/page"], ... egress_info={"name": "New York, NY"}, ... task_params={"output_mhtml": True} # ← MHTML format ... )

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.API_COMMAND, api_command="create_harvest_task")
def create_harvest_task(
    self,
    task_type: str,
    urls: list[str],
    egress_info: dict[str, Any],
    dest_path: str = "/",
    dest_name: str | None = None,
    dest_bucket_id: str | None = None,
    task_params: dict[str, Any] | None = None,
    run_after: int | str | None = None,
) -> str:
    """Create a new harvest task for web content collection.

    Creates a harvesting task that will collect content from the specified URL
    using the given parameters and store the results in the designated location.

    Args:
        task_type: Type of harvest task. Must be one of ``"asset"``,
            ``"visual"``, ``"video"``, or ``"single"``.

            - ``"visual"`` — Full-page capture. Output format controlled by ``task_params``:

              - Screenshot (default): no params, or ``{"output_image": True}``
              - PDF: ``{"output_pdf": True}``
              - MHTML: ``{"output_mhtml": True}``

              **IMPORTANT:** ``"screenshot"``, ``"pdf"``, and ``"mhtml"`` are **not**
              valid ``task_type`` values. Always use ``"visual"`` with ``task_params``.

            - ``"asset"`` — Recursive asset download (HTML, JS, CSS, images).
              Control depth/filters via ``task_params`` (wget-style options).
            - ``"video"`` — Video recording of page interaction (youtube-dl-style options).

        urls: List containing exactly one URL string to harvest.
        egress_info: Egress location dict with required ``"name"`` key
            (e.g., ``{"name": "New York, NY"}``). Optional ``"categories"``
            key accepts ``"connectivity"``, ``"availability"``, ``"protocol"``.
            For non-``"asset"`` task types, ``"connectivity"``/``"protocol"``
            are validated against what is actually available at ``"name"``
            (see :mod:`silo_sdk.harvesting.egress_validation`). Does not
            apply to ``"asset"`` tasks.
        dest_path: Destination path in storage (defaults to ``"/"``)
        dest_name: Name for the result file (auto-generated if not provided)
        dest_bucket_id: Destination bucket ID (uses default if not provided)
        task_params: Task-specific output parameters. Most keys are
            converted to a list of ``{"name": key}`` or
            ``{"name": key, "value": val}`` dicts sent under the
            appropriate wire key (``vis_params``, ``wget_params``,
            ``vid_params``). The following keys are instead injected
            directly at the top level of the wire ``task_params`` object:

            - ``return-cookies`` — send back cookies from the session
            - ``cookies`` — Netscape-format cookie file content (string)
            - ``all_files`` — include all fetched files alongside the main
              content. Valid for ``"visual"``, ``"single"``, and ``"video"``
              tasks. Raises ``ValidationError`` for ``"asset"`` tasks.


            Key visual params: ``output_pdf``, ``output_mhtml``,
            ``output_image``, ``scale``, ``landscape``, ``paper``.
            See class ``Attributes`` for full param sets per task type.
        run_after: Optional schedule time — Unix epoch integer
            (e.g., ``1707849000``) or ISO 8601 UTC string
            (e.g., ``"2026-02-13T18:30:00Z"``). Task enters ``"wait"``
            status until the specified time.

    Note:
        Wire format parameter mappings:

        - ``task_type`` → ``task_params.request_type``
        - ``urls`` → ``task_params.urls``
        - ``task_params`` (Python dict) → list of ``{"name": key}`` or
          ``{"name": key, "value": val}`` dicts sent as:

          - ``task_params.vis_params`` for ``"visual"`` tasks
          - ``task_params.wget_params`` for ``"asset"`` tasks
          - ``task_params.vid_params`` for ``"video"`` tasks

        - ``egress_info``, ``dest_bucket_id``, ``dest_path``, ``dest_name``,
          ``dest_auth_token`` remain at the top-level command dict
        - ``run_after`` is injected at **top-level command dict** alongside
          ``egress_info``, **NOT** inside ``task_params``

        Wire format structure for visual task::

            {
                "command": "create_harvest_task",
                "egress_info": {"name": "New York, NY"},
                "dest_bucket_id": "bucket_abc123",
                "dest_path": "/",
                "dest_name": "capture.pdf",
                "dest_auth_token": "<file_token>",
                "task_params": {
                    "request_type": "visual",
                    "urls": ["https://example.com"],
                    "vis_params": [
                        {"name": "output_pdf"},
                        {"name": "paper", "value": "A4"}
                    ]
                },
                "run_after": 1707849000
            }

    Returns:
        Task ID string for monitoring and managing the harvest task

    Raises:
        HarvesterAPIError: If task creation fails
        ValidationError: If parameters are invalid

    Example:
        >>> # Create asset harvest task
        >>> task_id = api.create_harvest_task(
        ...     task_type="asset",
        ...     urls=["https://example.com/files/"],
        ...     egress_info={
        ...         "name": "New York City",
        ...         "categories": {
        ...             "connectivity": "datacenter",
        ...             "availability": "private",
        ...             "protocol": "direct"
        ...         }
        ...     },
        ...     dest_path="/downloads/",
        ...     dest_name="harvested_files.zip",
        ...     task_params={
        ...         "recursive": True,
        ...         "level": 2,
        ...         "accept": "*.pdf,*.doc,*.docx"
        ...     }
        ... )
        >>>
        >>> # Create visual PDF capture
        >>> task_id = api.create_harvest_task(
        ...     task_type="visual",  # ← NOT "pdf"
        ...     urls=["https://example.com/report"],
        ...     egress_info={"name": "Singapore"},
        ...     dest_name="report_capture.pdf",
        ...     task_params={
        ...         "output_pdf": True,  # ← controls output format
        ...         "paper": "A4",
        ...         "landscape": False
        ...     }
        ... )
        >>>
        >>> # Create visual MHTML capture
        >>> task_id = api.create_harvest_task(
        ...     task_type="visual",  # ← NOT "mhtml"
        ...     urls=["https://example.com/page"],
        ...     egress_info={"name": "New York, NY"},
        ...     task_params={"output_mhtml": True}  # ← MHTML format
        ... )
    """
    task_params = self._normalize_task_params(task_params)
    top_level_keys = self._top_level_keys_for_type(task_type)

    # Validate all input parameters
    validate_task_creation_params(
        task_type,
        urls,
        egress_info,
        dest_path,
        task_params,
        self.VALID_TASK_TYPES,
        self._valid_params_map,
        top_level_keys=top_level_keys,
    )
    if run_after is not None:
        validate_run_after(run_after)

    # Prepare normalized data for API
    normalized_egress_info = normalize_egress_info(egress_info)
    bucket_id = dest_bucket_id or self.dest_bucket_id
    final_dest_name = generate_dest_name_if_needed(dest_name, task_type)

    # Build and execute the task creation request
    payload = build_task_creation_payload(
        task_type,
        urls,
        normalized_egress_info,
        bucket_id,
        dest_path,
        final_dest_name,
        task_params or {},
        self.file_api.auth_token,
        run_after=run_after,
        top_level_keys=top_level_keys,
    )

    return self._execute_task_creation(
        task_type, urls, normalized_egress_info, payload
    )

find_harvest_task

find_harvest_task(task_id: str, use_cache: bool = True, finished: bool | None = None) -> dict[str, Any]

Retrieve the status and details of a specific harvest task.

Gets comprehensive information about a harvest task including its current status, progress, results, and any error information. Supports caching to reduce API calls for frequently polled tasks.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to retrieve	required
use_cache	bool	Whether to use cached results (default: True)	True
finished	bool | None	If True, return only completed tasks. If False, return only in-progress tasks. If omitted, return regardless of completion state.	None

Returns:

Type	Description
dict[str, Any]	Dictionary containing task details including:
dict[str, Any]	task_id: Task identifier
dict[str, Any]	status: Current task status (wait, claimed, running, done, error)
dict[str, Any]	progress: Task progress percentage (if available)
dict[str, Any]	created_ts: Task creation timestamp
dict[str, Any]	started_ts: Task start timestamp (if started)
dict[str, Any]	completed_ts: Task completion timestamp (if completed)
dict[str, Any]	result_file_id: File ID of the result (if completed successfully)
dict[str, Any]	last_error: Error message (if status is error)
dict[str, Any]	task_params: Original task parameters

Raises:

Type	Description
HarvesterAPIError	If task retrieval fails
ValidationError	If task_id is invalid

Example

task_info = api.find_harvest_task("task_123") print(f"Status: {task_info['status']}") if task_info['status'] == 'done': ... print(f"Result file: {task_info['result_file_id']}") elif task_info['status'] == 'error': ... print(f"Error: {task_info['last_error']}")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.API_COMMAND, api_command="find_harvest_task")
def find_harvest_task(
    self,
    task_id: str,
    use_cache: bool = True,
    finished: bool | None = None,
) -> dict[str, Any]:
    """Retrieve the status and details of a specific harvest task.

    Gets comprehensive information about a harvest task including its current
    status, progress, results, and any error information. Supports caching
    to reduce API calls for frequently polled tasks.

    Args:
        task_id: The ID of the task to retrieve
        use_cache: Whether to use cached results (default: True)
        finished: If True, return only completed tasks. If False, return only
            in-progress tasks. If omitted, return regardless of completion state.

    Returns:
        Dictionary containing task details including:
        - task_id: Task identifier
        - status: Current task status (wait, claimed, running, done, error)
        - progress: Task progress percentage (if available)
        - created_ts: Task creation timestamp
        - started_ts: Task start timestamp (if started)
        - completed_ts: Task completion timestamp (if completed)
        - result_file_id: File ID of the result (if completed successfully)
        - last_error: Error message (if status is error)
        - task_params: Original task parameters

    Raises:
        HarvesterAPIError: If task retrieval fails
        ValidationError: If task_id is invalid

    Example:
        >>> task_info = api.find_harvest_task("task_123")
        >>> print(f"Status: {task_info['status']}")
        >>> if task_info['status'] == 'done':
        ...     print(f"Result file: {task_info['result_file_id']}")
        >>> elif task_info['status'] == 'error':
        ...     print(f"Error: {task_info['last_error']}")
    """
    self._require_task_id(task_id)

    # Check cache if enabled and requested
    if use_cache and self._status_cache_enabled and self._status_cache is not None:
        cached_result = get_cached_task_status(
            self._status_cache, task_id, self._status_cache_ttl
        )
        if cached_result is not None:
            self.logger.debug("Retrieved task status from cache: %s", task_id)
            return cached_result

    command: dict[str, Any] = {"command": "find_harvest_task", "task_id": task_id}
    if finished is not None:
        command["finished"] = finished
    payload = [command]

    try:
        self.logger.debug("Finding harvest task: %s", task_id)
        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        # Debug logging for response structure (only in debug mode)
        self.logger.debug("Processing response for task: %s", task_id)

        # Handle different response formats more robustly
        try:
            result = self._extract_api_result(response)
        except Exception as extract_error:
            # Fallback to direct response processing if extraction fails
            self.logger.debug(
                "Standard extraction failed, trying direct response processing: %s",
                extract_error,
            )
            if isinstance(response, list) and len(response) > 1:
                result = response[1]
            else:
                raise HarvesterAPIError(
                    f"Unexpected response format: {response}"
                ) from extract_error

        # Handle different result formats
        if isinstance(result, list) and len(result) > 0:
            task_details = result[0]
        elif isinstance(result, dict):
            task_details = result
        elif isinstance(result, str):
            # Handle string responses that might contain error information
            if "task produced no output" in result or "TypeError" in result:
                # Preserve the original server error message without additional wrapping
                raise HarvesterAPIError(
                    f"Task {task_id} failed with server error: {result}"
                )
            else:
                raise HarvesterAPIError(
                    f"Unexpected string response for task_id: {task_id} - {result}"
                )
        else:
            raise HarvesterAPIError(
                f"No task details found for task_id: {task_id} - "
                f"got result type: {type(result)}"
            )

        # Ensure task_details is a dictionary
        if not isinstance(task_details, dict):
            # Handle string task details that might contain error information
            if isinstance(task_details, str) and (
                "task produced no output" in task_details
                or "TypeError" in task_details
            ):
                raise HarvesterAPIError(
                    f"Task {task_id} failed with server error: {task_details}"
                )
            else:
                raise HarvesterAPIError(
                    f"Task details must be a dictionary, "
                    f"got {type(task_details)}: {task_details}"
                )

        self.logger.debug("Retrieved task details for: %s", task_id)

        # Cache the result if caching is enabled
        if self._status_cache_enabled:
            cache_task_status(
                self._status_cache, task_id, task_details, self._status_cache_ttl
            )

        return task_details

    except Exception as e:
        if isinstance(e, HarvesterAPIError):
            raise
        raise HarvesterAPIError(f"Failed to find harvest task: {e}") from e

delete_harvest_task

delete_harvest_task(task_id: str) -> bool

Delete a specific harvest task.

Permanently removes a harvest task and its associated data. This action cannot be undone. If the task is still running, it will be cancelled.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to delete	required

Returns:

Type	Description
bool	True if deletion was successful, False otherwise

Raises:

Type	Description
HarvesterAPIError	If task deletion fails
ValidationError	If task_id is invalid

Example

success = api.delete_harvest_task("task_123") if success: ... print("Task deleted successfully")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.API_COMMAND, api_command="delete_harvest_task")
def delete_harvest_task(self, task_id: str) -> bool:
    """Delete a specific harvest task.

    Permanently removes a harvest task and its associated data. This action
    cannot be undone. If the task is still running, it will be cancelled.

    Args:
        task_id: The ID of the task to delete

    Returns:
        True if deletion was successful, False otherwise

    Raises:
        HarvesterAPIError: If task deletion fails
        ValidationError: If task_id is invalid

    Example:
        >>> success = api.delete_harvest_task("task_123")
        >>> if success:
        ...     print("Task deleted successfully")
    """
    self._require_task_id(task_id)

    payload = [{"command": "delete_harvest_task", "task_id": task_id}]

    try:
        self.logger.debug("Deleting harvest task: %s", task_id)
        response = self._make_api_request("POST", "api/", self.auth_token, payload)

        result = self._extract_api_result(response)
        if isinstance(result, dict) and result.get("deleted"):
            self.logger.info("Successfully deleted harvest task: %s", task_id)
            return True

        self.logger.warning("Task deletion may have failed: %s", result)
        return False

    except Exception as e:
        if isinstance(e, HarvesterAPIError):
            raise
        raise HarvesterAPIError(f"Failed to delete harvest task: {e}") from e

wait_for_completion

wait_for_completion(task_id: str, max_wait_time: int = 1800, check_interval: int = 10, progress_callback: Callable[..., Any] | None = None) -> dict[str, Any]

Wait for a harvest task to complete, with optional progress monitoring.

Monitors a harvest task until it completes successfully, encounters an error, or the maximum wait time is exceeded. Provides optional progress callbacks for real-time monitoring.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to monitor	required
max_wait_time	int	Maximum time to wait in seconds (default: 30 minutes)	1800
check_interval	int	Time between status checks in seconds (default: 10)	10
progress_callback	Callable[..., Any] | None	Optional function to call with progress updates	None

Returns:

Type	Description
dict[str, Any]	Final task status dictionary when completed

Raises:

Type	Description
HarvesterAPIError	If task fails or times out
ValidationError	If parameters are invalid

Example

def progress_handler(task_info): ... print(f"Status: {task_info['status']}") ... if 'progress' in task_info: ... print(f"Progress: {task_info['progress']}%")

result = api.wait_for_completion( ... task_id="task_123", ... max_wait_time=600, # 10 minutes ... progress_callback=progress_handler ... )

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task"])
def wait_for_completion(
    self,
    task_id: str,
    max_wait_time: int = 1800,  # 30 minutes
    check_interval: int = 10,
    progress_callback: Callable[..., Any] | None = None,
) -> dict[str, Any]:
    """Wait for a harvest task to complete, with optional progress monitoring.

    Monitors a harvest task until it completes successfully, encounters an error,
    or the maximum wait time is exceeded. Provides optional progress callbacks
    for real-time monitoring.

    Args:
        task_id: The ID of the task to monitor
        max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
        check_interval: Time between status checks in seconds (default: 10)
        progress_callback: Optional function to call with progress updates

    Returns:
        Final task status dictionary when completed

    Raises:
        HarvesterAPIError: If task fails or times out
        ValidationError: If parameters are invalid

    Example:
        >>> def progress_handler(task_info):
        ...     print(f"Status: {task_info['status']}")
        ...     if 'progress' in task_info:
        ...         print(f"Progress: {task_info['progress']}%")
        >>>
        >>> result = api.wait_for_completion(
        ...     task_id="task_123",
        ...     max_wait_time=600,  # 10 minutes
        ...     progress_callback=progress_handler
        ... )
    """
    self._require_task_id(task_id)

    if not isinstance(max_wait_time, int) or max_wait_time <= 0:
        raise ValidationError("max_wait_time must be a positive integer")

    if not isinstance(check_interval, int) or check_interval <= 0:
        raise ValidationError("check_interval must be a positive integer")

    start_time = time.time()
    checks_performed = 0

    self.logger.info(
        "Waiting for task completion: %s (max %d seconds)", task_id, max_wait_time
    )

    try:
        while True:
            # Check if we've exceeded the maximum wait time
            elapsed_time = time.time() - start_time
            if elapsed_time > max_wait_time:
                raise HarvesterAPIError(
                    f"Task {task_id} did not complete within {max_wait_time} seconds"
                )

            # Get current task status
            try:
                task_status = self.find_harvest_task(task_id)
                if not isinstance(task_status, dict):
                    raise HarvesterAPIError(
                        f"Invalid task status format: {type(task_status)}"
                    )

                status = task_status.get("status")
                checks_performed += 1

                # Call progress callback if provided
                if progress_callback is not None:
                    try:
                        progress_callback(task_status)
                    except Exception as e:
                        self.logger.warning("Progress callback failed: %s", e)

                # Check terminal states
                if status == "done":
                    self.logger.info(
                        "Task %s completed successfully after %d checks (%.1f seconds)",
                        task_id,
                        checks_performed,
                        elapsed_time,
                    )
                    return task_status

                elif status == "error":
                    error_msg = task_status.get("last_error", "Unknown error")
                    raise HarvesterAPIError(
                        f"Task {task_id} failed with server error: {error_msg}"
                    )

            except HarvesterAPIError as e:
                # Re-check the original error message format
                error_str = str(e)
                if (
                    "task produced no output" in error_str
                    and "TypeError" in error_str
                ):
                    # This indicates a response parsing issue from find_harvest_task
                    self.logger.error(
                        "Task status retrieval failed due to response format issue: %s",
                        e,
                    )
                    # Don't re-wrap the error message - just re-raise the original
                    raise
                else:
                    raise

            if status in ["wait", "claimed", "running"]:
                # Task is still in progress
                self.logger.debug(
                    "Task %s status: %s (check %d, %.1f seconds elapsed)",
                    task_id,
                    status,
                    checks_performed,
                    elapsed_time,
                )
                time.sleep(check_interval)

            elif status not in ["done", "error"]:
                raise HarvesterAPIError(f"Unexpected task status: {status}")

    except KeyboardInterrupt as exc:
        self.logger.info("Task monitoring interrupted by user")
        raise HarvesterAPIError("Task monitoring was interrupted") from exc

wait_for_completion_with_progress

wait_for_completion_with_progress(task_id: str, task_type: str, max_wait_time: int = 1800, check_interval: int = 10, progress_callback: Callable[..., Any] | None = None) -> TaskProgress

Wait for a harvest task to complete with enhanced progress tracking.

Monitors a harvest task with detailed progress metrics including timing, status transitions, and estimated completion times.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to monitor	required
task_type	str	Type of harvest task	required
max_wait_time	int	Maximum time to wait in seconds (default: 30 minutes)	1800
check_interval	int	Time between status checks in seconds (default: 10)	10
progress_callback	Callable[..., Any] | None	Optional function to call with TaskProgress updates	None

Returns:

Type	Description
TaskProgress	TaskProgress object with detailed metrics when completed

Raises:

Type	Description
HarvesterAPIError	If task fails or times out
ValidationError	If parameters are invalid

Example

def progress_handler(progress: TaskProgress): ... print(f"Status: {progress.status} ({progress.progress_percentage}%)") ... if progress.estimated_completion: ... eta = datetime.fromtimestamp(progress.estimated_completion) ... print(f"ETA: {eta}")

progress = api.wait_for_completion_with_progress( ... task_id="task_123", ... task_type="asset", ... progress_callback=progress_handler ... ) print(f"Completed in {progress.get_duration():.1f} seconds")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task"])
def wait_for_completion_with_progress(
    self,
    task_id: str,
    task_type: str,
    max_wait_time: int = 1800,  # 30 minutes
    check_interval: int = 10,
    progress_callback: Callable[..., Any] | None = None,
) -> TaskProgress:
    """Wait for a harvest task to complete with enhanced progress tracking.

    Monitors a harvest task with detailed progress metrics including timing,
    status transitions, and estimated completion times.

    Args:
        task_id: The ID of the task to monitor
        task_type: Type of harvest task
        max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
        check_interval: Time between status checks in seconds (default: 10)
        progress_callback: Optional function to call with TaskProgress updates

    Returns:
        TaskProgress object with detailed metrics when completed

    Raises:
        HarvesterAPIError: If task fails or times out
        ValidationError: If parameters are invalid

    Example:
        >>> def progress_handler(progress: TaskProgress):
        ...     print(f"Status: {progress.status} ({progress.progress_percentage}%)")
        ...     if progress.estimated_completion:
        ...         eta = datetime.fromtimestamp(progress.estimated_completion)
        ...         print(f"ETA: {eta}")
        >>>
        >>> progress = api.wait_for_completion_with_progress(
        ...     task_id="task_123",
        ...     task_type="asset",
        ...     progress_callback=progress_handler
        ... )
        >>> print(f"Completed in {progress.get_duration():.1f} seconds")
    """
    self._require_task_id(task_id)

    if not isinstance(max_wait_time, int) or max_wait_time <= 0:
        raise ValidationError("max_wait_time must be a positive integer")

    if not isinstance(check_interval, int) or check_interval <= 0:
        raise ValidationError("check_interval must be a positive integer")

    # Initialize progress tracker
    progress = TaskProgress(task_id, task_type)

    start_time = time.time()
    checks_performed = 0

    self.logger.info(
        "Waiting for task completion with progress tracking: %s (max %d seconds)",
        task_id,
        max_wait_time,
    )

    try:
        while True:
            # Check if we've exceeded the maximum wait time
            elapsed_time = time.time() - start_time
            if elapsed_time > max_wait_time:
                progress.set_error(f"Task timeout after {max_wait_time} seconds")
                raise HarvesterAPIError(
                    f"Task {task_id} did not complete within {max_wait_time} seconds"
                )

            # Get current task status
            task_status = self.find_harvest_task(task_id)
            status: str = task_status.get("status", "unknown")
            checks_performed += 1

            # Update progress tracker
            progress.update_status(status, task_status.get("progress"))
            progress.retry_count = checks_performed - 1

            # Call progress callback if provided
            if progress_callback is not None:
                try:
                    progress_callback(progress)
                except Exception as e:
                    self.logger.warning("Progress callback failed: %s", e)

            # Check terminal states
            if status == "done":
                self.logger.info(
                    "Task %s completed successfully after %d checks (%.1f seconds)",
                    task_id,
                    checks_performed,
                    elapsed_time,
                )
                return progress

            elif status == "error":
                error_msg = task_status.get("last_error", "Unknown error")
                progress.set_error(error_msg)
                raise HarvesterAPIError(
                    f"Task {task_id} failed with server error: {error_msg}"
                )

            elif status in ["wait", "claimed", "running"]:
                # Task is still in progress
                self.logger.debug(
                    "Task %s status: %s (check %d, %.1f seconds elapsed)",
                    task_id,
                    status,
                    checks_performed,
                    elapsed_time,
                )
                time.sleep(check_interval)

            else:
                progress.set_error(f"Unexpected task status: {status}")
                raise HarvesterAPIError(f"Unexpected task status: {status}")

    except KeyboardInterrupt as exc:
        progress.set_error("Task monitoring interrupted by user")
        self.logger.info("Task monitoring interrupted by user")
        raise HarvesterAPIError("Task monitoring was interrupted") from exc

download_task_result

download_task_result(task_id: str, output_path: str, verify_completion: bool = True) -> str

Download the result of a completed harvest task.

Downloads the harvested content from a completed task to the local filesystem. Optionally verifies that the task is completed before attempting download.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the completed task	required
output_path	str	Local path where to save the downloaded file	required
verify_completion	bool	Whether to verify task completion before download	True

Returns:

Type	Description
str	Path of the downloaded file

Raises:

Type	Description
HarvesterAPIError	If download fails or task is not completed
ValidationError	If parameters are invalid

Example

downloaded_file = api.download_task_result( ... task_id="task_123", ... output_path="/local/downloads/harvest_result.zip" ... ) print(f"Downloaded to: {downloaded_file}")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.CONVENIENCE, wraps=["find_harvest_task", "getfile"])
def download_task_result(
    self,
    task_id: str,
    output_path: str,
    verify_completion: bool = True,
) -> str:
    """Download the result of a completed harvest task.

    Downloads the harvested content from a completed task to the local filesystem.
    Optionally verifies that the task is completed before attempting download.

    Args:
        task_id: The ID of the completed task
        output_path: Local path where to save the downloaded file
        verify_completion: Whether to verify task completion before download

    Returns:
        Path of the downloaded file

    Raises:
        HarvesterAPIError: If download fails or task is not completed
        ValidationError: If parameters are invalid

    Example:
        >>> downloaded_file = api.download_task_result(
        ...     task_id="task_123",
        ...     output_path="/local/downloads/harvest_result.zip"
        ... )
        >>> print(f"Downloaded to: {downloaded_file}")
    """
    self._require_task_id(task_id)

    if not output_path:
        raise ValidationError("output_path must be provided")

    # Verify task completion if requested
    if verify_completion:
        task_status = self.find_harvest_task(task_id)
        if task_status.get("status") != "done":
            raise HarvesterAPIError(
                f"Task {task_id} is not completed. "
                f"Current status: {task_status.get('status')}"
            )

        result_file_id = task_status.get("result_file_id")
        if not result_file_id:
            raise HarvesterAPIError(f"No result file found for task {task_id}")
    else:
        # Get task details to find result file ID
        task_status = self.find_harvest_task(task_id)
        result_file_id = task_status.get("result_file_id")
        if not result_file_id:
            raise HarvesterAPIError(f"No result file found for task {task_id}")

    try:
        self.logger.debug("Downloading result for task: %s", task_id)
        downloaded_path = self.file_api.download_file(result_file_id, output_path)

        self.logger.info(
            "Successfully downloaded task result: %s -> %s",
            task_id,
            downloaded_path,
        )
        return downloaded_path

    except FileAPIError as e:
        raise HarvesterAPIError(f"Failed to download task result: {e}") from e

get_valid_params classmethod

get_valid_params(task_type: str) -> set[str]

Get valid parameters for a specific task type.

Parameters:

Name	Type	Description	Default
task_type	str	Type of harvest task ("asset", "visual", or "video")	required

Returns:

Type	Description
set[str]	Set of valid parameter names for the task type

Raises:

Type	Description
ValidationError	If task_type is invalid

Example

asset_params = HarvesterAPI.get_valid_params("asset") print("Valid asset parameters:") for param in sorted(asset_params): ... print(f" - {param}")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.UTILITY)
@classmethod
def get_valid_params(cls, task_type: str) -> set[str]:
    """Get valid parameters for a specific task type.

    Args:
        task_type: Type of harvest task ("asset", "visual", or "video")

    Returns:
        Set of valid parameter names for the task type

    Raises:
        ValidationError: If task_type is invalid

    Example:
        >>> asset_params = HarvesterAPI.get_valid_params("asset")
        >>> print("Valid asset parameters:")
        >>> for param in sorted(asset_params):
        ...     print(f"  - {param}")
    """
    if task_type in {"asset", "single"}:
        return cls.VALID_ASSET_PARAMS.copy()
    elif task_type == "visual":
        return cls.VALID_VISUAL_PARAMS.copy()
    elif task_type == "video":
        return cls.VALID_VIDEO_PARAMS.copy()
    else:
        raise ValidationError(
            f"Invalid task_type: {task_type}. "
            f"Valid types are: {', '.join(sorted(cls.VALID_TASK_TYPES))}"
        )

get_valid_task_types classmethod

get_valid_task_types() -> set[str]

Get all valid task types.

Returns:

Type	Description
set[str]	Set of valid task type names

Example

task_types = HarvesterAPI.get_valid_task_types() print("Supported task types:") for task_type in sorted(task_types): ... print(f" - {task_type}")

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(MethodType.UTILITY)
@classmethod
def get_valid_task_types(cls) -> set[str]:
    """Get all valid task types.

    Returns:
        Set of valid task type names

    Example:
        >>> task_types = HarvesterAPI.get_valid_task_types()
        >>> print("Supported task types:")
        >>> for task_type in sorted(task_types):
        ...     print(f"  - {task_type}")
    """
    return cls.VALID_TASK_TYPES.copy()

create_harvest_task_async async

create_harvest_task_async(task_type: str, urls: list[str], egress_info: dict[str, Any], dest_path: str = '/', dest_name: str | None = None, dest_bucket_id: str | None = None, task_params: dict[str, Any] | None = None, max_retries: int = 5, backoff_factor: int = 2, run_after: int | str | None = None) -> str

Asynchronously create a new harvest task for web content collection.

This async version allows for non-blocking task creation and can be used in concurrent workflows for creating multiple tasks simultaneously.

Parameters:

Name	Type	Description	Default
task_type	str	Type of harvest task ("asset", "single", "visual", or "video")	required
urls	list[str]	Array containing exactly one URL string to harvest	required
egress_info	dict[str, Any]	Egress location information containing: - name: Egress location name (required, e.g., "New York, NY") - categories: Optional object with connectivity, availability, protocol Each category field can be a string value or null. For non-"asset" task types, connectivity/protocol are validated against what is actually available at name.	required
dest_path	str	Destination path in storage (defaults to root "/")	'/'
dest_name	str | None	Name for the result file (auto-generated if not provided)	None
dest_bucket_id	str | None	Destination bucket ID (uses default if not provided)	None
task_params	dict[str, Any] | None	Dictionary of task-specific parameters	None
max_retries	int	Maximum number of retries for explicit HTTP 429 rate-limit responses (default: 5). Ambiguous transport failures are not retried because task creation is non-idempotent.	5
backoff_factor	int	Positive integer exponential backoff multiplier for rate-limit retries (default: 2)	2
run_after	int | str | None	Optional schedule time — Unix epoch integer (e.g., 1707849000) or ISO 8601 UTC string (e.g., "2026-02-13T18:30:00Z"). When provided the task enters "wait" status until the specified time.	None

Returns:

Type	Description
str	Task ID string for monitoring and managing the harvest task

Raises:

Type	Description
HarvesterAPIError	If task creation fails
ValidationError	If parameters are invalid

Example

import asyncio async def create_tasks(): ... harvester = HarvesterAPI(config) ... task_id = await harvester.create_harvest_task_async( ... task_type="asset", ... urls=["https://example.com/files/"], ... egress_info={"name": "New York, NY"}, ... dest_path="/downloads/" ... ) ... return task_id task_id = asyncio.run(create_tasks())

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(
    MethodType.ASYNC_VARIANT,
    api_command="create_harvest_task",
    sync_of="create_harvest_task",
)
async def create_harvest_task_async(
    self,
    task_type: str,
    urls: list[str],
    egress_info: dict[str, Any],
    dest_path: str = "/",
    dest_name: str | None = None,
    dest_bucket_id: str | None = None,
    task_params: dict[str, Any] | None = None,
    max_retries: int = 5,
    backoff_factor: int = 2,
    run_after: int | str | None = None,
) -> str:
    """Asynchronously create a new harvest task for web content collection.

    This async version allows for non-blocking task creation and can be used
    in concurrent workflows for creating multiple tasks simultaneously.

    Args:
        task_type: Type of harvest task ("asset", "single", "visual", or "video")
        urls: Array containing exactly one URL string to harvest
        egress_info: Egress location information containing:
            - name: Egress location name (required, e.g., "New York, NY")
            - categories: Optional object with connectivity, availability, protocol
              Each category field can be a string value or null. For
              non-"asset" task types, connectivity/protocol are validated
              against what is actually available at name.
        dest_path: Destination path in storage (defaults to root "/")
        dest_name: Name for the result file (auto-generated if not provided)
        dest_bucket_id: Destination bucket ID (uses default if not provided)
        task_params: Dictionary of task-specific parameters
        max_retries: Maximum number of retries for explicit HTTP 429
            rate-limit responses (default: 5). Ambiguous transport failures
            are not retried because task creation is non-idempotent.
        backoff_factor: Positive integer exponential backoff multiplier for
            rate-limit retries (default: 2)
        run_after: Optional schedule time — Unix epoch integer
            (e.g., 1707849000) or ISO 8601 UTC string
            (e.g., "2026-02-13T18:30:00Z"). When provided the task
            enters "wait" status until the specified time.

    Returns:
        Task ID string for monitoring and managing the harvest task

    Raises:
        HarvesterAPIError: If task creation fails
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def create_tasks():
        ...     harvester = HarvesterAPI(config)
        ...     task_id = await harvester.create_harvest_task_async(
        ...         task_type="asset",
        ...         urls=["https://example.com/files/"],
        ...         egress_info={"name": "New York, NY"},
        ...         dest_path="/downloads/"
        ...     )
        ...     return task_id
        >>> task_id = asyncio.run(create_tasks())
    """
    validate_positive_int(max_retries, "max_retries")
    validate_positive_int(backoff_factor, "backoff_factor")

    task_params = self._normalize_task_params(task_params)
    top_level_keys = self._top_level_keys_for_type(task_type)
    validate_task_creation_params(
        task_type,
        urls,
        egress_info,
        dest_path,
        task_params,
        self.VALID_TASK_TYPES,
        self._valid_params_map,
        top_level_keys=top_level_keys,
    )

    if run_after is not None:
        validate_run_after(run_after)

    normalized_egress_info = normalize_egress_info(egress_info)
    bucket_id = dest_bucket_id or self.dest_bucket_id
    final_dest_name = generate_dest_name_if_needed(dest_name, task_type)

    payload_commands = build_task_creation_payload(
        task_type,
        urls,
        normalized_egress_info,
        bucket_id,
        dest_path,
        final_dest_name,
        task_params or {},
        self.file_api.auth_token,
        run_after=run_after,
        top_level_keys=top_level_keys,
    )

    # Build API payload
    api_url = f"{self.base_url.rstrip('/')}/api/"
    payload = [{"command": "setauth", "data": self.auth_token}] + payload_commands

    async with httpx.AsyncClient(verify=self._ssl_context) as client:
        for attempt in range(max_retries):
            try:
                response = await client.post(
                    api_url,
                    json=payload,
                    headers={
                        "Accept": "application/json",
                        "Content-Type": "application/json",
                    },
                    timeout=httpx.Timeout(30),
                )
                if response.status_code == 200:
                    task_response = response.json()
                    if (
                        isinstance(task_response, list)
                        and len(task_response) > 1
                        and isinstance(task_response[1], dict)
                    ):
                        result_inner = task_response[1].get("result")
                        result_task_id: str | None = (
                            result_inner.get("task_id")
                            if isinstance(result_inner, dict)
                            else None
                        )
                        if result_task_id:
                            self.logger.info(
                                "Created async harvest task: %s", result_task_id
                            )
                            return result_task_id

                    raise HarvesterAPIError(
                        f"Unexpected response format: {task_response}"
                    )

                elif response.status_code == 429:
                    if attempt == max_retries - 1:
                        break
                    try:
                        retry_after = int(
                            response.headers.get(
                                "Retry-After", backoff_factor**attempt
                            )
                        )
                    except ValueError:
                        retry_after = int(backoff_factor**attempt)
                    self.logger.warning(
                        "Rate limit hit, retrying after %s seconds...", retry_after
                    )
                    await asyncio.sleep(retry_after)
                else:
                    raise HarvesterAPIError(
                        f"API request failed with status code {response.status_code}"
                    )

            except httpx.HTTPError as e:
                raise HarvesterAPIError(
                    "Harvest task creation failed after the request may have "
                    "reached the server; the task may have been created. "
                    "The request was not retried to avoid creating a duplicate: "
                    f"{e}"
                ) from e
            except HarvesterAPIError:
                raise
            except Exception as e:
                raise HarvesterAPIError(
                    f"Failed to create async harvest task: {e}"
                ) from e

    raise HarvesterAPIError("Rate-limit retries exhausted; task was not created")

wait_for_completion_async async

wait_for_completion_async(task_id: str, max_wait_time: int = 1800, check_interval: int = 10, progress_callback: Callable[..., Any] | None = None) -> dict[str, Any]

Asynchronously wait for a harvest task to complete.

This async version allows for non-blocking task monitoring and can be used to monitor multiple tasks concurrently.

Parameters:

Name	Type	Description	Default
task_id	str	The ID of the task to monitor	required
max_wait_time	int	Maximum time to wait in seconds (default: 30 minutes)	1800
check_interval	int	Time between status checks in seconds (default: 10)	10
progress_callback	Callable[..., Any] | None	Optional function to call with progress updates	None

Returns:

Name	Type	Description
	dict[str, Any]	Final task status dictionary when completed. Shape matches the
synchronous	dict[str, Any]	meth:wait_for_completion: it is the raw task
	dict[str, Any]	dictionary as returned by the server (status, progress,
	dict[str, Any]	task_params, worker_id, and any other server-reported
	dict[str, Any]	fields), merged with the normalized convenience fields produced
by	dict[str, Any]	func:~silo_sdk.utils.harvester_utils.parse_task_response
	dict[str, Any]	(task_metadata, and normalized task_id,
	dict[str, Any]	result_file_id, last_error, run_after, created_ts,
	dict[str, Any]	started_ts, completed_ts). No server-reported fields are
	dict[str, Any]	dropped.

Raises:

Type	Description
HarvesterAPIError	If task fails or times out
ValidationError	If parameters are invalid

Example

import asyncio async def monitor_task(): ... harvester = HarvesterAPI(config) ... def progress_handler(task_info): ... print(f"Status: {task_info['status']}") ... ... result = await harvester.wait_for_completion_async( ... task_id="task_123", ... progress_callback=progress_handler ... ) ... return result result = asyncio.run(monitor_task())

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(
    MethodType.ASYNC_VARIANT,
    api_command="find_harvest_task",
    sync_of="wait_for_completion",
)
async def wait_for_completion_async(
    self,
    task_id: str,
    max_wait_time: int = 1800,  # 30 minutes
    check_interval: int = 10,
    progress_callback: Callable[..., Any] | None = None,
) -> dict[str, Any]:
    """Asynchronously wait for a harvest task to complete.

    This async version allows for non-blocking task monitoring and can be used
    to monitor multiple tasks concurrently.

    Args:
        task_id: The ID of the task to monitor
        max_wait_time: Maximum time to wait in seconds (default: 30 minutes)
        check_interval: Time between status checks in seconds (default: 10)
        progress_callback: Optional function to call with progress updates

    Returns:
        Final task status dictionary when completed. Shape matches the
        synchronous :meth:`wait_for_completion`: it is the raw task
        dictionary as returned by the server (``status``, ``progress``,
        ``task_params``, ``worker_id``, and any other server-reported
        fields), merged with the normalized convenience fields produced
        by :func:`~silo_sdk.utils.harvester_utils.parse_task_response`
        (``task_metadata``, and normalized ``task_id``,
        ``result_file_id``, ``last_error``, ``run_after``, ``created_ts``,
        ``started_ts``, ``completed_ts``). No server-reported fields are
        dropped.

    Raises:
        HarvesterAPIError: If task fails or times out
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def monitor_task():
        ...     harvester = HarvesterAPI(config)
        ...     def progress_handler(task_info):
        ...         print(f"Status: {task_info['status']}")
        ...
        ...     result = await harvester.wait_for_completion_async(
        ...         task_id="task_123",
        ...         progress_callback=progress_handler
        ...     )
        ...     return result
        >>> result = asyncio.run(monitor_task())
    """
    self._require_task_id(task_id)

    if not isinstance(max_wait_time, int) or max_wait_time <= 0:
        raise ValidationError("max_wait_time must be a positive integer")

    if not isinstance(check_interval, int) or check_interval <= 0:
        raise ValidationError("check_interval must be a positive integer")

    max_retries = max_wait_time // check_interval
    api_url = f"{self.base_url.rstrip('/')}/api/"

    self.logger.info("Starting async status check loop for task ID: %s", task_id)
    retries = 0
    last_response = None
    start_time = time.time()

    try:
        async with httpx.AsyncClient(verify=self._ssl_context) as client:
            while retries < max_retries:
                # Bound the loop by wall-clock elapsed time in addition to
                # retry count. Retry count alone assumes each iteration
                # costs ~check_interval seconds, but a slow API response
                # (up to the 30s per-request timeout below) isn't counted
                # against max_wait_time, which lets the real wait exceed
                # it. This check only ever exits *earlier* than the retry
                # count would (never later), so it doesn't change
                # behavior when checks are fast.
                if time.time() - start_time > max_wait_time:
                    self.logger.warning(
                        "Wall-clock max_wait_time (%s seconds) exceeded for "
                        "task %s before the retry budget was exhausted.",
                        max_wait_time,
                        task_id,
                    )
                    break

                self.logger.debug(
                    f"Checking status of task {task_id} "
                    f"(attempt {retries + 1}/{max_retries})"
                )

                payload = [
                    {"command": "setauth", "data": self.auth_token},
                    {"command": "find_harvest_task", "task_id": task_id},
                ]

                try:
                    response = await client.post(
                        api_url,
                        json=payload,
                        headers={
                            "Accept": "application/json",
                            "Content-Type": "application/json",
                        },
                        timeout=httpx.Timeout(30),
                    )
                    response.raise_for_status()
                    last_response = response.json()

                    if (
                        last_response
                        and isinstance(last_response, list)
                        and len(last_response) > 1
                        and isinstance(last_response[1], dict)
                    ):
                        task_result = last_response[1].get("result", [])
                    else:
                        self.logger.warning(
                            f"Invalid response format for task {task_id}: "
                            f"{last_response}"
                        )
                        raise HarvesterAPIError(
                            f"Invalid response format for task {task_id}"
                        )

                    if task_result:
                        task_status = task_result[0].get("status", "")
                        self.logger.info(f"Task {task_id} status: {task_status}")

                        # Call progress callback if provided
                        if progress_callback is not None:
                            try:
                                progress_callback(task_result[0])
                            except Exception as e:
                                self.logger.warning(
                                    "Progress callback failed: %s", e
                                )

                        if task_status == "done":
                            self.logger.info(
                                f"Task {task_id} completed successfully."
                            )
                            task_info = parse_task_response(last_response)
                            if not task_info:
                                raise HarvesterAPIError(
                                    f"Failed to parse task response for {task_id}"
                                )
                            # Align this method's return shape with the
                            # synchronous wait_for_completion(), which
                            # returns the full raw server task dict (see
                            # find_harvest_task/task_result[0] above).
                            # parse_task_response() only surfaces 9
                            # hardcoded keys, silently dropping fields
                            # like progress/task_params/worker_id — merge
                            # the raw dict as the base and layer the
                            # normalized fields on top so nothing the
                            # server sent is lost.
                            full_task_info: dict[str, Any] = {
                                **task_result[0],
                                **task_info,
                            }
                            return full_task_info
                        elif task_status == "error":
                            error_msg = task_result[0].get(
                                "last_error", "Unknown error"
                            )
                            raise HarvesterAPIError(
                                f"Task {task_id} failed with server error: {error_msg}"
                            )
                        elif task_status == "wait":
                            run_after = task_result[0].get("run_after")
                            if run_after:
                                run_after_time = datetime.fromtimestamp(
                                    run_after, tz=UTC
                                )
                                self.logger.info(
                                    f"Task {task_id} is in 'wait' status, "
                                    f"scheduled to run at: {run_after_time} UTC."
                                )
                            else:
                                self.logger.info(
                                    f"Task {task_id} is in 'wait' status, "
                                    f"but no scheduled time is available."
                                )
                        elif task_status in ["claimed", "running"]:
                            self.logger.info(
                                f"Task {task_id} is in progress... "
                                f"Status: {task_status}"
                            )
                        else:
                            self.logger.warning(
                                f"Unexpected task status for task {task_id}: "
                                f"{task_status}"
                            )
                    else:
                        self.logger.warning(
                            f"No task result found for task {task_id}"
                        )

                except httpx.HTTPError as e:
                    self.logger.error(
                        f"Error: Request failed for task {task_id} with exception {e}."
                    )
                    raise HarvesterAPIError(
                        f"Request failed for task {task_id}: {e}"
                    ) from e

                await asyncio.sleep(check_interval)
                retries += 1

        # Either the retry budget or the wall-clock max_wait_time was
        # exhausted without the task reaching a terminal state.
        self.logger.error(
            f"Timed out waiting for task {task_id}. "
            f"Task may not have completed successfully."
        )
        raise HarvesterAPIError(
            f"Task {task_id} did not complete within {max_wait_time} seconds"
        )

    except HarvesterAPIError:
        raise
    except Exception as e:
        raise HarvesterAPIError(f"Failed to monitor async task: {e}") from e

process_harvest_workflow_async async

process_harvest_workflow_async(task_type: str, urls: list[str], egress_info: dict[str, Any], output_dir: str, dest_path: str = '/', dest_name: str | None = None, task_params: dict[str, Any] | None = None, use_chunked_download: bool = False, max_wait_time: int = 1800, check_interval: int = 20) -> dict[str, Any] | None

Process a complete harvest workflow asynchronously.

This high-level async method orchestrates the entire harvest workflow: 1. Creates a harvest task 2. Monitors its status until completion 3. Downloads the result file 4. Returns comprehensive task information

Parameters:

Name	Type	Description	Default
task_type	str	Type of harvest task ("asset", "visual", or "video")	required
urls	list[str]	Array containing exactly one URL string to harvest	required
egress_info	dict[str, Any]	Egress location information containing: - name: Egress location name (required, e.g., "New York, NY") - categories: Optional object with connectivity, availability, protocol Each category field can be a string value or null. For non-"asset" task types, connectivity/protocol are validated against what is actually available at name.	required
output_dir	str	Directory to save downloaded results	required
dest_path	str	Destination path in storage (default: "/")	'/'
dest_name	str | None	Name for result file (auto-generated if not provided)	None
task_params	dict[str, Any] | None	Dictionary of task-specific parameters	None
use_chunked_download	bool	Whether to use chunked download (default: False)	False
max_wait_time	int	Maximum time to wait for completion (default: 30 minutes)	1800
check_interval	int	Time between status checks (default: 20 seconds)	20

Returns:

Type	Description
dict[str, Any] | None	Dictionary with task information and results, None if failed

Raises:

Type	Description
HarvesterAPIError	If workflow fails
ValidationError	If parameters are invalid

Example

import asyncio async def harvest_site(): ... harvester = HarvesterAPI(config) ... result = await harvester.process_harvest_workflow_async( ... task_type="asset", ... urls=["https://example.com/documents/"], ... egress_info={"name": "New York, NY"}, ... output_dir="./downloads", ... task_params={"recursive": True, "level": 2} ... ) ... return result result = asyncio.run(harvest_site())

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(
    MethodType.ASYNC_VARIANT,
    api_command="create_harvest_task",
    sync_of="create_harvest_task",
)
async def process_harvest_workflow_async(
    self,
    task_type: str,
    urls: list[str],
    egress_info: dict[str, Any],
    output_dir: str,
    dest_path: str = "/",
    dest_name: str | None = None,
    task_params: dict[str, Any] | None = None,
    use_chunked_download: bool = False,
    max_wait_time: int = 1800,
    check_interval: int = 20,
) -> dict[str, Any] | None:
    """Process a complete harvest workflow asynchronously.

    This high-level async method orchestrates the entire harvest workflow:
    1. Creates a harvest task
    2. Monitors its status until completion
    3. Downloads the result file
    4. Returns comprehensive task information

    Args:
        task_type: Type of harvest task ("asset", "visual", or "video")
        urls: Array containing exactly one URL string to harvest
        egress_info: Egress location information containing:
            - name: Egress location name (required, e.g., "New York, NY")
            - categories: Optional object with connectivity, availability, protocol
              Each category field can be a string value or null. For
              non-"asset" task types, connectivity/protocol are validated
              against what is actually available at name.
        output_dir: Directory to save downloaded results
        dest_path: Destination path in storage (default: "/")
        dest_name: Name for result file (auto-generated if not provided)
        task_params: Dictionary of task-specific parameters
        use_chunked_download: Whether to use chunked download (default: False)
        max_wait_time: Maximum time to wait for completion (default: 30 minutes)
        check_interval: Time between status checks (default: 20 seconds)

    Returns:
        Dictionary with task information and results, None if failed

    Raises:
        HarvesterAPIError: If workflow fails
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def harvest_site():
        ...     harvester = HarvesterAPI(config)
        ...     result = await harvester.process_harvest_workflow_async(
        ...         task_type="asset",
        ...         urls=["https://example.com/documents/"],
        ...         egress_info={"name": "New York, NY"},
        ...         output_dir="./downloads",
        ...         task_params={"recursive": True, "level": 2}
        ...     )
        ...     return result
        >>> result = asyncio.run(harvest_site())
    """
    # Validate egress_info with task-type-aware validation
    validate_egress_for_task_type(task_type, egress_info)

    # Create output directory
    output_path = Path(output_dir)
    output_path.mkdir(parents=True, exist_ok=True)

    self.logger.info(
        "Processing %s task for URL: %s", task_type, urls[0] if urls else "<empty>"
    )

    try:
        # Step 1: Create harvest task
        task_id = await self.create_harvest_task_async(
            task_type=task_type,
            urls=urls,
            egress_info=egress_info,
            dest_path=dest_path,
            dest_name=dest_name,
            task_params=task_params,
        )

        self.logger.info("Started %s task with ID: %s", task_type, task_id)

        # Step 2: Monitor task status
        task_info = await self.wait_for_completion_async(
            task_id=task_id,
            max_wait_time=max_wait_time,
            check_interval=check_interval,
        )

        if not task_info:
            self.logger.error("No status response received for task %s", task_id)
            return None

        result_file_id = task_info.get("result_file_id")
        if not result_file_id:
            self.logger.error("Missing result file ID for task %s", task_id)
            return task_info

        # Step 3: Download result file
        output_file_path = output_path / f"harvest-{task_id}.zip"

        try:
            self.logger.info("Downloading result for task %s", task_id)

            if use_chunked_download:
                downloaded_file = await self.file_api.download_file_chunked_async(
                    file_id=result_file_id, output_path=str(output_file_path)
                )
            else:
                downloaded_file = (
                    await self.file_api.download_file_with_retry_async(
                        file_id=result_file_id, output_path=str(output_file_path)
                    )
                )

            if downloaded_file:
                self.logger.info("Downloaded file saved to: %s", downloaded_file)
                task_info["downloaded_file"] = str(downloaded_file)
                task_info["download_success"] = True
            else:
                self.logger.error(
                    "Failed to download result file for task %s", task_id
                )
                task_info["download_success"] = False

        except Exception as e:
            self.logger.error("Error downloading file for task %s: %s", task_id, e)
            task_info["download_success"] = False
            task_info["download_error"] = str(e)

        return task_info

    except Exception as e:
        if isinstance(e, (HarvesterAPIError, ValidationError)):
            raise
        raise HarvesterAPIError(
            f"Failed to process async harvest workflow: {e}"
        ) from e

bulk_create_and_monitor_async async

bulk_create_and_monitor_async(tasks: list[dict[str, Any]], output_dir: str, max_concurrent: int = 5, egress_info: dict[str, Any] | None = None) -> list[dict[str, Any] | None]

Create and monitor multiple harvest tasks concurrently. This method processes multiple harvest tasks in parallel with configurable concurrency limits to efficiently handle bulk harvesting operations.

Parameters:

Name	Type	Description	Default
tasks	list[dict[str, Any]]	List of task dictionaries, each containing: - task_type: Type of harvest task - urls: List containing exactly one URL string to harvest (canonical key — matches :meth:create_harvest_task). A single url string key is also accepted as a convenience alias and is wrapped into a one-item list. - egress_info: Optional per-task egress location dict. When provided, it overrides the bulk-level egress_info for this task only — this is how you mix asset tasks (which need one of the 6 datacenter locations) with non-asset tasks in the same call. - params: Task-specific parameters (dict format) - dest_path: Optional destination path (default: "/") - dest_name: Optional destination name	required
output_dir	str	Directory to save downloaded results	required
max_concurrent	int	Positive maximum number of concurrent tasks (default: 5)	5
egress_info	dict[str, Any] | None	Default egress location information used for any task that doesn't specify its own egress_info (default: {"name": "New York, NY"}). Note: asset tasks require locations from the 6-location datacenter list (e.g., "New York City", not "New York, NY"). If mixing asset and non-asset tasks, set egress_info per-task (see above) or use separate calls.	None

Returns:

Type	Description
list[dict[str, Any] | None]	List of task results (same order as input), None for failed tasks

Raises:

Type	Description
HarvesterAPIError	If bulk processing fails
ValidationError	If parameters are invalid

Example

import asyncio async def bulk_harvest(): ... harvester = HarvesterAPI(config) ... tasks = [ ... { ... "task_type": "asset", ... "urls": ["https://site1.com"], ... "egress_info": {"name": "New York City"}, ... "params": {"recursive": True} ... }, ... { ... "task_type": "visual", ... "urls": ["https://site2.com"], ... "params": {"output_pdf": True} ... } ... ] ... results = await harvester.bulk_create_and_monitor_async( ... tasks=tasks, ... output_dir="./downloads", ... max_concurrent=3 ... ) ... return results results = asyncio.run(bulk_harvest())

Source code in silo_sdk/harvesting/harvester_api.py

@api_tag(
    MethodType.ASYNC_VARIANT,
    api_command="create_harvest_task",
    sync_of="create_harvest_task",
)
async def bulk_create_and_monitor_async(
    self,
    tasks: list[dict[str, Any]],
    output_dir: str,
    max_concurrent: int = 5,
    egress_info: dict[str, Any] | None = None,
) -> list[dict[str, Any] | None]:
    """Create and monitor multiple harvest tasks concurrently.
    This method processes multiple harvest tasks in parallel with configurable
    concurrency limits to efficiently handle bulk harvesting operations.

    Args:
        tasks: List of task dictionaries, each containing:
            - task_type: Type of harvest task
            - urls: List containing exactly one URL string to harvest
              (canonical key — matches :meth:`create_harvest_task`).
              A single ``url`` string key is also accepted as a
              convenience alias and is wrapped into a one-item list.
            - egress_info: Optional per-task egress location dict. When
              provided, it overrides the bulk-level ``egress_info`` for
              this task only — this is how you mix asset tasks (which
              need one of the 6 datacenter locations) with non-asset
              tasks in the same call.
            - params: Task-specific parameters (dict format)
            - dest_path: Optional destination path (default: "/")
            - dest_name: Optional destination name
        output_dir: Directory to save downloaded results
        max_concurrent: Positive maximum number of concurrent tasks (default: 5)
        egress_info: Default egress location information used for any
            task that doesn't specify its own ``egress_info``
            (default: {"name": "New York, NY"}).
            Note: asset tasks require locations from the 6-location datacenter
            list (e.g., "New York City", not "New York, NY"). If mixing asset
            and non-asset tasks, set ``egress_info`` per-task (see above) or
            use separate calls.

    Returns:
        List of task results (same order as input), None for failed tasks

    Raises:
        HarvesterAPIError: If bulk processing fails
        ValidationError: If parameters are invalid

    Example:
        >>> import asyncio
        >>> async def bulk_harvest():
        ...     harvester = HarvesterAPI(config)
        ...     tasks = [
        ...         {
        ...             "task_type": "asset",
        ...             "urls": ["https://site1.com"],
        ...             "egress_info": {"name": "New York City"},
        ...             "params": {"recursive": True}
        ...         },
        ...         {
        ...             "task_type": "visual",
        ...             "urls": ["https://site2.com"],
        ...             "params": {"output_pdf": True}
        ...         }
        ...     ]
        ...     results = await harvester.bulk_create_and_monitor_async(
        ...         tasks=tasks,
        ...         output_dir="./downloads",
        ...         max_concurrent=3
        ...     )
        ...     return results
        >>> results = asyncio.run(bulk_harvest())
    """
    validate_positive_int(max_concurrent, "max_concurrent")

    # Set default egress_info if not provided
    if egress_info is None:
        egress_info = {"name": "New York, NY"}

    # Create semaphore to limit concurrent tasks
    semaphore = asyncio.Semaphore(max_concurrent)

    async def process_single_task(
        task_config: dict[str, Any],
    ) -> dict[str, Any] | None:
        """Process a single task with semaphore control."""
        async with semaphore:
            try:
                # Get task_params from either 'params' or 'task_params' key
                task_params = task_config.get("task_params") or task_config.get(
                    "params", {}
                )

                # Accept both 'urls' (list — canonical, matches
                # create_harvest_task/create_harvest_task_async) and a
                # single 'url' string as a convenience alias, so callers
                # don't hit a KeyError from a docstring/param mismatch.
                task_urls = task_config.get("urls")
                if task_urls is None:
                    single_url = task_config.get("url")
                    task_urls = [single_url] if single_url is not None else None
                if not task_urls:
                    raise ValidationError(
                        "Each task dict must include a non-empty 'urls' "
                        "list (or a single 'url' string)"
                    )

                # Honor this task's own egress_info when provided,
                # falling back to the bulk-level default. Previously the
                # outer-scope egress_info was always used, making it
                # impossible to mix asset tasks (which require one of
                # the 6 datacenter locations) with non-asset tasks in a
                # single bulk call, despite the docstring advertising
                # per-task egress as the way to do so.
                task_egress_info = task_config.get("egress_info", egress_info)

                return await self.process_harvest_workflow_async(
                    task_type=task_config["task_type"],
                    urls=task_urls,
                    egress_info=task_egress_info,
                    output_dir=output_dir,
                    dest_path=task_config.get("dest_path", "/"),
                    dest_name=task_config.get("dest_name"),
                    task_params=(
                        task_params if isinstance(task_params, dict) else None
                    ),
                )
            except Exception as e:
                self.logger.error("Task failed: %s", e)
                return None

    # Process all tasks concurrently
    self.logger.info(
        "Processing %d tasks with max concurrency: %d", len(tasks), max_concurrent
    )

    try:
        results = await asyncio.gather(
            *[process_single_task(task) for task in tasks]
        )

        processed_results: list[dict[str, Any] | None] = [
            r if isinstance(r, dict) else None for r in results
        ]

        successful_tasks = sum(1 for r in processed_results if r is not None)
        self.logger.info(
            f"Completed bulk processing: {successful_tasks}/{len(tasks)} tasks successful"
        )

        return processed_results

    except Exception as e:
        if isinstance(e, (HarvesterAPIError, ValidationError)):
            raise
        raise HarvesterAPIError(f"Failed to process bulk async harvest: {e}") from e