AaronPo/the-biergarten-app
1
0
Fork 0
mirror of https://github.com/aaronpo97/the-biergarten-app.git synced 2026-10-15 01:54:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

Formatted src/biergarten_pipeline_orchestrator

Browse Source
This commit is contained in:
Aaron Po
2026-07-19 04:10:33 -04:00
parent 03467cbb65
commit 1885d9859f
5 changed files with 340 additions and 335 deletions
Download Patch File Download Diff File Expand all files Collapse all files

141
tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_breweries.cc
View File

@@ -12,80 +12,83 @@
void BiergartenPipelineOrchestrator::GenerateBreweries( void BiergartenPipelineOrchestrator::GenerateBreweries(
std::span<const EnrichedCity> cities) { std::span<const EnrichedCity> cities) {
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = "=== SAMPLE BREWERY GENERATION ==="}); .message = "=== SAMPLE BREWERY GENERATION ==="});
generated_breweries_.clear(); generated_breweries_.clear();
size_t skipped_count = 0; size_t skipped_count = 0;
size_t export_failed_count = 0; size_t export_failed_count = 0;
const auto generate_record = const auto generate_record =
[this, &skipped_count]( [this, &skipped_count](
const EnrichedCity& enriched_city) -> std::optional<BreweryRecord> { const EnrichedCity& enriched_city) -> std::optional<BreweryRecord> {
try { try {
const BreweryResult brewery = generator_->GenerateBrewery(enriched_city); const BreweryResult brewery =
const std::string postal_code = generator_->GenerateBrewery(enriched_city);
postal_code_service_->GeneratePostalCode(enriched_city.location); const std::string postal_code =
return BreweryRecord{ postal_code_service_->GeneratePostalCode(enriched_city.location);
.address = BreweryAddress{.city = enriched_city.location, return BreweryRecord{
.postal_code = postal_code}, .address = BreweryAddress{.city = enriched_city.location,
.brewery = brewery}; .postal_code = postal_code},
} catch (const std::exception& e) { .brewery = brewery};
++skipped_count; } catch (const std::exception& e) {
++skipped_count;
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format(
"[Pipeline] Skipping city '{}' ({}): brewery "
"generation failed: {}",
enriched_city.location.city,
enriched_city.location.country, e.what())});
return std::nullopt;
}
};
const auto export_record =
[this, &export_failed_count](const BreweryRecord& record) {
try {
exporter_->ProcessRecord(record);
} catch (const std::exception& export_exception) {
++export_failed_count;
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format(
"[Pipeline] Generated brewery for '{}' ({}) "
"but SQLite export failed: {}",
record.address.city.city, record.address.city.country,
export_exception.what())});
}
};
for (const EnrichedCity& enriched_city : cities) {
const std::optional<BreweryRecord> record =
generate_record(enriched_city);
if (!record.has_value()) {
continue;
}
generated_breweries_.push_back(*record);
export_record(*record);
}
if (skipped_count > 0) {
logger_->Log( logger_->Log(
{.level = LogLevel::Warn, {.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration, .phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("[Pipeline] Skipping city '{}' ({}): brewery " .message = std::format(
"generation failed: {}", "[Pipeline] Skipped {} city/cities due to generation errors",
enriched_city.location.city, skipped_count)});
enriched_city.location.country, e.what())}); }
return std::nullopt;
}
};
const auto export_record = [this, &export_failed_count]( if (export_failed_count > 0) {
const BreweryRecord& record) { logger_->Log(
try { {.level = LogLevel::Warn,
exporter_->ProcessRecord(record); .phase = PipelinePhase::Teardown,
} catch (const std::exception& export_exception) { .message = std::format("[Pipeline] Failed to export {} generated "
++export_failed_count; "brewery/breweries to SQLite",
logger_->Log({.level = LogLevel::Warn, export_failed_count)});
.phase = PipelinePhase::BreweryAndBeerGeneration, }
.message = std::format(
"[Pipeline] Generated brewery for '{}' ({}) "
"but SQLite export failed: {}",
record.address.city.city, record.address.city.country,
export_exception.what())});
}
};
for (const EnrichedCity& enriched_city : cities) {
const std::optional<BreweryRecord> record = generate_record(enriched_city);
if (!record.has_value()) {
continue;
}
generated_breweries_.push_back(*record);
export_record(*record);
}
if (skipped_count > 0) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format(
"[Pipeline] Skipped {} city/cities due to generation errors",
skipped_count)});
}
if (export_failed_count > 0) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::Teardown,
.message = std::format("[Pipeline] Failed to export {} generated "
"brewery/breweries to SQLite",
export_failed_count)});
}
} }

