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
|
||||
from logging.config import dictConfig
|
||||
from pathlib import Path
|
||||
from typing import Optional
|
||||
|
||||
import pkg_resources
|
||||
from appdirs import AppDirs
|
||||
from dynaconf import Dynaconf, Validator
|
||||
|
||||
|
@ -48,17 +47,18 @@ def init_logging(settings: Optional[Dynaconf] = None):
|
|||
def get_settings_files_paths() -> Optional[str]:
|
||||
|
||||
dirs = AppDirs(appname="mobilizon-reshare", version=current_version())
|
||||
bundled_settings_path = pkg_resources.resource_filename(
|
||||
"mobilizon_reshare", "settings.toml"
|
||||
)
|
||||
for config_path in [
|
||||
Path(dirs.user_config_dir, "mobilizon_reshare.toml").absolute(),
|
||||
Path(dirs.site_config_dir, "mobilizon_reshare.toml").absolute(),
|
||||
bundled_settings_path,
|
||||
]:
|
||||
if config_path and Path(config_path).exists():
|
||||
logger.debug(f"Loading configuration from {config_path}")
|
||||
return config_path
|
||||
bundled_settings_ref = importlib.resources.files(
|
||||
"mobilizon_reshare"
|
||||
) / "settings.toml"
|
||||
with importlib.resources.as_file(bundled_settings_ref) as bundled_settings_path:
|
||||
for config_path in [
|
||||
Path(dirs.user_config_dir, "mobilizon_reshare.toml").absolute(),
|
||||
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}")
|
||||
return config_path
|
||||
|
||||
|
||||
def build_settings(validators: Optional[list[Validator]] = None) -> Dynaconf:
|
||||
|
|
|
@ -1,3 +1,4 @@
|
|||
import importlib
|
||||
import inspect
|
||||
import logging
|
||||
from abc import ABC, abstractmethod
|
||||
|
@ -124,6 +125,33 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
|||
"""
|
||||
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:
|
||||
self._validate_event(event)
|
||||
self._validate_message(self.get_message_from_event(event))
|
||||
|
@ -148,21 +176,20 @@ class AbstractEventFormatter(LoggerMixin, ConfLoaderMixin):
|
|||
"""
|
||||
Retrieves publisher's message template.
|
||||
"""
|
||||
template_path = self.conf.msg_template_path or self.default_template_path
|
||||
return JINJA_ENV.get_template(template_path)
|
||||
return self._get_template(self.conf.msg_template_path, self.get_default_template_path)
|
||||
|
||||
def get_recap_header(self):
|
||||
template_path = (
|
||||
self.conf.recap_header_template_path
|
||||
or self.default_recap_header_template_path
|
||||
def get_recap_header(self) -> Template:
|
||||
return self._get_template(
|
||||
self.conf.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:
|
||||
template_path = (
|
||||
self.conf.recap_template_path or self.default_recap_template_path
|
||||
return self._get_template(
|
||||
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:
|
||||
"""
|
||||
|
|
|
@ -1,7 +1,6 @@
|
|||
from typing import Optional
|
||||
|
||||
import facebook
|
||||
import pkg_resources
|
||||
from facebook import GraphAPIError
|
||||
|
||||
from mobilizon_reshare.dataclasses import MobilizonEvent
|
||||
|
@ -19,19 +18,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
|||
|
||||
|
||||
class FacebookFormatter(AbstractEventFormatter):
|
||||
|
||||
_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:
|
||||
text = event.description
|
||||
|
|
|
@ -1,7 +1,6 @@
|
|||
from typing import Optional
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import pkg_resources
|
||||
import requests
|
||||
from requests import Response
|
||||
|
||||
|
@ -20,19 +19,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
|||
|
||||
|
||||
class MastodonFormatter(AbstractEventFormatter):
|
||||
|
||||
_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:
|
||||
text = event.description
|
||||
|
|
|
@ -1,7 +1,6 @@
|
|||
import re
|
||||
from typing import Optional
|
||||
|
||||
import pkg_resources
|
||||
import requests
|
||||
from bs4 import BeautifulSoup
|
||||
from requests import Response
|
||||
|
@ -20,18 +19,6 @@ from mobilizon_reshare.publishers.exceptions import (
|
|||
|
||||
|
||||
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")
|
||||
|
||||
def _validate_event(self, event: MobilizonEvent) -> None:
|
||||
|
|
|
@ -1,6 +1,5 @@
|
|||
from typing import Optional
|
||||
|
||||
import pkg_resources
|
||||
from tweepy import OAuthHandler, API, TweepyException
|
||||
from tweepy.models import Status
|
||||
|
||||
|
@ -17,19 +16,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
|||
|
||||
|
||||
class TwitterFormatter(AbstractEventFormatter):
|
||||
|
||||
_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:
|
||||
pass # pragma: no cover
|
||||
|
|
|
@ -1,7 +1,6 @@
|
|||
from typing import Optional
|
||||
from urllib.parse import urljoin
|
||||
|
||||
import pkg_resources
|
||||
import requests
|
||||
from requests import Response
|
||||
from requests.auth import HTTPBasicAuth
|
||||
|
@ -23,19 +22,7 @@ from mobilizon_reshare.publishers.exceptions import (
|
|||
|
||||
|
||||
class ZulipFormatter(AbstractEventFormatter):
|
||||
|
||||
_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:
|
||||
text = event.description
|
||||
|
|
|
@ -1,7 +1,7 @@
|
|||
import logging
|
||||
from pathlib import Path
|
||||
|
||||
import pkg_resources
|
||||
import importlib
|
||||
import urllib3.util
|
||||
from aerich import Command
|
||||
from tortoise import Tortoise
|
||||
|
@ -47,9 +47,9 @@ TORTOISE_ORM = get_tortoise_orm()
|
|||
class MoReDB:
|
||||
def get_migration_location(self):
|
||||
scheme = get_db_url().scheme
|
||||
return pkg_resources.resource_filename(
|
||||
"mobilizon_reshare", f"migrations/{scheme}"
|
||||
)
|
||||
scheme_ref = importlib.resources.files("mobilizon_reshare") / "migrations" / f"{scheme}"
|
||||
with importlib.resources.as_file(scheme_ref) as scheme_path:
|
||||
return scheme_path
|
||||
|
||||
async def _implement_db_changes(self):
|
||||
logging.info("Performing aerich migrations.")
|
||||
|
|
Loading…
Reference in New Issue