citests_httpdata_manager

#!/usr/bin/env python
"""
CI Tests data manager

Use this script to download all possible tests data to be used by the local http server fixture

>>> citests_httpdata_manager --help
>>> citests_httpdata_manager --destination

>>> citests_httpdata_manager -a check

>>> citests_httpdata_manager -a download
>>> citests_httpdata_manager -a download --force
>>> citests_httpdata_manager -a download --force --refresh

>>> citests_httpdata_manager -a dry -t erddap --refresh

>>> citests_httpdata_manager -a download -t erddap
>>> citests_httpdata_manager -a download -t erddap_bgc
>>> citests_httpdata_manager -a download -t ifremer_gdac
>>> citests_httpdata_manager -a download -t ifremer_gdac_bgc
>>> citests_httpdata_manager -a download -t argovis

>>> citests_httpdata_manager -a download -t topo
>>> citests_httpdata_manager -a download -t oops
>>> citests_httpdata_manager -a download -t nvs
>>> citests_httpdata_manager -a download -t docs

>>> citests_httpdata_manager -a download -t ea_dataselection
>>> citests_httpdata_manager -a download -t ifremer_api
>>> citests_httpdata_manager -a download -t github
>>> citests_httpdata_manager -a download -t altim


If some downloads (eg erddap) are difficult:
>>> citests_httpdata_manager -t erddap_bgc -a check  # Delete files raising errors
>>> citests_httpdata_manager -t erddap_bgc -a download  # Try another download
>>> citests_httpdata_manager -t erddap_bgc -a check  # Delete files raising errors, then trying again download

Clean up all data for a fresh start:
>>> citests_httpdata_manager -a clear --refresh



"""
import os
import sys
import argparse
import aiohttp
import asyncio
import aiofiles
import json
from collections import UserList
import pandas as pd
import xarray as xr
from urllib.parse import urlparse, parse_qs, unquote
import logging
import hashlib
import numpy as np
from pathlib import Path
import glob

sys.path.append(os.path.join(os.path.dirname(__file__), ".."))
import argopy
from argopy import DataFetcher, clear_cache
from argopy.related import (
    TopoFetcher,
    OceanOPSDeployments,
    ArgoNVSReferenceTables,
    ArgoDocs,
)


log = logging.getLogger("argopy.cli.test_data")

# Where to save remote data to:
DATA_FOLDER = os.path.dirname(os.path.realpath(__file__)).replace(
    "cli", "argopy/tests/test_data"
)

# DB file to be used by the http server fixture to load content:
DB_FILE = os.path.join(DATA_FOLDER, "httpmocked_uri_index.json")

# Dictionary mapping of file extension with http data content type
CONTENT_TYPE = {
    "js": "application/json",
    "json": "application/json",
    "yaml": "text/yaml",
    "ld+json": "application/ld+json",
    "rdf+xml": "application/rdf+xml",
    "xml": "application/rdf+xml",
    "text/turtle": "text/turtle",
    "turtle": "text/turtle",
    "nc": "application/x-netcdf",
    "ncHeader": "text/plain",
    "txt": "text/plain",
    "html": "text/html",
    "png": "image/png",
    "gz": "application/x-gzip",
}

DEFAULT_TARGETS = [
    "erddap",
    "erddap_bgc",
    "ifremer_gdac",
    "ifremer_gdac_bgc",
    "argovis",
    "topo",
    "oops",
    "nvs",
    "docs",
    "altim",
    "ea_dataselection",
    "ifremer_api",
    "github",
    "altim",
]


start_with = lambda f, x: (
    f[0 : len(x)] == x if len(x) <= len(f) else False
)  # noqa: E731


