Compare commits
9 Commits
1147ed7d05
...
79946741e9
Author | SHA1 | Date |
---|---|---|
Giacomo Leidi | 79946741e9 | |
Giacomo Leidi | 2ee3e6238a | |
Giacomo Leidi | 95e508f70e | |
Giacomo Leidi | 028590c224 | |
Giacomo Leidi | 5238a3cd18 | |
Giacomo Leidi | dd80299cff | |
Giacomo Leidi | 101fe9ca37 | |
Giacomo Leidi | 2106b99b42 | |
Giacomo Leidi | 7a9aee1819 |
|
@ -1,10 +1,9 @@
|
||||||
import importlib.resources
|
import importlib
|
||||||
import logging
|
import logging
|
||||||
from logging.config import dictConfig
|
from logging.config import dictConfig
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import pkg_resources
|
|
||||||
from appdirs import AppDirs
|
from appdirs import AppDirs
|
||||||
from dynaconf import Dynaconf, Validator
|
from dynaconf import Dynaconf, Validator
|
||||||
|
|
||||||
|
@ -48,17 +47,18 @@ def init_logging(settings: Optional[Dynaconf] = None):
|
||||||
def get_settings_files_paths() -> Optional[str]:
|
def get_settings_files_paths() -> Optional[str]:
|
||||||
|
|
||||||
dirs = AppDirs(appname="mobilizon-reshare", version=current_version())
|
dirs = AppDirs(appname="mobilizon-reshare", version=current_version())
|
||||||
bundled_settings_path = pkg_resources.resource_filename(
|
bundled_settings_ref = importlib.resources.files(
|
||||||
"mobilizon_reshare", "settings.toml"
|
"mobilizon_reshare"
|
||||||
)
|
) / "settings.toml"
|
||||||
for config_path in [
|
with importlib.resources.as_file(bundled_settings_ref) as bundled_settings_path:
|
||||||
Path(dirs.user_config_dir, "mobilizon_reshare.toml").absolute(),
|
for config_path in [
|
||||||
Path(dirs.site_config_dir, "mobilizon_reshare.toml").absolute(),
|
Path(dirs.user_config_dir, "mobilizon_reshare.toml").absolute(),
|
||||||
bundled_settings_path,
|
Path(dirs.site_config_dir, "mobilizon_reshare.toml").absolute(),
|
||||||
]:
|
bundled_settings_path.absolute(),
|
||||||
if config_path and Path(config_path).exists():
|
]:
|
||||||
logger.debug(f"Loading configuration from {config_path}")
|
if config_path and Path(config_path).exists():
|
||||||
return config_path
|
logger.debug(f"Loading configuration from {config_path}")
|
||||||
|
return config_path
|
||||||
|
|
||||||
|
|
||||||
def build_settings(validators: Optional[list[Validator]] = None) -> Dynaconf:
|
def build_settings(validators: Optional[list[Validator]] = None) -> Dynaconf:
|
||||||
|
|
|
@ -1,3 +1,4 @@
|
||||||
|
import importlib
|
||||||
import inspect
|
import inspect
|
||||||
import logging
|
import logging
|
||||||
from abc import ABC, abstractmethod
|
from abc import ABC, abstractmethod
|
||||||
|
@ -124,6 +125,33 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
||||||
"""
|
"""
|
||||||
raise NotImplementedError # pragma: no cover
|
raise NotImplementedError # pragma: no cover
|
||||||
|
|
||||||
|
def _get_name(self) -> str:
|
||||||
|
return self._conf[1]
|
||||||
|
|
||||||
|
|
||||||
|
def _get_template(self, configured_template, default_generator) -> Template:
|
||||||
|
if configured_template:
|
||||||
|
return JINJA_ENV.get_template(configured_template)
|
||||||
|
else:
|
||||||
|
template_ref = default_generator()
|
||||||
|
with importlib.resources.as_file(template_ref) as template_path:
|
||||||
|
return JINJA_ENV.get_template(template_path.as_posix())
|
||||||
|
|
||||||
|
|
||||||
|
def get_default_template_path(self, type=""):
|
||||||
|
return importlib.resources.files(
|
||||||
|
"mobilizon_reshare.publishers.templates"
|
||||||
|
) / f"{self._get_name()}{type}.tmpl.j2"
|
||||||
|
|
||||||
|
|
||||||
|
def get_default_recap_template_path(self):
|
||||||
|
return self.get_default_template_path(type="_recap")
|
||||||
|
|
||||||
|
|
||||||
|
def get_default_recap_header_template_path(self):
|
||||||
|
return self.get_default_template_path(type="_recap_header")
|
||||||
|
|
||||||
|
|
||||||
def validate_event(self, event: _MobilizonEvent) -> None:
|
def validate_event(self, event: _MobilizonEvent) -> None:
|
||||||
self._validate_event(event)
|
self._validate_event(event)
|
||||||
self._validate_message(self.get_message_from_event(event))
|
self._validate_message(self.get_message_from_event(event))
|
||||||
|
@ -148,21 +176,20 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
||||||
"""
|
"""
|
||||||
Retrieves publisher's message template.
|
Retrieves publisher's message template.
|
||||||
"""
|
"""
|
||||||
template_path = self.conf.msg_template_path or self.default_template_path
|
return self._get_template(self.conf.msg_template_path, self.get_default_template_path)
|
||||||
return JINJA_ENV.get_template(template_path)
|
|
||||||
|
|
||||||
def get_recap_header(self):
|
def get_recap_header(self) -> Template:
|
||||||
template_path = (
|
return self._get_template(
|
||||||
self.conf.recap_header_template_path
|
self.conf.recap_header_template_path,
|
||||||
or self.default_recap_header_template_path
|
self.get_default_recap_header_template_path
|
||||||
)
|
)
|
||||||
return JINJA_ENV.get_template(template_path).render()
|
|
||||||
|
|
||||||
def get_recap_fragment_template(self) -> Template:
|
def get_recap_fragment_template(self) -> Template:
|
||||||
template_path = (
|
return self._get_template(
|
||||||
self.conf.recap_template_path or self.default_recap_template_path
|
self.conf.recap_template_path,
|
||||||
|
self.get_default_recap_template_path
|
||||||
)
|
)
|
||||||
return JINJA_ENV.get_template(template_path)
|
|
||||||
|
|
||||||
def get_recap_fragment(self, event: _MobilizonEvent) -> str:
|
def get_recap_fragment(self, event: _MobilizonEvent) -> str:
|
||||||
"""
|
"""
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import facebook
|
import facebook
|
||||||
import pkg_resources
|
|
||||||
from facebook import GraphAPIError
|
from facebook import GraphAPIError
|
||||||
|
|
||||||
from mobilizon_reshare.dataclasses import MobilizonEvent
|
from mobilizon_reshare.dataclasses import MobilizonEvent
|
||||||
|
@ -19,19 +18,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
|
||||||
|
|
||||||
class FacebookFormatter(AbstractEventFormatter):
|
class FacebookFormatter(AbstractEventFormatter):
|
||||||
|
|
||||||
_conf = ("publisher", "facebook")
|
_conf = ("publisher", "facebook")
|
||||||
default_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "facebook.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "facebook_recap.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_header_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "facebook_recap_header.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
text = event.description
|
text = event.description
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from urllib.parse import urljoin
|
from urllib.parse import urljoin
|
||||||
|
|
||||||
import pkg_resources
|
|
||||||
import requests
|
import requests
|
||||||
from requests import Response
|
from requests import Response
|
||||||
|
|
||||||
|
@ -20,19 +19,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
|
||||||
|
|
||||||
class MastodonFormatter(AbstractEventFormatter):
|
class MastodonFormatter(AbstractEventFormatter):
|
||||||
|
|
||||||
_conf = ("publisher", "mastodon")
|
_conf = ("publisher", "mastodon")
|
||||||
default_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "mastodon.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "mastodon_recap.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_header_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "mastodon_recap_header.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
text = event.description
|
text = event.description
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
import re
|
import re
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import pkg_resources
|
|
||||||
import requests
|
import requests
|
||||||
from bs4 import BeautifulSoup
|
from bs4 import BeautifulSoup
|
||||||
from requests import Response
|
from requests import Response
|
||||||
|
@ -20,18 +19,6 @@ from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
|
||||||
|
|
||||||
class TelegramFormatter(AbstractEventFormatter):
|
class TelegramFormatter(AbstractEventFormatter):
|
||||||
default_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "telegram.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "telegram_recap.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_header_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "telegram_recap_header.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
_conf = ("publisher", "telegram")
|
_conf = ("publisher", "telegram")
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
|
|
|
@ -1,6 +1,5 @@
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
|
|
||||||
import pkg_resources
|
|
||||||
from tweepy import OAuthHandler, API, TweepyException
|
from tweepy import OAuthHandler, API, TweepyException
|
||||||
from tweepy.models import Status
|
from tweepy.models import Status
|
||||||
|
|
||||||
|
@ -17,19 +16,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
|
||||||
|
|
||||||
class TwitterFormatter(AbstractEventFormatter):
|
class TwitterFormatter(AbstractEventFormatter):
|
||||||
|
|
||||||
_conf = ("publisher", "twitter")
|
_conf = ("publisher", "twitter")
|
||||||
default_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "twitter.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "twitter_recap.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_header_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "twitter_recap_header.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
pass # pragma: no cover
|
pass # pragma: no cover
|
||||||
|
|
|
@ -1,7 +1,6 @@
|
||||||
from typing import Optional
|
from typing import Optional
|
||||||
from urllib.parse import urljoin
|
from urllib.parse import urljoin
|
||||||
|
|
||||||
import pkg_resources
|
|
||||||
import requests
|
import requests
|
||||||
from requests import Response
|
from requests import Response
|
||||||
from requests.auth import HTTPBasicAuth
|
from requests.auth import HTTPBasicAuth
|
||||||
|
@ -23,19 +22,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
||||||
|
|
||||||
|
|
||||||
class ZulipFormatter(AbstractEventFormatter):
|
class ZulipFormatter(AbstractEventFormatter):
|
||||||
|
|
||||||
_conf = ("publisher", "zulip")
|
_conf = ("publisher", "zulip")
|
||||||
default_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "zulip.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "zulip_recap.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
default_recap_header_template_path = pkg_resources.resource_filename(
|
|
||||||
"mobilizon_reshare.publishers.templates", "zulip_recap_header.tmpl.j2"
|
|
||||||
)
|
|
||||||
|
|
||||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||||
text = event.description
|
text = event.description
|
||||||
|
|
|
@ -1,7 +1,7 @@
|
||||||
import logging
|
import logging
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
|
|
||||||
import pkg_resources
|
import importlib
|
||||||
import urllib3.util
|
import urllib3.util
|
||||||
from aerich import Command
|
from aerich import Command
|
||||||
from tortoise import Tortoise
|
from tortoise import Tortoise
|
||||||
|
@ -47,9 +47,9 @@ TORTOISE_ORM = get_tortoise_orm()
|
||||||
class MoReDB:
|
class MoReDB:
|
||||||
def get_migration_location(self):
|
def get_migration_location(self):
|
||||||
scheme = get_db_url().scheme
|
scheme = get_db_url().scheme
|
||||||
return pkg_resources.resource_filename(
|
scheme_ref = importlib.resources.files("mobilizon_reshare") / "migrations" / f"{scheme}"
|
||||||
"mobilizon_reshare", f"migrations/{scheme}"
|
with importlib.resources.as_file(scheme_ref) as scheme_path:
|
||||||
)
|
return scheme_path
|
||||||
|
|
||||||
async def _implement_db_changes(self):
|
async def _implement_db_changes(self):
|
||||||
logging.info("Performing aerich migrations.")
|
logging.info("Performing aerich migrations.")
|
||||||
|
|
Loading…
Reference in New Issue