Download and parse ethnologue.com language code tables

ethnologue.py

"""Download and parse code tables from https://www.ethnologue.com download link."""

from collections.abc import Callable, Iterable, Iterator, Mapping
import contextlib
import csv
import fnmatch
import functools
import html.parser
import http.client
import io
import itertools
import os
import pathlib
import pprint
import re
import shutil
from typing import Literal, NamedTuple, Self
import urllib.request
import zipfile

__all__ = ['languagecodes',
           'countrycodes',
           'languageindex']

LINKS_URL = 'https://www.ethnologue.com/codes/'

HEADERS = {'User-Agent': ('Mozilla/5.0 (X11; U; Linux i686)'
                          ' Gecko/20071127 Firefox/2.0.0.11')}

FILENAME_GLOB = 'Language_Code_Data_????????.zip'

LINK_SELECT_MODE = 'max'

CSV_NAME_PATTERN = re.compile(r'(?:.*/)?(?P<tablename>[\w-]+)\.tab',
                              flags=re.IGNORECASE | re.VERBOSE)

CSV_DIALECT = csv.excel_tab

CSV_ENCODING = 'utf-8'


def get_path(filename_glob: str = FILENAME_GLOB, /, *,
             try_download: bool = False,
             links_url: str = LINKS_URL,
             link_select_mode: Literal['singleton', 'max'] = LINK_SELECT_MODE,
             headers: Mapping[str, str] | None = HEADERS) -> pathlib.Path:
    """Return path of the latest glob matching filename or download from selected link."""
    for retry in ([True, False] if try_download else [False]):
        matching_paths = pathlib.Path().glob(filename_glob)
        try:
            return get_one(matching_paths, strategy='max')
        except ValueError:
            if not retry:
                raise RuntimeError('download failed or disabled') from None

        with urlopen(links_url, headers=headers) as response:
            encoding = response.headers.get_content_charset()
            with io.TextIOWrapper(response, encoding=encoding) as lines:
                found_urls = list(read_links(lines, base_href=response.url))

        match_glob = functools.partial(fnmatch.fnmatch, pat=filename_glob)
        matching_urls = [url for url in found_urls if match_glob(url.posix_path.name)]
        url_parse = get_one(matching_urls, strategy=link_select_mode)
        urlretrieve(url_parse.geturl(), url_parse.posix_path.name, headers=headers)


def get_one[T](items: Iterable[T], /, *, strategy: Literal['singleton', 'max']) -> T:
    """Return a single element from an iterable."""
    if strategy == 'singleton':
        (result,) = set(items)
    elif strategy == 'max':
        result = max(items)
    else:
        raise TypeError
    return result


@contextlib.contextmanager
def urlopen(url: str, /, *, headers: Mapping[str, str] | None) -> http.client.HTTPResponse:
    """Open the URL and return the response object."""
    print(url)
    request = urllib.request.Request(url, headers=headers)
    with urllib.request.urlopen(request) as response:
        if response.url != url:  # redirect
            print('->', response.url)
        yield response


def read_links(chunks: Iterable[str], /, *, base_href: str | None) -> Iterator['UrlParse']:
    """Yield parsed link target URLs from HTML chunks."""
    link_reader = HTMLTagAttributeReader(tag='a', attrname='href')
    hrefs = link_reader(chunks)
    if base_href:
        add_base_href = functools.partial(urllib.parse.urljoin, base_href)
        hrefs = map(add_base_href, hrefs)
    return map(UrlParse.from_string, hrefs)


class HTMLTagAttributeReader(html.parser.HTMLParser):
    """HTML tag attribute value collector."""

    def __init__(self, *args, tag: str, attrname: str, **kwargs) -> None:
        super().__init__(*args, **kwargs)
        self.tag = tag
        self.attrname = attrname
        self._collected_values: list[str] = []

    def handle_starttag(self, tag, attrs) -> None:
        """If the tag matches, collect the first matching attribute value."""
        if tag == self.tag:
            for key, value in attrs:
                if key == self.attrname:
                    self._collected_values.append(value)
                    return

    def __call__(self, chunks: Iterable[str] | str) -> Iterator[str]:
        """Yield parsed tag attribute values from HTML chunks."""
        if isinstance(chunks, str):
            chunks = [chunks]
        self.reset()
        values = self._collected_values
        with contextlib.closing(self):
            values.clear()
            for data in chunks:
                self.feed(data)
                if values:
                    yield from values
                    values.clear()


class UrlParse(urllib.parse.ParseResult):
    """A 6-tuple of parsed URL components with a posix_path attribute."""

    @classmethod
    def from_string(cls, url: str) -> Self:
        return cls._make(urllib.parse.urlparse(url))

    @functools.cached_property
    def posix_path(self) -> pathlib.PurePosixPath:
        return pathlib.PurePosixPath(self.path)


def urlretrieve(url: str, /, filename: os.PathLike[str] | str, *,
                headers: Mapping[str, str] | None = None) -> pathlib.Path:
    """Download the URL to the given target filename."""
    with (urlopen(url, headers=headers) as response,
          open(filename, mode='wb') as target):
        shutil.copyfileobj(response, target)
    return pathlib.Path(filename)


def itertables(zip_path: os.PathLike[str] | str, /, *,
               filename_pattern: re.Pattern[str] = CSV_NAME_PATTERN,
               encoding: str = CSV_ENCODING
               ) -> Iterator[tuple[str, Iterator[tuple[str, ...]]]]:
    """Yield (table name, csv.reader) pairs from matching CSV files in ZIP archive."""
    def key_func(zinfo: zipfile.ZipInfo) -> str | None:
        if (match := filename_pattern.fullmatch(zinfo.filename)) is not None:
            return match.group(match.lastindex or 0)

    for name, buffer in iterzipfile(zip_path, key_func=key_func):
        tablename = name.lower().replace('-', '_')
        with io.TextIOWrapper(buffer, encoding=encoding, newline='') as lines:
            yield tablename, read_csv(lines, namedtuple_name=tablename.capitalize())


def iterzipfile(path: os.PathLike[str] | str, /, *,
                key_func: Callable[[zipfile.ZipInfo], str | None] = lambda x: x.filename,
                ) -> Iterator[tuple[str, zipfile.ZipExtFile]]:
    """Yield matching (key, file-like object) pairs from ZIP archive."""
    with zipfile.ZipFile(path) as z:
        for i in z.infolist():
            if (key := key_func(i)) is not None:
                with z.open(i) as f:
                    yield key, f


def read_csv(lines: Iterable[str], /, *,
             dialect: csv.Dialect | type[csv.Dialect] | str = CSV_DIALECT,
             namedtuple_name: str = 'Row',
             value_type: type = str) -> Iterator[tuple[str, ...]]:
    """Yield namedtuple rows from CSV lines."""
    reader = csv.reader(lines, dialect=dialect)
    header = next(reader)
    fields = [(h, value_type) for h in header]
    row_cls = NamedTuple(namedtuple_name, fields)
    return map(row_cls._make, reader)


path = get_path(try_download=True)
print(path)

for name, rows in itertables(path):
    rows = list(itertools.islice(rows, 5))
    print(f'\nTABLE {name}', rows[0]._fields)
    pprint.pp(rows)

print()
tables = {name: list(rows) for name, rows in itertables(path)}
pprint.pp({name: len(rows) for name, rows in tables.items()})

globals().update({_: tables[_] for _ in __all__})