hkc
/
mastoposter
2
0
Fork 2
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Uhh I think I added logging and some other stuff 

Yeah so basically now initialization of integrations and filters is
moved to from_section instead of __init__ because muh better imo
This commit is contained in:
Casey 2022-11-01 12:55:23 +03:00
parent e4cd94b7c3
commit d861b2fe45
Signed by: hkc
GPG Key ID: F0F6CFE11CDB0960
16 changed files with 209 additions and 71 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

14
README.md
View File

@ -175,15 +175,13 @@ expression to match your spoiler text.
Filter to match post content against either a regular expression, or a list of
tags. Matching is done on the plaintext version of the post.
`mode` property determines the type of operation. Can be either `regexp`,
`tag` or `hashtag` (two last ones are the same).
You can have one of two properties (but not both because fuck you): `tags` or
`regexp`. It's obvious what does what: if you have `regexp` set, plaintext
version of status is checked against that regular expression, and if you have
`tags` set, then only statuses that have those tags will be allowed.
In `mode = regexp` you have to specify the `regexp` option with a regular
expression to match against.
In `mode = tag` or `mode = hashtag`, `tags` option should contain a
space-separated list of tags. If any of the tags are present in that list,
filter will be triggered.
Please note that in case of tags, you should NOT use `#` symbol in front of
them.
#### `type = visibility`
Simple filter that just checks for post visibility.

22
mastoposter/__init__.py
View File

