Added a all other filters
This commit is contained in:
parent
d19a3d2005
commit
bba6168f2b
|
@ -1,4 +1,14 @@
|
|||
from typing import List
|
||||
|
||||
from mastoposter.types import Status
|
||||
from .base import BaseFilter # NOQA
|
||||
from mastoposter.filters.boost import BoostFilter # NOQA
|
||||
from mastoposter.filters.combined import CombinedFilter # NOQA
|
||||
from mastoposter.filters.mention import MentionFilter # NOQA
|
||||
from mastoposter.filters.media import MediaFilter # NOQA
|
||||
from mastoposter.filters.text import TextFilter # NOQA
|
||||
from mastoposter.filters.spoiler import SpoilerFilter # NOQA
|
||||
|
||||
|
||||
def run_filters(filters: List[BaseFilter], status: Status) -> bool:
|
||||
return all((fil(status) for fil in filters))
|
||||
|
|
|
@ -0,0 +1,27 @@
|
|||
from configparser import SectionProxy
|
||||
from typing import Set
|
||||
from mastoposter.filters.base import BaseFilter
|
||||
from mastoposter.types import Status
|
||||
|
||||
|
||||
class MediaFilter(BaseFilter, filter_name="media"):
|
||||
def __init__(self, section: SectionProxy):
|
||||
super().__init__(section)
|
||||
self.valid_media: Set[str] = set(section.get("valid_media").split())
|
||||
self.mode = section.get("mode", "include")
|
||||
if self.mode not in ("include", "exclude", "only"):
|
||||
raise ValueError(f"{self.mode=} is not valid")
|
||||
|
||||
def __call__(self, status: Status) -> bool:
|
||||
if not status.media_attachments:
|
||||
return False
|
||||
|
||||
types: Set[str] = {a.type for a in status.media_attachments}
|
||||
|
||||
if self.mode == "include":
|
||||
return len(types & self.valid_media) > 0
|
||||
elif self.mode == "exclude":
|
||||
return len(types & self.valid_media) == 0
|
||||
elif self.mode == "only":
|
||||
return len((types ^ self.valid_media) & types) == 0
|
||||
raise ValueError(f"{self.mode=} is not valid")
|
|
@ -0,0 +1,13 @@
|
|||
from configparser import SectionProxy
|
||||
from re import Pattern, compile as regexp
|
||||
from mastoposter.filters.base import BaseFilter
|
||||
from mastoposter.types import Status
|
||||
|
||||
|
||||
class SpoilerFilter(BaseFilter, filter_name="spoiler"):
|
||||
def __init__(self, section: SectionProxy):
|
||||
super().__init__(section)
|
||||
self.regexp: Pattern = regexp(section["regexp"])
|
||||
|
||||
def __call__(self, status: Status) -> bool:
|
||||
return self.regexp.match(status.spoiler_text) is not None
|
|
@ -0,0 +1,51 @@
|
|||
from configparser import SectionProxy
|
||||
from re import Pattern, compile as regexp
|
||||
from typing import Optional, Set
|
||||
|
||||
from bs4 import BeautifulSoup, PageElement, Tag
|
||||
from mastoposter.filters.base import BaseFilter
|
||||
from mastoposter.types import Status
|
||||
|
||||
|
||||
class TextFilter(BaseFilter, filter_name="content"):
|
||||
def __init__(self, section: SectionProxy):
|
||||
super().__init__(section)
|
||||
self.mode = section["mode"]
|
||||
self.tags: Set[str] = set()
|
||||
self.regexp: Optional[Pattern] = None
|
||||
|
||||
if self.mode == "regexp":
|
||||
self.regexp = regexp(section["regexp"])
|
||||
elif self.mode == "hashtag":
|
||||
self.tags = set(section["tags"].split())
|
||||
else:
|
||||
raise ValueError(f"Invalid filter mode {self.mode}")
|
||||
|
||||
@classmethod
|
||||
def node_to_text(cls, el: PageElement) -> str:
|
||||
if isinstance(el, Tag):
|
||||
if el.name == "br":
|
||||
return "\n"
|
||||
elif el.name == "p":
|
||||
return (
|
||||
str.join("", map(cls.node_to_text, el.children)) + "\n\n"
|
||||
)
|
||||
return str.join("", map(cls.node_to_text, el.children))
|
||||
return str(el)
|
||||
|
||||
@classmethod
|
||||
def html_to_plain(cls, html: str) -> str:
|
||||
soup = BeautifulSoup(html, "lxml")
|
||||
return cls.node_to_text(soup).rstrip()
|
||||
|
||||
def __call__(self, status: Status) -> bool:
|
||||
source = status.reblog or status
|
||||
if self.regexp is not None:
|
||||
return (
|
||||
self.regexp.match(self.html_to_plain(source.content))
|
||||
is not None
|
||||
)
|
||||
elif self.tags:
|
||||
return len(self.tags & {t.name for t in source.tags}) > 0
|
||||
else:
|
||||
raise ValueError("Neither regexp or tags were set. Why?")
|
|
@ -0,0 +1,12 @@
|
|||
from configparser import SectionProxy
|
||||
from mastoposter.filters.base import BaseFilter
|
||||
from mastoposter.types import Status
|
||||
|
||||
|
||||
class VisibilityFilter(BaseFilter, filter_name="visibility"):
|
||||
def __init__(self, section: SectionProxy):
|
||||
super().__init__(section)
|
||||
self.options = tuple(section["options"].split())
|
||||
|
||||
def __call__(self, status: Status) -> bool:
|
||||
return status.visibility in self.options
|
|
@ -263,6 +263,7 @@ class Status:
|
|||
favourites_count: int
|
||||
replies_count: int
|
||||
mentions: List[Mention]
|
||||
tags: List[Tag]
|
||||
application: Optional[Application] = None
|
||||
url: Optional[str] = None
|
||||
in_reply_to_id: Optional[str] = None
|
||||
|
@ -300,6 +301,7 @@ class Status:
|
|||
language=data.get("language"),
|
||||
text=data.get("text"),
|
||||
mentions=[Mention.from_dict(m) for m in data.get("mentions", [])],
|
||||
tags=[Tag.from_dict(m) for m in data.get("tags", [])],
|
||||
)
|
||||
|
||||
@property
|
||||
|
|
Loading…
Reference in New Issue