320
tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc
View File

@@ -20,203 +20,205 @@
namespace { namespace {
std::string Sanitize(std::string_view value) { std::string Sanitize(std::string_view value) {
std::string out; std::string out;
out.reserve(value.size()); out.reserve(value.size());
for (const char character : value) { for (const char character : value) {
if (std::isalnum(static_cast<unsigned char>(character)) != 0) { if (std::isalnum(static_cast<unsigned char>(character)) != 0) {
out.push_back(static_cast<char>( out.push_back(static_cast<char>(
std::tolower(static_cast<unsigned char>(character)))); std::tolower(static_cast<unsigned char>(character))));
} }
} }
return out; return out;
} }
std::string BuildEmail(const Name& name, std::string BuildEmail(const Name& name,
std::unordered_set<std::string>& used_local_parts) { std::unordered_set<std::string>& used_local_parts) {
const std::string base = const std::string base = std::format("{}.{}", Sanitize(name.first_name),
std::format("{}.{}", Sanitize(name.first_name), Sanitize(name.last_name)); Sanitize(name.last_name));
std::string local_part = base; std::string local_part = base;
uint32_t suffix = 1; uint32_t suffix = 1;
while (used_local_parts.contains(local_part)) { while (used_local_parts.contains(local_part)) {
local_part = std::format("{}{}", base, suffix); local_part = std::format("{}{}", base, suffix);
++suffix; ++suffix;
} }
used_local_parts.insert(local_part); used_local_parts.insert(local_part);
return std::format("{}@thebiergarten.app", local_part); return std::format("{}@thebiergarten.app", local_part);
} }
std::string GenerateDateOfBirth(std::mt19937& rng) { std::string GenerateDateOfBirth(std::mt19937& rng) {
using namespace std::chrono; using namespace std::chrono;
constexpr int kMinAge = 19; constexpr int kMinAge = 19;
constexpr int kMaxAge = 48; constexpr int kMaxAge = 48;
constexpr int kMaxDayOffset = 364; constexpr int kMaxDayOffset = 364;
std::uniform_int_distribution<int> age_dist(kMinAge, kMaxAge); std::uniform_int_distribution<int> age_dist(kMinAge, kMaxAge);
std::uniform_int_distribution<int> day_offset_dist(0, kMaxDayOffset); std::uniform_int_distribution<int> day_offset_dist(0, kMaxDayOffset);
const year_month_day today{floor<days>(system_clock::now())}; const year_month_day today{floor<days>(system_clock::now())};
const year_month_day birth_year_anchor{today.year() - years{age_dist(rng)}, const year_month_day birth_year_anchor{today.year() - years{age_dist(rng)},
today.month(), today.day()}; today.month(), today.day()};
const sys_days birth_date = const sys_days birth_date =
sys_days{birth_year_anchor} - days{day_offset_dist(rng)}; sys_days{birth_year_anchor} - days{day_offset_dist(rng)};
const year_month_day birth_ymd{birth_date}; const year_month_day birth_ymd{birth_date};
return std::format("{:04}-{:02}-{:02}", static_cast<int>(birth_ymd.year()), return std::format("{:04}-{:02}-{:02}", static_cast<int>(birth_ymd.year()),
static_cast<unsigned>(birth_ymd.month()), static_cast<unsigned>(birth_ymd.month()),
static_cast<unsigned>(birth_ymd.day())); static_cast<unsigned>(birth_ymd.day()));
} }
std::optional<Name> SampleName( std::optional<Name> SampleName(
const ForenamesByCountryMap& forenames_by_country, const ForenamesByCountryMap& forenames_by_country,
const SurnamesByCountryMap& surnames_by_country, const SurnamesByCountryMap& surnames_by_country,
const std::string& iso3166_1, std::mt19937& rng) { const std::string& iso3166_1, std::mt19937& rng) {
const auto forenames_it = forenames_by_country.find(iso3166_1); const auto forenames_it = forenames_by_country.find(iso3166_1);
const auto surnames_it = surnames_by_country.find(iso3166_1); const auto surnames_it = surnames_by_country.find(iso3166_1);
if (forenames_it == forenames_by_country.end() || if (forenames_it == forenames_by_country.end() ||
surnames_it == surnames_by_country.end() || surnames_it == surnames_by_country.end() ||
forenames_it->second.empty() || surnames_it->second.empty()) { forenames_it->second.empty() || surnames_it->second.empty()) {
return std::nullopt; return std::nullopt;
} }
const ForenameList& forenames = forenames_it->second; const ForenameList& forenames = forenames_it->second;
const SurnameList& surnames = surnames_it->second; const SurnameList& surnames = surnames_it->second;
std::uniform_int_distribution<size_t> forename_dist(0, forenames.size() - 1); std::uniform_int_distribution<size_t> forename_dist(0, forenames.size() - 1);
std::uniform_int_distribution<size_t> surname_dist(0, surnames.size() - 1); std::uniform_int_distribution<size_t> surname_dist(0, surnames.size() - 1);
auto forename_it = forenames.begin(); auto forename_it = forenames.begin();
std::advance(forename_it, forename_dist(rng)); std::advance(forename_it, forename_dist(rng));
auto surname_it = surnames.begin(); auto surname_it = surnames.begin();
std::advance(surname_it, surname_dist(rng)); std::advance(surname_it, surname_dist(rng));
return Name{.first_name = forename_it->name, return Name{.first_name = forename_it->name,
.last_name = *surname_it, .last_name = *surname_it,
.gender = forename_it->gender}; .gender = forename_it->gender};
} }
} // namespace } // namespace
void BiergartenPipelineOrchestrator::GenerateUsers( void BiergartenPipelineOrchestrator::GenerateUsers(
std::span<const EnrichedCity> cities) { std::span<const EnrichedCity> cities) {
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration, .phase = PipelinePhase::UserGeneration,
.message = "=== SAMPLE USER GENERATION ==="}); .message = "=== SAMPLE USER GENERATION ==="});
const PersonasList& personas = curated_data_service_->LoadPersonas(); const PersonasList& personas = curated_data_service_->LoadPersonas();
if (personas.empty()) { if (personas.empty()) {
throw std::runtime_error( throw std::runtime_error(
"No personas available in personas.json for user generation"); "No personas available in personas.json for user generation");
} }
const ForenamesByCountryMap& forenames_by_country = const ForenamesByCountryMap& forenames_by_country =
curated_data_service_->LoadForenamesByCountry(); curated_data_service_->LoadForenamesByCountry();
const SurnamesByCountryMap& surnames_by_country = const SurnamesByCountryMap& surnames_by_country =
curated_data_service_->LoadSurnamesByCountry(); curated_data_service_->LoadSurnamesByCountry();
std::mt19937 rng(std::random_device{}()); std::mt19937 rng(std::random_device{}());
std::uniform_int_distribution<size_t> persona_dist(0, personas.size() - 1); std::uniform_int_distribution<size_t> persona_dist(0, personas.size() - 1);
generated_users_.clear(); generated_users_.clear();
size_t skipped_count = 0; size_t skipped_count = 0;
size_t export_failed_count = 0; size_t export_failed_count = 0;
std::unordered_set<std::string> used_email_local_parts; std::unordered_set<std::string> used_email_local_parts;
const auto generate_record = const auto generate_record =
[this, &rng, &skipped_count, &used_email_local_parts]( [this, &rng, &skipped_count, &used_email_local_parts](
const EnrichedCity& city, const UserPersona& persona, const EnrichedCity& city, const UserPersona& persona,
const Name& sampled_name) -> std::optional<UserRecord> { const Name& sampled_name) -> std::optional<UserRecord> {
try { try {
const UserResult user = const UserResult user =
generator_->GenerateUser(city, persona, sampled_name); generator_->GenerateUser(city, persona, sampled_name);
const std::string postal_code = const std::string postal_code =
postal_code_service_->GeneratePostalCode(city.location); postal_code_service_->GeneratePostalCode(city.location);
return UserRecord{ return UserRecord{
.address = .address =
UserAddress{.city = city.location, .postal_code = postal_code}, UserAddress{.city = city.location, .postal_code = postal_code},
.user = user, .user = user,
.email = BuildEmail(sampled_name, used_email_local_parts), .email = BuildEmail(sampled_name, used_email_local_parts),
.date_of_birth = GenerateDateOfBirth(rng), .date_of_birth = GenerateDateOfBirth(rng),
}; };
} catch (const std::exception& e) { } catch (const std::exception& e) {
++skipped_count; ++skipped_count;
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format("[Pipeline] Skipping city '{}' ({}): "
"user generation failed: {}",
city.location.city, city.location.country,
e.what())});
return std::nullopt;
}
};
const auto export_record = [this,
&export_failed_count](const UserRecord& record) {
try {
exporter_->ProcessRecord(record);
} catch (const std::exception& export_exception) {
++export_failed_count;
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Generated user for '{}' ({}) but "
"SQLite export failed: {}",
record.address.city.city, record.address.city.country,
export_exception.what())});
}
};
for (const auto& city : cities) {
const std::optional<Name> sampled_name =
SampleName(forenames_by_country, surnames_by_country,
city.location.iso3166_1, rng);
if (!sampled_name.has_value()) {
++skipped_count;
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipping city '{}' ({}): no names "
"available for country '{}'",
city.location.city, city.location.country,
city.location.iso3166_1)});
continue;
}
const UserPersona& persona = personas[persona_dist(rng)];
const std::optional<UserRecord> record =
generate_record(city, persona, *sampled_name);
if (!record.has_value()) {
continue;
}
generated_users_.push_back(*record);
export_record(*record);
}
if (skipped_count > 0) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipped {} city/cities during user generation",
skipped_count)});
}
if (export_failed_count > 0) {
logger_->Log({.level = LogLevel::Warn, logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration, .phase = PipelinePhase::Teardown,
.message = std::format( .message = std::format("[Pipeline] Failed to export {} "
"[Pipeline] Skipping city '{}' ({}): " "generated user/users to SQLite",
"user generation failed: {}", export_failed_count)});
city.location.city, city.location.country, e.what())}); }
return std::nullopt;
}
};
const auto export_record = [this,
&export_failed_count](const UserRecord& record) {
try {
exporter_->ProcessRecord(record);
} catch (const std::exception& export_exception) {
++export_failed_count;
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Generated user for '{}' ({}) but "
"SQLite export failed: {}",
record.address.city.city, record.address.city.country,
export_exception.what())});
}
};
for (const auto& city : cities) {
const std::optional<Name> sampled_name =
SampleName(forenames_by_country, surnames_by_country,
city.location.iso3166_1, rng);
if (!sampled_name.has_value()) {
++skipped_count;
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipping city '{}' ({}): no names "
"available for country '{}'",
city.location.city, city.location.country,
city.location.iso3166_1)});
continue;
}
const UserPersona& persona = personas[persona_dist(rng)];
const std::optional<UserRecord> record =
generate_record(city, persona, *sampled_name);
if (!record.has_value()) {
continue;
}
generated_users_.push_back(*record);
export_record(*record);
}
if (skipped_count > 0) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipped {} city/cities during user generation",
skipped_count)});
}
if (export_failed_count > 0) {
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Teardown,
.message = std::format("[Pipeline] Failed to export {} "
"generated user/users to SQLite",
export_failed_count)});
}
} }

