1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
|
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 = None
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)]
def updated_last_done_datestamps(rows: list[ProcessedRow]) -> list[str]:
return [
row.next_datestamp if _is_this_week(row) else row.last_datestamp for row in rows
]
|
