code: def get_log_uri(
      *,
      cluster: dict[str, Any] | None = None,
      emr_client: boto3.client = None,
      job_flow_id: str | None = None,
  ) -> str | None:
      """
      Retrieve the S3 URI to the EMR Job logs.
  
      Requires either the output of a describe_cluster call or both an EMR Client and a job_flow_id..
      """
      if not exactly_one(bool(cluster), emr_client and job_flow_id):
          raise AirflowException(
              "Requires either the output of a describe_cluster call or both an EMR Client and a job_flow_id."
          )
  
      cluster_info = (
          cluster or emr_client.describe_cluster(ClusterId=job_flow_id)
      )["Cluster"]
      if "LogUri" not in cluster_info:
          return None
      log_uri = S3Hook.parse_s3_url(cluster_info["LogUri"])
      return "/".join(log_uri)
  
file_path: airflow/airflow/providers/amazon/aws/links/emr.py
imports: from __future__ import annotations
  from typing import TYPE_CHECKING, Any
  from airflow.exceptions import AirflowException
  from airflow.providers.amazon.aws.hooks.s3 import S3Hook
  from airflow.providers.amazon.aws.links.base_aws import BASE_AWS_CONSOLE_LINK, BaseAwsLink
  from airflow.utils.helpers import exactly_one
label_desc: Use image node_id 1 for calling a global function with example usage: get_log_uri() and returns: str, None