feat(Location): script to fill in OSM brand & version (#592)

open_prices/common/openstreetmap.py

@@ -1,4 +1,4 @@
-from OSMPythonTools.api import Api
+from OSMPythonTools.api import Api, ApiResult
 from OSMPythonTools.nominatim import Nominatim
 
 OSM_FIELDS_FROM_NOMINATIM = ["name", "display_name", "lat", "lon"]
@@ -32,9 +32,34 @@ def get_location_from_openstreetmap(osm_id: int, osm_type: str) -> dict:
         "lat": response.lat(),
         "lon": response.lon(),
         "version": response.version(),
+        "tags": response.tags(),
     }
 
 
+def get_location_with_history_from_openstreetmap(
+    osm_id: int, osm_type: str
+) -> ApiResult:
+    api = Api()
+    response = api.query(f"{osm_type.lower()}/{osm_id}", history=True)
+    return response
+
+
+def get_historical_location_from_openstreetmap(
+    osm_id: int, osm_type: str, historical_datetime: str
+) -> ApiResult:
+    """
+    Loop until we find a version that is more recent than the historical_datetime  # noqa
+    And return the previous version
+    """
+    response = get_location_with_history_from_openstreetmap(osm_id, osm_type)
+    if len(response.history()) == 1:
+        return response.history()[0]
+    for index, location_version in enumerate(response.history()):
+        if historical_datetime < location_version.timestamp():
+            return response.history()[index - 1]
+    return response.history()[-1]
+
+
 def get_location_dict(location):
     location_dict = dict()
     # fetch data from Nominatim

open_prices/locations/management/commands/set_location_osm_brand_and_version.py

@@ -0,0 +1,30 @@
+import time
+
+from django.core.management.base import BaseCommand
+
+from open_prices.common import openstreetmap as common_openstreetmap
+from open_prices.locations.models import Location
+
+
+class Command(BaseCommand):
+    help = "Init Location osm_brand & osm_version fields (depending on it's creation date!)."
+
+    def handle(self, *args, **options) -> None:  # type: ignore
+        qs = Location.objects.has_type_osm()
+        self.stdout.write(f"Found {qs.count()} OSM locations...")
+        qs = qs.filter(osm_version=None)
+        self.stdout.write(
+            f"Of which {qs.count()} have their osm_version field empty..."
+        )
+
+        for location in qs.all():
+            response = common_openstreetmap.get_historical_location_from_openstreetmap(
+                location.osm_id, location.osm_type, location.created
+            )
+            if response:
+                location.osm_brand = response.tag("brand")
+                location.osm_version = response.version()
+                location.save(update_fields=["osm_brand", "osm_version"])
+            else:
+                self.stdout.write(f"Could not find historical data for {location}")
+            time.sleep(1)  # be nice to the OSM API