readabs.abs_catalogue

  1"""Catalogue map for ABS data."""
  2
  3from functools import cache
  4from io import StringIO
  5
  6from pandas import DataFrame, Index, Series, read_html
  7
  8from readabs.download_cache import CacheError, HttpError, get_file
  9
 10# Constants
 11ABS_CATALOGUE_URL = "https://www.abs.gov.au/about/data-services/help/abs-time-series-directory"
 12ABS_STATISTICS_ROOT = "https://www.abs.gov.au/statistics/"
 13EXPECTED_COLUMNS = ["Theme", "Parent Topic", "Topic"]
 14CATALOGUE_INDEX_NAME = "Catalogue ID"
 15CEASED_MARKER = "Ceased"
 16DEFAULT_ENCODING = "utf-8"
 17
 18
 19class CatalogueError(Exception):
 20    """Error processing ABS catalogue data."""
 21
 22
 23@cache
 24def abs_catalogue(*, cache_only: bool = False, verbose: bool = False) -> DataFrame:
 25    """Return a DataFrame of ABS Catalogue numbers.
 26
 27    Downloads catalogue data from the ABS website on first call and caches
 28    for future use. The returned DataFrame contains catalogue numbers with
 29    their topics, themes, URLs, and status.
 30
 31    Parameters
 32    ----------
 33    cache_only : bool, default False
 34        If True, only use cached data and don't attempt to download.
 35    verbose : bool, default False
 36        If True, print progress messages.
 37
 38    Returns
 39    -------
 40    DataFrame
 41        DataFrame with columns ['Theme', 'Parent Topic', 'Topic', 'URL', 'Status']
 42        and index of catalogue IDs.
 43
 44    Raises
 45    ------
 46    CatalogueError
 47        If the catalogue data cannot be retrieved or parsed.
 48    HttpError
 49        If there's a network error downloading the catalogue.
 50    CacheError
 51        If cache_only=True but no cached data is available.
 52
 53    Example
 54    -------
 55    >>> import readabs as ra
 56    >>> catalogue = ra.abs_catalogue()
 57    >>> print(catalogue.head())
 58
 59    """
 60    try:
 61        # Download ABS catalogue page
 62        abs_bytes = get_file(ABS_CATALOGUE_URL, cache_only=cache_only, verbose=verbose)
 63
 64        if not abs_bytes:
 65            raise CatalogueError("No data retrieved from ABS catalogue URL")
 66
 67        # Parse HTML content
 68        try:
 69            html_content = abs_bytes.decode(DEFAULT_ENCODING, errors="replace")
 70        except UnicodeDecodeError as e:
 71            raise CatalogueError(f"Failed to decode HTML content: {e}") from e
 72
 73        # Extract tables from HTML
 74        try:
 75            tables = read_html(StringIO(html_content), extract_links="body")
 76            if not tables:
 77                raise CatalogueError("No tables found in HTML content")
 78            links = tables[-1]  # Get the last table
 79        except (ValueError, IndexError) as e:
 80            raise CatalogueError(f"Failed to parse HTML tables: {e}") from e
 81
 82        # Validate required columns exist
 83        required_cols = ["Catalogue number", "Topic"]
 84        missing_cols = [col for col in required_cols if col not in links.columns]
 85        if missing_cols:
 86            raise CatalogueError(f"Missing required columns: {missing_cols}")
 87
 88        # Extract catalogue numbers and URLs
 89        try:
 90            cats = links["Catalogue number"].apply(Series)[0]
 91            urls = links["Topic"].apply(Series)[1]
 92        except (KeyError, IndexError) as e:
 93            raise CatalogueError(f"Failed to extract catalogue data: {e}") from e
 94
 95        # Process topic URLs to create hierarchical structure
 96        url_snippets = _process_topic_urls(urls)
 97
 98        # Create main DataFrame with hierarchical topic structure
 99        frame = _create_topic_frame(url_snippets)
100        frame["URL"] = urls
101
102        # Align catalogue numbers with processed frame
103        cats = cats[frame.index]
104
105        # Process catalogue status (active vs ceased)
106        cat_index, status = _process_catalogue_status(cats)
107
108        frame["Status"] = status
109        frame.index = Index(cat_index)
110        frame.index.name = CATALOGUE_INDEX_NAME
111
112    except (HttpError, CacheError, ValueError) as e:
113        raise CatalogueError(f"Error retrieving ABS catalogue: {e}") from e
114
115    return frame
116
117
118def _process_topic_urls(urls: Series) -> Series:
119    """Process topic URLs to extract clean topic hierarchy."""
120    # Remove root URL prefix
121    snippets = urls.str.replace(ABS_STATISTICS_ROOT, "", regex=False)
122
123    # Filter out invalid URLs and clean formatting
124    valid_snippets = snippets[~snippets.str.contains("http", na=False)]
125    return valid_snippets.str.replace("-", " ").str.title()
126
127
128def _create_topic_frame(snippets: Series) -> DataFrame:
129    """Create DataFrame with topic hierarchy from URL snippets."""
130    # Split URL paths into hierarchical components
131    frame = snippets.str.split("/", expand=True).iloc[:, :3]
132    frame.columns = Index(EXPECTED_COLUMNS)
133
134    return frame
135
136
137def _process_catalogue_status(cats: Series) -> tuple[Series, Series]:
138    """Process catalogue numbers to extract IDs and status."""
139    # Extract clean catalogue IDs (remove ceased marker)
140    cat_index = cats.str.replace(CEASED_MARKER, "", regex=False).str.strip()
141
142    # Determine status based on presence of ceased marker
143    status = Series("Active", index=cats.index)
144    ceased_mask = cats.str.contains(CEASED_MARKER, na=False)
145    status.loc[ceased_mask] = "Ceased"
146
147    return cat_index, status
148
149
150if __name__ == "__main__":
151    print(abs_catalogue())