def setup_args():
    icons_help_string = """CI Tests data manager for the http mocked server"""

    parser = argparse.ArgumentParser(
        description="argopy http mocked server test data manager",
        formatter_class=argparse.RawTextHelpFormatter,
        epilog="%s\n(c) Argo-France/Ifremer/LOPS, 2023-2024" % icons_help_string,
    )

    def action_choicesDescriptions():
        return """
       download     - download all test data
       list         - list available test data
       clear        - delete all test data and associated files
       dry          - list & store URI to be accessed
       check        - consolidate test data (check file content and possibly remove file if content not expected)
       search       - search record data for a given url
    """

    def action_getChoices():
        return ["download", "list", "clear", "dry", "check", "search"]

    parser.add_argument(
        "-a",
        "--action",
        choices=action_getChoices(),
        help="action to perform among: %s" % action_choicesDescriptions(),
        metavar="",
    )

    def target_choicesDescriptions():
        return """
       erddap           - URIs accessed by ErddapArgoDataFetcher with ds='phy'/'ref' from test_fetchers_data_erddap/Test_Backend and test_xarray_accessor/*
       erddap_bgc       - URIs accessed by ErddapArgoDataFetcher with ds='bgc'/'bgc-s' ifromn test_fetchers_data_erddap_bgc/Test_Backend
       ifremer_gdac     - URIs accessed by FTPArgoDataFetcher with ds='phy' from test_fetchers_data_gdac.py/Test_Backend
       ifremer_gdac_bgc - URIs accessed by FTPArgoDataFetcher with ds='bgc'/'bgc-s'/'bgc-b'
       argovis          - URIs accessed by ArgovisDataFetcher from test_fetchers_data_argovis/Test_Backend
        
       topo             - URIs accessed by TopoFetcher with ds='gebco' from test_related/Test_TopoFetcher
       oops             - URIs accessed by OceanOPSDeployments from test_related/Test_OceanOPSDeployments
       nvs              - URIs accessed by ArgoNVSReferenceTables from test_related/Test_ArgoNVSReferenceTables
       docs             - URIs accessed by ArgoDocs from test_related/Test_ArgoDocs
       altim            - URIs accessed by open_sat_altim_report from test_related/Test_open_sat_altim_report
       
       ea_dataselection - URIs from https://dataselection.euro-argo.eu/api/trajectory accessed by test_plot_dashboards and test_related
       ifremer_api      - URIs from https://api.ifremer.fr
       github           - URIs from https://github.com/euroargodev/argopy-data/raw/master    
    """

    def target_getChoices():
        return [
            "erddap",
            "erddap_bgc",
            "ifremer_gdac",
            "ifremer_gdac_bgc",
            "argovis",
            "topo",
            "oops",
            "nvs",
            "docs",
            "altim",
            "ea_dataselection",
            "ifremer_api",
            "github",
        ]

    parser.add_argument(
        "-t",
        "--target",
        choices=target_getChoices(),
        help="HTTP target server among: %s" % target_choicesDescriptions(),
        metavar="",
    )
    parser.add_argument(
        "-r",
        "--refresh",
        help="Refresh URI list if necessary",
        action="store_true",
    )
    parser.add_argument(
        "-f",
        "--force",
        help="Force write on data files even if they already exist",
        action="store_true",
    )
    parser.add_argument(
        "-d",
        "--destination",
        help="Return absolute path to test data storage folder",
        action="store_true",
    )
    parser.add_argument("--url", nargs="?", help="Url to search in database")
    return parser


def can_xr_be_opened(src, file):
    try:
        xr.open_dataset(file)
        return src
    except:
        # print("This source can't be opened with xarray: %s" % src)
        return src


def log_file_desc(file, data, src):
    size = float(os.stat(file).st_size / (1024 * 1024))
    prt_size = lambda x: "Size < 1Mb" if x < 1 else "Size = %0.2dMb" % x
    msg = []
    # msg.append("- %s, %s" % (file.replace(DATA_FOLDER, '<DATA_FOLDER>'), data[0:3]))
    if "erddap.ifremer.fr" in src["uri"]:
        if src["ext"] != "json":
            msg.append("🤖 ERDDAP: %s" % parse_qs(src["uri"]))
        else:
            msg.append("🤖 ERDDAP: %s" % src["uri"])

    elif "coastwatch.pfeg.noaa.gov/erddap" in src["uri"]:
        # msg.append("🤖 NOAA ERDDAP: %s" % urlparse(unquote(src["uri"])).query)
        msg.append("🤖 NOAA ERDDAP: %s" % unquote(src["uri"]))

    # elif "github.com/euroargodev/argopy-data" in src['uri']:
    #     msg.append("🤖 GITHUB ARGOPY-DATA: %s" % parse_qs(src["uri"]))

    elif start_with(src["uri"], "https://www.ocean-ops.org/api/1"):
        # msg.append("🤖 Ocean-OPS API: %s" % unquote(src["uri"]))
        msg.append("🤖 Ocean-OPS API: %s" % parse_qs(src["uri"]))

    elif start_with(src["uri"], "https://argovis-api.colorado.edu"):
        msg.append("🤖 Argovis API: %s" % unquote(src["uri"]))

    else:
        msg.append("🔸 %s" % src["uri"])

    msg.append(
        "%s*, %s, %s" % (file.replace(DATA_FOLDER, "")[1:13], data[0:3], prt_size(size))
    )
    return ", ".join(msg)
    # return " 🔸 ".join(msg)


def ls_files(load=True, target=None):
    if target is None:
        targets = DEFAULT_TARGETS
    else:
        targets = [target]

    L = Lister()
    for t in targets:
        if Path(DB_FILE).exists():
            print("Listing data for '%s':" % t)
            with open(DB_FILE, "r") as f:
                URIs = json.load(f)
            for item in L.load(t, refresh=False):
                test_data_file = Path(DATA_FOLDER).joinpath(
                    "%s.%s" % (item["sha"], item["ext"])
                )
                for db_item in URIs:
                    if db_item["uri"] == item["uri"]:
                        if load:
                            with open(test_data_file, mode="rb") as file:
                                data = file.read()
                        else:
                            data = "---"
                        print(log_file_desc(str(test_data_file), data, db_item))
                        break


