WhatTheMike/Fox-V3
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Use customdate, check expired before scheduling

Browse Source
This commit is contained in:
bobloy 2020-11-25 10:15:36 -05:00
parent d85f166062
commit 51dc2e62d4
1 changed files with 21 additions and 8 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

29
fifo/task.py
View File

@ -9,10 +9,12 @@ from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.date import DateTrigger from apscheduler.triggers.date import DateTrigger
from apscheduler.triggers.interval import IntervalTrigger from apscheduler.triggers.interval import IntervalTrigger
from discord.utils import time_snowflake from discord.utils import time_snowflake
from pytz import timezone import pytz
from redbot.core import Config, commands from redbot.core import Config, commands
from redbot.core.bot import Red from redbot.core.bot import Red
from fifo.date_trigger import CustomDateTrigger
log = logging.getLogger("red.fox_v3.fifo.task") log = logging.getLogger("red.fox_v3.fifo.task")
@ -26,7 +28,7 @@ def get_trigger(data):
return IntervalTrigger(days=parsed_time.days, seconds=parsed_time.seconds) return IntervalTrigger(days=parsed_time.days, seconds=parsed_time.seconds)
if data["type"] == "date": if data["type"] == "date":
return DateTrigger(data["time_data"], timezone=data["tzinfo"]) return CustomDateTrigger(data["time_data"], timezone=data["tzinfo"])
if data["type"] == "cron": if data["type"] == "cron":
return CronTrigger.from_crontab(data["time_data"], timezone=data["tzinfo"]) return CronTrigger.from_crontab(data["time_data"], timezone=data["tzinfo"])
@ -34,14 +36,25 @@ def get_trigger(data):
return False return False
def check_expired_trigger(trigger: BaseTrigger):
return trigger.get_next_fire_time(None, datetime.now(pytz.utc)) is not None
def parse_triggers(data: Union[Dict, None]): def parse_triggers(data: Union[Dict, None]):
if data is None or not data.get("triggers", False): # No triggers if data is None or not data.get("triggers", False): # No triggers
return None return None
if len(data["triggers"]) > 1: # Multiple triggers if len(data["triggers"]) > 1: # Multiple triggers
return OrTrigger([get_trigger(t_data) for t_data in data["triggers"]]) triggers_list = [get_trigger(t_data) for t_data in data["triggers"]]
triggers_list = [t for t in triggers_list if not check_expired_trigger(t)]
if not triggers_list:
return None
return OrTrigger(triggers_list)
else: else:
return get_trigger(data["triggers"][0]) trigger = get_trigger(data["triggers"][0])
if check_expired_trigger(trigger):
return None
return trigger
class FakeMessage: class FakeMessage:
@ -66,11 +79,11 @@ def neuter_message(message: FakeMessage):
class Task: class Task:
default_task_data = {"triggers": [], "command_str": ""} default_task_data = {"triggers": [], "command_str": "", "expired_triggers": []}
default_trigger = { default_trigger = {
"type": "", "type": "",
"time_data": None, # Used for Interval and Date Triggers "time_data": None,
"tzinfo": None, "tzinfo": None,
} }
@ -138,7 +151,7 @@ class Task:
# First decode timezone if there is one # First decode timezone if there is one
if t["tzinfo"] is not None: if t["tzinfo"] is not None:
t["tzinfo"] = timezone(t["tzinfo"]) t["tzinfo"] = pytz.timezone(t["tzinfo"])
if t["type"] == "interval": # Convert into timedelta if t["type"] == "interval": # Convert into timedelta
t["time_data"] = timedelta(**t["time_data"]) t["time_data"] = timedelta(**t["time_data"])
@ -174,7 +187,7 @@ class Task:
await self._decode_time_triggers() await self._decode_time_triggers()
return self.data return self.data
async def get_triggers(self) -> List[Union[IntervalTrigger, DateTrigger]]: async def get_triggers(self) -> List[BaseTrigger]:
if not self.data: if not self.data:
await self.load_from_config() await self.load_from_config()