<commit_before>#  Copyright (c) 2022. The Pycroft Authors. See the AUTHORS file.
#  This file is part of the Pycroft project and licensed under the terms of
#  the Apache License, Version 2.0. See the LICENSE file for details
from __future__ import annotations

import json
import traceback
import typing
from functools import partial

import jsonschema

from .babel import gettext, dgettext, dngettext, ngettext
from .formatting import format_param
from .options import Options
from .serde import Serializable, serialize_param, deserialize_param


# TODO remove in py3.11 (replace `TSelf` by `Self`)
TMessage = typing.TypeVar("TMessage", bound="Message")


# TODO turn into ABC && properly type subclasses
class Message:
    __slots__ = ("domain", "args", "kwargs")

    @classmethod
    def from_json(cls, json_string: str) -> Message:
        try:
            obj = json.loads(json_string)
        except ValueError:
            return ErroneousMessage(json_string)
        try:
            jsonschema.validate(obj, schema)
        except jsonschema.ValidationError as e:
            return ErroneousMessage(
                "Message validation failed: {} for " "message {}".format(e, json_string)
            )
        args = obj.get("args", ())
        kwargs = obj.get("kwargs", {})
        try:
            args = tuple(deserialize_param(a) for a in args)
            kwargs = {k: deserialize_param(v) for k, v in kwargs.items()}
        except (TypeError, ValueError) as e:
            error = "".join(traceback.format_exception_only(type(e), e))
            return ErroneousMessage(
                "Parameter deserialization error: {} in "
                "message: {}".format(error, json_string)
            )
        m: Message
        if "plural" in obj:
            m = NumericalMessage(
                obj["singular"], obj["plural"], obj["n"], obj.get("domain")
            )
        else:
            m = SimpleMessage(obj["message"], obj.get("domain"))
        m.args = args
        m.kwargs = kwargs
        return m

    def __init__(self, domain: str | None = None):
        self.domain = domain
        self.args: typing.Iterable[Serializable] = ()
        self.kwargs: dict[str, Serializable] = {}

    def _base_dict(self) -> dict[str, typing.Any]:
        raise NotImplementedError()

    def _gettext(self) -> str:
        raise NotImplementedError()

    def to_json(self) -> str:
        obj = self._base_dict()
        if self.domain is not None:
            obj["domain"] = self.domain
        if self.args:
            obj["args"] = tuple(serialize_param(a) for a in self.args)
        if self.kwargs:
            obj["kwargs"] = {k: serialize_param(v) for k, v in self.kwargs.items()}<commit_msg>[helpers] turn Message into abc<commit_after>#  Copyright (c) 2022. The Pycroft Authors. See the AUTHORS file.
#  This file is part of the Pycroft project and licensed under the terms of
#  the Apache License, Version 2.0. See the LICENSE file for details
from __future__ import annotations

import abc
import json
import traceback
import typing
from functools import partial

import jsonschema

from .babel import gettext, dgettext, dngettext, ngettext
from .formatting import format_param
from .options import Options
from .serde import Serializable, serialize_param, deserialize_param


# TODO remove in py3.11 (replace `TSelf` by `Self`)
TMessage = typing.TypeVar("TMessage", bound="Message")


class Message(abc.ABC):
    __slots__ = ("domain", "args", "kwargs")

    @classmethod
    def from_json(cls, json_string: str) -> Message:
        try:
            obj = json.loads(json_string)
        except ValueError:
            return ErroneousMessage(json_string)
        try:
            jsonschema.validate(obj, schema)
        except jsonschema.ValidationError as e:
            return ErroneousMessage(
                "Message validation failed: {} for " "message {}".format(e, json_string)
            )
        args = obj.get("args", ())
        kwargs = obj.get("kwargs", {})
        try:
            args = tuple(deserialize_param(a) for a in args)
            kwargs = {k: deserialize_param(v) for k, v in kwargs.items()}
        except (TypeError, ValueError) as e:
            error = "".join(traceback.format_exception_only(type(e), e))
            return ErroneousMessage(
                "Parameter deserialization error: {} in "
                "message: {}".format(error, json_string)
            )
        m: Message
        if "plural" in obj:
            m = NumericalMessage(
                obj["singular"], obj["plural"], obj["n"], obj.get("domain")
            )
        else:
            m = SimpleMessage(obj["message"], obj.get("domain"))
        m.args = args
        m.kwargs = kwargs
        return m

    def __init__(self, domain: str | None = None):
        self.domain = domain
        self.args: typing.Iterable[Serializable] = ()
        self.kwargs: dict[str, Serializable] = {}

    @abc.abstractmethod
    def _base_dict(self) -> dict[str, typing.Any]:
        ...

    @abc.abstractmethod
    def _gettext(self) -> str:
        ...

    def to_json(self) -> str:
        obj = self._base_dict()
        if self.domain is not None:
            obj["domain"] = self.domain
        if self.args:
            obj["args"] = tuple(serialize_param(a) for a in self.args)
        if self.kwargs:
            obj["kwargs"] = {k: serialize_param(v) for k, v in self.kwargs.items()}
{'fraction_non_alphanumeric': 0.10945821854912764, 'fraction_numerical': 0.003305785123966942, 'mean_word_length': 2.3596545342381248, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 11, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '6589911', 'n_tokens_mistral': 1584, 'n_tokens_neox': 1565, 'n_words': 472}