class UriRegistry(UserList):
    """

    >>> R = UriRegistry(name='My_Registry')
    >>> R.exists(uri)
    >>> R.commit(uri)
    >>> R.delete(uri)
    >>> R.save()  # to 'My_Registry.json'
    >>> R.load()  # from 'My_Registry.json'

    >>> UriRegistry().commit({'uri': 'a'}) + UriRegistry().commit({'uri': 'b'})
    >>> UriRegistry().commit({'uri': 'a'}).commit({'uri': 'b'}) - UriRegistry().commit({'uri': 'a'})

    """

    def __init__(
        self,
        name: str = "UriRegistry",
        checking_key: str = "uri",
        overwrite: bool = False,
    ):
        self.checking_key = checking_key
        self.overwrite = overwrite
        self.name = name
        super().__init__([])

    def exists(self, entry):
        for item in self.data:
            if item.get(self.checking_key) == entry.get(self.checking_key):
                return True
        return False

    def commit(self, entry, overwrite: bool = None):
        overwrite = self.overwrite if overwrite is None else overwrite
        if not self.exists(entry):
            super().append(entry)
        elif overwrite:
            self.delete(entry)
            super().append(entry)
        log.debug("commit: %s" % entry["uri"])
        return self

    def delete(self, entry):
        for item in self.data:
            if item.get(self.checking_key) == entry.get(self.checking_key):
                super().remove(item)
        return self

    def save(self, file=None):
        file = "%s.json" % self.name if file is None else file
        with open(file, "w") as f:
            json.dump(self.data, f, indent=4)
        return self

    def load(self, file=None):
        file = "%s.json" % self.name if file is None else file
        with open(file, "r") as f:
            data = json.load(f)
        super().__init__(data)
        return self

    def __add__(self, obj):
        if isinstance(obj, UriRegistry):
            for item in obj.data:
                self += item
                super().append(item)
        else:
            raise TypeError("Can only add a UriRegistry with another UriRegistry")
        return self

    def __sub__(self, obj):
        if isinstance(obj, UriRegistry):
            for item in obj.data:
                self.delete(item)
        else:
            raise TypeError("Can only subtract a UriRegistry from another UriRegistry")
        return self


async def place_file(session: aiohttp.ClientSession, source: dict) -> None:
    """Download remote file and save it locally"""
    test_data_file = os.path.join(DATA_FOLDER, "%s.%s" % (source["sha"], source["ext"]))
    if OVERWRITE or not os.path.exists(test_data_file):
        async with session.get(source["uri"], ssl=False, timeout=60 * 10) as r:
            if r.content_type not in CONTENT_TYPE.values():
                print(
                    "Unexpected content type (%s) with this GET request: %s (%s extension)"
                    % (
                        r.content_type,
                        source["uri"],
                        os.path.splitext(urlparse(source["uri"]).path)[1],
                    )
                )

            async with aiofiles.open(test_data_file, "wb") as f:
                data = await r.content.read(n=-1)  # load all read bytes !
                # if data[0:3] in [b'Err', b'<!D']:
                #     raise ValueError(source)
                await f.write(data)
                print(log_file_desc(f.name, data, source))
                return can_xr_be_opened(source, test_data_file)
    else:
        # print("%s already exists !" % test_data_file)
        return can_xr_be_opened(source, test_data_file)


