mirror of
https://github.com/Tech-Workers-Coalition-Italia/mobilizon-reshare.git
synced 2025-02-11 09:10:51 +01:00
4dc1e4080a
* filtering publications with inactive publishers * filtering publications with inactive publishers * WIP: Generate publications at runtime. TODO: - change `MobilizonEvent.compute_status`'s contract and break everything - while we're at it we should remove `PublicationStatus.WAITING` - test `storage.query.create_publications_for_publishers` * cli: inspect_events: Unnest if-then-else. * publishers: abstract: Remove `EventPublication.make`. * fixed tests * split query.py file * added tests for get_unpublished_events * added tests * more tests * added start test * main: start: Remove filter_publications_with_inactive_publishers. Co-authored-by: Giacomo Leidi <goodoldpaul@autistici.org>
120 lines
3.6 KiB
Python
120 lines
3.6 KiB
Python
import logging
|
|
from abc import ABC, abstractmethod
|
|
from typing import List, Optional
|
|
|
|
import arrow
|
|
|
|
from mobilizon_reshare.config.config import get_settings
|
|
from mobilizon_reshare.event.event import MobilizonEvent
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
class EventSelectionStrategy(ABC):
|
|
def select(
|
|
self,
|
|
published_events: List[MobilizonEvent],
|
|
unpublished_events: List[MobilizonEvent],
|
|
) -> Optional[MobilizonEvent]:
|
|
|
|
if not self.is_in_publishing_window():
|
|
logger.info("Outside of publishing window, no event will be published.")
|
|
return None
|
|
selected = self._select(published_events, unpublished_events)
|
|
if selected:
|
|
return selected[0]
|
|
else:
|
|
return None
|
|
|
|
def is_in_publishing_window(self) -> bool:
|
|
settings = get_settings()
|
|
window_beginning = settings["publishing"]["window"]["begin"]
|
|
window_end = settings["publishing"]["window"]["end"]
|
|
now_hour = arrow.now().datetime.hour
|
|
if window_beginning <= window_end:
|
|
return window_beginning <= now_hour < window_end
|
|
else:
|
|
return now_hour >= window_beginning or now_hour < window_end
|
|
|
|
@abstractmethod
|
|
def _select(
|
|
self,
|
|
published_events: List[MobilizonEvent],
|
|
unpublished_events: List[MobilizonEvent],
|
|
) -> Optional[List[MobilizonEvent]]:
|
|
pass
|
|
|
|
|
|
class SelectNextEventStrategy(EventSelectionStrategy):
|
|
def _select(
|
|
self,
|
|
published_events: List[MobilizonEvent],
|
|
unpublished_events: List[MobilizonEvent],
|
|
) -> Optional[List[MobilizonEvent]]:
|
|
|
|
# if there are no unpublished events, there's nothing I can do
|
|
if not unpublished_events:
|
|
logger.debug("No event to publish.")
|
|
return []
|
|
|
|
# if there's no published event (first execution) I return the next in queue
|
|
if not published_events:
|
|
logger.debug(
|
|
"First Execution with an available event. Picking next event in the queue."
|
|
)
|
|
return unpublished_events
|
|
|
|
last_published_event = published_events[-1]
|
|
now = arrow.now()
|
|
last_published_event_most_recent_publication_time = max(
|
|
last_published_event.publication_time.values()
|
|
)
|
|
|
|
assert last_published_event_most_recent_publication_time < now, (
|
|
f"Last published event has been published in the future\n"
|
|
f"{last_published_event_most_recent_publication_time}\n"
|
|
f"{now}"
|
|
)
|
|
if (
|
|
last_published_event_most_recent_publication_time.shift(
|
|
minutes=get_settings()[
|
|
"selection.strategy_options.break_between_events_in_minutes"
|
|
]
|
|
)
|
|
> now
|
|
):
|
|
|
|
logger.debug(
|
|
"Last event was published recently. No event is going to be published."
|
|
)
|
|
return []
|
|
|
|
return unpublished_events
|
|
|
|
|
|
STRATEGY_NAME_TO_STRATEGY_CLASS = {"next_event": SelectNextEventStrategy}
|
|
|
|
|
|
def select_unpublished_events(
|
|
published_events: List[MobilizonEvent],
|
|
unpublished_events: List[MobilizonEvent],
|
|
):
|
|
|
|
strategy = STRATEGY_NAME_TO_STRATEGY_CLASS[
|
|
get_settings()["selection"]["strategy"]
|
|
]()
|
|
|
|
return strategy._select(published_events, unpublished_events)
|
|
|
|
|
|
def select_event_to_publish(
|
|
published_events: List[MobilizonEvent],
|
|
unpublished_events: List[MobilizonEvent],
|
|
):
|
|
|
|
strategy = STRATEGY_NAME_TO_STRATEGY_CLASS[
|
|
get_settings()["selection"]["strategy"]
|
|
]()
|
|
|
|
return strategy.select(published_events, unpublished_events)
|