update awb_bad_kreuznach to support the new app

This commit is contained in:
5ila5
2023-06-14 20:17:15 +02:00
parent 08a2d96366
commit 283af0c027
2 changed files with 193 additions and 9 deletions

View File

@@ -1,5 +1,6 @@
import datetime
import json
import logging
import requests
from waste_collection_schedule import Collection # type: ignore[attr-defined]
@@ -7,20 +8,175 @@ from waste_collection_schedule import Collection # type: ignore[attr-defined]
TITLE = "AWB Bad Kreuznach"
DESCRIPTION = "Source for AWB Bad Kreuznach."
URL = "https://app.awb-bad-kreuznach.de/"
NEW_URL = "https://blupassionsystem.de/city/rest/garbageregion/filterRegion"
TEST_CASES = {
"Hargesheim": {"ort": "Hargesheim", "strasse": "Winzenheimer Straße", "nummer": 16}
"Hargesheim": {"ort": "Hargesheim"},
"Bad Kreuznach": {"ort": "Bad Kreuznach", "strasse": "adalbert-stifter-straße", "nummer": "3"},
"Bad Kreuznach OT Bad Münster-Ebernburg": {"ort": "Bad Kreuznach OT Bad Münster-Ebernburg", "strasse": "am Götzenfels"},
"Stromberg": {"ort": "Stromberg", "stadtteil": "Schindeldorf"},
}
TYPES = ("restmuell", "bio", "wert", "papier")
ICON_MAP = {
"Restabfall": "mdi:trash-can",
"Bioabfall": "mdi:leaf",
"Gelber Sack/Gelbe Tonne": "mdi:sack",
"Papiertonne": "mdi:package-variant",
"Bildschirm-/Kühlgeräte": "mdi:television-classic",
"Schadstoffsammlung": "mdi:biohazard",
"altmetalle": "mdi:nail",
}
LOGGER = logging.getLogger(__name__)
def compare_str(a: str, b: str):
return a.lower().replace(" ", "").replace("-", "") == b.lower().replace("-", "").replace(" ", "")
class Source:
def __init__(self, ort, strasse, nummer):
def __init__(self, ort, strasse=None, nummer=None, stadtteil=None):
self._stadtteil = stadtteil if stadtteil else None
self._ort = ort
self._strasse = strasse
self._nummer = nummer
self._strasse = strasse if strasse else None
self._nummer = str(nummer) if nummer else None
def fetch_new_app(self):
params = {"active": True, "appId": 44, "hugeDataInfo": "", "regionId": None,
"cityId": None, "streetId": None, "partOfCityId": None, "districtId": None}
r = requests.post(NEW_URL, json=params)
r.raise_for_status()
data = r.json()
if not ("data" in data and "citys" in data["data"]):
raise ValueError("Unexpected response")
if "regions" in data["data"]:
params["regionId"] = data["data"]["regions"][0]["id"]
r = requests.post(NEW_URL, json=params)
r.raise_for_status()
data = r.json()
found = False
for city in data["data"]["citys"]:
if compare_str(city["name"], self._ort):
params["cityId"] = city["id"]
found = True
break
if not found:
raise ValueError("Ort not found")
r = requests.post(NEW_URL, json=params)
r.raise_for_status()
data = r.json()
if data["data"]["partOfCitys"] != []:
if self._stadtteil == None:
raise ValueError("Stadtteil required")
elif self._stadtteil != None:
LOGGER.warning("stadtteil provided but not needed")
if data["data"]["partOfCitys"]:
found = False
for part_of_city in data["data"]["partOfCitys"]:
if compare_str(part_of_city["name"], self._stadtteil):
params["partOfCityId"] = part_of_city["id"]
found = True
break
if not found:
raise ValueError("Stadtteil not found")
r = requests.post(NEW_URL, json=params)
r.raise_for_status()
data = r.json()
if data["data"]["streets"] != []:
if self._strasse == None:
raise ValueError("strasse required")
elif self._strasse != None:
LOGGER.warning("strasse provided but not needed")
if data["data"]["streets"]:
found = False
for street in data["data"]["streets"]:
if compare_str(street["name"], self._strasse):
params["streetId"] = street["id"]
found = True
break
if not found:
raise ValueError("Strasse not found")
r = requests.post(NEW_URL, json=params)
r.raise_for_status()
data = r.json()
if data["data"]["houseNumbers"] != []:
if self._nummer == None:
raise ValueError("nummer required")
elif self._nummer != None:
LOGGER.warning("nummer provided but not needed")
if data["data"]["houseNumbers"]:
found = False
for house_number in data["data"]["houseNumbers"]:
if compare_str(str(house_number["name"]), str(self._nummer)):
params["streetId"] = house_number["id"]
found = True
break
if not found:
raise ValueError("Nummer not found")
from_time = datetime.datetime.now()
from_time = from_time.replace(
hour=0, minute=0, second=0, microsecond=0)
from_time = from_time - datetime.timedelta(days=1)
to_time = from_time + datetime.timedelta(days=365)
params["fromTime"] = int(from_time.timestamp() * 1000)
params["toTime"] = int(to_time.timestamp() * 1000)
header = {"Accept": "application/json, text/plain, */*"}
# r = requests.get("https://blupassionsystem.de/city/rest/garbagestation/getNameRegion", params=params, headers=header)
# r.raise_for_status()
r = requests.get(
"https://blupassionsystem.de/city/rest/garbagestation/getAllGarbageCalendar", params=params, headers=header)
r.raise_for_status()
data = r.json()
if "data" not in data or "calendars" not in data["data"]:
raise ValueError("Unexpected response")
entries = []
for cal_entry in data["data"]["calendars"]:
entries.append(Collection(datetime.date.fromtimestamp(
cal_entry["fromDate"] / 1000), cal_entry["name"], ICON_MAP.get(cal_entry["name"])))
return entries
def fetch(self):
try:
to_return = self.fetch_new_app()
except Exception as e:
LOGGER.warning(
"Error fetching new app: with Exception '%s' using old method", e)
try:
to_return = self.old_fetch()
except Exception as e2:
LOGGER.warning(
"Error fetching old app: with Exception '%s' raising Exception from new API", e2)
raise e
return to_return
def old_fetch(self):
args = {
"ort": self._ort,
"strasse": self._strasse,

View File

@@ -1,6 +1,6 @@
# AWB Bad Kreuznach
Support for schedules provided by [AWB Bad Kreuznach](https://app.awb-bad-kreuznach.de/), Germany.
Support for schedules provided by [AWB Bad Kreuznach](https://abfall-app-bad-kreuznach), Germany.
## Configuration via configuration.yaml
@@ -20,10 +20,15 @@ waste_collection_schedule:
*(string) (required)*
**strasse**
*(string) (required)*
*(string) (optional)*
**nummer**
*(integer) (required)*
*(string|integer) (optional)*
**stadtteil**
*(string) (optional)*
`strasse`, `nummer` and `stadtteil` are only needed if the web interface needs them.
## Example
@@ -33,12 +38,35 @@ waste_collection_schedule:
- name: awb_bad_kreuznach_de
args:
ort: "Hargesheim"
strasse: "Winzenheimer Straße"
nummer: 16
```
```yaml
waste_collection_schedule:
sources:
- name: awb_bad_kreuznach_de
args:
ort: "Bad Kreuznach"
strasse: "adalbert-stifter-straße"
nummer: 3
```
```yaml
waste_collection_schedule:
sources:
- name: awb_bad_kreuznach_de
args:
ort: "Stromberg"
stadtteil: "Schindeldorf"
```
## How to get the source arguments
1. Go to your calendar at `https://abfall-app-bad-kreuznach`.
2. Enter your location.
3. Copy the exact values from the select boxes as `stadt`, `stadtteil`, `strasse` and `nummer` in the source configuration (`stadtteil`, `strasse` and `nummer` are only needed for some addresses).
### Old App
1. Go to your calendar at `https://app.awb-bad-kreuznach.de/`.
2. Enter your location.
3. Copy the exact values from the 3 textboxes as `ort`, `strasse` and `nummer` in the source configuration.