class Lister:
    def __init__(self, session: aiohttp.ClientSession = None):
        self.session = session

    def dfile(self, url: str, fmt: str) -> dict:
        """Return a dictionary to store one URI asset into a UriRegistry instance

        For gzip files we should also modify the header: Content-Encoding: gzip
        """
        return {
            "uri": url,
            "ext": fmt,
            "sha": hashlib.sha256(url.encode()).hexdigest(),
            "type": CONTENT_TYPE[fmt],
            "ts": pd.to_datetime("now", utc=True).strftime("%F %T"),
        }

    def respond(self, URIs: UriRegistry) -> list:
        R = UriRegistry(overwrite=False, name="citests_httpdata_manager_%s" % URIs.name)
        [R.commit(x) for x in URIs]
        R.save()
        return R.data

    def src(self, target: str = None):
        return "citests_httpdata_manager_%s.json" % target

    def load(self, target, refresh=False):
        if os.path.exists("citests_httpdata_manager_%s.json" % target) and not refresh:
            log.debug("Found a URI list for target='%s'" % target)
            return UriRegistry(name="citests_httpdata_manager_%s" % target).load().data
        else:
            log.debug("Refreshing URI list for target='%s'" % target)
            return getattr(self, target)

    @property
    def erddap(self):
        """From test_fetchers_data_erddap.py"""
        this_URI = UriRegistry(name="erddap")

        # This should correspond to all the possible erddap requests made during CI tests.
        # And because of the erddap fetcher N_POINT attribute, we also need to fetch ".ncHeader" on top of ".nc" files
        requests_phy = {
            "float": [[1901393], [1901393, 6902746]],
            "profile": [
                [6902746, 34],
                [6902746, [1, 12]],
            ],
            "region": [
                [-20, -16.0, 0, 1, 0, 100.0],
                [-20, -16.0, 0, 1, 0, 100.0, "2004-01-01", "2004-01-31"],
            ],
        }
        requests_ref = {
            "region": [
                [-25, -10, 36, 40, 0, 10.0],
                [-25, -10, 36, 40, 0, 10.0, "20180101", "20190101"],
            ]
        }
        requests_seq = {"phy": requests_phy, "ref": requests_ref}

        requests_par = {
            "float": [[1900468, 1900117, 1900386]],
            "region": [
                [-60, -55, 40.0, 45.0, 0.0, 20.0],
                [-60, -55, 40.0, 45.0, 0.0, 20.0, "2007-08-01", "2007-09-01"],
            ],
        }

        def add_to_URI(facade):
            uri = facade.uri
            for ii, url in enumerate(uri):
                this_URI.commit(self.dfile(url, "nc"))
                this_URI.commit(
                    self.dfile(
                        url.replace("." + facade.fetcher.erddap.response, ".ncHeader"),
                        "ncHeader",
                    )
                )

        for ds in requests_seq:
            for mode in ["expert", "standard", "research"]:
                fetcher = DataFetcher(src="erddap", ds=ds, mode=mode)
                for access_point in requests_seq[ds]:
                    [
                        add_to_URI(fetcher.profile(*cfg))
                        for cfg in requests_seq[ds][access_point]
                        if access_point == "profile"
                    ]
                    [
                        add_to_URI(fetcher.float(cfg))
                        for cfg in requests_seq[ds][access_point]
                        if access_point == "float"
                    ]
                    [
                        add_to_URI(fetcher.region(cfg))
                        for cfg in requests_seq[ds][access_point]
                        if access_point == "region"
                    ]

        for mode in ["expert", "standard", "research"]:
            fetcher = DataFetcher(
                src="erddap",
                ds="phy",
                mode=mode,
                parallel=True,
                parallel_method="thread",
                chunks_maxsize={"lon": 2.5, "lat": 2.5, "wmo": 1},
            )
            for access_point in requests_par:
                [
                    add_to_URI(fetcher.profile(*cfg))
                    for cfg in requests_par[access_point]
                    if access_point == "profile"
                ]
                [
                    add_to_URI(fetcher.float(cfg))
                    for cfg in requests_par[access_point]
                    if access_point == "float"
                ]
                [
                    add_to_URI(fetcher.region(cfg))
                    for cfg in requests_par[access_point]
                    if access_point == "region"
                ]

        # Add more URI accessed by the erddap data fetcher:
        this_URI.commit(
            self.dfile(
                "https://erddap.ifremer.fr/erddap/info/ArgoFloats/index.json", "json"
            )
        )
        this_URI.commit(
            self.dfile(
                "https://erddap.ifremer.fr/erddap/info/ArgoFloats-reference/index.json",
                "json",
            )
        )
        this_URI.commit(
            self.dfile("https://erddap.ifremer.fr/erddap/info/index.json", "json")
        )

        return self.respond(this_URI)

    @property
    def erddap_bgc(self):
        """From test_fetchers_data_erddap_bgc.py"""
        this_URI = UriRegistry(name="erddap_bgc")

        # This should correspond to all the possible erddap requests made during CI tests.
        # And because of the erddap fetcher N_POINT attribute, we also need to fetch ".ncHeader" on top of ".nc" files
        requests_seq = {
            # "float": [[6904240]],
            "float": [[5903248], [6904240], [5903248, 6904241]],
            "profile": [
                [5903248, 34],
                [5903248, np.arange(12, 14)],
                [5903248, [1, 12]],
            ],
            "region": [
                [-55, -47, 55, 57, 0, 10],
                [-55, -47, 55, 57, 0, 10, "2022-05-1", "2023-07-01"],
            ],
        }

        requests_par = {
            "float": [[5903248, 6904241]],
            "region": [
                [-55, -47, 55, 57, 0, 10, "2022-05-1", "2023-07-01"],
            ],
        }

        def add_to_URI(facade):
            uri = facade.uri
            for ii, url in enumerate(uri):
                this_URI.commit(self.dfile(url, "nc"))
                this_URI.commit(
                    self.dfile(
                        url.replace("." + facade.fetcher.erddap.response, ".ncHeader"),
                        "ncHeader",
                    )
                )

            uri = facade.fetcher.fs.urls_registry
            for ii, url in enumerate(uri):
                suffix = Path(urlparse(url).path).suffix
                if suffix == ".gz":
                    this_URI.commit(self.dfile(url, "gz"))

        clear_cache()
        indexfs = argopy.ArgoIndex(
            index_file="argo_synthetic-profile_index.txt",  # corresponds to dataset in the erddap
            cache=True,
        )
        for mode in ["expert", "standard", "research"]:
            for measured in [None, "all", "DOXY"]:
                for params in ["all", "DOXY"]:
                    fetcher = DataFetcher(
                        src="erddap",
                        ds="bgc",
                        cache=True,
                        mode=mode,
                        params=params,
                        measured=measured,
                        indexfs=indexfs,
                    )
                    for access_point in requests_seq:
                        [
                            add_to_URI(fetcher.profile(*cfg))
                            for cfg in requests_seq[access_point]
                            if access_point == "profile"
                        ]
                        [
                            add_to_URI(fetcher.float(cfg))
                            for cfg in requests_seq[access_point]
                            if access_point == "float"
                        ]
                        [
                            add_to_URI(fetcher.region(cfg))
                            for cfg in requests_seq[access_point]
                            if access_point == "region"
                        ]

        measured = None
        for mode in ["expert", "standard", "research"]:
            for params in ["all", "DOXY"]:
                fetcher = DataFetcher(
                    src="erddap",
                    ds="bgc",
                    cache=True,
                    parallel=True,
                    parallel_method="thread",
                    chunks_maxsize={"lon": 2.5, "lat": 2.5, "wmo": 1},
                    mode=mode,
                    params=params,
                    measured=measured,
                    indexfs=indexfs,
                )
                for access_point in requests_par:
                    [
                        add_to_URI(fetcher.profile(*cfg))
                        for cfg in requests_par[access_point]
                        if access_point == "profile"
                    ]
                    [
                        add_to_URI(fetcher.float(cfg))
                        for cfg in requests_par[access_point]
                        if access_point == "float"
                    ]
                    [
                        add_to_URI(fetcher.region(cfg))
                        for cfg in requests_par[access_point]
                        if access_point == "region"
                    ]

        # Add more URI from the erddap:
        this_URI.commit(
            self.dfile(
                "https://erddap.ifremer.fr/erddap/info/ArgoFloats-synthetic-BGC/index.json",
                "json",
            )
        )

        return self.respond(this_URI)

    @property
    def ifremer_gdac(self):
        """From test_fetchers_data_gdac.py"""
        this_URI = UriRegistry(name="ifremer_gdac")
        server = "https://data-argo.ifremer.fr"

        requests_seq = {
            "float": [[13857]],
            "profile": [[13857, 90]],
            "region": [
                [-20, -16.0, 0, 1, 0, 100.0],
                [-20, -16.0, 0, 1, 0, 100.0, "1997-07-01", "1997-09-01"],
            ],
        }

        def add_to_URI(facade):
            [this_URI.commit(self.dfile(uri, "nc")) for uri in facade.uri]

        fetcher = DataFetcher(src="gdac", gdac=server, ds="phy", MAX_FILES=100)
        for access_point in requests_seq:
            [
                add_to_URI(fetcher.profile(*cfg))
                for cfg in requests_seq[access_point]
                if access_point == "profile"
            ]
            [
                add_to_URI(fetcher.float(cfg))
                for cfg in requests_seq[access_point]
                if access_point == "float"
            ]
            [
                add_to_URI(fetcher.region(cfg))
                for cfg in requests_seq[access_point]
                if access_point == "region"
            ]

        # Add more URI from the IFREMER GDAC:
        this_URI.commit(self.dfile("%s/ar_index_global_prof.txt.gz" % server, "gz"))
        this_URI.commit(self.dfile("%s/dac" % server, "html"))

        return self.respond(this_URI)

    @property
    def ifremer_gdac_bgc(self):
        this_URI = UriRegistry(name="ifremer_gdac_bgc")
        server = "https://data-argo.ifremer.fr"

        requests_seq = {
            "float": [[5904989], [3902131], [6903247]],
            "profile": [[5904989, 12]],
        }

        def add_to_URI(facade):
            [this_URI.commit(self.dfile(uri, "nc")) for uri in facade.uri]

        fetcher = DataFetcher(
            src="gdac", gdac=server, ds="bgc", mode="expert", MAX_FILES=100
        )
        for access_point in requests_seq:
            [
                add_to_URI(fetcher.profile(*cfg))
                for cfg in requests_seq[access_point]
                if access_point == "profile"
            ]
            [
                add_to_URI(fetcher.float(cfg))
                for cfg in requests_seq[access_point]
                if access_point == "float"
            ]
            [
                add_to_URI(fetcher.region(cfg))
                for cfg in requests_seq[access_point]
                if access_point == "region"
            ]

        # Add more URI from the IFREMER GDAC:
        this_URI.commit(self.dfile("%s/argo_bio-profile_index.txt.gz" % server, "gz"))
        this_URI.commit(
            self.dfile("%s/argo_synthetic-profile_index.txt.gz" % server, "gz")
        )

        return self.respond(this_URI)

    @property
    def argovis(self):
        this_URI = UriRegistry(name="argovis")

        requests_phy = {
            "float": [[1901393], [1901393, 6902746]],
            "profile": [
                [6902746, 34],
                [6902746, [1, 12]],
            ],
            "region": [
                [-70, -65, 35.0, 40.0, 0, 10.0, "2012-01", "2012-03"],
                [-70, -65, 35.0, 40.0, 0, 10.0, "2012-01", "2012-06"],
            ],
        }

        requests_seq = {"phy": requests_phy}

        requests_par = {
            "float": [[1900468, 1900117, 1900386]],
            "region": [
                [-70, -65, 35.0, 40.0, 0.0, 10.0, "2012-01", "2012-06"],
                [-60, -55, 40.0, 45.0, 0.0, 20.0, "2007-08-01", "2007-09-01"],
            ],
        }

        def add_to_URI(facade):
            encode = facade.fetcher.url_encode
            [this_URI.commit(self.dfile(uri, "js")) for uri in encode(facade.uri)]

        for ds in requests_seq:
            fetcher = DataFetcher(src="argovis", ds=ds)
            for access_point in requests_seq[ds]:
                [
                    add_to_URI(fetcher.profile(*cfg))
                    for cfg in requests_seq[ds][access_point]
                    if access_point == "profile"
                ]
                [
                    add_to_URI(fetcher.float(cfg))
                    for cfg in requests_seq[ds][access_point]
                    if access_point == "float"
                ]
                [
                    add_to_URI(fetcher.region(cfg))
                    for cfg in requests_seq[ds][access_point]
                    if access_point == "region"
                ]

        fetcher = DataFetcher(
            src="argovis", parallel=True, chunks_maxsize={"lon": 2.5, "lat": 2.5}
        )
        for access_point in requests_par:
            [
                add_to_URI(fetcher.profile(*cfg))
                for cfg in requests_par[access_point]
                if access_point == "profile"
            ]
            [
                add_to_URI(fetcher.float(cfg))
                for cfg in requests_par[access_point]
                if access_point == "float"
            ]
            [
                add_to_URI(fetcher.region(cfg))
                for cfg in requests_par[access_point]
                if access_point == "region"
            ]

        # Add more URI from the ARGOVIS server:
        this_URI.commit(self.dfile("https://argovis-api.colorado.edu/ping", "json"))

        return self.respond(this_URI)

    @property
    def topo(self):
        """https://coastwatch.pfeg.noaa.gov/erddap"""
        this_URI = UriRegistry(name="topo")

        box = [81, 123, -67, -54]
        fetcher = TopoFetcher(box, ds="gebco", stride=[10, 10], cache=True)
        this_URI.commit(self.dfile(unquote(fetcher.uri[0]), "nc"))

        return self.respond(this_URI)

    @property
    def oops(self):
        this_URI = UriRegistry(name="oops")

        scenarios = [
            # (None, False),  # Can't be handled by the mocked server (test date is surely different from the test data date)
            # ([-90, 0, 0, 90], False),  # Can't be handled by the mocked server (test date is surely different from the test data date)
            ([-90, 0, 0, 90, "2022-01"], True),
            ([-90, 0, 0, 90, "2022-01"], False),
            ([None, 0, 0, 90, "2022-01-01", "2023-01-01"], True),
            ([None, 0, 0, 90, "2022-01-01", "2023-01-01"], False),
            ([-90, None, 0, 90, "2022-01-01", "2023-01-01"], True),
            ([-90, None, 0, 90, "2022-01-01", "2023-01-01"], False),
            ([-90, 0, None, 90, "2022-01-01", "2023-01-01"], True),
            ([-90, 0, None, 90, "2022-01-01", "2023-01-01"], False),
            ([-90, 0, 0, None, "2022-01-01", "2023-01-01"], True),
            ([-90, 0, 0, None, "2022-01-01", "2023-01-01"], False),
            ([-90, 0, 0, 90, None, "2023-01-01"], True),
            ([-90, 0, 0, 90, None, "2023-01-01"], False),
            ([-90, 0, 0, 90, "2022-01-01", None], True),
            ([-90, 0, 0, 90, "2022-01-01", None], False),
        ]

        for sc in scenarios:
            box, deployed_only = sc
            oops = OceanOPSDeployments(box, deployed_only=deployed_only)
            this_URI.commit(self.dfile(unquote(oops.uri), "json"))

        this_URI.commit(
            self.dfile("https://www.ocean-ops.org/api/1/oceanops-api.yaml", "yaml")
        )

        return self.respond(this_URI)

    @property
    def nvs(self):
        this_URI = UriRegistry(name="nvs")
        nvs = ArgoNVSReferenceTables()

        for rtid in nvs.valid_ref:
            url = nvs.get_url(rtid)
            this_URI.commit(self.dfile(unquote(url), "json"))

        fmts = {"ld+json": "json", "rdf+xml": "xml", "text/turtle": "txt"}
        for fmt in fmts.keys():
            url = nvs.get_url(3, fmt=fmt)
            this_URI.commit(self.dfile(unquote(url), fmts[fmt]))

        return self.respond(this_URI)

    @property
    def docs(self):
        this_URI = UriRegistry(name="docs")

        docs = ArgoDocs().list
        for ii, doc in docs.iterrows():
            Ad = ArgoDocs(doc["id"])

            doi = "https://dx.doi.org/%s" % Ad.js["doi"]
            this_URI.commit(self.dfile(unquote(doi), "html"))

        return self.respond(this_URI)

    @property
    def altim(self):
        # Used by Test_open_sat_altim_report
        this_URI = UriRegistry(name="altim")

        server = "https://data-argo.ifremer.fr"
        for scenario in [[2901623], [2901623, 6901929]]:
            for wmo in scenario:
                url = "%s/etc/argo-ast9-item13-AltimeterComparison/figures/%i.png" % (
                    server,
                    wmo,
                )
                this_URI.commit(self.dfile(url, "png"))

        return self.respond(this_URI)

    @property
    def ea_dataselection(self):
        this_URI = UriRegistry(name="ea_dataselection")

        WMO_list = []
        WMO_list.append(6901929)  # Used by test_related
        WMO_list.append(5904797)  # Used by test_plot_dashboards.py
        WMO_list.append(6902755)  # Used by test_plot_dashboards.py
        WMO_list.append(
            2901623
        )  # Used by test_fetchers_facade_data.py::Test_Facade::test_to_index_coriolis

        for uri in [
            "https://dataselection.euro-argo.eu/api/trajectory/%i" % wmo
            for wmo in WMO_list
        ]:
            this_URI.commit(self.dfile(uri, "json"))

        return self.respond(this_URI)

    @property
    def ifremer_api(self):
        this_URI = UriRegistry(name="ifremer_api")

        repo = "https://api.ifremer.fr"
        uris = ["argopy/data/ARGO-FULL.json", "argopy/data/ARGO-BGC.json"]
        for uri in uris:
            file_extension = uri.split(".")[-1]
            this_URI.commit(self.dfile("%s/%s" % (repo, uri), file_extension))

        return self.respond(this_URI)

    @property
    def github(self):
        this_URI = UriRegistry(name="github")

        repo = "https://github.com/euroargodev/argopy-data/raw/master"
        uris = [
            "ftp/dac/csiro/5900865/5900865_prof.nc",
            "ftp/ar_index_global_prof.txt",
            "ftp/dac/csiro/5900865/profiles/D5900865_001.nc",
            "ftp/dac/csiro/5900865/profiles/D5900865_002.nc",
        ]
        for uri in uris:
            file_extension = uri.split(".")[-1]
            this_URI.commit(self.dfile("%s/%s" % (repo, uri), file_extension))

        return self.respond(this_URI)


