arm-ka-fixtures/synthgen/config.py

"""Настройки генератора (спека §3): YAML-файл + переопределения `--set ключ=значение`.

Неизвестный ключ или значение вне диапазона — ConfigError до начала генерации.
"""

import re
from dataclasses import dataclass, fields
from decimal import Decimal, InvalidOperation
from pathlib import Path
from typing import Any

import yaml


class ConfigError(ValueError):
    """Ошибка настроек: неизвестный ключ, неверный тип или значение вне диапазона."""


@dataclass(frozen=True)
class Config:
    group_size: int = 5
    prefix: str = "SYN"
    periods: int = 4
    start: str = "2025-Q1"
    osv_counterparties_mean: int = 60
    no_inn_companies: int = 1
    vgo_mismatch_share: Decimal = Decimal("0.1")
    seed: int = 1
    out: Path | None = None


_PREFIX_RE = re.compile(r"[0-9A-Za-zА-Яа-яЁё]{2,12}")
_START_RE = re.compile(r"(\d{4})-Q([1-4])")


def _as_int(name: str, value: Any) -> int:
    if isinstance(value, bool):
        raise ConfigError(f"{name}: ожидается целое число, получено {value!r}")
    if isinstance(value, int):
        return value
    if isinstance(value, str) and re.fullmatch(r"[+-]?\d+", value.strip()):
        return int(value.strip())
    raise ConfigError(f"{name}: ожидается целое число, получено {value!r}")


def _as_str(name: str, value: Any) -> str:
    if isinstance(value, bool):
        raise ConfigError(f"{name}: ожидается строка, получено {value!r}")
    if isinstance(value, (str, int)):
        return str(value)
    raise ConfigError(f"{name}: ожидается строка, получено {value!r}")


def _as_decimal(name: str, value: Any) -> Decimal:
    if isinstance(value, bool) or not isinstance(value, (int, float, str)):
        raise ConfigError(f"{name}: ожидается число, получено {value!r}")
    try:
        number = Decimal(str(value).strip())
    except InvalidOperation as exc:
        raise ConfigError(f"{name}: ожидается число, получено {value!r}") from exc
    if not number.is_finite():
        raise ConfigError(f"{name}: ожидается конечное число, получено {value!r}")
    return number


def _as_path(name: str, value: Any) -> Path | None:
    if value is None or value == "":
        return None
    if isinstance(value, str):
        return Path(value)
    raise ConfigError(f"{name}: ожидается путь, получено {value!r}")


_COERCERS = {
    "group_size": _as_int,
    "prefix": _as_str,
    "periods": _as_int,
    "start": _as_str,
    "osv_counterparties_mean": _as_int,
    "no_inn_companies": _as_int,
    "vgo_mismatch_share": _as_decimal,
    "seed": _as_int,
    "out": _as_path,
}


def _validate(cfg: Config) -> None:
    if not 2 <= cfg.group_size <= 40:
        raise ConfigError(f"group_size: вне диапазона 2–40, получено {cfg.group_size}")
    if not _PREFIX_RE.fullmatch(cfg.prefix):
        raise ConfigError(f"prefix: 2–12 букв/цифр без пробелов, получено {cfg.prefix!r}")
    if not 1 <= cfg.periods <= 12:
        raise ConfigError(f"periods: вне диапазона 1–12, получено {cfg.periods}")
    match = _START_RE.fullmatch(cfg.start)
    if match is None:
        raise ConfigError(f"start: ожидается ГГГГ-Qn, получено {cfg.start!r}")
    start_key = (int(match.group(1)), int(match.group(2)))
    if not (2020, 1) <= start_key <= (2030, 4):
        raise ConfigError(f"start: вне диапазона 2020-Q1 … 2030-Q4, получено {cfg.start!r}")
    if not 0 <= cfg.osv_counterparties_mean <= 150:
        raise ConfigError(f"osv_counterparties_mean: вне диапазона 0–150, получено {cfg.osv_counterparties_mean}")
    if not 0 <= cfg.no_inn_companies <= cfg.group_size:
        raise ConfigError(
            f"no_inn_companies: вне диапазона 0–group_size ({cfg.group_size}), получено {cfg.no_inn_companies}"
        )
    if not Decimal(0) <= cfg.vgo_mismatch_share <= Decimal(1):
        raise ConfigError(f"vgo_mismatch_share: вне диапазона 0–1, получено {cfg.vgo_mismatch_share}")


def load_config(path: Path | None, overrides: list[str]) -> Config:
    """Читает YAML (если задан) и накладывает переопределения `ключ=значение`."""
    known = {f.name for f in fields(Config)}
    data: dict[str, Any] = {}
    if path is not None:
        try:
            loaded = yaml.safe_load(Path(path).read_text(encoding="utf-8"))
        except OSError as exc:
            raise ConfigError(f"не удалось прочитать {path}: {exc}") from exc
        except yaml.YAMLError as exc:
            raise ConfigError(f"некорректный YAML в {path}: {exc}") from exc
        if loaded is None:
            loaded = {}
        if not isinstance(loaded, dict):
            raise ConfigError(f"{path}: верхний уровень должен быть отображением ключ: значение")
        data.update(loaded)
    for item in overrides:
        if "=" not in item:
            raise ConfigError(f"--set: ожидается ключ=значение, получено {item!r}")
        key, value = item.split("=", 1)
        data[key.strip()] = value
    kwargs: dict[str, Any] = {}
    for key, value in data.items():
        if key not in known:
            raise ConfigError(f"неизвестный ключ: {key!r}")
        kwargs[key] = _COERCERS[key](key, value)
    cfg = Config(**kwargs)
    _validate(cfg)
    return cfg


def dump_config(cfg: Config) -> str:
    """YAML с ключами в порядке полей Config; значение `out` None → null."""
    data: dict[str, Any] = {}
    for f in fields(Config):
        value = getattr(cfg, f.name)
        if isinstance(value, Path):
            value = str(value)
        elif isinstance(value, Decimal):
            value = float(value)
        data[f.name] = value
    return yaml.safe_dump(data, sort_keys=False, allow_unicode=True, default_flow_style=False)