Move brewery reverse geocoding from C# seeding to C++ pipeline (#249)

tooling/pipeline/src/biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc

@@ -13,10 +13,12 @@ BiergartenPipelineOrchestrator::BiergartenPipelineOrchestrator(
     std::unique_ptr<DataGenerator> generator,
     std::unique_ptr<IExportService> exporter,
     std::unique_ptr<ICuratedDataService> curated_data_service,
+    std::unique_ptr<IAddressService> address_service,
     const ApplicationOptions& application_options)
     : logger_(std::move(logger)),
       context_service_(std::move(context_service)),
       generator_(std::move(generator)),
       exporter_(std::move(exporter)),
       curated_data_service_(std::move(curated_data_service)),
+      address_service_(std::move(address_service)),
       application_options_(application_options) {}

tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_breweries.cc

@@ -7,14 +7,21 @@
 #include <format>
 #include <optional>
 #include <random>
+#include <string>
+#include <string_view>
 
 #include "biergarten_pipeline_orchestrator.h"
 #include "biergarten_pipeline_orchestrator/coords.h"
+#include "services/address/address_service.h"
 #include "services/logging/logger.h"
 
 using biergarten_pipeline_orchestrator_internal::Coords;
 using biergarten_pipeline_orchestrator_internal::GetRandomCoordsWithinRange;
 