async def fetch_download_links(session: aiohttp.ClientSession, target=None):
    """Gather the list of URIs to download

    The return list is a list of dictionaries with all the necessary keys to save and
    retrieve requests offline using the fixture of the local HTTP server

    Lister(session).target

    """
    URI = []

    if target is None:
        targets = DEFAULT_TARGETS
    else:
        targets = [target]

    L = Lister(session)
    for t in targets:
        [URI.append(link) for link in L.load(t, refresh=REFRESH)]

    # Return the list of dictionaries
    log.debug("Listed %i URIs for %i targets" % (len(URI), len(targets)))
    return URI


async def download(target=None):
    connector = aiohttp.TCPConnector(limit=1, force_close=True)
    async with aiohttp.ClientSession(
        connector=connector, timeout=aiohttp.ClientTimeout(total=60 * 10)
    ) as session:
        urls = await fetch_download_links(session, target=target)
        return await asyncio.gather(*[place_file(session, url) for url in urls])


async def dry_download(target=None):
    connector = aiohttp.TCPConnector(limit=1, force_close=True)
    async with aiohttp.ClientSession(
        connector=connector, timeout=aiohttp.ClientTimeout(total=60 * 10)
    ) as session:
        return await fetch_download_links(session, target=target)


def clear_data(target=None):
    """For a given target: (i) delete files on disk and (ii) delete records in DB

    User --refresh to further remove the URI list registry file
    """

    if target is None:
        targets = DEFAULT_TARGETS
    else:
        targets = [target]

    L = Lister()
    for t in targets:
        n_delete = 0
        if Path(DB_FILE).exists():
            print("Clearing data for '%s'" % t)
            with open(DB_FILE, "r") as f:
                URIs = json.load(f)
            for item in L.load(t, refresh=False):
                file_path = Path(DATA_FOLDER).joinpath(
                    "%s.%s" % (item["sha"], item["ext"])
                )
                file_path.unlink(missing_ok=True)
                for db_item in URIs:
                    if db_item["uri"] == item["uri"]:
                        URIs.remove(db_item)
                        n_delete += 1
                        break
            with open(DB_FILE, "w") as f:
                json.dump(URIs, f, indent=4)
            print("\tCleared %i files on disk and from registry" % n_delete)
        if REFRESH and Path(L.src(t)).exists():
            Path(L.src(t)).unlink()
            print("\tCleared URI registry")

    if Path(DB_FILE).exists():
        with open(DB_FILE, "r") as f:
            URIS = json.load(f)
        if len(URIS) == 0:
            Path(DATA_FOLDER).joinpath(DB_FILE).unlink()