78
tooling/pipeline/src/biergarten_pipeline_orchestrator/log_results.cc
View File

@@ -12,48 +12,48 @@
#include "services/logging/logger.h" #include "services/logging/logger.h"
void BiergartenPipelineOrchestrator::LogResults() const { void BiergartenPipelineOrchestrator::LogResults() const {
boost::json::array brewery_output; boost::json::array brewery_output;
for (const auto& [address, brewery] : generated_breweries_) { for (const auto& [address, brewery] : generated_breweries_) {
const City& city = address.city; const City& city = address.city;
brewery_output.push_back(boost::json::object{ brewery_output.push_back(boost::json::object{
{"name_en", brewery.name_en}, {"name_en", brewery.name_en},
{"description_en", brewery.description_en}, {"description_en", brewery.description_en},
{"name_local", brewery.name_local}, {"name_local", brewery.name_local},
{"description_local", brewery.description_local}, {"description_local", brewery.description_local},
{"address", boost::json::object{ {"address", boost::json::object{
{"city", city.city}, {"city", city.city},
{"country", city.country}, {"country", city.country},
{"state_province", city.state_province}, {"state_province", city.state_province},
{"iso3166_2", city.iso3166_2}, {"iso3166_2", city.iso3166_2},
{"postal_code", address.postal_code}, {"postal_code", address.postal_code},
}}}); }}});
} }
std::ostringstream brewery_oss; std::ostringstream brewery_oss;
PrettyPrint(brewery_oss, brewery_output); PrettyPrint(brewery_oss, brewery_output);
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = brewery_oss.str()}); .message = brewery_oss.str()});
boost::json::array user_output; boost::json::array user_output;
for (const auto& generated_user : generated_users_) { for (const auto& generated_user : generated_users_) {
user_output.push_back(boost::json::object{ user_output.push_back(boost::json::object{
{"first_name", generated_user.user.first_name}, {"first_name", generated_user.user.first_name},
{"last_name", generated_user.user.last_name}, {"last_name", generated_user.user.last_name},
{"gender", generated_user.user.gender}, {"gender", generated_user.user.gender},
{"username", generated_user.user.username}, {"username", generated_user.user.username},
{"bio", generated_user.user.bio}, {"bio", generated_user.user.bio},
{"activity_weight", generated_user.user.activity_weight}, {"activity_weight", generated_user.user.activity_weight},
{"email", generated_user.email}, {"email", generated_user.email},
{"date_of_birth", generated_user.date_of_birth}, {"date_of_birth", generated_user.date_of_birth},
}); });
} }
std::ostringstream user_oss; std::ostringstream user_oss;
PrettyPrint(user_oss, user_output); PrettyPrint(user_oss, user_output);
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = user_oss.str()}); .message = user_oss.str()});
} }

