lastfm

def fetch_dataset(self) -> None:
    """Check if dataset is present, if not download.

    This method overrides the base class to handle the special case where
    the zipfile may exist but the extracted file doesn't.
    """
    zip_path = os.path.join(
        self.config.default_base_path, f"{self.config.remote_zipname}.zip"
    )

    if not os.path.exists(zip_path):
        logger.debug(f"{self.name} dataset zipfile not found in {zip_path}.")
        self._download_dataset()
    elif not os.path.exists(self.file_path):
        logger.debug(
            f"{self.name} dataset file not found, but zipfile already downloaded. "
            f"Extracting file from zipfile."
        )
        with zipfile.ZipFile(zip_path, "r") as zip_ref:
            zip_ref.extract(self.config.remote_filename, self.config.default_base_path)
    else:
        logger.debug("Data zipfile is in memory and in dir specified.")

def fetch_dataset_force(self) -> None:
    """Force re-download of the dataset."""
    logger.debug(f"{self.name} force re-download of dataset.")
    self._download_dataset()

def get_timestamp_range_in_epoch(self) -> tuple[int, int]:
    """Get the minimum and maximum timestamps in the dataset.

    Returns:
        A tuple of (min_timestamp, max_timestamp).

    Raises:
        RuntimeError: If load() has not been called yet.
    """
    return self.timestamp_min, self.timestamp_max

def get_timestamp_range_in_datetime(self) -> tuple[datetime, datetime]:
    """Get the minimum and maximum timestamps in the dataset.

    Returns:
        A tuple of (min_timestamp, max_timestamp).

    Raises:
        RuntimeError: If load() has not been called yet.
    """
    min_dt = datetime.fromtimestamp(self.timestamp_min)
    max_dt = datetime.fromtimestamp(self.timestamp_max)
    return min_dt, max_dt

def add_filter(self, filter_: Filter) -> None:
    """Add a filter to be applied when loading the data.

    Utilize :class:`DataFramePreprocessor` class to add filters to the
    dataset to load. The filter will be applied when the data is loaded into
    an :class:`InteractionMatrix` object when :meth:`load` is called.

    :param filter_: Filter to be applied to the loaded DataFrame
                processing to interaction matrix.
    :type filter_: Filter
    """
    self.preprocessor.add_filter(filter_)

def load(self, apply_filters: bool = True, use_cache: bool = True) -> InteractionMatrix:
    """Loads data into an InteractionMatrix object.

    Data is loaded into a DataFrame using the :func:`_load_dataframe` function.
    Resulting DataFrame is parsed into an :class:`InteractionMatrix` object. If
    :data:`apply_filters` is set to True, the filters set will be applied to the
    dataset and mapping of user and item ids will be done. This is advised
    even if there is no filter set, as it will ensure that the user and item
    ids are incrementing in the order of time.

    Args:
        apply_filters: To apply the filters set and preprocessing,
            defaults to True
        use_cache: Whether to use cached processed data, defaults to True

    Returns:
        Resulting interaction matrix.
    """
    logger.info(f"{self.name} is loading dataset...")
    start = time.time()
    try:
        df = self._load_dataframe_from_cache() if use_cache else self._load_dataframe()
    except FileNotFoundError:
        logger.warning("Processed cache not found, loading raw dataframe.")
        df = self._load_dataframe()
        self._cache_processed_dataframe(df)
    if apply_filters:
        logger.debug(f"{self.name} applying filters set.")
        im = self.preprocessor.process(df)
    else:
        im = self._dataframe_to_matrix(df)
        logger.warning(
            "No filters applied, user and item ids may not be incrementing in the order of time. "
            "Classes that use this dataset may not work as expected."
        )
    self._timestamp_min = int(df[self.config.timestamp_ix].min())
    self._timestamp_max = int(df[self.config.timestamp_ix].max())

    if self.fetch_metadata:
        user_id_mapping, item_id_mapping = (
            self.preprocessor.user_id_mapping,
            self.preprocessor.item_id_mapping,
        )
        self._fetch_dataset_metadata(user_id_mapping=user_id_mapping, item_id_mapping=item_id_mapping)

    end = time.time()
    logger.info(f"{self.name} dataset loaded - Took {end - start:.3}s")
    return im