refacto(json_feed_client): utilise le file downloader pour mutualiser le code

geotribu_cli/constants.py

@@ -96,6 +96,7 @@ class GeotribuDefaults:
     site_git_default_branch: str = "master"
     site_git_project: str = "website"
     site_search_index: str = "search/search_index.json"
+    site_search_tags_remote_path = "tags.json"
 
     # CDN
     cdn_base_url: str = "https://cdn.geotribu.fr/"
@@ -105,6 +106,10 @@ class GeotribuDefaults:
     # comments
     comments_base_url: str = "https://comments.geotribu.fr/"
 
+    # JSON Feed
+    json_path_created = "feed_json_created.json"
+    json_path_updated = "feed_json_updated.json"
+
     # RSS
     rss_path_created: str = "feed_rss_created.xml"
     rss_path_updated: str = "feed_rss_updated.xml"
@@ -137,6 +142,15 @@ def cdn_search_index_full_url(self) -> str:
         """
         return f"{self.cdn_base_url}{self.cdn_base_path}/{self.cdn_search_index}"
 
+    @property
+    def json_created_full_url(self) -> str:
+        """Returns website JSON Feed full URL for latest created contents.
+
+        Returns:
+            str: URL as string
+        """
+        return f"{self.site_base_url}{self.json_path_created}"
+
     @property
     def rss_created_full_url(self) -> str:
         """Returns website RSS full URL for latest created contents.
@@ -155,6 +169,15 @@ def site_search_index_full_url(self) -> str:
         """
         return f"{self.site_base_url}{self.site_search_index}"
 
+    @property
+    def site_search_tags_full_url(self) -> str:
+        """Returns website search tags full URL.
+
+        Returns:
+            str: URL as string
+        """
+        return f"{self.site_base_url}{self.site_search_tags_remote_path}"
+
     def site_git_source_base_url(
         self, mode: Literal["blob", "edit", "raw"] = "blob", url_path: str = ""
     ) -> str:

geotribu_cli/json/json_client.py

@@ -1,56 +1,109 @@
+#! python3  # noqa: E265
+
+
+"""JSON Feed client."""
+
+
+# ############################################################################
+# ########## IMPORTS #############
+# ################################
+
+# standard
+import logging
+from pathlib import Path
 from typing import Any
 
-import requests
-from requests import Response
+# 3rd party
+import orjson
 
-from geotribu_cli.__about__ import __executable_name__, __version__
+# project
+from geotribu_cli.constants import GeotribuDefaults
+from geotribu_cli.utils.file_downloader import download_remote_file_to_local
 
-JSON_FEED_URL = "https://geotribu.fr/feed_json_created.json"
-JSON_TAGS_URL = "https://geotribu.fr/tags.json"
+# ############################################################################
+# ########## GLOBALS #############
+# ################################
 
-HEADERS: dict = {
-    b"Accept": b"application/json",
-    b"User-Agent": bytes(f"{__executable_name__}/{__version__}", "utf8"),
-}
+logger = logging.getLogger(__name__)
+defaults_settings = GeotribuDefaults()
+
+
+# ############################################################################
+# ########## CLASSES #############
+# ################################
 
 
 class JsonFeedClient:
-    def __init__(self, url: str = JSON_FEED_URL, tags_url: str = JSON_TAGS_URL):
-        """
-        Class initialization
+    """JSON Feed client."""
+
+    def __init__(
+        self,
+        json_feed_url: str = defaults_settings.json_created_full_url,
+        tags_url: str = defaults_settings.site_search_tags_full_url,
+        expiration_rotating_hours: int = 24,
+    ):
+        """Class initialization.
 
         Args:
             url: JSON feed URL, defaults to https://geotribu.fr/feed_json_created.json
             tags_url: JSON tags URL, defaults to https://geotribu.fr/tags.json
+            expiration_rotating_hours: nombre d'heures à partir duquel considérer
+                le fichier local comme périmé. Defaults to 24.
         """
-        self.url = url
+        # params as attributes
+        self.json_feed_url = json_feed_url
         self.tags_url = tags_url
+        self.expiration_rotating_hours = expiration_rotating_hours
+
+        # attributes
+        self.local_json_feed_path: Path = (
+            defaults_settings.geotribu_working_folder.joinpath("rss/json_feed.json")
+        )
+        self.local_json_feed_path.parent.mkdir(parents=True, exist_ok=True)
+        self.local_tags_path: Path = defaults_settings.geotribu_working_folder.joinpath(
+            "search/tags.json"
+        )
+        self.local_tags_path.parent.mkdir(parents=True, exist_ok=True)
 
     def items(self) -> list[dict[str, Any]]:
-        """
-        Fetch Geotribu JSON feed items
+        """Fetch Geotribu JSON feed latest created items.
 
         Returns:
             List of dicts representing raw JSON feed items
         """
-        r: Response = requests.get(self.url, headers=HEADERS)
-        r.raise_for_status()
-        return r.json()["items"]
+
+        local_json_feed = download_remote_file_to_local(
+            remote_url_to_download=self.json_feed_url,
+            local_file_path=self.local_json_feed_path,
+            expiration_rotating_hours=self.expiration_rotating_hours,
+        )
+
+        with local_json_feed.open("rb") as fd:
+            json_feed = orjson.loads(fd.read())
+
+        return json_feed.get("items")
 
     def tags(self, should_sort: bool = False) -> list[str]:
-        """
-        Fetch Geotribu used tags
+        """Fetch Geotribu used tags.
 
         Args:
-            should_sort: if the list of returned tags should be alphabetically sorted
+            should_sort: if the list of returned tags should be alphabetically sorted.
+                Defaults to False.
 
         Returns:
             List of tags used by Geotribu
         """
-        r: Response = requests.get(self.tags_url, headers=HEADERS)
-        r.raise_for_status()
+        local_tags = download_remote_file_to_local(
+            remote_url_to_download=self.tags_url,
+            local_file_path=self.local_tags_path,
+            expiration_rotating_hours=self.expiration_rotating_hours,
+        )
+
+        with local_tags.open("rb") as fd:
+            search_tags = orjson.loads(fd.read())
+
         tags = set()
-        for item in r.json()["mappings"]:
+        for item in search_tags.get("mappings"):
             for tag in item["tags"]:
                 tags.add(tag)
         return sorted(tags) if should_sort else list(tags)