Skip to content

Commit

Permalink
awb_main_bingen_de now get all ical links (next year)
Browse files Browse the repository at this point in the history
  • Loading branch information
@5ila5
5ila5 committed Dec 28, 2023
1 parent eb20bc9 commit af5d191
Showing 1 changed file with 60 additions and 46 deletions.
106 changes: 60 additions & 46 deletions ...ponents/waste_collection_schedule/waste_collection_schedule/source/awb_mainz_bingen_de.py
View file
Original file line number Diff line number Diff line change
@@ -1,18 +1,18 @@
import re

import requests
from bs4 import BeautifulSoup
from waste_collection_schedule import Collection # type: ignore[attr-defined]
from waste_collection_schedule.service.ICS import ICS

from bs4 import BeautifulSoup
import re

TITLE = "Abfallwirtschaftsbetrieb LK Mainz-Bingen"
DESCRIPTION = "Source for Abfallwirtschaftsbetrieb LK Mainz-Bingen."
URL = "https://www.awb-mainz-bingen.de/"
TEST_CASES = {
"Stadt Ingelheim Ingelheim Süd Albert-Schweitzer-Straße": {
"bezirk": "Stadt Ingelheim",
"ort": "Ingelheim Süd",
"strasse": "Albert-Schweitzer-Straße"
"strasse": "Albert-Schweitzer-Straße",
},
"Verbandsgemeinde Rhein-Selz, Mommenheim": {
"bezirk": "Verbandsgemeinde Rhein-Selz",
Expand All @@ -21,7 +21,7 @@
"Stadt Bingen, Bingen-Stadt, Martinstraße (Haus-Nr.: 5 - 11, 10 - 18)": {
"bezirk": "Stadt Bingen",
"ort": "Bingen-Stadt",
"strasse": "Martinstraße (Haus-Nr.: 5 - 11, 10 - 18)"
"strasse": "Martinstraße (Haus-Nr.: 5 - 11, 10 - 18)",
},
}

Expand All @@ -40,10 +40,10 @@


class Source:
def __init__(self, bezirk: str, ort: str, strasse: str = None):
def __init__(self, bezirk: str, ort: str, strasse: str | None = None):
self._bezirk: str = bezirk
self._ort: str = ort
self._strasse: str = strasse
self._strasse: str | None = strasse
self._ics = ICS()

def fetch(self):
Expand All @@ -55,28 +55,32 @@ def fetch(self):

soup = BeautifulSoup(r.text, "html.parser")
bezirk = soup.find("select", {"id": "Abfuhrbezirk"}).find(
"option", text=re.compile(re.escape(self._bezirk), re.IGNORECASE))
"option", text=re.compile(re.escape(self._bezirk), re.IGNORECASE)
)

if not bezirk:
found = [i.text for i in soup.find_all("option")][1:]
raise Exception(
f"No matching bezirk found search for: {self._bezirk} found: {str(found)}")
f"No matching bezirk found search for: {self._bezirk} found: {str(found)}"
)

bezirk_id = bezirk.get("value")

# set arguemnts to imitate xajax call
# set arguments to imitate xajax call
xjxargs_string = "<xjxobj>"
for key, value in {
"Abfuhrbezirk": "{bezirk_id}",
"Ortschaft": "{ort_id}",
"Strasse": "{strasse_id}",
}.items():
xjxargs_string += "<e><k>"+key+"</k><v>S"+value+"</v></e>"
xjxargs_string += "<e><k>" + key + "</k><v>S" + value + "</v></e>"
xjxargs_string += "</xjxobj>"

args = {
"xjxfun": "show_ortsteil_dropdown",
"xjxargs[]": xjxargs_string.format(bezirk_id=bezirk_id, ort_id=0, strasse_id=0),
"xjxargs[]": xjxargs_string.format(
bezirk_id=bezirk_id, ort_id=0, strasse_id=0
),
}

# send request to get dropdown with for ort id
Expand All @@ -85,24 +89,29 @@ def fetch(self):

soup = BeautifulSoup(r.text, "xml")
teilorte_div = soup.find("cmd", {"id": "divTeilort"})

if not teilorte_div:
raise Exception("invalid resonse from server", soup)

teilorte = BeautifulSoup(teilorte_div.text.replace(
"<![CDATA[", "").replace("]]>", ""), "html.parser")
raise Exception("invalid response from server", soup)

