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
|
# horrible little module that fetches ICS files for the local trash public service.
#
# It tries its best to not overwrite existing ICS files in case the upstream goes down
# or returns empty ICS files.
import sys
import httpx
import asyncio
import icalendar
from datetime import datetime
import syslog
import os.path
# Internal id for the street (extracted from the ics download url)
ortsteil_id = "e9c32ab3-df25-4660-b88e-abda91897d7a"
# They are using a numeric encoding to refer to different kinds of trash
fraktionen = {
"restmüll": "1",
"bio": "5",
"papier": "7",
"gelbe_tonne": "13",
"problemmüllsammlung": "20"
}
def ics_url(year):
frakt = ','.join(fraktionen.values())
return f'https://awido.cubefour.de/Customer/aic-fdb/KalenderICS.aspx?oid={ortsteil_id}&jahr={year}&fraktionen={frakt}&reminder=1.12:00'
def fetchers_for_years(start_year, no_of_years_in_future):
"""given a starting year, and a number of years in the future,
return the years for which to fetch ics files"""
current_year = datetime.now().year
max_year = current_year + no_of_years_in_future
return {
"passed_years": range(start_year, current_year),
"this_and_future_years": range(current_year, 1 + max_year)
}
async def fetch_ics(c, url):
"""fetch an ICS file from an URL"""
try:
resp = await c.get(url)
except Exception as e:
return { "ics_does_not_exist_exc": e }
if resp.is_error:
return { "ics_does_not_exist": resp }
else:
try:
ics = icalendar.Calendar.from_ical(resp.content)
return { "ics": { "ics_parsed": ics, "ics_bytes": resp.content } }
except ValueError as e:
return { "ics_cannot_be_parsed": e }
def ics_has_events(ics):
"""Determine if there is any event in the ICS, otherwise we can assume it’s an empty file"""
for item in ics.walk():
if isinstance(item, icalendar.Event):
return True
return False
async def write_nonempty_ics(directory, year, ics):
# only overwrite if the new ics has any events
if ics_has_events(ics['ics_parsed']):
path = os.path.join(directory, f"{year}.ics")
with open(path, "wb") as f:
f.write(ics['ics_bytes'])
info(f"wrote ics for year {year} to file {path}")
else:
info(f"ics for year {year} was empty, skipping")
def main():
ics_directory = os.getenv("ICS_DIRECTORY", None)
if not ics_directory:
critical("please set ICS_DIRECTORY")
start_year = int(os.getenv("ICS_START_YEAR", 2022))
future_years = int(os.getenv("ICS_FUTURE_YEARS", 2))
years = fetchers_for_years(start_year, no_of_years_in_future=future_years)
async def go():
async with httpx.AsyncClient(follow_redirects=True) as c:
info(f"fetching ics for passed years: {years['passed_years']}")
for year in years["passed_years"]:
match await fetch_ics(c, ics_url(year)):
case { "ics_does_not_exist_exc": error }:
warn(f"The ics for the year {year} is gone, error when requesting: {error} for url {ics_url(year)}")
case { "ics_does_not_exist": resp }:
warn(f"The ics for the year {year} is gone, server returned status {resp.status} for url {ics_url(year)}")
case { "ics_cannot_be_parsed": error }:
warn(f"The returned ICS could not be parsed: {error} for url {ics_url(year)}")
case { "ics": ics }:
info(f"fetched ics from {ics_url(year)}")
await write_nonempty_ics(ics_directory, year, ics)
case _:
critical("unknown case for ics result")
info(f"fetching ics for current and upcoming years: {years['this_and_future_years']}")
for year in years["this_and_future_years"]:
match await fetch_ics(c, ics_url(year)):
case { "ics_does_not_exist_exc": error }:
critical(f"The ics for the year {year} is not available, error when requesting: {error} for url {ics_url(year)}")
case { "ics_does_not_exist": resp }:
critical(f"The ics for the year {year} is not available, server returned status {resp.status} for url {ics_url(year)}")
case { "ics_cannot_be_parsed": error }:
critical(f"The returned ICS could not be parsed: {error} for url {ics_url(year)}")
case { "ics": ics }:
info(f"fetched ics from {ics_url(year)}")
await write_nonempty_ics(ics_directory, year, ics)
case _:
critical("unknown case for ics result")
asyncio.run(go())
def info(msg):
syslog.syslog(syslog.LOG_INFO, msg)
def critical(msg):
syslog.syslog(syslog.LOG_CRIT, msg)
sys.exit(1)
def warn(msg):
syslog.syslog(syslog.LOG_WARNING, msg)
def debug(msg):
syslog.syslog(syslog.LOG_DEBUG, msg)
if __name__ == "__main__":
main()
|
