citiesapps_com: Update to citiesapp v2 api (#3461)

* use citiesapp api v2

* update wizard to be compatible with citiesapp v2 updates

* rework to be able to also support cities that still use garbage calendars in api v1

* cr changes, add suggestion to re-check app if no calendar was found, update Rudersdorf test case

* reformatting

---------

Co-authored-by: 5ila5 <[email protected]>

.pre-commit-config.yaml

@@ -27,7 +27,7 @@ repos:
     hooks:
       - id: flake8
         args:
-          - --ignore=D100,D101,D102,D103,D104,D105,D107,E501,W503
+          - --ignore=D100,D101,D102,D103,D104,D105,D106,D107,E501,W503
         additional_dependencies:
           - flake8-docstrings==1.5.0
           - pydocstyle==5.0.2

custom_components/waste_collection_schedule/waste_collection_schedule/service/CitiesAppsCom.py

@@ -3,9 +3,7 @@
 import urllib.parse
 
 import requests
-from waste_collection_schedule.exceptions import (
-    SourceArgumentNotFoundWithSuggestions,
-)
+from waste_collection_schedule.exceptions import SourceArgumentNotFoundWithSuggestions
 
 SERVICE_MAP = [
     {
@@ -1418,7 +1416,7 @@ def __init__(self, password, email=None, phone=None) -> None:
         if email is not None and phone is not None:
             raise Exception("Only provide one of email or phone not both")
 
-        # get authentication as guest
+        # get authentication
         self._session = requests.Session()
         self._session.headers.update(
             {
@@ -1459,58 +1457,38 @@ def get_specific_citiy(self, search: str) -> dict:
             if city["name"].lower().strip() == search.lower().strip():
                 return city
         raise SourceArgumentNotFoundWithSuggestions(
-            "city", [city["name"] for city in cities]
+            "city", search, [city["name"] for city in cities]
         )
 
-    def get_garbage_calendars(self, city_id: str) -> list:
-        params = {
-            "filter": json.dumps(
-                {"entityid": {"$in": [city_id]}}, separators=(",", ":")
-            ),
-        }
-        params_str = urllib.parse.urlencode(params, safe=":$")
+    def get_uses_garbage_calendar_v2(self, city_id: str) -> bool:
+        r = self._session.get(f"https://api.citiesapps.com/entities/{city_id}/services")
+        r.raise_for_status()
+        essentials = r.json()["essentials"]
 
-        r = self._session.get(
-            "https://api.citiesapps.com/garbagecalendars/filter", params=params_str
+        return essentials["garbage_calendar_v2"]
+
+    def fetch_garbage_plans(self, city: str, calendar: str):
+        city_dict = self.get_specific_citiy(city)
+        api: CitiesApps.GarbageApiV2 | CitiesApps.GarbageApiV1 = (
+            CitiesApps.GarbageApiV1(self._session)
         )
-        r.raise_for_status()
-        return r.json()["garbage_calendars"]
+        is_v2 = self.get_uses_garbage_calendar_v2(city_dict["_id"])
 
-    def get_streets(self, city_id):
-        params = {
-            "entityid": [city_id],
-        }
+        if is_v2:
+            api = CitiesApps.GarbageApiV2(self._session)
 
-        r = self._session.get("https://api.citiesapps.com/garbageareas", params=params)
-        r.raise_for_status()
-        return r.json()
+        return {"is_v2": is_v2, "data": api.fetch_garbage_plans(city_dict, calendar)}
 
-    def get_specific_calendar(self, city_id: str, search: str) -> dict:
-        calendars = self.get_garbage_calendars(city_id)
-        for calendar in calendars:
-            if calendar["name"].lower().strip() == search.lower().strip():
-                return calendar
-        raise SourceArgumentNotFoundWithSuggestions(
-            "calendar", [calendar["name"] for calendar in calendars]
+    def get_garbage_calendars(self, city_id: str) -> dict[str, bool | list]:
+        api: CitiesApps.GarbageApiV2 | CitiesApps.GarbageApiV1 = self.GarbageApiV1(
+            self._session
         )
+        is_v2 = self.get_uses_garbage_calendar_v2(city_id)
 
-    def get_garbage_plans(self, garbage_calendar: dict) -> list:
-        r = self._session.get(
-            "https://api.citiesapps.com/garbagecalendars/",
-            params={"full": "true", "ids": garbage_calendar["_id"]},
-        )
-        r.raise_for_status()
-        garbage_plans = []
-        for cal in r.json():
-            garbage_plans += cal["garbage_plans"]
-        return garbage_plans
+        if is_v2:
+            api = CitiesApps.GarbageApiV2(self._session)
 
-    def fetch_garbage_plans(self, city: str, calendar: str):
-        city_dict = self.get_specific_citiy(city)
-        city_id = city_dict["_id"]
-        specific_calendar = self.get_specific_calendar(city_id, calendar)
-
-        return self.get_garbage_plans(specific_calendar)
+        return {"is_v2": is_v2, "data": api.get_garbage_calendars(city_id)}
 
     def get_supported_cities(self) -> dict[str, list]:
         supported_dict: dict[str, list] = {"supported": [], "not_supported": []}
@@ -1562,6 +1540,104 @@ def generate_service_map(self) -> list[dict[str, str]]:
             service_map.append({"title": city["name"], "url": url, "country": "at"})
         return service_map
 
+    class GarbageApiV1:
+        def __init__(self, session: requests.Session) -> None:
+            self._session = session
+
+        def fetch_garbage_plans(self, city_dict: dict, calendar: str):
+            city_id = city_dict["_id"]
+            specific_calendar = self.get_specific_calendar(city_id, calendar)
+
+            return self.get_garbage_plans(specific_calendar)
+
+        def get_specific_calendar(self, city_id: str, search: str) -> dict:
+            calendars = self.get_garbage_calendars(city_id)
+            for calendar in calendars:
+                if calendar["name"].lower().strip() == search.lower().strip():
+                    return calendar
+            raise SourceArgumentNotFoundWithSuggestions(
+                "calendar", search, [calendar["name"] for calendar in calendars]
+            )
+
+        def get_garbage_plans(self, garbage_calendar: dict) -> list:
+            r = self._session.get(
+                "https://api.citiesapps.com/garbagecalendars/",
+                params={"full": "true", "ids": garbage_calendar["_id"]},
+            )
+            r.raise_for_status()
+            garbage_plans = []
+            for cal in r.json():
+                garbage_plans += cal["garbage_plans"]
+            return garbage_plans
+
+        def get_garbage_calendars(self, city_id: str) -> list:
+            params = {
+                "filter": json.dumps(
+                    {"entityid": {"$in": [city_id]}}, separators=(",", ":")
+                ),
+            }
+            params_str = urllib.parse.urlencode(params, safe=":$")
+
+            r = self._session.get(
+                "https://api.citiesapps.com/garbagecalendars/filter", params=params_str
+            )
+            r.raise_for_status()
+            return r.json()["garbage_calendars"]
+
+    class GarbageApiV2:
+        def __init__(self, session: requests.Session) -> None:
+            self._session = session
+
+        def fetch_garbage_plans(self, city_dict: dict, calendar: str):
+            city_id = city_dict["_id"]
+            specific_calendar = self.get_specific_calendar(city_id, calendar)
+
+            return self.get_garbage_plans(specific_calendar)
+
+        def get_specific_calendar(self, city_id: str, search: str) -> dict:
+            calendars = self.get_garbage_calendars_with_search(city_id, search)
+            for calendar in calendars:
+                if calendar["street"].lower().strip() == search.lower().strip():
+                    return calendar
+
+            suggestions = [calendar["street"] for calendar in calendars]
+
+            if len(suggestions) == 0:
+                suggestions = [
+                    "Recheck your CitiesApp, the name of the calendar might have changed"
+                ]
+
+            raise SourceArgumentNotFoundWithSuggestions("calendar", search, suggestions)
+
+        def get_garbage_plans(self, garbage_calendar: dict) -> list:
+            r = self._session.get(
+                f"https://api.v2.citiesapps.com/waste-management/areas/{garbage_calendar['_id']}/calendar"
+            )
+            r.raise_for_status()
+
+            return r.json()["garbageCollectionDays"]
+
+        def get_garbage_calendars_with_search(self, city_id: str, search: str) -> list:
+            r = self._session.get(
+                f"https://api.v2.citiesapps.com/waste-management/by-city/{city_id}/areas/search/autocomplete?query={search}&limit=100"
+            )
+            r.raise_for_status()
+            return r.json()["garbageAreas"]
+
+        def get_garbage_calendars(self, city_id: str) -> list:
+            calendars = []
+            next_url = f"/waste-management/by-city/{city_id}/areas?pagination=limit:100"
+
+            while next_url:
+                r = self._session.get(f"https://api.v2.citiesapps.com/{next_url}")
+                r.raise_for_status()
+                j = r.json()
+                next_url = j["nextUrl"]
+                calendars += j["data"]
+
+            r.raise_for_status()
+            return calendars
+
 
 if __name__ == "__main__":
     c = CitiesApps(email=input("email: "), password=input("password: "))

custom_components/waste_collection_schedule/waste_collection_schedule/source/citiesapps_com.py

@@ -1,10 +1,7 @@
 from datetime import datetime
 
 from waste_collection_schedule import Collection  # type: ignore[attr-defined]
-from waste_collection_schedule.service.CitiesAppsCom import (
-    SERVICE_MAP,
-    CitiesApps,
-)
+from waste_collection_schedule.service.CitiesAppsCom import SERVICE_MAP, CitiesApps
 
 EXTRA_INFO = SERVICE_MAP
 TITLE = "App CITIES"
@@ -25,9 +22,9 @@
         "phone": "!secret citiesapps_com_phone",
         "password": "!secret citiesapps_com_password",
     },
-    "Rudersdorf Rudersdorf 3": {
+    "Rudersdorf Wiesengasse": {
         "city": "Rudersdorf",
-        "calendar": "Rudersdorf 3",
+        "calendar": "Wiesengasse",
         "email": "!secret citiesapps_com_email",
         "phone": "!secret citiesapps_com_phone",
         "password": "!secret citiesapps_com_password",
@@ -70,8 +67,28 @@ def fetch(self):
         cities_apps = CitiesApps(
             email=self._email, phone=self._phone, password=self._password
         )
+
         garbage_plans = cities_apps.fetch_garbage_plans(self._city, self._calendar)
 
+        if garbage_plans["is_v2"]:
+            return self.convert_v2(garbage_plans["data"])
+        else:
+            return self.convert_v1(garbage_plans["data"])
+
+    def convert_v2(self, garbage_plans):
+        entries = []
+        for garbage_plan in garbage_plans:
+            bin_type = garbage_plan["garbageTypeSettings"]["displayName"]
+            icon = ICON_MAP.get(bin_type.split(" ")[0])  # Collection icon
+
+            date = datetime.strptime(
+                garbage_plan["date"], "%Y-%m-%dT%H:%M:%S.%fZ"
+            ).date()
+            entries.append(Collection(date=date, t=bin_type, icon=icon))
+
+        return entries
+
+    def convert_v1(self, garbage_plans):
         entries = []
         for garbage_plan in garbage_plans:
             bin_type = garbage_plan["garbage_type"]["name"]

custom_components/waste_collection_schedule/waste_collection_schedule/wizard/citiesapps_com.py

@@ -1,6 +1,7 @@
 #!/usr/bin/env python3
 import site
 from pathlib import Path
+from datetime import datetime
 
 import inquirer
 
@@ -29,16 +30,21 @@ def ask_city():
     return [city_id, city]
 
 
-def ask_calendar(city_id, allow_search=True):
+def ask_calendar(city_id):
     calendars = app.get_garbage_calendars(city_id)
+
+    calendar_name_entry = "name"
+
+    if calendars["is_v2"]:
+        calendar_name_entry = "street"
+
     if len(calendars) == 1:
         print("# Only one calendar found, using that one")
-        cal = calendars[0]["name"]
+        cal = calendars[0]["data"][calendar_name_entry]
     else:
-        choices = [c["name"] for c in calendars]
+        choices = [c[calendar_name_entry] for c in calendars["data"]]
         choices.sort()
-        if allow_search:
-            choices = ["Search By Street", *choices]
+
         questions = [
             inquirer.List(
                 "cal",
@@ -51,25 +57,6 @@ def ask_calendar(city_id, allow_search=True):
     return cal
 
 
-def ask_by_street(city_id):
-    streets = app.get_streets(city_id)
-    if not streets["streets"]:
-        print("City does not support searching by street")
-        return ask_calendar(city_id, allow_search=False)
-
-    streetlist = streets["streets"]
-
-    streetlist.sort(key=lambda d: d["full_names"])
-    calendars = {
-        s_cal["garbage_areaid"]: s_cal["name"] for s_cal in streets["calendars"]
-    }
-    choices = [
-        (" ".join(c["full_names"]), calendars[c["areaids"][0]]) for c in streetlist
-    ]
-    questions = [inquirer.List("cal", choices=choices, message="Select a street")]
-    return inquirer.prompt(questions)["cal"]
-
-
 def ask_login():
     questions = [
         inquirer.List(
@@ -93,9 +80,6 @@ def main(password, email, phone):
     city_id, city = ask_city()
     cal = ask_calendar(city_id)
 
-    if cal == "Search By Street":
-        cal = ask_by_street(city_id)
-
     print(
         f"""waste_collection_schedule:
     sources:
@@ -107,7 +91,6 @@ def main(password, email, phone):
         {"email: " + email if email else "phone: " + phone}"""
     )
 
-
 if __name__ == "__main__":
     credentials = ask_login()
     app = CitiesAppsCom.CitiesApps(