ort = teilorte.find("option", text=re.compile(
re.escape(self._ort), re.IGNORECASE))
teilorte = BeautifulSoup(
teilorte_div.text.replace("<![CDATA[", "").replace("]]>", ""), "html.parser"
)

ort = teilorte.find(
"option", text=re.compile(re.escape(self._ort), re.IGNORECASE)
)
if not ort:
raise Exception(
f"No matching ort found. Searched for: {self._ort}. Found {str([i.text for i in teilorte.find_all('option')][1:])})")
f"No matching ort found. Searched for: {self._ort}. Found {str([i.text for i in teilorte.find_all('option')][1:])})"
)

ort_id = ort.get("value")

args = {
"xjxfun": "show_strasse_dropdown_or_abfuhrtermine",
"xjxargs[]": xjxargs_string.format(bezirk_id=bezirk_id, ort_id=ort_id, strasse_id=0),
"xjxargs[]": xjxargs_string.format(
bezirk_id=bezirk_id, ort_id=ort_id, strasse_id=0
),
}

r = session.post(API_URL, data=args)
Expand All @@ -112,27 +121,33 @@ def fetch(self):
div_strasse = soup.find("cmd", {"id": "divStrasse"})

if not div_strasse:
raise Exception("invalid resonse from server")
raise Exception("invalid response from server")

strassen_soup = BeautifulSoup(div_strasse.text.replace(
"<![CDATA[", "").replace("]]>", ""), "html.parser")
strassen_soup = BeautifulSoup(
div_strasse.text.replace("<![CDATA[", "").replace("]]>", ""), "html.parser"
)

# If strasse is needed
if strassen_soup.find("option"):
if not self._strasse:
raise Exception("Street needed but not provided")

# get strasse id
strasse_id = strassen_soup.find("option", text=re.compile(
re.escape(self._strasse), re.IGNORECASE))
strasse_id = strassen_soup.find(
"option", text=re.compile(re.escape(self._strasse), re.IGNORECASE)
)
if not strasse_id:
found = [i.text for i in strassen_soup.find_all("option")][1:]
raise Exception(
f"Street wanted but no matching street found. Searched for: {self._strasse}. Found {str(found)})")
f"Street wanted but no matching street found. Searched for: {self._strasse}. Found {str(found)})"
)

strasse_id = strasse_id.get("value")
xjxargs = {"bezirk_id": bezirk_id,
"ort_id": ort_id, "strasse_id": strasse_id}
xjxargs = {
"bezirk_id": bezirk_id,
"ort_id": ort_id,
"strasse_id": strasse_id,
}
args = {
"xjxfun": "show_abfuhrtermine",
"xjxargs[]": xjxargs_string.format(**xjxargs),
Expand All @@ -147,26 +162,25 @@ def fetch(self):

if not cal_wrapper:
raise Exception("No calendar found", r.text)
cal_soup = BeautifulSoup(cal_wrapper.text.replace(
"<![CDATA[", "").replace("]]>", ""), "html.parser")
cal_soup = BeautifulSoup(
cal_wrapper.text.replace("<![CDATA[", "").replace("]]>", ""), "html.parser"
)

entries = []
# get ical file url
ical_path = cal_soup.find(
"a", {"href": re.compile("ical")}).get("href")

# get ical file
r = requests.get(API_URL+ical_path)
r.raise_for_status()
r.encoding = "utf-8"
for ical_path in cal_soup.findAll("a", {"href": re.compile("ical")}):
ical_path = ical_path.get("href")
# get ical file
r = requests.get(API_URL + ical_path)
r.raise_for_status()
r.encoding = "utf-8"

# remove DURATION because the returned icalendar has invalid DURATION syntax
ical_string = re.sub(r'^DURATION.*\n?', "", r.text, flags=re.MULTILINE)
# remove DURATION because the returned icalendar has invalid DURATION syntax
ical_string = re.sub(r"^DURATION.*\n?", "", r.text, flags=re.MULTILINE)

dates = self._ics.convert(ical_string)
entries = []
for d in dates:
bin_type = d[1].split(" am ")[0].replace(
"Abfuhrtermin", "").strip()
entries.append(Collection(d[0], bin_type, ICON_MAP.get(bin_type)))
dates = self._ics.convert(ical_string)
for d in dates:
bin_type = d[1].split(" am ")[0].replace("Abfuhrtermin", "").strip()
entries.append(Collection(d[0], bin_type, ICON_MAP.get(bin_type)))

return entries

0 comments on commit af5d191

Please sign in to comment.