53
tooling/pipeline/src/biergarten_pipeline_orchestrator/query_cities_with_countries.cc
View File

@@ -14,37 +14,36 @@
#include "services/curated_data/curated_json_data_service.h" #include "services/curated_data/curated_json_data_service.h"
#include "services/logging/logger.h" #include "services/logging/logger.h"
std::vector<City> std::vector<City> BiergartenPipelineOrchestrator::QueryLocations() {
BiergartenPipelineOrchestrator::QueryLocations() { logger_->Log({.level = LogLevel::Info,
logger_->Log({.level = LogLevel::Info, .phase = PipelinePhase::Startup,
.phase = PipelinePhase::Startup, .message = "=== GEOGRAPHIC DATA OVERVIEW ==="});
.message = "=== GEOGRAPHIC DATA OVERVIEW ==="});
const std::vector<City>& all_locations = const std::vector<City>& all_locations =
curated_data_service_->LoadLocations(); curated_data_service_->LoadLocations();
const size_t sample_count = std::min( const size_t sample_count = std::min(
static_cast<size_t>(application_options_.pipeline.location_count), static_cast<size_t>(application_options_.pipeline.location_count),
all_locations.size()); all_locations.size());
const auto sample_count_signed = const auto sample_count_signed =
static_cast<std::iter_difference_t<decltype(all_locations.cbegin())>>( static_cast<std::iter_difference_t<decltype(all_locations.cbegin())>>(
sample_count); sample_count);
std::vector<City> sampled_locations; std::vector<City> sampled_locations;
sampled_locations.reserve(sample_count); sampled_locations.reserve(sample_count);
std::random_device random_generator; std::random_device random_generator;
std::ranges::sample(all_locations, std::back_inserter(sampled_locations), std::ranges::sample(all_locations, std::back_inserter(sampled_locations),
sample_count_signed, random_generator); sample_count_signed, random_generator);
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = std::format(" Locations available: {}", .message = std::format(" Locations available: {}",
all_locations.size())}); all_locations.size())});
logger_->Log({.level = LogLevel::Info, logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup, .phase = PipelinePhase::Startup,
.message = std::format(" Sampled locations: {}", .message = std::format(" Sampled locations: {}",
sampled_locations.size())}); sampled_locations.size())});
return sampled_locations; return sampled_locations;
} }

