ecoharmonogram_pl supports more town by supporting various android apps

custom_components/waste_collection_schedule/waste_collection_schedule/service/EcoHarmonogramPL.py

@@ -2,35 +2,52 @@ import sys
 
 import requests
 
-towns_url = "https://ecoharmonogram.pl/api/api.php?action=getTowns"
-community_towns_url = (
-    "https://ecoharmonogram.pl/api/api.php?action=getTownsForCommunity"
-)
-scheduled_periods_url = (
-    "https://ecoharmonogram.pl/api/api.php?action=getSchedulePeriods"
-)
-streets_url = "https://ecoharmonogram.pl/api/api.php?action=getStreets"
-schedules_url = "https://ecoharmonogram.pl/api/api.php?action=getSchedules"
+API_URL = "https://ecoharmonogram.pl/api/api.php"
 
-headers = {
-    "Content-Type": "application/json; charset=utf-8",
-    "Accept": "application/json",
-}
+SUPPORTED_APPS = [
+    "eco-przyszlosc",
+    "ogrodzieniec",
+    "gdansk",
+    "hajnowka",
+    "niemce",
+    "zgk-info",
+    "ilza",
+    "swietochlowice",
+    "popielow",
+    "mierzecice",
+    "bialapodlaska" "slupsk",
+    "trzebownisko",
+    "zory",
+]
 
 
 class Ecoharmonogram:
-    @staticmethod
-    def fetch_schedules(sp, streetId):
+    def __init__(self, app: str | None = None):
+        self._headers = {
+            "Content-Type": "application/json; charset=utf-8",
+            "Accept": "application/json",
+        }
+        self._app = app if app else None
+
+    def do_request(
+        self, action: str, payload: dict[str, str], url: str = API_URL
+    ) -> requests.Response:
+        params = payload.copy()
+        params["action"] = action
+        if self._app:
+            params["customApp"] = self._app
+        response = requests.get(url, headers=self._headers, params=params)
+        response.encoding = "utf-8-sig"
+        return response
+
+    def fetch_schedules(self, sp, streetId):
         payload = {"streetId": streetId, "schedulePeriodId": sp.get("id")}
-        schedules_response = requests.get(
-            schedules_url, headers=headers, params=payload
-        )
+        schedules_response = self.do_request("getSchedules", payload)
         schedules_response.encoding = "utf-8-sig"
         schedules_response = schedules_response.json()
         return schedules_response
 
-    @staticmethod
-    def fetch_streets(sp, town, street, house_number):
+    def fetch_streets(self, sp, town, street, house_number):
         payload = {
             "streetName": str(street),
             "number": str(house_number),
@@ -38,43 +55,35 @@ class Ecoharmonogram:
             "schedulePeriodId": sp.get("id"),
         }
 
-        streets_response = requests.get(streets_url, headers=headers, params=payload)
+        streets_response = self.do_request("getStreets", payload)
         streets_response.encoding = "utf-8-sig"
         streets = streets_response.json().get("streets")
         return streets
 
-    @staticmethod
-    def fetch_scheduled_periods(town):
+    def fetch_scheduled_periods(self, town):
         payload = {"townId": town.get("id")}
-        scheduled_perionds_response = requests.get(
-            scheduled_periods_url, headers=headers, params=payload
-        )
+        scheduled_perionds_response = self.do_request("getSchedulePeriods", payload)
         scheduled_perionds_response.encoding = "utf-8-sig"
         schedule_periods_data = scheduled_perionds_response.json()
         return schedule_periods_data
 
-    @staticmethod
-    def fetch_town():
-        town_response = requests.get(towns_url, headers=headers)
+    def fetch_town(self):
+        town_response = self.do_request("getTowns", {})
         town_response.encoding = "utf-8-sig"
         town_data = town_response.json()
         return town_data
 
-    @staticmethod
-    def fetch_town_with_community(community):
+    def fetch_town_with_community(self, community):
         payload = {"communityId": community}
-        town_response = requests.get(
-            community_towns_url, headers=headers, params=payload
-        )
+        town_response = self.do_request(action="getTownsForCommunity", payload=payload)
         town_response.encoding = "utf-8-sig"
         town_data = town_response.json()
         return town_data
 
-    @staticmethod
     def print_possible_sides(
-        town_input, district_input, street_input, house_number_input
+        self, town_input, district_input, street_input, house_number_input
     ):