def consolidate(target=None):
    """
    Check file content, if not what is expected, or some known errors, delete file and record in DB
    """
    log.debug("Consolidate ...")

    if target is None:
        targets = DEFAULT_TARGETS
    else:
        targets = [target]

    L = Lister()
    for t in targets:
        n_delete = 0
        if Path(DB_FILE).exists():
            with open(DB_FILE, "r") as f:
                URIs = json.load(f)
            for item in URIs:
                file_path = Path(DATA_FOLDER).joinpath(
                    "%s.%s" % (item["sha"], item["ext"])
                )
                with open(file_path, mode="rb") as file:
                    data = file.read()
                if data[0:3] in [b"Err", b"<!D"] and item["ext"] in ["nc"]:
                    print("🚨 Bad %s in %s" % (item["type"], file_path))
                    print("▶ Record: ", item)
                    print("▶ Content:", data[0:100])
                    txt = "?"
                    while txt.lower() not in ["y", "n", ""]:
                        txt = input("Do you confirm to suppress this record ? [y]/n ")
                    if txt == "y" or txt == "":
                        n_delete += 1
                        file_path.unlink()
                        URIs.remove(item)
            if n_delete > 0:
                with open(DB_FILE, "w") as f:
                    json.dump(URIs, f, indent=4)
                print(
                    "\tCleared %i files on disk and from registry because of bad content"
                    % n_delete
                )
    if n_delete == 0:
        print("Everything looks good so far !")


