fix: download non zipped report if zip does not exist

Author: Guillaume Latour

pyproject.toml

@@ -22,7 +22,7 @@ maintainers = [
 name = "tomtom_api"
 readme = "readme.md"
 requires-python = ">=3.8"
-version = "2024.7.4"
+version = "2024.7.9"
 
 [project.urls]
 repository = "https://stash.macq.eu/projects/RDDS/repos/tomtom-api"

src/tomtom_api/traffic_stats/models/responses.py

@@ -2,6 +2,7 @@
 
 In this module are grouped all the dataclasses that are hydrated when the API is responding something.
 """
+
 from __future__ import annotations
 
 import datetime as dt
@@ -47,7 +48,13 @@ class TomtomResponseStatus(JSONWizard, LoadMixin):
     response_status: str
     urls: Optional[List[str]]
 
-    def __init__(self, job_id: int, job_state: TomtomJobState, response_status: str, urls: Optional[List[str]] = None):
+    def __init__(
+        self,
+        job_id: int,
+        job_state: TomtomJobState,
+        response_status: str,
+        urls: Optional[List[str]] = None,
+    ):
         self.job_id = job_id
         self.job_state = job_state
         self.response_status = response_status
@@ -57,7 +64,7 @@ def load_to_enum(o, base_type: Type[Enum]) -> Enum:
         return base_type.from_str(o)
 
     def display_info(self) -> str:
-        return f'{self.job_id} ({self.job_state.name})'
+        return f"{self.job_id} ({self.job_state.name})"
 
     def download(self, file_type: TomtomDownloadFileType) -> bytes:
         """Download the job report, in the given format.
@@ -82,24 +89,38 @@ def download(self, file_type: TomtomDownloadFileType) -> bytes:
             This exception can also be triggered if there is an error while downloading the file.
         """
         if self.job_state != TomtomJobState.DONE:
-            raise DownloadException(f'The "DONE" state is required for downloading the report ({self.job_state}).')
+            raise DownloadException(
+                f'The "DONE" state is required for downloading the report ({self.job_state}).'
+            )
 
         if self.urls is None:
-            raise DownloadException('There is no download url for this job.')
+            raise DownloadException("There is no download url for this job.")
+
+        filtered_links = [link for link in self.urls if f"{file_type.value}?" in link]
+
+        # if nothing is found, remove the compression extension
+        if len(filtered_links) == 0:
+            filtered_links = [
+                link
+                for link in self.urls
+                if f"{file_type.value.replace('.zip', '')}?" in link
+            ]
 
-        filtered_links = [link for link in self.urls if f'{file_type.value}?' in link]
+        # then perform a sanity check
         if len(filtered_links) != 1:
-            log.debug(f'Available links: {self.urls}.')
-            log.debug(f'Provided file type: {file_type}.')
-            raise DownloadException(f'Impossible to find the desired file in the following links: {filtered_links}.')
+            log.debug(f"Available links: {self.urls}.")
+            log.debug(f"Provided file type: {file_type}.")
+            raise DownloadException(
+                f"Impossible to find the desired file in the following links: {filtered_links}."
+            )
 
         link = filtered_links[0]
-        log.debug(f'Performing request to {link}')
+        log.debug(f"Performing request to {link}")
         response = requests.get(link)
 
         if response.status_code != 200:
-            log.debug(f'Response: {response}')
-            raise DownloadException(f'Error while downloading {link}')
+            log.debug(f"Response: {response}")
+            raise DownloadException(f"Error while downloading {link}")
 
         return response.content
 
@@ -124,10 +145,10 @@ def write(self, file: Path, file_type: TomtomDownloadFileType) -> None:
         """
         content = self.download(file_type)
         if file.exists():
-            log.warning(f'The following file will be overwritten: {file}')
+            log.warning(f"The following file will be overwritten: {file}")
 
-        log.debug(f'Writing file {file}')
-        with open(file, 'wb') as f:
+        log.debug(f"Writing file {file}")
+        with open(file, "wb") as f:
             f.write(content)
 
 
@@ -145,25 +166,29 @@ class TomtomJobInfo(JSONWizard, LoadMixin):
     name: str
     created_at: dt.datetime
     state: TomtomJobState
-    job_id: int = json_field('id', all=True)
-    job_type: str = json_field('type', all=True)
-    completed_at: Optional[dt.datetime] = json_field('completed_at', default=None)
+    job_id: int = json_field("id", all=True)
+    job_type: str = json_field("type", all=True)
+    completed_at: Optional[dt.datetime] = json_field("completed_at", default=None)
 
     def load_to_enum(o, base_type: Type[Enum]) -> Enum:
         return base_type.from_str(o)
 
     def display_info(self) -> str:
-        datetime_fmt = '%Y-%m-%d %H:%M:%S'
+        datetime_fmt = "%Y-%m-%d %H:%M:%S"
         start = self.created_at.strftime(datetime_fmt)
-        end = 'None' if self.completed_at is None else self.completed_at.strftime(datetime_fmt)
-        return f'┌─ [{self.job_id}] {self.name} // {self.state.name}\n└─ {self.job_type} <{start} ⟶ {end}>'
+        end = (
+            "None"
+            if self.completed_at is None
+            else self.completed_at.strftime(datetime_fmt)
+        )
+        return f"┌─ [{self.job_id}] {self.name} // {self.state.name}\n└─ {self.job_type} <{start} ⟶ {end}>"
 
 
 @dataclass
 class Sort(JSONWizard):
-    is_sorted: bool = json_field('sorted', all=True)
-    is_unsorted: bool = json_field('unsorted', all=True)
-    is_empty: bool = json_field('empty', all=True)
+    is_sorted: bool = json_field("sorted", all=True)
+    is_unsorted: bool = json_field("unsorted", all=True)
+    is_empty: bool = json_field("empty", all=True)
 
 
 @dataclass
@@ -194,5 +219,5 @@ class TomtomResponseSearchJobs(JSONWizard):
 @dataclass
 class JsonIpResponse(JSONWizard):
     ip: str
-    geo_ip: str = json_field('geo-ip', all=True)
-    api_help: str = json_field('API Help', all=True)
+    geo_ip: str = json_field("geo-ip", all=True)
+    api_help: str = json_field("API Help", all=True)