maven-sync-setup/sync/src/config.py

import yaml
import logging
from dataclasses import dataclass
from pathlib import Path

logger = logging.getLogger(__name__)


@dataclass
class KotlinVersion:
    version: str
    name: str
    plugins: dict[str, str]
    packages: list[str]


@dataclass
class Config:
    kotlin_versions: list[KotlinVersion]
    mirrors: list[str]


def handle_packages(section) -> list[str]:
    ignore = ["_kotlin-version", '_versions']
    result = []

    for entry in section:
        if entry not in ignore:
            if ':' in entry:
                versions = section[entry]
                if isinstance(versions, str):
                    result.append(f'{entry}:{versions}')
                elif isinstance(versions, list):
                    for version in versions:
                        result.append(f'{entry}:{version}')
                else:
                    logger.warning(f'Invalid version "{versions}" for "{entry}". Should be a string or list.')
            elif isinstance(section[entry], dict):
                group_section = section[entry]
                default_versions = group_section['_versions']

                for artifact in group_section:
                    if artifact not in ignore:
                        versions = group_section[artifact]

                        if not versions and default_versions:
                            versions = default_versions

                        if isinstance(versions, str):
                            result.append(f'{entry}:{artifact}:{versions}')
                        elif isinstance(versions, list) and versions:
                            for version in versions:
                                result.append(f'{entry}:{artifact}:{version}')
                        else:
                            logger.warning(f'Invalid versions "{versions}" for "{entry}:{artifact}"')
            else:
                logger.warning(f'Invalid package spec "{entry}". Should be a full spec or a group ID')

    return result


def handle_plugins(section) -> dict[str, list[str]]:
    result = {}

    for entry in section:
        versions = section[entry]

        if isinstance(versions, str):
            result[entry] = [versions]
        elif isinstance(versions, list):
            result[entry] = versions
        else:
            logger.warning(f'Invalid plugin version "{versions}" for {entry}')

    return result


def parse_config(path: Path) -> Config:
    with path.open('r') as f:
        data = yaml.safe_load(f)

        parsed = {}

        for version, section in data['kotlin'].items():
            plugins = handle_plugins(section.get('plugins', {}))
            packages = handle_packages(section.get('packages', {}))
            parsed[version] = {
                'plugins': plugins,
                'packages': packages,
            }

        # Merge the general plugins and packages with that of the default version
        if general := data.get('general'):
            if default_version := general.get('_kotlin-version'):
                parsed.setdefault(default_version, {})

                parsed[default_version]['plugins'].update(
                    handle_plugins(general.get('plugins', {}))
                )

                parsed[default_version]['packages'].extend(
                    handle_packages(general.get('packages', {}))
                )
            else:
                logger.warning(f'"general" section exists, but "_kotlin-version" is not defined. Skipping.')

        kotlin_versions = []

        for version, section in parsed.items():
            kotlin_versions.append(KotlinVersion(version, f'version-packages', {}, section['packages']))

            for plugin, plugin_versions in section['plugins'].items():
                for plugin_version in plugin_versions:
                    kotlin_versions.append(KotlinVersion(version, f'version-plugins-{plugin}', {plugin: plugin_version}, []))

        return Config(
            kotlin_versions,
            data.get('maven', {}).get('mirrors', []),
        )