def search_data(url=None):
    if url is not None and Path(DB_FILE).exists():
        with open(DB_FILE, "r") as f:
            URIs = json.load(f)
        found = False
        results = []
        for item in URIs:
            if unquote(url) in item["uri"] or url in item["uri"]:
                results.append(item)
                found = True
        if found:
            print("\n🎉 Found %i item(s) corresponding to your search:" % len(results))
            for r in results:
                print("▶ ", r)
        else:
            print("\n❌ No items correspond to your search")


if __name__ == "__main__":
    logging.getLogger("matplotlib").setLevel(logging.ERROR)
    logging.getLogger("fsspec").setLevel(logging.ERROR)
    logging.getLogger("pyftpdlib").setLevel(logging.ERROR)
    # logging.getLogger("MainThread").setLevel(logging.ERROR)
    logging.getLogger("s3fs").setLevel(logging.ERROR)
    logging.getLogger("urllib3").setLevel(logging.ERROR)
    logging.getLogger("aiobotocore").setLevel(logging.ERROR)
    logging.getLogger("botocore").setLevel(logging.ERROR)

    logging.basicConfig(
        level=logging.DEBUG,
        format="%(asctime)s [%(levelname)s] [%(name)s] %(filename)s:%(lineno)d: %(message)s",
        datefmt="%m/%d/%Y %I:%M:%S %p",
        handlers=[logging.FileHandler("citests_httpdata_manager.log", mode="w")],
    )

    ARGS = setup_args().parse_args()
    OVERWRITE = ARGS.force
    REFRESH = ARGS.refresh

    # Create destination folder if necessary
    Path(DATA_FOLDER).mkdir(parents=True, exist_ok=True)

    if ARGS.destination:
        print("Tests data are stored in:\n%s" % DATA_FOLDER)

    if ARGS.action == "dry":
        log.debug("Dry run ...")
        # Dry download of all remote resources (just list urls):
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        URIs = loop.run_until_complete(dry_download(target=ARGS.target))
        for uri in URIs:
            uri.pop("ext")
            uri.pop("sha")
            print("\n-", uri)
        print(
            "\nFound %i URIs to download to populate the http mocked server" % len(URIs)
        )

    if ARGS.action == "download":
        log.debug("Downloading ...")
        # Async download of all remote resources:
        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        URIs = loop.run_until_complete(download(target=ARGS.target))

        # Save the URI list in a json file to be loaded and used by the HTTP server fixture:
        if ARGS.target is not None:
            # Update the DB file
            if os.path.exists(DB_FILE):
                with open(DB_FILE, "r") as f:
                    URIS_before = json.load(f)
                unique_uris_before = [x["uri"] for x in URIS_before]
                unique_uris_downloaded = [x["uri"] for x in URIs]

                URIS_after = []
                for entry in URIs:
                    if entry["uri"] in unique_uris_before:
                        if OVERWRITE:
                            URIS_after.append(entry)
                    else:
                        URIS_after.append(entry)

                for entry in URIS_before:
                    if entry["uri"] not in unique_uris_downloaded:
                        URIS_after.append(entry)
                    elif entry["uri"] in unique_uris_downloaded and not OVERWRITE:
                        URIS_after.append(entry)

                with open(DB_FILE, "w") as f:
                    json.dump(URIS_after, f, indent=4)
                print("Updated existing DB file")

            else:
                print("Created new DB file")
                with open(DB_FILE, "w") as f:
                    json.dump(URIs, f, indent=4)

        else:
            # Create a full DB file
            with open(DB_FILE, "w") as f:
                json.dump(URIs, f, indent=4)
            print("Created new DB file")

        # URLS = [url for url in URLS if url is not None]
        print("\nSaved %i URIs" % len(URIs))

    if ARGS.action == "list":
        ls_files(load=False, target=ARGS.target)

    if ARGS.action == "check":
        consolidate(target=ARGS.target)

    if ARGS.action == "clear":
        clear_data(target=ARGS.target)

    if ARGS.action == "search":
        search_data(ARGS.url)