mirror of
https://github.com/sascha-hemi/hacs_waste_collection_schedule.git
synced 2026-03-21 05:06:33 +01:00
fix abfallnavi_de
abfallnavi.de changed their service api. Now it is no sufficient any more to use the extracted street or house number id to fetch the dates because the ids frequently change. Therefore it is now mandatory to specify city, street and house number and the scraper has to query the current id every time.
This commit is contained in:
mampfes
@@ -0,0 +1,155 @@
|
||||
#!/usr/bin/env python3
|
||||
|
||||
import requests
|
||||
import json
|
||||
import datetime
|
||||
|
||||
SERVICE_DOMAINS = {
|
||||
"aachen": "Aachen",
|
||||
"zew2": "AWA Entsorgungs GmbH",
|
||||
"aw-bgl2": "Bergisch Gladbach",
|
||||
"bav": "Bergischer Abfallwirtschaftverbund",
|
||||
"din": "Dinslaken",
|
||||
"dorsten": "Dorsten",
|
||||
"gt2": "Gütersloh",
|
||||
"hlv": "Halver",
|
||||
"coe": "Kreis Coesfeld",
|
||||
"krhs": "Kreis Heinsberg",
|
||||
"pi": "Kreis Pinneberg",
|
||||
"krwaf": "Kreis Warendorf",
|
||||
"lindlar": "Lindlar",
|
||||
"stl": "Lüdenscheid",
|
||||
"nds": "Norderstedt",
|
||||
"nuernberg": "Nürnberg",
|
||||
"roe": "Roetgen",
|
||||
"wml2": "EGW Westmünsterland",
|
||||
}
|
||||
|
||||
class AbfallnaviDe(object):
|
||||
def __init__(self, service_domain):
|
||||
self._service_domain = service_domain
|
||||
self._service_url = f"https://{service_domain}-abfallapp.regioit.de/abfall-app-{service_domain}/rest"
|
||||
|
||||
def _fetch(self, path, **kwargs):
|
||||
r = requests.get(f"{self._service_url}/{path}", **kwargs)
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
return r.text
|
||||
|
||||
def _fetch_json(self, path, **kwargs):
|
||||
return json.loads(self._fetch(path, **kwargs))
|
||||
|
||||
def get_cities(self):
|
||||
"""Return all cities of service domain."""
|
||||
cities = self._fetch_json("orte")
|
||||
result = {}
|
||||
for city in cities:
|
||||
result[city["id"]] = city["name"]
|
||||
return result
|
||||
|
||||
def get_city_id(self, city):
|
||||
"""Return id for given city string."""
|
||||
cities = self.get_cities()
|
||||
return self._find_in_inverted_dict(cities, city)
|
||||
|
||||
def get_streets(self, city_id):
|
||||
"""Return all streets of a city."""
|
||||
streets = self._fetch_json(f"orte/{city_id}/strassen")
|
||||
result = {}
|
||||
for street in streets:
|
||||
result[street["id"]] = street["name"]
|
||||
return result
|
||||
|
||||
def get_street_id(self, city_id, street):
|
||||
"""Return id for given street string."""
|
||||
streets = self.get_streets(city_id)
|
||||
return self._find_in_inverted_dict(streets, street)
|
||||
|
||||
def get_house_numbers(self, street_id):
|
||||
"""Return all house numbers of a street."""
|
||||
house_numbers = self._fetch_json(f"strassen/{street_id}")
|
||||
result = {}
|
||||
for hausNr in house_numbers.get("hausNrList", {}):
|
||||
# {"id":5985445,"name":"Adalbert-Stifter-Straße","hausNrList":[{"id":5985446,"nr":"1"},
|
||||
result[hausNr["id"]] = hausNr["nr"]
|
||||
return result
|
||||
|
||||
def get_house_number_id(self, street_id, house_number):
|
||||
"""Return id for given house number string."""
|
||||
house_numbers = self.get_house_numbers(street_id)
|
||||
return self._find_in_inverted_dict(house_numbers, house_number)
|
||||
|
||||
def get_waste_types(self):
|
||||
waste_types = self._fetch_json("fraktionen")
|
||||
result = {}
|
||||
for waste_type in waste_types:
|
||||
result[waste_type["id"]] = waste_type["name"]
|
||||
return result
|
||||
|
||||
def _get_dates(self, target, id, waste_types=None):
|
||||
# retrieve appointments
|
||||
args = []
|
||||
|
||||
if waste_types is None:
|
||||
waste_types = self.get_waste_types()
|
||||
|
||||
for f in waste_types.keys():
|
||||
args.append(("fraktion", f))
|
||||
|
||||
r = requests.get(f"{self._service_url}/{target}/{id}/termine", params=args)
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
results = json.loads(r.text)
|
||||
|
||||
entries = []
|
||||
for r in results:
|
||||
date = datetime.datetime.strptime(r["datum"], "%Y-%m-%d").date()
|
||||
fraktion = waste_types[r["bezirk"]["fraktionId"]]
|
||||
entries.append([date, fraktion])
|
||||
return entries
|
||||
|
||||
def get_dates_by_street_id(self, street_id):
|
||||
return self._get_dates("strassen", street_id, waste_types=None)
|
||||
|
||||
def get_dates_by_house_number_id(self, house_number_id):
|
||||
return self._get_dates("hausnummern", house_number_id, waste_types=None)
|
||||
|
||||
def get_dates(self, city, street, house_number=None):
|
||||
"""Convenient function to get dates by strings only."""
|
||||
# find city_id
|
||||
city_id = self.get_city_id(city)
|
||||
if city_id is None:
|
||||
raise Exception(f"No id found for city: {city}")
|
||||
|
||||
# find street_id
|
||||
street_id = self.get_street_id(city_id, street)
|
||||
if street_id is None:
|
||||
raise Exception(f"No id found for street: {street}")
|
||||
|
||||
# find house_number_id (which is optional: not all house number do have an id)
|
||||
house_number_id = self.get_house_number_id(street_id, house_number)
|
||||
|
||||
# return dates for specific house number of street if house number
|
||||
# doesn't have an own id
|
||||
if house_number_id is not None:
|
||||
return self.get_dates_by_house_number_id(house_number_id)
|
||||
else:
|
||||
return self.get_dates_by_street_id(street_id)
|
||||
|
||||
|
||||
def _find_in_inverted_dict(self, mydict, value):
|
||||
inverted_dict = dict(map(reversed, mydict.items()))
|
||||
return inverted_dict.get(value)
|
||||
|
||||
|
||||
def main():
|
||||
aachen = AbfallnaviDe("aachen")
|
||||
print(aachen.get_dates("Aachen", "Abteiplatz", "7"))
|
||||
|
||||
lindlar = AbfallnaviDe("lindlar")
|
||||
print(lindlar.get_dates("Lindlar", "Aggerweg"))
|
||||
|
||||
roe = AbfallnaviDe("roe")
|
||||
print(roe.get_dates("Roetgen", "Am Sportplatz", "2"))
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
|
||||
@@ -1,10 +1,7 @@
|
||||
import requests
|
||||
import datetime
|
||||
import icalendar
|
||||
import json
|
||||
from collections import OrderedDict
|
||||
|
||||
from ..helpers import CollectionAppointment
|
||||
from ..service.AbfallnaviDe import AbfallnaviDe
|
||||
|
||||
|
||||
DESCRIPTION = "Source for AbfallNavi (= regioit.de) based services"
|
||||
@@ -13,43 +10,25 @@ TEST_CASES = OrderedDict(
|
||||
[
|
||||
(
|
||||
"Aachen, Abteiplatz 7",
|
||||
{"service": "aachen", "strasse": 6654812, "hausnummer": 6654817},
|
||||
{"service": "aachen", "ort": "Aachen", "strasse": "Abteiplatz", "hausnummer": "7"},
|
||||
),
|
||||
("Lindlar, Aggerweg", {"service": "lindlar", "strasse": 63202}),
|
||||
("Roetgen, Am Sportplatz 2", {"service": "roe", "strasse": 52073}),
|
||||
("Lindlar, Aggerweg", {"service": "lindlar", "ort": "Lindlar", "strasse": "Aggerweg"}),
|
||||
("Roetgen, Am Sportplatz 2", {"service": "roe", "ort": "Roetgen", "strasse": "Am Sportplatz", "hausnummer": "2"}),
|
||||
]
|
||||
)
|
||||
|
||||
|
||||
class Source:
|
||||
def __init__(self, service, strasse, hausnummer=None):
|
||||
self._url = f"https://{service}-abfallapp.regioit.de/abfall-app-{service}/rest"
|
||||
if hausnummer is not None:
|
||||
self._url += f"/hausnummern/{hausnummer}"
|
||||
else:
|
||||
self._url += f"/strassen/{strasse}"
|
||||
def __init__(self, service, ort, strasse, hausnummer=None):
|
||||
self._api = AbfallnaviDe(service)
|
||||
self._ort = ort
|
||||
self._strasse = strasse
|
||||
self._hausnummer = hausnummer
|
||||
|
||||
def fetch(self):
|
||||
# get fraktionen
|
||||
r = requests.get(f"{self._url}/fraktionen")
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
fraktionen_list = json.loads(r.text)
|
||||
fraktionen = {}
|
||||
for fraktion in fraktionen_list:
|
||||
fraktionen[fraktion["id"]] = fraktion["name"]
|
||||
|
||||
# retrieve appointments
|
||||
args = []
|
||||
for f in fraktionen.keys():
|
||||
args.append(("fraktion", f))
|
||||
|
||||
r = requests.get(f"{self._url}/termine", params=args)
|
||||
results = json.loads(r.text)
|
||||
dates = self._api.get_dates(self._ort, self._strasse, self._hausnummer)
|
||||
|
||||
entries = []
|
||||
for r in results:
|
||||
date = datetime.datetime.strptime(r["datum"], "%Y-%m-%d").date()
|
||||
fraktion = fraktionen[r["bezirk"]["fraktionId"]]
|
||||
entries.append(CollectionAppointment(date, fraktion))
|
||||
|
||||
for d in dates:
|
||||
entries.append(CollectionAppointment(d[0], d[1]))
|
||||
return entries
|
||||
|
||||
@@ -4,86 +4,77 @@ import inquirer
|
||||
import requests
|
||||
import json
|
||||
|
||||
import sys
|
||||
import os
|
||||
|
||||
PACKAGE_PARENT = '..'
|
||||
SCRIPT_DIR = os.path.dirname(os.path.realpath(os.path.join(os.getcwd(), os.path.expanduser(__file__))))
|
||||
sys.path.append(os.path.normpath(os.path.join(SCRIPT_DIR, PACKAGE_PARENT)))
|
||||
|
||||
from service.AbfallnaviDe import AbfallnaviDe, SERVICE_DOMAINS
|
||||
|
||||
def convert_dict_to_array(d):
|
||||
a = []
|
||||
for item in d.items():
|
||||
a.append((item[1], item[0]))
|
||||
return a
|
||||
|
||||
|
||||
def main():
|
||||
# select service
|
||||
service_choices = [
|
||||
("Aachen", "aachen"),
|
||||
("AWA Entsorgungs GmbH", "zew2"),
|
||||
("Bergisch Gladbach", "aw-bgl2"),
|
||||
("Bergischer Abfallwirtschaftverbund", "bav"),
|
||||
("Dinslaken", "din"),
|
||||
("Dorsten", "dorsten"),
|
||||
("Gütersloh", "gt2"),
|
||||
("Halver", "hlv"),
|
||||
("Kreis Coesfeld", "coe"),
|
||||
("Kreis Heinsberg", "krhs"),
|
||||
("Kreis Pinneberg", "pi"),
|
||||
("Kreis Warendorf", "krwaf"),
|
||||
("Lindlar", "lindlar"),
|
||||
("Lüdenscheid", "stl"),
|
||||
("Norderstedt", "nds"),
|
||||
("Nürnberg", "nuernberg"),
|
||||
("Roetgen", "roe"),
|
||||
("EGW Westmünsterland", "wml2"),
|
||||
]
|
||||
args = {}
|
||||
|
||||
# select service domain
|
||||
questions = [
|
||||
inquirer.List(
|
||||
"service",
|
||||
choices=service_choices,
|
||||
"service_id",
|
||||
choices=convert_dict_to_array(SERVICE_DOMAINS),
|
||||
message="Select service provider for district [Landkreis]",
|
||||
)
|
||||
]
|
||||
answers = inquirer.prompt(questions)
|
||||
service_id = inquirer.prompt(questions)["service_id"]
|
||||
args["service"] = service_id
|
||||
|
||||
SERVICE_URL = f"https://{answers['service']}-abfallapp.regioit.de/abfall-app-{answers['service']}"
|
||||
# create service
|
||||
api = AbfallnaviDe(service_id)
|
||||
|
||||
# get cities
|
||||
r = requests.get(f"{SERVICE_URL}/rest/orte")
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
cities = json.loads(r.text)
|
||||
city_choices = []
|
||||
for city in cities:
|
||||
city_choices.append((city["name"], city["id"]))
|
||||
SERVICE_URL = f"https://{service_id}-abfallapp.regioit.de/abfall-app-{service_id}"
|
||||
|
||||
# select city
|
||||
cities = api.get_cities()
|
||||
questions = [
|
||||
inquirer.List(
|
||||
"city_id", choices=city_choices, message="Select municipality [Kommune/Ort]"
|
||||
"city_id",
|
||||
choices=convert_dict_to_array(cities),
|
||||
message="Select municipality [Kommune/Ort]"
|
||||
)
|
||||
]
|
||||
ort = inquirer.prompt(questions)["city_id"]
|
||||
|
||||
# get streets
|
||||
r = requests.get(f"{SERVICE_URL}/rest/orte/{ort}/strassen")
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
streets = json.loads(r.text)
|
||||
street_choices = []
|
||||
for street in streets:
|
||||
street_choices.append((street["name"], street["id"]))
|
||||
city_id = inquirer.prompt(questions)["city_id"]
|
||||
args["ort"] = cities[city_id]
|
||||
|
||||
# select street
|
||||
streets = api.get_streets(city_id)
|
||||
questions = [
|
||||
inquirer.List("strasse", choices=street_choices, message="Select street")
|
||||
inquirer.List(
|
||||
"street_id",
|
||||
choices=convert_dict_to_array(streets),
|
||||
message="Select street"
|
||||
)
|
||||
]
|
||||
answers.update(inquirer.prompt(questions))
|
||||
street_id = inquirer.prompt(questions)["street_id"]
|
||||
args["strasse"] = streets[street_id]
|
||||
|
||||
# get list of house numbers
|
||||
r = requests.get(f"{SERVICE_URL}/rest/strassen/{answers['strasse']}")
|
||||
r.encoding = "utf-8" # requests doesn't guess the encoding correctly
|
||||
house_numbers = json.loads(r.text)
|
||||
house_number_choices = []
|
||||
for hausNr in house_numbers.get("hausNrList", {}):
|
||||
# {"id":5985445,"name":"Adalbert-Stifter-Straße","hausNrList":[{"id":5985446,"nr":"1"},
|
||||
house_number_choices.append((hausNr["nr"], hausNr["id"]))
|
||||
|
||||
if len(house_number_choices) > 0:
|
||||
house_numbers = api.get_house_numbers(street_id)
|
||||
if len(house_numbers) > 0:
|
||||
questions = [
|
||||
inquirer.List(
|
||||
"hausnummer",
|
||||
choices=house_number_choices,
|
||||
"house_number_id",
|
||||
choices=convert_dict_to_array(house_numbers),
|
||||
message="Select house number",
|
||||
)
|
||||
]
|
||||
answers.update(inquirer.prompt(questions))
|
||||
house_number_id = inquirer.prompt(questions)["house_number_id"]
|
||||
args["hausnummer"] = house_numbers[house_number_id]
|
||||
|
||||
print("Copy the following statements into your configuration.yaml:\n")
|
||||
print("# waste_collection_schedule source configuration")
|
||||
@@ -91,7 +82,7 @@ def main():
|
||||
print(" sources:")
|
||||
print(" - name: abfallnavi_de")
|
||||
print(" args:")
|
||||
for key, value in answers.items():
|
||||
for key, value in args.items():
|
||||
print(f" {key}: {value}")
|
||||
|
||||
|
||||
|
||||
@@ -10,6 +10,7 @@ waste_collection_schedule:
|
||||
- name: abfallnavi_de
|
||||
args:
|
||||
service: SERVICE
|
||||
ort: SERVICE
|
||||
strasse: STRASSE
|
||||
hausnummer: hausnummer
|
||||
```
|
||||
@@ -19,11 +20,14 @@ waste_collection_schedule:
|
||||
**service**<br>
|
||||
*(string) (required)*
|
||||
|
||||
**ort**<br>
|
||||
*(string) (required)*
|
||||
|
||||
**strasse**<br>
|
||||
*(integer) (required)*
|
||||
*(string) (required)*
|
||||
|
||||
**hausnummer**<br>
|
||||
*(integer) (optional)*
|
||||
*(string) (optional)*
|
||||
|
||||
## Example
|
||||
|
||||
@@ -32,8 +36,9 @@ waste_collection_schedule:
|
||||
sources:
|
||||
- name: abfallnavi_de
|
||||
args:
|
||||
service: lindlar
|
||||
strasse: 53585
|
||||
service: coe
|
||||
ort: Coesfeld
|
||||
strasse: Ahornweg
|
||||
```
|
||||
|
||||
## How to get the source arguments
|
||||
@@ -42,4 +47,4 @@ There is a script with an interactive command line interface which generates the
|
||||
|
||||
[https://github.com/mampfes/hacs_waste_collection_schedule/blob/master/custom_components/waste_collection_schedule/package/wizard/abfallnavi_de.py](https://github.com/mampfes/hacs_waste_collection_schedule/blob/master/custom_components/waste_collection_schedule/package/wizard/abfallnavi_de.py).
|
||||
|
||||
Just run this script from a shell and answer the questions.
|
||||
Just run this script from a shell and answer the questions.
|
||||
|
||||
Reference in New Issue
Block a user