From 6a30da929153f5710fa8cd0c6be3e824e9cd9916 Mon Sep 17 00:00:00 2001 From: Brunno Vanelli Date: Sun, 12 May 2024 17:29:36 +0200 Subject: [PATCH] feat: Add stringification to Schedule and include evaluation on rules. --- actual/rules.py | 2 + actual/schedules.py | 158 ++++++++++++++++++++++++++++++++++------ tests/test_schedules.py | 123 +++++++++++++++++++++++++++---- 3 files changed, 246 insertions(+), 37 deletions(-) diff --git a/actual/rules.py b/actual/rules.py index 175a76f..1e4c7ae 100644 --- a/actual/rules.py +++ b/actual/rules.py @@ -129,6 +129,8 @@ def condition_evaluation( # Actual uses two days as reference # https://github.com/actualbudget/actual/blob/98a7aac73667241da350169e55edd2fc16a6687f/packages/loot-core/src/server/accounts/rules.ts#L302-L304 interval = datetime.timedelta(days=2) + if isinstance(self_value, Schedule): + return self_value.is_approx(true_value, interval) else: # Actual uses 7.5% of the value as threshold # https://github.com/actualbudget/actual/blob/243703b2f70532ec1acbd3088dda879b5d07a5b3/packages/loot-core/src/shared/rules.ts#L261-L263 diff --git a/actual/schedules.py b/actual/schedules.py index da23d0d..111a5a3 100644 --- a/actual/schedules.py +++ b/actual/schedules.py @@ -1,5 +1,6 @@ import datetime import enum +import typing import pydantic from dateutil.rrule import ( @@ -14,6 +15,22 @@ ) +def date_to_datetime(date: typing.Optional[datetime.date]) -> typing.Optional[datetime.datetime]: + """Converts one object from date to datetime object. The reverse is possible directly by calling datetime.date().""" + if date is None: + return None + return datetime.datetime.combine(date, datetime.time.min) + + +def day_to_ordinal(day: int) -> str: + """Converts an integer day to an ordinal number, i.e. 1 -> 1st, 32 -> 32nd""" + if 11 <= (day % 100) <= 13: + suffix = "th" + else: + suffix = ["th", "st", "nd", "rd", "th"][min(day % 10, 4)] + return f"{day}{suffix}" + + class EndMode(enum.Enum): AFTER_N_OCCURRENCES = "after_n_occurrences" ON_DATE = "on_date" @@ -26,7 +43,7 @@ class Frequency(enum.Enum): MONTHLY = "monthly" YEARLY = "yearly" - def as_dateutil(self): + def as_dateutil(self) -> int: frequency_map = {"YEARLY": YEARLY, "MONTHLY": MONTHLY, "WEEKLY": WEEKLY, "DAILY": DAILY} return frequency_map[self.name] @@ -52,19 +69,44 @@ def as_dateutil(self) -> weekday: class Pattern(pydantic.BaseModel): + model_config = pydantic.ConfigDict(validate_assignment=True) + value: int type: PatternType + def __str__(self) -> str: + if self.value == -1: + qualifier = "last" + else: + qualifier = day_to_ordinal(self.value) + type_str = "" + if self.type != PatternType.DAY: + type_str = f" {self.type.name.lower().capitalize()}" + elif self.value == -1: + type_str = " day" + return f"{qualifier}{type_str}" + class Schedule(pydantic.BaseModel): + """ + Implements basic schedules. They are described in https://actualbudget.org/docs/budgeting/schedules/ + + Schedules are part of a rule which then compares if the date found would fit within the schedule by the .is_approx() + method. If it does, and the other conditions match, the transaction will then be linked with the schedule id + (stored in the database). + """ + + model_config = pydantic.ConfigDict(validate_assignment=True) + start: datetime.date = pydantic.Field(..., description="Start date of the schedule.") interval: int = pydantic.Field(1, description="Repeat every interval at frequency unit.") frequency: Frequency = pydantic.Field(Frequency.MONTHLY, description="Unit for the defined interval.") patterns: list[Pattern] = pydantic.Field(default_factory=list) skip_weekend: bool = pydantic.Field( - alias="skipWeekend", description="If should move schedule before or after a weekend." + False, alias="skipWeekend", description="If should move schedule before or after a weekend." ) weekend_solve_mode: WeekendSolveMode = pydantic.Field( + WeekendSolveMode.AFTER, alias="weekendSolveMode", description="When skipping weekend, the value should be set before or after the weekend interval.", ) @@ -74,14 +116,65 @@ class Schedule(pydantic.BaseModel): description="If the schedule should run forever or end at a certain date or number of occurrences.", ) end_occurrences: int = pydantic.Field( - WeekendSolveMode.AFTER, alias="endOccurrences", description="Number of occurrences before the schedule ends." + 1, alias="endOccurrences", description="Number of occurrences before the schedule ends." ) - end_date: datetime.date = pydantic.Field(alias="endDate") - - def is_approx(self, date: datetime.date) -> bool: - pass + end_date: datetime.date = pydantic.Field(None, alias="endDate") + + def __str__(self) -> str: + # evaluate frequency: handle the case where DAILY convert to 'dai' instead of 'day' + interval = "day" if self.frequency == Frequency.DAILY else self.frequency.value.rstrip("ly") + frequency = interval if self.interval == 1 else f"{self.interval} {interval}s" + # evaluate + if self.frequency == Frequency.YEARLY: + target = f" on {self.start.strftime('%b %d')}" + elif self.frequency == Frequency.MONTHLY: + if not self.patterns: + target = f" on the {day_to_ordinal(self.start.day)}" + else: + patterns_str = [] + for pattern in self.patterns: + patterns_str.append(str(pattern)) + target = " on the " + ", ".join(patterns_str) + elif self.frequency == Frequency.WEEKLY: + target = f" on {self.start.strftime('%A')}" + else: # DAILY + target = "" + # end date part + if self.end_mode == EndMode.ON_DATE: + end = f", until {self.end_date}" + elif self.end_mode == EndMode.AFTER_N_OCCURRENCES: + end = ", once" if self.end_occurrences == 1 else f", {self.end_occurrences} times" + else: + end = "" + # weekend skips + move = f" ({self.weekend_solve_mode.value} weekend)" if self.skip_weekend else "" + return f"Every {frequency}{target}{end}{move}" + + @pydantic.model_validator(mode="after") + def validate_end_date(self): + if self.end_mode == EndMode.ON_DATE and self.end_date is None: + raise ValueError("endDate cannot be 'None' when ") + if self.end_date is None: + self.end_date = self.start + return self + + def is_approx(self, date: datetime.date, interval: datetime.timedelta = datetime.timedelta(days=2)) -> bool: + """This function checks if the input date could fit inside of this schedule. It will use the interval as the + maximum threshold before and after the specified date to look for. This defaults on Actual to 2 days.""" + if date < self.start or (self.end_mode == EndMode.ON_DATE and self.end_date < date): + return False + before = self.before(date) + after = self.xafter(date, 1) + if before and (before - interval <= date <= before + interval): + return True + if after and (after[0] - interval <= date <= after[0] + interval): + return True + return False def rruleset(self) -> rruleset: + """Returns the rruleset from dateutil library. This is used internally to calculate the schedule dates. + + For information on how to use this object check the official documentation https://dateutil.readthedocs.io""" rule_sets_configs = [] config = dict(freq=self.frequency.as_dateutil(), dtstart=self.start, interval=self.interval) # add termination options @@ -96,11 +189,14 @@ def rruleset(self) -> rruleset: by_month_day.append(p.value) else: # it's a weekday by_weekday.append(p.type.as_dateutil()(p.value)) + # for the month or weekday rules, add a different rrule to the ruleset. This is because otherwise the rule + # would only look for, for example, days that are 15 that are also Fridays, and that is not desired if by_month_day: monthly_config = config.copy() | {"bymonthday": by_month_day} rule_sets_configs.append(monthly_config) if by_weekday: rule_sets_configs.append(config.copy() | {"byweekday": by_weekday}) + # if ruleset does not contain multiple rules, add the current rule as default if not rule_sets_configs: rule_sets_configs.append(config) # create rule set @@ -109,6 +205,33 @@ def rruleset(self) -> rruleset: rs.rrule(rrule(**cfg)) return rs + def do_skip_weekend( + self, dt_start: datetime.datetime, value: datetime.datetime + ) -> typing.Optional[datetime.datetime]: + if value.weekday() in (5, 6) and self.skip_weekend: + if self.weekend_solve_mode == WeekendSolveMode.AFTER: + value = value + datetime.timedelta(days=7 - value.weekday()) + if self.end_mode == EndMode.ON_DATE and value > date_to_datetime(self.end_date): + return None + else: # BEFORE + value_after = value - datetime.timedelta(days=value.weekday() - 4) + if value_after < dt_start: + # value is in the past, skip and look for another + return None + value = value_after + return value + + def before(self, date: datetime.date = None) -> typing.Optional[datetime.date]: + if not date: + date = datetime.date.today() + dt_start = date_to_datetime(date) + # we also always use the day before since today can also be a valid entry for our time + rs = self.rruleset() + before_datetime = rs.before(dt_start) + if not before_datetime: + return None + return self.do_skip_weekend(dt_start, before_datetime).date() + def xafter(self, date: datetime.date = None, count: int = 1) -> list[datetime.date]: if not date: date = datetime.date.today() @@ -117,22 +240,11 @@ def xafter(self, date: datetime.date = None, count: int = 1) -> list[datetime.da # we also always use the day before since today can also be a valid entry for our time rs = self.rruleset() - ret, i = [], 0 - for value in rs: - value: datetime.datetime - if value.weekday() in (5, 6) and self.skip_weekend: - if self.weekend_solve_mode == WeekendSolveMode.AFTER: - value = value + datetime.timedelta(days=7 - value.weekday()) - else: # BEFORE - value_after = value - datetime.timedelta(days=value.weekday() - 4) - if value_after < dt_start: - # value is in the past, skip and look for another - continue - value = value_after - i += 1 - dt = value - # convert back to date - ret.append(dt.date()) + ret = [] + for value in rs.xafter(dt_start, count, inc=True): + if value := self.do_skip_weekend(dt_start, value): + # convert back to date + ret.append(value.date()) if len(ret) == count: break return sorted(ret) diff --git a/tests/test_schedules.py b/tests/test_schedules.py index 14265f9..47cac18 100644 --- a/tests/test_schedules.py +++ b/tests/test_schedules.py @@ -1,9 +1,32 @@ -import datetime +from datetime import date -from actual.schedules import Schedule +import pytest + +from actual.schedules import Schedule, date_to_datetime def test_basic_schedules(): + s = Schedule.parse_obj( + { + "start": "2024-05-12", + "frequency": "monthly", + "skipWeekend": False, + "endMode": "after_n_occurrences", + "endOccurrences": 3, + "interval": 1, + } + ) + assert s.before(date(2024, 5, 13)) == date(2024, 5, 12) + assert s.xafter(date(2024, 5, 12), 4) == [ + date(2024, 5, 12), + date(2024, 6, 12), + date(2024, 7, 12), + ] + + assert str(s) == "Every month on the 12th, 3 times" + + +def test_complex_schedules(): s = Schedule.parse_obj( { "start": "2024-05-08", @@ -23,20 +46,92 @@ def test_basic_schedules(): "interval": 1, } ) - assert s.xafter(datetime.date(2024, 5, 10), count=5) == [ - datetime.date(2024, 5, 10), - datetime.date(2024, 5, 13), - datetime.date(2024, 5, 27), - datetime.date(2024, 5, 31), - datetime.date(2024, 6, 5), + assert s.xafter(date(2024, 5, 10), count=5) == [ + date(2024, 5, 10), + date(2024, 5, 13), + date(2024, 5, 27), + date(2024, 5, 31), + date(2024, 6, 5), ] # change the solve mode to before s.weekend_solve_mode = "before" - assert s.xafter(datetime.date(2024, 5, 10), count=5) == [ - datetime.date(2024, 5, 10), + assert s.xafter(date(2024, 5, 10), count=5) == [ + date(2024, 5, 10), # according to frontend, this entry happens twice - datetime.date(2024, 5, 10), - datetime.date(2024, 5, 24), - datetime.date(2024, 5, 31), - datetime.date(2024, 6, 5), + date(2024, 5, 10), + date(2024, 5, 24), + date(2024, 5, 31), + date(2024, 6, 5), ] + + assert str(s) == "Every month on the last Sunday, 2nd Saturday, 10th, 31st, 5th (before weekend)" + + +def test_is_approx(): + # create schedule for every 1st and last day of the month (30th or 31st) + s = Schedule.parse_obj( + { + "start": "2024-05-10", + "frequency": "monthly", + "patterns": [ + {"value": 1, "type": "day"}, + {"value": -1, "type": "day"}, + ], + "skipWeekend": True, + "weekendSolveMode": "after", + "endMode": "on_date", + "endOccurrences": 1, + "endDate": "2024-07-01", + "interval": 1, + } + ) + # make sure the xafter is correct + assert s.xafter(date(2024, 6, 1), 5) == [ + date(2024, 6, 3), + date(2024, 7, 1), + date(2024, 7, 1), + ] + # compare is_approx + assert s.is_approx(date(2024, 5, 1)) is False # before starting period + assert s.is_approx(date(2024, 5, 30)) is True + assert s.is_approx(date(2024, 5, 31)) is True + assert s.is_approx(date(2024, 6, 1)) is True + assert s.is_approx(date(2024, 6, 3)) is True # because 1st is also included + + # 30th June is a sunday, so the right date would be 1st of June + assert s.is_approx(date(2024, 6, 28)) is False + assert s.is_approx(date(2024, 6, 30)) is True + assert s.is_approx(date(2024, 7, 1)) is True + + # after end date we reject everything + assert s.is_approx(date(2024, 7, 2)) is False + assert s.is_approx(date(2024, 7, 31)) is False + + assert str(s) == "Every month on the 1st, last day, until 2024-07-01 (after weekend)" + + +def test_date_to_datetime(): + dt = date(2024, 5, 1) + assert date_to_datetime(dt).date() == dt + assert date_to_datetime(None) is None + + +def test_exceptions(): + with pytest.raises(ValueError): + # on_date is set but no date is provided + Schedule.parse_obj( + { + "start": "2024-05-12", + "frequency": "monthly", + "skipWeekend": False, + "endMode": "on_date", + "endOccurrences": 3, + "interval": 1, + } + ) + + +def test_strings(): + assert str(Schedule(start="2024-05-12", frequency="yearly")) == "Every year on May 12" + assert str(Schedule(start="2024-05-12", frequency="weekly")) == "Every week on Sunday" + assert str(Schedule(start="2024-05-12", frequency="daily")) == "Every day"