from dataclasses import dataclass import datetime import re from dataclass_defs import ProcessedRow, SheetsRow INTERVAL_ABBREVS = { "pv": datetime.timedelta(days=1), "vk": datetime.timedelta(days=7), "kk": datetime.timedelta(days=30), "v": datetime.timedelta(days=365), } def _calculate_interval(interval_str: str): match = re.search(r"(\d+)(\w+)", interval_str) number = int(match[1]) abbrev = match[2] try: interval = INTERVAL_ABBREVS[abbrev] except KeyError: raise Exception(f'"{interval_str}" is an invalid interval string!') total_interval = datetime.timedelta(seconds=interval.total_seconds() * number) return total_interval def _timestamp_to_date(timestamp: str) -> datetime.datetime: return datetime.datetime.strptime(timestamp, "%Y-%m-%d") def _date_to_timestamp(date: datetime.datetime) -> str: return datetime.datetime.strftime(date, "%Y-%m-%d") def _process_row(row: SheetsRow): interval = _calculate_interval(row.interval) if row.last_done is not None: last_done_date = _timestamp_to_date(row.last_done) next_date = last_done_date + interval next_date_str = _date_to_timestamp(next_date) else: last_done_date = datetime.datetime.now() next_date = datetime.datetime.now() next_date_str = _date_to_timestamp(next_date) return ProcessedRow( name=row.name, next_datestamp=next_date_str, next_date=next_date, last_datestamp=row.last_done, last_date=last_done_date, interval=row.interval, ) def process_rows(rows: list[SheetsRow]) -> list[ProcessedRow]: return [_process_row(row) for row in rows] def _is_this_week(processed_row: ProcessedRow): next_date = processed_row.next_date today = datetime.datetime.now() current_week_start = today - datetime.timedelta(days=today.weekday()) current_week_end = current_week_start + datetime.timedelta(days=7) return next_date > current_week_start and next_date < current_week_end def filter_only_this_week(rows: list[ProcessedRow]) -> list[ProcessedRow]: return [row for row in rows if _is_this_week(row)]