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
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
|
import os
import re
from datetime import datetime, timezone, timedelta
def parse_timestamp(timestamp, tzinfo=timezone.utc):
"""Parse a ISO 8601 timestamp string.
For naive/unaware dt, assume it is in tzinfo timezone (default: UTC).
"""
dt = datetime.fromisoformat(timestamp)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=tzinfo)
return dt
def parse_local_timestamp(timestamp, tzinfo=None):
"""Parse a ISO 8601 timestamp string.
For naive/unaware dt, assume it is in local timezone.
Convert to tzinfo timezone (the default None means: local timezone).
"""
dt = datetime.fromisoformat(timestamp)
if dt.tzinfo is None:
dt = dt.astimezone(tz=tzinfo)
return dt
def timestamp(s):
"""Convert a --timestamp=s argument to a datetime object"""
try:
# is it pointing to a file / directory?
ts = safe_s(os.stat(s).st_mtime)
return datetime.fromtimestamp(ts, tz=timezone.utc)
except OSError:
# didn't work, try parsing as a ISO timestamp. if no TZ is given, we assume local timezone.
return parse_local_timestamp(s)
# Not too rarely, we get crappy timestamps from the fs, that overflow some computations.
# As they are crap anyway (valid filesystem timestamps always refer to the past up to
# the present, but never to the future), nothing is lost if we just clamp them to the
# maximum value we can support.
# As long as people are using borg on 32bit platforms to access borg archives, we must
# keep this value True. But we can expect that we can stop supporting 32bit platforms
# well before coming close to the year 2038, so this will never be a practical problem.
SUPPORT_32BIT_PLATFORMS = True # set this to False before y2038.
if SUPPORT_32BIT_PLATFORMS:
# second timestamps will fit into a signed int32 (platform time_t limit).
# nanosecond timestamps thus will naturally fit into a signed int64.
# subtract last 48h to avoid any issues that could be caused by tz calculations.
# this is in the year 2038, so it is also less than y9999 (which is a datetime internal limit).
# msgpack can pack up to uint64.
MAX_S = 2**31 - 1 - 48 * 3600
MAX_NS = MAX_S * 1000000000
else:
# nanosecond timestamps will fit into a signed int64.
# subtract last 48h to avoid any issues that could be caused by tz calculations.
# this is in the year 2262, so it is also less than y9999 (which is a datetime internal limit).
# round down to 1e9 multiple, so MAX_NS corresponds precisely to a integer MAX_S.
# msgpack can pack up to uint64.
MAX_NS = (2**63 - 1 - 48 * 3600 * 1000000000) // 1000000000 * 1000000000
MAX_S = MAX_NS // 1000000000
def safe_s(ts):
if 0 <= ts <= MAX_S:
return ts
elif ts < 0:
return 0
else:
return MAX_S
def safe_ns(ts):
if 0 <= ts <= MAX_NS:
return ts
elif ts < 0:
return 0
else:
return MAX_NS
def safe_timestamp(item_timestamp_ns):
t_ns = safe_ns(item_timestamp_ns)
return datetime.fromtimestamp(t_ns / 1e9, timezone.utc) # return tz-aware utc datetime obj
def format_time(ts: datetime, format_spec=""):
"""
Convert *ts* to a human-friendly format with textual weekday.
"""
return ts.strftime("%a, %Y-%m-%d %H:%M:%S %z" if format_spec == "" else format_spec)
def format_timedelta(td):
"""Format timedelta in a human friendly format"""
ts = td.total_seconds()
s = ts % 60
m = int(ts / 60) % 60
h = int(ts / 3600) % 24
txt = "%.3f seconds" % s
if m:
txt = "%d minutes %s" % (m, txt)
if h:
txt = "%d hours %s" % (h, txt)
if td.days:
txt = "%d days %s" % (td.days, txt)
return txt
def calculate_relative_offset(format_string, from_ts, earlier=False):
"""
Calculates offset based on a relative marker. 7d (7 days), 8m (8 months)
earlier: whether offset should be calculated to an earlier time.
"""
if from_ts is None:
from_ts = archive_ts_now()
if format_string is not None:
offset_regex = re.compile(r"(?P<offset>\d+)(?P<unit>[md])")
match = offset_regex.search(format_string)
if match:
unit = match.group("unit")
offset = int(match.group("offset"))
offset *= -1 if earlier else 1
if unit == "d":
return from_ts + timedelta(days=offset)
elif unit == "m":
return offset_n_months(from_ts, offset)
raise ValueError(f"Invalid relative ts offset format: {format_string}")
def offset_n_months(from_ts, n_months):
def get_month_and_year_from_total(total_completed_months):
month = (total_completed_months % 12) + 1
year = total_completed_months // 12
return month, year
# Calculate target month and year by getting completed total_months until target_month
total_months = (from_ts.year * 12) + from_ts.month + n_months - 1
target_month, target_year = get_month_and_year_from_total(total_months)
# calculate the max days of the target month by subtracting a day from the next month
following_month, year_of_following_month = get_month_and_year_from_total(total_months + 1)
max_days_in_month = (datetime(year_of_following_month, following_month, 1) - timedelta(1)).day
return datetime(day=min(from_ts.day, max_days_in_month), month=target_month, year=target_year).replace(
tzinfo=from_ts.tzinfo
)
class OutputTimestamp:
def __init__(self, ts: datetime):
self.ts = ts
def __format__(self, format_spec):
# we want to output a timestamp in the user's local timezone
return format_time(self.ts.astimezone(), format_spec=format_spec)
def __str__(self):
return f"{self}"
def isoformat(self):
# we want to output a timestamp in the user's local timezone
return self.ts.astimezone().isoformat(timespec="microseconds")
to_json = isoformat
def archive_ts_now():
"""return tz-aware datetime obj for current time for usage as archive timestamp"""
return datetime.now(timezone.utc) # utc time / utc timezone
|