+namespace {
+constexpr std::string_view kPlaceholderAddressLine1 = "Address unavailable";
+}  // namespace
+
 void BiergartenPipelineOrchestrator::GenerateBreweries(
     std::span<const EnrichedCity> cities) {
    logger_->Log({.level = LogLevel::Info,
@@ -39,10 +46,32 @@ void BiergartenPipelineOrchestrator::GenerateBreweries(
          const Coords brewery_coords = GetRandomCoordsWithinRange(
              city_centre, kMaxDistanceFromCentreKm, rng);
 
+         const std::optional<AddressLookupResult> geocoded =
+             address_service_->ReverseGeocode(brewery_coords.longitude,
+                                              brewery_coords.latitude);
+
+         if (!geocoded.has_value()) {
+            logger_->Log(
+                {.level = LogLevel::Warn,
+                 .phase = PipelinePhase::BreweryAndBeerGeneration,
+                 .message = std::format(
+                     "[Pipeline] Could not reverse geocode brewery in '{}' "
+                     "({}, {}); using a placeholder address.",
+                     enriched_city.location.city, brewery_coords.latitude,
+                     brewery_coords.longitude)});
+         }
+
          return BreweryRecord{
-             .address = BreweryAddress{.city = enriched_city.location,
-                                       .longitude = brewery_coords.longitude,
-                                       .latitude = brewery_coords.latitude},
+             .address =
+                 BreweryAddress{
+                     .city = enriched_city.location,
+                     .longitude = brewery_coords.longitude,
+                     .latitude = brewery_coords.latitude,
+                     .address_line1 = geocoded.has_value()
+                                          ? geocoded->address_line1
+                                          : std::string(kPlaceholderAddressLine1),
+                     .postal_code =
+                         geocoded.has_value() ? geocoded->postal_code : ""},
              .brewery = brewery};
       } catch (const std::exception& e) {
          ++skipped_count;

tooling/pipeline/src/main.cc

@@ -166,6 +166,26 @@ int main(const int argc, char** argv) {
                      inj.template create<std::unique_ptr<WebClient>>(),
                      log_producer);
               }),
+          di::bind<IAddressService>().to(
+              [options, &log_producer](
+                  const auto& inj) -> std::unique_ptr<IAddressService> {
+                 if (options.generator.mode == GeneratorMode::kMock) {
+                    log_producer->Log({.level = LogLevel::Info,
+                                       .phase = PipelinePhase::Startup,
+                                       .message = "Address service: mock"});
+
+                    return std::make_unique<MockAddressService>();
+                 }
+
+                 log_producer->Log(
+                     {.level = LogLevel::Info,
+                      .phase = PipelinePhase::Startup,
+                      .message = "Address service: Nominatim"});
+
+                 return std::make_unique<NominatimAddressService>(
+                     inj.template create<std::unique_ptr<WebClient>>(),
+                     log_producer);
+              }),
           di::bind<DataGenerator>().to([&options, &model_path, &sampling,
                                         &prompt_directory,
                                         &log_producer](const auto& inj)

tooling/pipeline/src/services/address/nominatim_address_service.cc

@@ -0,0 +1,133 @@
+/**
+ * @file services/address/nominatim_address_service.cc
+ * @brief NominatimAddressService implementation.
+ */
+
+#include "services/address/nominatim_address_service.h"
+
+#include <boost/json.hpp>
+#include <chrono>
+#include <format>
+#include <stdexcept>
+#include <string>
+#include <thread>
+#include <utility>
+
+using namespace boost;
+
+NominatimAddressService::NominatimAddressService(
+    std::unique_ptr<WebClient> client, std::shared_ptr<ILogger> logger)
+    : client_(std::move(client)), logger_(std::move(logger)) {}
+
+std::optional<AddressLookupResult> NominatimAddressService::ReverseGeocode(
+    double longitude, double latitude) {
+   if (!client_) {
+      if (logger_) {
+         logger_->Log({.level = LogLevel::Warn,
+                       .phase = PipelinePhase::Enrichment,
+                       .message = "Nominatim client is nullptr."});
+      }
+      return std::nullopt;
+   }
+
+   const std::string url = std::format(
+       "https://nominatim.openstreetmap.org/reverse?format=jsonv2&lat={}&"
+       "lon={}&zoom=18&addressdetails=1",
+       latitude, longitude);
+
+   std::string body;
+   try {
+      body = client_->Get(url);
+   } catch (const std::runtime_error& e) {
+      if (logger_) {
+         logger_->Log({.level = LogLevel::Warn,
+                       .phase = PipelinePhase::Enrichment,
+                       .message = std::format(
+                           "NominatimAddressService: request failed for "
+                           "({}, {}): {}",
+                           latitude, longitude, e.what())});
+      }
+      return std::nullopt;
+   }
+
+   {
+      using namespace std::literals::chrono_literals;
+      std::this_thread::sleep_for(1s);
+   }
+
+   system::error_code ec;
+   const json::value doc = json::parse(body, ec);
+
+   if (ec) {
+      if (logger_) {
+         logger_->Log({.level = LogLevel::Warn,
+                       .phase = PipelinePhase::Enrichment,
+                       .message = std::format(
+                           "NominatimAddressService: JSON parse error for "
+                           "({}, {}): {}",
+                           latitude, longitude, ec.message())});
+      }
+      return std::nullopt;
+   }
+
+   const json::object* obj = doc.if_object();
+   if ((obj == nullptr) || obj->contains("error")) {
+      if (logger_) {
+         logger_->Log(
+             {.level = LogLevel::Warn,
+              .phase = PipelinePhase::Enrichment,
+              .message = std::format("NominatimAddressService: no address "
+                                     "found for ({}, {})",
+                                     latitude, longitude)});
+      }
+      return std::nullopt;
+   }
+
+   const json::value* address_ptr = obj->if_contains("address");
+   if ((address_ptr == nullptr) || !address_ptr->is_object()) {
+      if (logger_) {
+         logger_->Log(
+             {.level = LogLevel::Warn,
+              .phase = PipelinePhase::Enrichment,
+              .message = std::format("NominatimAddressService: missing "
+                                     "'address' for ({}, {})",
+                                     latitude, longitude)});
+      }
+      return std::nullopt;
+   }
+
+   const json::object& address = address_ptr->get_object();
+
+   const auto extract_string = [&address](std::string_view key) -> std::string {
+      const json::value* value_ptr = address.if_contains(key);
+      if ((value_ptr == nullptr) || !value_ptr->is_string()) {
+         return {};
+      }
+      return std::string(value_ptr->get_string());
+   };
+
+   const std::string house_number = extract_string("house_number");
+   const std::string road = extract_string("road");
+   const std::string postal_code = extract_string("postcode");
+
+   std::string address_line1 = road;
+   if (!house_number.empty()) {
+      address_line1 = address_line1.empty() ? house_number
+                                             : house_number + " " + road;
+   }
+
+   if (address_line1.empty()) {
+      if (logger_) {
+         logger_->Log(
+             {.level = LogLevel::Warn,
+              .phase = PipelinePhase::Enrichment,
+              .message = std::format("NominatimAddressService: no usable "
+                                     "street address for ({}, {})",
+                                     latitude, longitude)});
+      }
+      return std::nullopt;
+   }
+
+   return AddressLookupResult{.address_line1 = address_line1,
+                              .postal_code = postal_code};
+}

tooling/pipeline/src/services/sqlite/process_record.cc

@@ -184,6 +184,20 @@ uint64_t SqliteExportService::ProcessRecord(const BreweryRecord& brewery) {
                sqlite_export_service_internal::kBreweryAddressLatitudeBindIndex,
            .value = brewery.address.latitude,
            .action = "Failed to bind SQLite brewery address latitude"});
+   sqlite_export_service_internal::Bind(
+       insert_brewery_address_stmt_,
+       sqlite_export_service_internal::BoundParam<std::string_view>{
+           .index =
+               sqlite_export_service_internal::kBreweryAddressLine1BindIndex,
+           .value = brewery.address.address_line1,
+           .action = "Failed to bind SQLite brewery address line 1"});
+   sqlite_export_service_internal::Bind(
+       insert_brewery_address_stmt_,
+       sqlite_export_service_internal::BoundParam<std::string_view>{
+           .index = sqlite_export_service_internal::
+               kBreweryAddressPostalCodeBindIndex,
+           .value = brewery.address.postal_code,
+           .action = "Failed to bind SQLite brewery address postal code"});
 
    sqlite_export_service_internal::StepStatement(
        db_handle_, insert_brewery_address_stmt_,