import os
import re

from six.moves import urllib

from mlflow.entities.file_info import FileInfo
from mlflow.store.artifact_repo import ArtifactRepository
from mlflow.utils.file_utils import build_path, get_relative_path, TempDir


class AzureBlobArtifactRepository(ArtifactRepository):
    """
    Stores artifacts on Azure Blob Storage.

    This repository is used with URIs of the form
    ``wasbs://<container-name>@<ystorage-account-name>.blob.core.windows.net/<path>``,
    following the same URI scheme as Hadoop on Azure blob storage. It requires that your Azure
    storage access key be available in the environment variable ``AZURE_STORAGE_ACCESS_KEY``.
    """

    def __init__(self, artifact_uri, client=None):
        super(AzureBlobArtifactRepository, self).__init__(artifact_uri)

        # Allow override for testing
        if client:
            self.client = client
            return

        from azure.storage.blob import BlockBlobService
        (_, account, _) = AzureBlobArtifactRepository.parse_wasbs_uri(artifact_uri)
        if "AZURE_STORAGE_CONNECTION_STRING" in os.environ:
            self.client = BlockBlobService(
                account_name=account,
                connection_string=os.environ.get("AZURE_STORAGE_CONNECTION_STRING"))
        elif "AZURE_STORAGE_ACCESS_KEY" in os.environ:
            self.client = BlockBlobService(
                account_name=account,
                account_key=os.environ.get("AZURE_STORAGE_ACCESS_KEY"))
        else:
            raise Exception("You need to set one of AZURE_STORAGE_CONNECTION_STRING or "
                            "AZURE_STORAGE_ACCESS_KEY to access Azure storage.")

    @staticmethod
    def parse_wasbs_uri(uri):
        """Parse a wasbs:// URI, returning (container, storage_account, path)."""
        parsed = urllib.parse.urlparse(uri)
        if parsed.scheme != "wasbs":
            raise Exception("Not a WASBS URI: %s" % uri)
        match = re.match("([^@]+)@([^.]+)\\.blob\\.core\\.windows\\.net", parsed.netloc)
        if match is None:
            raise Exception("WASBS URI must be of the form "
                            "<container>@<account>.blob.core.windows.net")
        container = match.group(1)
        storage_account = match.group(2)
        path = parsed.path
        if path.startswith('/'):
            path = path[1:]
        return container, storage_account, path

    def log_artifact(self, local_file, artifact_path=None):
        (container, _, dest_path) = self.parse_wasbs_uri(self.artifact_uri)
        if artifact_path:
            dest_path = build_path(dest_path, artifact_path)
        dest_path = build_path(dest_path, os.path.basename(local_file))
        self.client.create_blob_from_path(container, dest_path, local_file)

    def log_artifacts(self, local_dir, artifact_path=None):
        (container, _, dest_path) = self.parse_wasbs_uri(self.artifact_uri)
        if artifact_path:
            dest_path = build_path(dest_path, artifact_path)
        local_dir = os.path.abspath(local_dir)
        for (root, _, filenames) in os.walk(local_dir):
            upload_path = dest_path
            if root != local_dir:
                rel_path = get_relative_path(local_dir, root)
                upload_path = build_path(dest_path, rel_path)
            for f in filenames:
                path = build_path(upload_path, f)
                self.client.create_blob_from_path(container, path, build_path(root, f))

    def list_artifacts(self, path=None):
        from azure.storage.blob.models import BlobPrefix
        (container, _, artifact_path) = self.parse_wasbs_uri(self.artifact_uri)
        dest_path = artifact_path
        if path:
            dest_path = build_path(dest_path, path)
        infos = []
        prefix = dest_path + "/"
        marker = None  # Used to make next list request if this one exceeded the result limit
        while True:
            results = self.client.list_blobs(container, prefix=prefix, delimiter='/', marker=marker)
            for r in results:
                if isinstance(r, BlobPrefix):   # This is a prefix for items in a subdirectory
                    subdir = r.name[len(artifact_path)+1:]
                    if subdir.endswith("/"):
                        subdir = subdir[:-1]
                    infos.append(FileInfo(subdir, True, None))
                else:  # Just a plain old blob
                    file_name = r.name[len(artifact_path)+1:]
                    infos.append(FileInfo(file_name, False, r.properties.content_length))
            # Check whether a new marker is returned, meaning we have to make another request
            if results.next_marker:
                marker = results.next_marker
            else:
                break
        return sorted(infos, key=lambda f: f.path)

    def download_artifacts(self, artifact_path):
        with TempDir(remove_on_exit=False) as tmp:
            return self._download_artifacts_into(artifact_path, tmp.path())

    def _download_artifacts_into(self, artifact_path, dest_dir):
        """Private version of download_artifacts that takes a destination directory."""
        basename = os.path.basename(artifact_path)
        local_path = build_path(dest_dir, basename)
        listing = self.list_artifacts(artifact_path)
        if len(listing) > 0:
            # Artifact_path is a directory, so make a directory for it and download everything
            os.mkdir(local_path)
            for file_info in listing:
                self._download_artifacts_into(file_info.path, local_path)
        else:
            (container, _, remote_path) = self.parse_wasbs_uri(self.artifact_uri)
            remote_path = build_path(remote_path, artifact_path)
            self.client.get_blob_to_path(container, remote_path, local_path)
        return local_path