@ -1,5 +1,6 @@
from asyncio import gather
from configparser import ConfigParser
from logging import getLogger
from typing import Dict, List, Optional
from mastoposter.filters import run_filters
from mastoposter.filters.base import BaseFilter, FilterInstance
@ -11,33 +12,49 @@ from mastoposter.integrations import (
)
from mastoposter.types import Status
logger = getLogger(__name__)
def load_integrations_from(config: ConfigParser) -> List[FilteredIntegration]:
modules: List[FilteredIntegration] = []
for module_name in config.get("main", "modules").split():
mod = config[f"module/{module_name}"]
logger.info(
"Configuring %s integration (type=%s)", module_name, mod["type"]
)
filters: Dict[str, FilterInstance] = {}
for filter_name in mod.get("filters", "").split():
filter_basename = filter_name.lstrip("~!")
logger.info(
"Loading filter %s for integration %s",
filter_basename,
module_name,
)
filters[filter_basename] = BaseFilter.new_instance(
filter_name, config[f"filter/{filter_basename}"]
)
for finst in list(filters.values()):
logger.info("Running post-initialization hook for %r", finst)
finst.filter.post_init(filters, config)
# TODO: make a registry of integrations
# INFO: mastoposter/integrations/base.py@__init__
if mod["type"] == "telegram":
modules.append(
FilteredIntegration(
TelegramIntegration(mod), list(filters.values())
TelegramIntegration.from_section(mod),
list(filters.values()),
)
)
elif mod["type"] == "discord":
modules.append(
FilteredIntegration(
DiscordIntegration(mod), list(filters.values())
DiscordIntegration.from_section(mod),
list(filters.values()),
)
)
else:
@ -48,6 +65,7 @@ def load_integrations_from(config: ConfigParser) -> List[FilteredIntegration]:
async def execute_integrations(
status: Status, sinks: List[FilteredIntegration]
) -> List[Optional[str]]:
logger.debug("Executing integrations...")
return await gather(
*[sink[0](status) for sink in sinks if run_filters(sink[1], status)],
return_exceptions=True,

32
mastoposter/__main__.py
View File

@ -1,6 +1,7 @@
#!/usr/bin/env python3
from asyncio import run
from configparser import ConfigParser, ExtendedInterpolation
from logging import getLogger
from mastoposter import execute_integrations, load_integrations_from
from mastoposter.integrations import FilteredIntegration
from mastoposter.sources import websocket_source
@ -8,10 +9,14 @@ from typing import AsyncGenerator, Callable, List
from mastoposter.types import Account, Status
from httpx import Client
from mastoposter.utils import normalize_config
WSOCK_TEMPLATE = "wss://{instance}/api/v1/streaming"
VERIFY_CREDS_TEMPLATE = "https://{instance}/api/v1/accounts/verify_credentials"
logger = getLogger()
async def listen(
source: Callable[..., AsyncGenerator[Status, None]],
@ -20,12 +25,19 @@ async def listen(
/,
**kwargs,
):
logger.info("Starting listening...")
async for status in source(**kwargs):
logger.debug("Got status: %r", status)
if status.account.id != user:
continue
# TODO: add option/filter to handle that
if status.visibility in ("direct",):
logger.info(
"Skipping post %s (status.visibility=%r)",
status.uri,
status.visibility,
)
continue
# TODO: find a better way to handle threads
@ -33,6 +45,10 @@ async def listen(
status.in_reply_to_account_id is not None
and status.in_reply_to_account_id != user
):
logger.info(
"Skipping post %s because it's a reply to another person",
status.uri,
)
continue
await execute_integrations(status, drains)
@ -42,21 +58,15 @@ def main(config_path: str):
conf = ConfigParser(interpolation=ExtendedInterpolation())
conf.read(config_path)
for section in conf.sections():
_remove = set()
for k, v in conf[section].items():
normalized_key = k.replace(" ", "_").replace("-", "_")
if k == normalized_key:
continue
conf[section][normalized_key] = v
_remove.add(k)
for k in _remove:
del conf[section][k]
normalize_config(conf)
modules: List[FilteredIntegration] = load_integrations_from(conf)
logger.info("Loaded %d integrations", len(modules))
user_id: str = conf["main"]["user"]
if user_id == "auto":
logger.info("config.main.user is set to auto, getting user ID")
with Client() as c:
rq = c.get(
VERIFY_CREDS_TEMPLATE.format(**conf["main"]),
@ -64,8 +74,10 @@ def main(config_path: str):
)
account = Account.from_dict(rq.json())
user_id = account.id
logger.info("account.id=%s", user_id)
url = "wss://{}/api/v1/streaming".format(conf["main"]["instance"])
run(
listen(
websocket_source,

21
mastoposter/filters/__init__.py
View File

@ -1,3 +1,4 @@
from logging import getLogger
from typing import List
from mastoposter.types import Status
@ -10,8 +11,26 @@ from mastoposter.filters.text import TextFilter # NOQA
from mastoposter.filters.spoiler import SpoilerFilter # NOQA
from mastoposter.filters.visibility import VisibilityFilter # NOQA
logger = getLogger(__name__)
def run_filters(filters: List[FilterInstance], status: Status) -> bool:
logger.debug("Running filters on %r", status)
if not filters:
logger.debug("No filters, returning True")
return True
return all((fil.filter(status) ^ fil.inverse for fil in filters))
results: List[bool] = []
for fil in filters:
result = fil.filter(status)
logger.debug(
"%r -> %r ^ %r -> %r",
fil.filter,
result,
fil.inverse,
result ^ fil.inverse,
)
results.append(result ^ fil.inverse)
logger.debug("Result: %r", all(results))
return all(results)

10
mastoposter/filters/base.py
View File

@ -23,8 +23,8 @@ class BaseFilter(ABC):
filter_name: ClassVar[str] = "_base"
def __init__(self, section: SectionProxy):
UNUSED(section)
def __init__(self):
pass
def __init_subclass__(cls, filter_name: str, **kwargs):
super().__init_subclass__(**kwargs)
@ -51,7 +51,11 @@ class BaseFilter(ABC):
def load_filter(cls, name: str, section: SectionProxy) -> "BaseFilter":
if name not in cls.FILTER_REGISTRY:
raise KeyError(f"no filter with name {name!r} was found")
return cls.FILTER_REGISTRY[name](section)
return cls.FILTER_REGISTRY[name].from_section(section)
@classmethod
def from_section(cls, section: SectionProxy) -> "BaseFilter":
raise NotImplementedError
@classmethod
def new_instance(cls, name: str, section: SectionProxy) -> FilterInstance:

11
mastoposter/filters/boost.py
View File

@ -1,13 +1,18 @@
from configparser import SectionProxy
from fnmatch import fnmatch
from typing import List
from mastoposter.filters.base import BaseFilter
from mastoposter.types import Status
class BoostFilter(BaseFilter, filter_name="boost"):
def __init__(self, section: SectionProxy):
super().__init__(section)
self.list = section.get("list", "").split()
def __init__(self, accounts: List[str]):
super().__init__()
self.list = accounts
@classmethod
def from_section(cls, section: SectionProxy) -> "BoostFilter":
return cls(section["list"].split())
@classmethod
def check_account(cls, acct: str, mask: str) -> bool:

15
mastoposter/filters/combined.py
View File

@ -11,12 +11,19 @@ class CombinedFilter(BaseFilter, filter_name="combined"):
"single": lambda d: sum(d) == 1,
}
def __init__(self, section: SectionProxy):
self.filter_names = section.get("filters", "").split()
self.operator = self.OPERATORS[section.get("operator", "all")]
self._operator_name = section.get("operator", "all")
def __init__(self, filter_names: List[str], operator: str):
self.filter_names = filter_names
self._operator_name = operator
self.operator = self.OPERATORS[self._operator_name]
self.filters: List[FilterInstance] = []
@classmethod
def from_section(cls, section: SectionProxy) -> "CombinedFilter":
return cls(
filter_names=section["filters"].split(),
operator=section.get("operator", "all"),
)
def post_init(
self, filters: Dict[str, FilterInstance], config: ConfigParser
):

24
mastoposter/filters/media.py
View File

@ -1,16 +1,26 @@
from configparser import SectionProxy
from typing import Set
from typing import Literal, 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 __init__(
self,
valid_media: Set[str],
mode: Literal["include", "exclude", "only"],
):
super().__init__()
self.valid_media: Set[str] = valid_media
self.mode = mode
assert self.mode in ("include", "exclude", "only")
@classmethod
def from_section(cls, section: SectionProxy) -> "MediaFilter":
return cls(
valid_media=set(section["valid_media"].split()),
mode=section.get("mode", "include"), # type: ignore
)
def __call__(self, status: Status) -> bool:
if not status.media_attachments:

21
mastoposter/filters/mention.py
View File

@ -1,6 +1,6 @@
from configparser import SectionProxy
from re import Pattern, compile as regexp
from typing import ClassVar
from typing import ClassVar, Set
from fnmatch import fnmatch
from mastoposter.filters.base import BaseFilter
from mastoposter.types import Status
@ -9,22 +9,27 @@ from mastoposter.types import Status
class MentionFilter(BaseFilter, filter_name="mention"):
MENTION_REGEX: ClassVar[Pattern] = regexp(r"@([^@]+)(@([^@]+))?")
def __init__(self, section: SectionProxy):
super().__init__(section)
self.list = section.get("list", "").split()
def __init__(self, accounts: Set[str]):
super().__init__()
self.accounts = accounts
@classmethod
def from_section(cls, section: SectionProxy) -> "MentionFilter":
return cls(set(section.get("list", "").split()))
@classmethod
def check_account(cls, acct: str, mask: str) -> bool:
return fnmatch("@" + acct, mask)
def __call__(self, status: Status) -> bool:
if not self.list and status.mentions:
if not self.accounts and status.mentions:
return True
# XXX: make it better somehow. and faster. and stronger.
return any(
(
any(
self.check_account(mention.acct, mask)
for mask in self.list
for mask in self.accounts
)
for mention in status.mentions
)
@ -32,5 +37,7 @@ class MentionFilter(BaseFilter, filter_name="mention"):
def __repr__(self):
return str.format(
"Filter:{name}({list!r})", name=self.filter_name, list=self.list
"Filter:{name}({list!r})",
name=self.filter_name,
list=self.accounts,
)

10
mastoposter/filters/spoiler.py
View File

@ -5,9 +5,13 @@ from mastoposter.types import Status
class SpoilerFilter(BaseFilter, filter_name="spoiler"):
def __init__(self, section: SectionProxy):
super().__init__(section)
self.regexp: Pattern = regexp(section.get("regexp", "^.*$"))
def __init__(self, regex: str = "^.*$"):
super().__init__()
self.regexp: Pattern = regexp(regex)
@classmethod
def from_section(cls, section: SectionProxy) -> "SpoilerFilter":
return cls(section.get("regexp", section.get("regex", "^.*$")))
def __call__(self, status: Status) -> bool:
return self.regexp.match(status.spoiler_text) is not None

29
mastoposter/filters/text.py
View File

@ -8,18 +8,24 @@ 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
def __init__(
self, regex: Optional[str] = None, tags: Optional[Set[str]] = None
):
super().__init__()
assert regex is not None or tags
if self.mode == "regexp":
self.regexp = regexp(section["regexp"])
elif self.mode in ("hashtag", "tag"):
self.tags = set(map(str.lower, section["tags"].split()))
else:
raise ValueError(f"Invalid filter mode {self.mode}")
self.tags: Optional[Set[str]] = tags
self.regexp: Optional[Pattern] = regexp(regex) if regex else None
@classmethod
def from_section(cls, section: SectionProxy) -> "TextFilter":
if "regexp" in section and "tags" in section:
raise AssertionError("you can't use both tags and regexp")
elif "regexp" in section:
return cls(regex=section["regexp"])
elif "tags" in section:
return cls(tags=set(section["tags"].split()))
raise AssertionError("neither regexp or tags were set")
@classmethod
def node_to_text(cls, el: PageElement) -> str:
@ -63,3 +69,4 @@ class TextFilter(BaseFilter, filter_name="content"):
name=self.filter_name,
tags=self.tags,
)
return "Filter:{name}(invalid)".format(name=self.filter_name)

11
mastoposter/filters/visibility.py
View File

@ -1,12 +1,17 @@
from configparser import SectionProxy
from typing import Set
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 = set(section["options"].split())
def __init__(self, options: Set[str]):
super().__init__()
self.options = options
@classmethod
def from_section(cls, section: SectionProxy) -> "BaseFilter":
return cls(set(section["options"].split()))
def __call__(self, status: Status) -> bool:
return status.visibility in self.options

7
mastoposter/integrations/base.py
View File

@ -6,9 +6,14 @@ from mastoposter.types import Status
class BaseIntegration(ABC):
def __init__(self, section: SectionProxy):
# TODO: make a registry of integrations
def __init__(self):
pass
@classmethod
def from_section(cls, section: SectionProxy) -> "BaseIntegration":
raise NotImplementedError
@abstractmethod
async def __call__(self, status: Status) -> Optional[str]:
raise NotImplementedError

8
mastoposter/integrations/discord/__init__.py
View File

@ -12,8 +12,12 @@ from mastoposter.types import Status
class DiscordIntegration(BaseIntegration):
def __init__(self, section: SectionProxy):
self.webhook = section.get("webhook", "")
def __init__(self, webhook: str):
self.webhook = webhook
@classmethod
def from_section(cls, section: SectionProxy) -> "DiscordIntegration":
return cls(section["webhook"])
async def execute_webhook(
self,

31
mastoposter/integrations/telegram.py
View File

@ -54,12 +54,31 @@ Boost from <a href="{{status.reblog.account.url}}">\
class TelegramIntegration(BaseIntegration):
def __init__(self, sect: SectionProxy):
self.token = sect.get("token", "")
self.chat_id = sect.get("chat", "")
self.silent = sect.getboolean("silent", True)
self.template: Template = Template(
emojize(sect.get("template", DEFAULT_TEMPLATE))
def __init__(
self,
token: str,
chat_id: str,
template: Optional[Template] = None,
silent: bool = True,
):
self.token = token
self.chat_id = chat_id
self.silent = silent
if template is None:
self.template = Template(emojize(DEFAULT_TEMPLATE))
else:
self.template = template
@classmethod
def from_section(cls, section: SectionProxy) -> "TelegramIntegration":
return cls(
token=section["token"],
chat_id=section["chat"],
silent=section.getboolean("silent", True),
template=Template(
emojize(section.get("template", DEFAULT_TEMPLATE))
),
)
async def _tg_request(self, method: str, **kwargs) -> TGResponse:

14
mastoposter/utils.py
View File

@ -1,3 +1,4 @@
from configparser import ConfigParser
from html import escape
from typing import Callable, Dict
from bs4.element import Tag, PageElement
@ -16,6 +17,19 @@ def md_escape(text: str) -> str:
)
def normalize_config(conf: ConfigParser):
for section in conf.sections():
_remove = set()
for k, v in conf[section].items():
normalized_key = k.replace(" ", "_").replace("-", "_")
if k == normalized_key:
continue
conf[section][normalized_key] = v
_remove.add(k)
for k in _remove:
del conf[section][k]
def node_to_html(el: PageElement) -> str:
TAG_TRANSFORMS: Dict[str, Callable[[Tag,], str]] = {
"a": lambda tag: '<a href="{}">{}</a>'.format(