Added combined filter

This commit is contained in:
Casey 2022-08-27 15:47:51 +03:00
parent ae8a1ddf34
commit 4d7a9be45f
Signed by: hkc
GPG Key ID: F0F6CFE11CDB0960
2 changed files with 49 additions and 6 deletions

View File

@ -1,19 +1,18 @@
from abc import ABC, abstractmethod
from configparser import SectionProxy
from typing import ClassVar, Dict, Type
from mastoposter.types import Status
from re import Pattern, compile as regexp
UNUSED = lambda *_: None # NOQA
class BaseFilter(ABC):
FILTER_REGISTRY: ClassVar[Dict[str, Type["BaseFilter"]]] = {}
FILTER_NAME_REGEX: Pattern = regexp(r"^([a-z_]+)$")
def __init__(self):
pass
@abstractmethod
def __call__(self, status: Status) -> bool:
raise NotImplementedError
def __init__(self, section: SectionProxy):
UNUSED(section)
def __init_subclass__(cls, filter_name: str, **kwargs):
super().__init_subclass__(**kwargs)
@ -22,3 +21,10 @@ class BaseFilter(ABC):
if filter_name in cls.FILTER_REGISTRY:
raise KeyError(f"{filter_name=!r} is already registered")
cls.FILTER_REGISTRY[filter_name] = cls
@abstractmethod
def __call__(self, status: Status) -> bool:
raise NotImplementedError
def post_init(self, filters: Dict[str, "BaseFilter"]):
UNUSED(filters)

View File

@ -0,0 +1,37 @@
from configparser import SectionProxy
from typing import Callable, ClassVar, Dict, List, NamedTuple
from functools import reduce
from mastoposter.filters.base import BaseFilter
from mastoposter.types import Status
class FilterType(NamedTuple):
inverse: bool
filter: BaseFilter
class CombinedFilter(BaseFilter, filter_name="combined"):
OPERATORS: ClassVar[Dict[str, Callable]] = {
"and": lambda a, b: a and b,
"or": lambda a, b: a or b,
"xor": lambda a, b: a ^ b,
}
def __init__(self, section: SectionProxy):
self.filter_names = section.get("filters", "").split()
self.operator = self.OPERATORS[section.get("operator", "and")]
self.filters: List[FilterType] = []
def post_init(self, filters: Dict[str, "BaseFilter"]):
super().post_init(filters)
for filter_name in self.filter_names:
self.filters.append(
FilterType(
filter_name[:1] in "~!", # inverse
filters[filter_name.rstrip("!~")],
)
)
def __call__(self, status: Status) -> bool:
results = [fil.filter(status) ^ fil.inverse for fil in self.filters]
return reduce(self.operator, results)