-        town_data = Ecoharmonogram.fetch_town()
+        town_data = self.fetch_town()
         matching_towns = filter(
             lambda x: town_input.lower() in x.get("name").lower(),
             town_data.get("towns"),
@@ -86,20 +95,54 @@ class Ecoharmonogram:
 
         town = list(matching_towns_district)[0]
 
-        schedule_periods_data = Ecoharmonogram.fetch_scheduled_periods(town)
+        schedule_periods_data = self.fetch_scheduled_periods(town)
         schedule_periods = schedule_periods_data.get("schedulePeriods")
 
         for sp in schedule_periods:
-            streets = Ecoharmonogram.fetch_streets(
-                sp, town, street_input, house_number_input
-            )
+            streets = self.fetch_streets(sp, town, street_input, house_number_input)
             for street in streets:
                 for streetId in street.get("id").split(","):
-                    schedules_response = Ecoharmonogram.fetch_schedules(sp, streetId)
+                    schedules_response = self.fetch_schedules(sp, streetId)
                     print(schedules_response.get("street").get("sides"))
 
 
+def print_markdown_table() -> None:
+    table_data: dict[str | None, list[str]] = {}
+
+    for app in [None] + SUPPORTED_APPS:
+        ecoharmonogram = Ecoharmonogram(app)
+        towns = ecoharmonogram.fetch_town()["towns"]
+        town_names = [t["name"] for t in towns]
+        table_data[app] = town_names
+
+    duplicates_count: dict[str, int] = {}
+
+    for app, towns in table_data.items():
+        if app is None:
+            continue
+
+        for town in towns:
+            if town in table_data[None]:
+                if app not in duplicates_count:
+                    duplicates_count[app] = 0
+                duplicates_count[app] += 1
+
+    duplicates_with_total = {
+        k: {"duplicates": v, "total": len(table_data[k])}
+        for k, v in duplicates_count.items()
+        if v > 0
+    }
+    if len(duplicates_with_total) > 0:
+        print(f"duplicate Towns wiht No App: {duplicates_with_total}")
+
+    print("|APP | TOWN|")
+    print("|-|-|")
+    for app, towns in table_data.items():
+        app = app or "NO APP (LEAVE EMPTY)"
+        print(f"|{app}|{', '.join(towns)}|")
+
+
 if __name__ == "__main__":
-    Ecoharmonogram.print_possible_sides(
+    Ecoharmonogram(sys.argv[5] if len(sys.argv) > 5 else None).print_possible_sides(
         sys.argv[1], sys.argv[2] or "", sys.argv[3] or "", sys.argv[4] or ""
     )

custom_components/waste_collection_schedule/waste_collection_schedule/source/ecoharmonogram_pl.py

@@ -49,6 +49,12 @@ TEST_CASES = {
         "district": "Czernica",
         "house_number": "1",
     },
+    "With app": {
+        "town": "Buczków",
+        "street": "Buczków",
+        "house_number": "1",
+        "app": "eco-przyszlosc",
+    },
 }
 
 
@@ -56,6 +62,7 @@ class Source:
     def __init__(
         self,
         town,
+        app=None,
         district="",
         street="",
         house_number="",
@@ -68,12 +75,18 @@ class Source:
         self.district_input = district
         self.additional_sides_matcher_input = additional_sides_matcher
         self.community_input = community
+        if app:
+            self._ecoharmonogram_pl = Ecoharmonogram(app)
+        else:
+            self._ecoharmonogram_pl = Ecoharmonogram()
 
     def fetch(self):
         if self.community_input == "":
-            town_data = Ecoharmonogram.fetch_town()
+            town_data = self._ecoharmonogram_pl.fetch_town()
         else:
-            town_data = Ecoharmonogram.fetch_town_with_community(self.community_input)
+            town_data = self._ecoharmonogram_pl.fetch_town_with_community(
+                self.community_input
+            )
 
         matching_towns = filter(
             lambda x: self.town_input.lower() in x.get("name").lower(),
@@ -119,7 +132,7 @@ class Source:
                     f"Found multiple matches but no exact match found {matches}"
                 )
 
-        schedule_periods_data = Ecoharmonogram.fetch_scheduled_periods(town)
+        schedule_periods_data = self._ecoharmonogram_pl.fetch_scheduled_periods(town)
         schedule_periods = schedule_periods_data.get("schedulePeriods")
 
         entries = []
@@ -127,20 +140,22 @@ class Source:
             entries.extend(self._create_entries(sp, town))
         return entries
 
-    def _entry_exists(self, dmy, name, entries: [Collection]):
+    def _entry_exists(self, dmy, name, entries: list[Collection]):
         for e in entries:
             if dmy == e.date and name == e.type:
                 return True
         return False
 
     def _create_entries(self, sp, town):
-        streets = Ecoharmonogram.fetch_streets(
+        streets = self._ecoharmonogram_pl.fetch_streets(
             sp, town, self.street_input, self.house_number_input
         )
         entries = []
         for street in streets:
             for streetId in street.get("id").split(","):
-                schedules_response = Ecoharmonogram.fetch_schedules(sp, streetId)
+                schedules_response = self._ecoharmonogram_pl.fetch_schedules(
+                    sp, streetId
+                )
                 schedules_raw = schedules_response.get("schedules")
                 if (
                     self.additional_sides_matcher_input.lower()
@@ -171,9 +186,7 @@ class Source:
                             dmy = datetime.date(int(year), int(month), int(d))
                             name = sch.get("name")
                             if not self._entry_exists(dmy, name, entries):
-                                entries.append(
-                                    Collection(dmy, name)
-                                )
+                                entries.append(Collection(dmy, name))
                 if self.additional_sides_matcher_input != "":
                     return entries
         return entries