83
tooling/pipeline/src/biergarten_pipeline_orchestrator/run.cc
View File

@@ -11,51 +11,52 @@
#include "services/logging/logger.h" #include "services/logging/logger.h"
bool BiergartenPipelineOrchestrator::Run() { bool BiergartenPipelineOrchestrator::Run() {
try { try {
exporter_->Initialize(); exporter_->Initialize();
std::vector<City> cities = QueryLocations(); std::vector<City> cities = QueryLocations();
std::vector<EnrichedCity> enriched; std::vector<EnrichedCity> enriched;
enriched.reserve(cities.size()); enriched.reserve(cities.size());
size_t skipped_count = 0; size_t skipped_count = 0;
for (auto& city : cities) { for (auto& city : cities) {
try { try {
std::string region_context = context_service_->GetLocationContext(city); std::string region_context =
context_service_->GetLocationContext(city);
enriched.push_back( enriched.push_back(
EnrichedCity{.location = std::move(city), EnrichedCity{.location = std::move(city),
.region_context = std::move(region_context)}); .region_context = std::move(region_context)});
} catch (const std::exception& exception) { } catch (const std::exception& exception) {
++skipped_count; ++skipped_count;
logger_->Log({.level = LogLevel::Warn, logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment, .phase = PipelinePhase::Enrichment,
.message = std::format( .message = std::format(
"[Pipeline] Skipping city '{}' ({}): context " "[Pipeline] Skipping city '{}' ({}): context "
"lookup failed: {}", "lookup failed: {}",
city.city, city.country, exception.what())}); city.city, city.country, exception.what())});
}
} }
}
if (skipped_count > 0) { if (skipped_count > 0) {
logger_->Log( logger_->Log(
{.level = LogLevel::Warn, {.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment, .phase = PipelinePhase::Enrichment,
.message = std::format("[Pipeline] Skipped {} city/cities due " .message = std::format("[Pipeline] Skipped {} city/cities due "
"to context lookup errors", "to context lookup errors",
skipped_count)}); skipped_count)});
} }
this->GenerateUsers(enriched); this->GenerateUsers(enriched);
this->GenerateBreweries(enriched); this->GenerateBreweries(enriched);
exporter_->Finalize(); exporter_->Finalize();
this->LogResults(); this->LogResults();
return true; return true;
} catch (const std::exception& e) { } catch (const std::exception& e) {
logger_->Log({.level = LogLevel::Error, logger_->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Teardown, .phase = PipelinePhase::Teardown,
.message = std::format( .message = std::format(
"Pipeline execution failed with error: {}", e.what())}); "Pipeline execution failed with error: {}", e.what())});
return false; return false;
} }
} }