AaronPo/the-biergarten-app
1
0
Fork 0
mirror of https://github.com/aaronpo97/the-biergarten-app.git synced 2026-07-16 17:47:22 +00:00
Code Issues Packages Projects Releases Wiki Activity

Pipeline: Add LLM and mocked user generation to the pipeline (#230)

Browse Source
This commit is contained in:
Aaron Po
2026-07-01 07:47:24 -04:00
committed by GitHub
parent 880f73e004
commit 2b8a900d12
90 changed files with 57254 additions and 800 deletions
Download Patch File Download Diff File Expand all files Collapse all files

50
tooling/pipeline/src/application_options/parse_arguments.cc
View File

@@ -15,8 +15,6 @@ std::optional<ApplicationOptions> ParseArguments(
opt("help,h", "Produce help message");
// Defaults sourced from SamplingOptions{} so the CLI and LlamaGenerator
// share a single source of truth — changing the struct updates both.
auto add_sampling_options = [&]() -> void {
const SamplingOptions sampling_defaults{};
opt("temperature",
@@ -53,7 +51,8 @@ std::optional<ApplicationOptions> ParseArguments(
prog_opts::value<std::string>()->default_value("pipeline.log"),
"Path for application logs");
opt("prompt-dir", prog_opts::value<std::string>()->default_value(""),
"Directory containing named prompt files (e.g. BREWERY_GENERATION.md)."
"Directory containing named prompt files (e.g. "
"BREWERY_GENERATION.md)."
" Required when not using --mocked.");
opt("location-count", prog_opts::value<uint32_t>()->default_value(10));
};
@@ -72,11 +71,11 @@ std::optional<ApplicationOptions> ParseArguments(
})();
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = title});
.phase = PipelinePhase::Startup,
.message = title});
logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = usage});
.phase = PipelinePhase::Startup,
.message = usage});
}
return std::nullopt;
}
@@ -91,8 +90,8 @@ std::optional<ApplicationOptions> ParseArguments(
help_stream << "\n" << desc;
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = help_stream.str()});
.phase = PipelinePhase::Startup,
.message = help_stream.str()});
}
return std::nullopt;
}
@@ -108,14 +107,16 @@ std::optional<ApplicationOptions> ParseArguments(
const std::string model_path = var_map["model"].as<std::string>();
const int n_gpu_layers = var_map["n-gpu-layers"].as<int>();
// Enforce mutual exclusivity before any further configuration is applied.
// Enforce mutual exclusivity before any further configuration is
// applied.
if (use_mocked && !model_path.empty()) {
const std::string msg =
"Invalid arguments: --mocked and --model are mutually exclusive";
"Invalid arguments: --mocked and --model are mutually "
"exclusive";
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
.message = msg});
.phase = PipelinePhase::Startup,
.message = msg});
} else {
std::cerr << msg << std::endl;
}
@@ -124,11 +125,12 @@ std::optional<ApplicationOptions> ParseArguments(
if (!use_mocked && model_path.empty()) {
const std::string msg =
"Invalid arguments: either --mocked or --model must be specified";
"Invalid arguments: either --mocked or --model must be "
"specified";
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
.message = msg});
.phase = PipelinePhase::Startup,
.message = msg});
} else {
std::cerr << msg << std::endl;
}
@@ -139,7 +141,8 @@ std::optional<ApplicationOptions> ParseArguments(
// generator has no use for it and should not require it to be present.
if (!use_mocked && options.pipeline.prompt_dir.empty()) {
const std::string msg =
"Invalid arguments: --prompt-dir is required when not using --mocked";
"Invalid arguments: --prompt-dir is required when not using "
"--mocked";
if (logger) {
logger->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
@@ -152,7 +155,6 @@ std::optional<ApplicationOptions> ParseArguments(
options.generator.use_mocked = use_mocked;
options.generator.model_path = model_path;
// options.generator.n_gpu_layers = n_gpu_layers;
// Only populate sampling config when the user explicitly overrides at
// least one value. Leaving it as std::nullopt lets LlamaGenerator fall
@@ -171,8 +173,8 @@ std::optional<ApplicationOptions> ParseArguments(
"Sampling parameters are ignored when using --mocked";
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Warn,
.phase = PipelinePhase::Startup,
.message = msg});
.phase = PipelinePhase::Startup,
.message = msg});
} else {
std::cerr << msg << std::endl;
}
@@ -197,8 +199,8 @@ std::optional<ApplicationOptions> ParseArguments(
exception.what();
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
.message = msg});
.phase = PipelinePhase::Startup,
.message = msg});
}
return std::nullopt;
} catch (...) {
@@ -206,8 +208,8 @@ std::optional<ApplicationOptions> ParseArguments(
"Failed to parse command-line arguments: unknown error";
if (logger) {
logger->Log(LogDTO{.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
.message = msg});
.phase = PipelinePhase::Startup,
.message = msg});
}
return std::nullopt;
}

8
tooling/pipeline/src/biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc
View File

@@ -1,6 +1,6 @@
/**
* @file biergarten_pipeline_orchestrator/biergarten_pipeline_orchestrator.cc
* @brief BiergartenDataGenerator constructor implementation.
* @brief BiergartenPipelineOrchestrator constructor implementation.
*/
#include "biergarten_pipeline_orchestrator.h"
@@ -12,9 +12,11 @@ BiergartenPipelineOrchestrator::BiergartenPipelineOrchestrator(
std::unique_ptr<IEnrichmentService> context_service,
std::unique_ptr<DataGenerator> generator,
std::unique_ptr<IExportService> exporter,
const ApplicationOptions &app_options)
std::unique_ptr<ICuratedDataService> curated_data_service,
const ApplicationOptions& application_options)
: logger_(std::move(logger)),
context_service_(std::move(context_service)),
generator_(std::move(generator)),
exporter_(std::move(exporter)),
application_options_(app_options) {}
curated_data_service_(std::move(curated_data_service)),
application_options_(application_options) {}

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

@@ -1,10 +1,11 @@
/**
* @file biergarten_pipeline_orchestrator/generate_breweries.cc
* @brief BiergartenDataGenerator::GenerateBreweries() implementation.
* @brief BiergartenPipelineOrchestrator::GenerateBreweries() implementation.
*/
#include <chrono>
#include <format>
#include <optional>
#include "biergarten_pipeline_orchestrator.h"
#include "services/logging/logger.h"
@@ -19,50 +20,69 @@ void BiergartenPipelineOrchestrator::GenerateBreweries(
size_t skipped_count = 0;
size_t export_failed_count = 0;
for (const auto& [location, region_context] : cities) {
const auto generate_record =
[this, &skipped_count](
const Location& location,
const std::string& region_context) -> std::optional<BreweryRecord> {
try {
const BreweryResult brewery =
generator_->GenerateBrewery(location, region_context);
const GeneratedBrewery gen{.location = location, .brewery = brewery};
generated_breweries_.push_back(gen);
try {
exporter_->ProcessRecord(gen);
} 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: {}",
location.city, location.country, export_exception.what())});
}
return BreweryRecord{.location = location, .brewery = brewery};
} catch (const std::exception& e) {
++skipped_count;
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("[Pipeline] Skipping city '{}' ({}): brewery generation failed: {}",
location.city, location.country, e.what())});
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("[Pipeline] Skipping city '{}' ({}): brewery "
"generation failed: {}",
location.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.location.city, record.location.country,
export_exception.what())});
}
};
for (const auto& [location, region_context] : cities) {
const std::optional<BreweryRecord> record =
generate_record(location, region_context);
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)});
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)});
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::Teardown,
.message = std::format("[Pipeline] Failed to export {} generated "
"brewery/breweries to SQLite",
export_failed_count)});
}
}

219
tooling/pipeline/src/biergarten_pipeline_orchestrator/generate_users.cc Normal file
View File

@@ -0,0 +1,219 @@
/**
* @file biergarten_pipeline_orchestrator/generate_users.cc
* @brief BiergartenPipelineOrchestrator::GenerateUsers() implementation.
*/
#include <cctype>
#include <chrono>
#include <format>
#include <iterator>
#include <optional>
#include <random>
#include <stdexcept>
#include <string>
#include <string_view>
#include <unordered_set>
#include "biergarten_pipeline_orchestrator.h"
#include "services/curated_data/curated_json_data_service.h"
#include "services/logging/logger.h"
namespace {
std::string Sanitize(std::string_view value) {
std::string out;
out.reserve(value.size());
for (const char character : value) {
if (std::isalnum(static_cast<unsigned char>(character)) != 0) {
out.push_back(static_cast<char>(
std::tolower(static_cast<unsigned char>(character))));
}
}
return out;
}
std::string BuildEmail(const Name& name,
std::unordered_set<std::string>& used_local_parts) {
const std::string base =
std::format("{}.{}", Sanitize(name.first_name), Sanitize(name.last_name));
std::string local_part = base;
uint32_t suffix = 1;
while (used_local_parts.contains(local_part)) {
local_part = std::format("{}{}", base, suffix);
++suffix;
}
used_local_parts.insert(local_part);
return std::format("{}@thebiergarten.app", local_part);
}
std::string GenerateDateOfBirth(std::mt19937& rng) {
using namespace std::chrono;
constexpr int kMinAge = 19;
constexpr int kMaxAge = 48;
constexpr int kMaxDayOffset = 364;
std::uniform_int_distribution<int> age_dist(kMinAge, kMaxAge);
std::uniform_int_distribution<int> day_offset_dist(0, kMaxDayOffset);
const year_month_day today{floor<days>(system_clock::now())};
const year_month_day birth_year_anchor{today.year() - years{age_dist(rng)},
today.month(), today.day()};
const sys_days birth_date =
sys_days{birth_year_anchor} - days{day_offset_dist(rng)};
const year_month_day birth_ymd{birth_date};
return std::format("{:04}-{:02}-{:02}", static_cast<int>(birth_ymd.year()),
static_cast<unsigned>(birth_ymd.month()),
static_cast<unsigned>(birth_ymd.day()));
}
std::optional<Name> SampleName(
const ForenamesByCountryMap& forenames_by_country,
const SurnamesByCountryMap& surnames_by_country,
const std::string& iso3166_1, std::mt19937& rng) {
const auto forenames_it = forenames_by_country.find(iso3166_1);
const auto surnames_it = surnames_by_country.find(iso3166_1);
if (forenames_it == forenames_by_country.end() ||
surnames_it == surnames_by_country.end() ||
forenames_it->second.empty() || surnames_it->second.empty()) {
return std::nullopt;
}
const ForenameList& forenames = forenames_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> surname_dist(0, surnames.size() - 1);
auto forename_it = forenames.begin();
std::advance(forename_it, forename_dist(rng));
auto surname_it = surnames.begin();
std::advance(surname_it, surname_dist(rng));
return Name{.first_name = forename_it->name,
.last_name = *surname_it,
.gender = forename_it->gender};
}
} // namespace
void BiergartenPipelineOrchestrator::GenerateUsers(
std::span<const EnrichedCity> cities) {
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration,
.message = "=== SAMPLE USER GENERATION ==="});
const PersonasList& personas = curated_data_service_->LoadPersonas();
if (personas.empty()) {
throw std::runtime_error(
"No personas available in personas.json for user generation");
}
const ForenamesByCountryMap& forenames_by_country =
curated_data_service_->LoadForenamesByCountry();
const SurnamesByCountryMap& surnames_by_country =
curated_data_service_->LoadSurnamesByCountry();
std::mt19937 rng(std::random_device{}());
std::uniform_int_distribution<size_t> persona_dist(0, personas.size() - 1);
generated_users_.clear();
size_t skipped_count = 0;
size_t export_failed_count = 0;
std::unordered_set<std::string> used_email_local_parts;
const auto generate_record =
[this, &rng, &skipped_count, &used_email_local_parts](
const EnrichedCity& city, const UserPersona& persona,
const Name& sampled_name) -> std::optional<UserRecord> {
try {
const UserResult user =
generator_->GenerateUser(city, persona, sampled_name);
return UserRecord{
.location = city.location,
.user = user,
.email = BuildEmail(sampled_name, used_email_local_parts),
.date_of_birth = GenerateDateOfBirth(rng),
};
} catch (const std::exception& e) {
++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.location.city, record.location.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)});
}
}

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

@@ -1,20 +1,21 @@
/**
* @file biergarten_pipeline_orchestrator/log_results.cc
* @brief BiergartenDataGenerator::LogResults() implementation.
* @brief BiergartenPipelineOrchestrator::LogResults() implementation.
*/
#include <boost/json/array.hpp>
#include <chrono>
#include <format>
#include "../../includes/json_handling/pretty_print.h"
#include "biergarten_pipeline_orchestrator.h"
#include "json_handling/pretty_print.h"
#include "services/logging/logger.h"
void BiergartenPipelineOrchestrator::LogResults() const {
boost::json::array output;
boost::json::array brewery_output;
for (const auto& [location, brewery] : generated_breweries_) {
output.push_back(boost::json::object{
brewery_output.push_back(boost::json::object{
{"name_en", brewery.name_en},
{"description_en", brewery.description_en},
{"name_local", brewery.name_local},
@@ -29,9 +30,30 @@ void BiergartenPipelineOrchestrator::LogResults() const {
}}});
}
std::ostringstream oss;
PrettyPrint(oss, output);
std::ostringstream brewery_oss;
PrettyPrint(brewery_oss, brewery_output);
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown,
.message = oss.str()});
.message = brewery_oss.str()});
boost::json::array user_output;
for (const auto& generated_user : generated_users_) {
user_output.push_back(boost::json::object{
{"first_name", generated_user.user.first_name},
{"last_name", generated_user.user.last_name},
{"gender", generated_user.user.gender},
{"username", generated_user.user.username},
{"bio", generated_user.user.bio},
{"activity_weight", generated_user.user.activity_weight},
{"email", generated_user.email},
{"date_of_birth", generated_user.date_of_birth},
});
}
std::ostringstream user_oss;
PrettyPrint(user_oss, user_output);
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown,
.message = user_oss.str()});
}

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

@@ -1,17 +1,17 @@
/**
* @file biergarten_pipeline_orchestrator/query_cities_with_countries.cc
* @brief BiergartenDataGenerator::QueryCitiesWithCountries() implementation.
* @brief BiergartenPipelineOrchestrator::QueryCitiesWithCountries()
* implementation.
*/
#include <algorithm>
#include <chrono>
#include <filesystem>
#include <format>
#include <iterator>
#include <random>
#include "biergarten_pipeline_orchestrator.h"
#include "json_handling/json_loader.h"
#include "services/curated_data/curated_json_data_service.h"
#include "services/logging/logger.h"
std::vector<Location>
@@ -20,9 +20,8 @@ BiergartenPipelineOrchestrator::QueryCitiesWithCountries() {
.phase = PipelinePhase::Startup,
.message = "=== GEOGRAPHIC DATA OVERVIEW ==="});
const std::filesystem::path locations_path = "locations.json";
auto all_locations = JsonLoader::LoadLocations(locations_path, logger_);
const std::vector<Location>& all_locations =
curated_data_service_->LoadLocations();
const size_t sample_count = std::min(
static_cast<size_t>(application_options_.pipeline.location_count),

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

@@ -1,6 +1,6 @@
/**
* @file biergarten_pipeline_orchestrator/run.cc
* @brief BiergartenDataGenerator::Run() implementation.
* @brief BiergartenPipelineOrchestrator::Run() implementation.
*/
#include <chrono>
@@ -22,42 +22,40 @@ bool BiergartenPipelineOrchestrator::Run() {
for (auto& city : cities) {
try {
std::string region_context = context_service_->GetLocationContext(city);
// logger_->Log(LogLevel::Debug, PipelinePhase::UserGeneration,
// "[Pipeline] Context for '" + city.city + "' (" +
// city.iso3166_2 + ") gathered:\n" + region_context);
enriched.push_back(
EnrichedCity{.location = std::move(city),
.region_context = std::move(region_context)});
} catch (const std::exception& exception) {
++skipped_count;
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipping city '{}' ({}): context lookup failed: {}",
city.city, city.country, exception.what())});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"[Pipeline] Skipping city '{}' ({}): context "
"lookup failed: {}",
city.city, city.country, exception.what())});
}
}
if (skipped_count > 0) {
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"[Pipeline] Skipped {} city/cities due to context lookup errors",
skipped_count)});
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format("[Pipeline] Skipped {} city/cities due "
"to context lookup errors",
skipped_count)});
}
this->GenerateUsers(enriched);
this->GenerateBreweries(enriched);
exporter_->Finalize();
this->LogResults();
return true;
} catch (const std::exception& e) {
logger_->Log(
{.level = LogLevel::Error,
.phase = PipelinePhase::Teardown,
.message =
std::format("Pipeline execution failed with error: {}", e.what())});
logger_->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Teardown,
.message = std::format(
"Pipeline execution failed with error: {}", e.what())});
return false;
}
}

57
tooling/pipeline/src/data_generation/llama/generate_brewery.cc
View File

@@ -12,6 +12,7 @@
#include <string_view>
#include <vector>
#include "data_generation/json_grammars.h"
#include "data_generation/llama_generator.h"
#include "data_generation/llama_generator_helpers.h"
@@ -32,19 +33,6 @@ static std::string FormatLocalLanguageCodes(
return formatted;
}
// GBNF grammar for structured brewery JSON output.
// @TODO move to a separate gbnf file if it grows in complexity or is shared
// across modules.
static constexpr std::string_view kBreweryJsonGrammar = R"json_brewery(
root ::= thought-block "{" ws "\"name_en\"" ws ":" ws string ws "," ws "\"description_en\"" ws ":" ws string ws "," ws "\"name_local\"" ws ":" ws string ws "," ws "\"description_local\"" ws ":" ws string ws "}" ws
thought-block ::= [^{]*
ws ::= [ \t\n\r]*
string ::= "\"" char+ "\""
char ::= [^"\\\x7F\x00-\x1F] | [\\] escape
escape ::= ["\\/bfnrt] | "u" hex hex hex hex
hex ::= [0-9a-fA-F]
)json_brewery";
static constexpr int kBreweryInitialMaxTokens = 2800;
BreweryResult LlamaGenerator::GenerateBrewery(
@@ -83,15 +71,14 @@ BreweryResult LlamaGenerator::GenerateBrewery(
/**
* RETRY LOOP with validation and error correction
* Attempts to generate valid brewery data up to 3 times, with feedback-based
* refinement
* Attempts to generate valid brewery data up to 3 times, with
* feedback-based refinement
*/
constexpr int max_attempts = 3;
std::string raw;
std::string last_error;
// Token budget: too small risks truncating valid JSON mid-string.
// Start conservatively but allow adaptive increases on truncation.
int max_tokens = kBreweryInitialMaxTokens;
// Limit output length to keep it concise and focused
@@ -104,7 +91,7 @@ BreweryResult LlamaGenerator::GenerateBrewery(
{.level = LogLevel::Debug,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("LlamaGenerator: raw output (attempt {}): {}",
attempt + 1, raw)});
attempt + 1, raw)});
}
// Validate output: parse JSON and check required fields
@@ -120,8 +107,9 @@ BreweryResult LlamaGenerator::GenerateBrewery(
logger_->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format("LlamaGenerator: successfully generated brewery data on attempt {}",
attempt + 1)});
.message = std::format("LlamaGenerator: successfully generated "
"brewery data on attempt {}",
attempt + 1)});
}
return brewery;
@@ -134,33 +122,38 @@ BreweryResult LlamaGenerator::GenerateBrewery(
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message =
std::format("LlamaGenerator: malformed brewery JSON (attempt {}): {}",
.message = std::format(
"LlamaGenerator: malformed brewery JSON (attempt {}): {}",
attempt + 1, *validation_error)});
}
// Update prompt with error details to guide LLM toward correct output.
user_prompt = std::format(
"Your previous response was invalid. Error: {}\nReturn the thought "
"process before the JSON if needed, then return ONLY valid JSON with "
"exactly these keys, in this exact order: {{\"name_en\": \"<English "
"process before the JSON if needed, then return ONLY valid JSON "
"with "
"exactly these keys, in this exact order: {{\"name_en\": "
"\"<English "
"brewery name>\", \"description_en\": \"<English single-paragraph "
"description>\", \"name_local\": \"<local-language brewery name>\", "
"description>\", \"name_local\": \"<local-language brewery "
"name>\", "
"\"description_local\": \"<local-language single-paragraph "
"description>\"}}.\nDo not include markdown, comments, extra keys, or "
"literal placeholder values.\n\nKeep the JSON strings concise enough "
"description>\"}}.\nDo not include markdown, comments, extra keys, "
"or "
"literal placeholder values.\n\nKeep the JSON strings concise "
"enough "
"to fit within the token budget.\n\n{}",
*validation_error, retry_location);
}
// All retry attempts exhausted: log failure and throw exception
if (logger_) {
logger_->Log(
{.level = LogLevel::Error,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format(
"LlamaGenerator: malformed brewery response after {} attempts: {}",
max_attempts, last_error.empty() ? raw : last_error)});
logger_->Log({.level = LogLevel::Error,
.phase = PipelinePhase::BreweryAndBeerGeneration,
.message = std::format(
"LlamaGenerator: malformed brewery "
"response after {} attempts: {}",
max_attempts, last_error.empty() ? raw : last_error)});
}
throw std::runtime_error("LlamaGenerator: malformed brewery response");
}

117
tooling/pipeline/src/data_generation/llama/generate_user.cc
View File

@@ -1,24 +1,115 @@
/**
* @file data_generation/llama/generate_user.cc
* @brief Generates locale-aware user profiles with strict two-line formatting,
* retry handling, and output sanitization for downstream parsing.
* @brief Builds persona/name-grounded user prompts, performs retry-based
* inference, and validates structured JSON output for user records.
*/
#include <format>
#include <optional>
#include <stdexcept>
#include <string>
#include <string_view>
#include "data_generation/json_grammars.h"
#include "data_generation/llama_generator.h"
#include "data_generation/llama_generator_helpers.h"
// TODO: Implement locale-aware user profile generation.
// Current implementation returns a hardcoded test value and ignores the
// locale parameter. Future implementation should:
// 1. Load a USER_GENERATION.md prompt template with locale context
// 2. Perform LLM inference with locale-specific username/bio generation
// 3. Parse and validate JSON output with retry handling (similar to brewery)
// 4. Return locale-aware username and biography
UserResult LlamaGenerator::GenerateUser(const std::string& locale) {
return {.username = "test_user",
.bio = std::format("This is a test user profile from {}.", locale)};
static constexpr int kUserInitialMaxTokens = 1200;
UserResult LlamaGenerator::GenerateUser(const EnrichedCity& city,
const UserPersona& persona,
const Name& name) {
std::string style_affinities;
for (const std::string& style : persona.style_affinities) {
if (!style_affinities.empty()) {
style_affinities += ", ";
}
style_affinities += style;
}
const std::string system_prompt = prompt_directory_->Load("USER_GENERATION");
std::string user_prompt = std::format(
"## NAME:\n\n{} {}\n\n"
"## GENDER:\n\n{}\n\n"
"## CITY:\n\n{}\n\n"
"## COUNTRY:\n\n{}\n\n"
"## PERSONA:\n\n{}\n\n"
"## PERSONA DESCRIPTION:\n\n{}\n\n"
"## STYLE AFFINITIES:\n\n{}",
name.first_name, name.last_name, name.gender, city.location.city,
city.location.country, persona.name, persona.description,
style_affinities);
const std::string retry_context = std::format(
"Name: {} {}\nCity: {}, {}\nPersona: {}", name.first_name, name.last_name,
city.location.city, city.location.country, persona.name);
constexpr int max_attempts = 3;
std::string raw;
std::string last_error;
int max_tokens = kUserInitialMaxTokens;
for (int attempt = 0; attempt < max_attempts; ++attempt) {
raw = this->Infer(system_prompt, user_prompt, max_tokens, kUserJsonGrammar);
if (logger_) {
logger_->Log(
{.level = LogLevel::Debug,
.phase = PipelinePhase::UserGeneration,
.message = std::format("LlamaGenerator: raw output (attempt {}): {}",
attempt + 1, raw)});
}
UserResult user;
const std::optional<std::string> validation_error =
ValidateUserJson(raw, user);
if (!validation_error.has_value()) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration,
.message = std::format("LlamaGenerator: successfully "
"generated user data on attempt {}",
attempt + 1)});
}
user.first_name = name.first_name;
user.last_name = name.last_name;
user.gender = name.gender;
return user;
}
last_error = *validation_error;
if (logger_) {
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"LlamaGenerator: malformed user JSON (attempt {}): {}",
attempt + 1, *validation_error)});
}
user_prompt = std::format(
"Your previous response was invalid. Error: {}\nReturn the thought "
"process before the JSON if needed, then return ONLY valid JSON "
"with "
"exactly these keys, in this exact order: {{\"username\": "
"\"<handle "
"derived from the given name>\", \"bio\": \"<first-person bio "
"grounded in the persona>\", \"activity_weight\": <number between "
"0 "
"and 1>}}.\nDo not include markdown, comments, extra keys, or "
"literal placeholder values.\n\n{}",
*validation_error, retry_context);
}
if (logger_) {
logger_->Log({.level = LogLevel::Error,
.phase = PipelinePhase::UserGeneration,
.message = std::format(
"LlamaGenerator: malformed user response "
"after {} attempts: {}",
max_attempts, last_error.empty() ? raw : last_error)});
}
throw std::runtime_error("LlamaGenerator: malformed user response");
}

62
tooling/pipeline/src/data_generation/llama/helpers.cc
View File

@@ -9,13 +9,16 @@
#include <boost/json.hpp>
#include <cctype>
#include <optional>
#include <span>
#include <stdexcept>
#include <string>
#include <string_view>
#include <vector>
#include <llama.h>
#include "data_generation/llama_generator_helpers.h"
#include "llama.h"
namespace {
/**
* String trimming: removes leading and trailing whitespace
@@ -76,7 +79,7 @@ bool ReadRequiredTrimmedStringField(const boost::json::object& obj,
return !out.empty();
}
bool HasSchemaPlaceholder(const std::array<std::string*, 4>& values) {
bool HasSchemaPlaceholder(std::span<std::string* const> values) {
for (const std::string* value : values) {
std::string lowered = *value;
std::ranges::transform(lowered, lowered.begin(),
@@ -207,3 +210,58 @@ std::optional<std::string> ValidateBreweryJson(const std::string& raw,
return std::nullopt;
}
std::optional<std::string> ValidateUserJson(const std::string& raw,
UserResult& user_out) {
boost::system::error_code error_code;
const std::string_view raw_view(raw);
const size_t opening_brace = raw_view.find('{');
if (opening_brace == std::string_view::npos) {
return "JSON parse error: missing opening brace '{'";
}
const std::string_view json_payload = raw_view.substr(opening_brace);
boost::json::value json_value = boost::json::parse(json_payload, error_code);
if (error_code) {
return "JSON parse error: " + error_code.message();
}
if (!json_value.is_object()) {
return "JSON root must be an object";
}
const auto& obj = json_value.get_object();
if (obj.size() != 3) {
return "JSON object must contain exactly three keys";
}
std::string validation_error;
if (!ReadRequiredTrimmedStringField(obj, "username", user_out.username,
&validation_error)) {
return validation_error;
}
if (!ReadRequiredTrimmedStringField(obj, "bio", user_out.bio,
&validation_error)) {
return validation_error;
}
const boost::json::value* activity_weight_field =
obj.if_contains("activity_weight");
if (activity_weight_field == nullptr || !activity_weight_field->is_number()) {
return "Missing or invalid numeric field: activity_weight";
}
const double activity_weight = activity_weight_field->to_number<double>();
if (activity_weight < 0.0 || activity_weight > 1.0) {
return "activity_weight must be between 0 and 1";
}
user_out.activity_weight = static_cast<float>(activity_weight);
const std::array schema_placeholders = {&user_out.username, &user_out.bio};
if (HasSchemaPlaceholder(schema_placeholders)) {
return "JSON appears to be a schema placeholder, not content";
}
return std::nullopt;
}

3
tooling/pipeline/src/data_generation/llama/infer.cc
View File

@@ -14,9 +14,10 @@
#include <string_view>
#include <vector>
#include <llama.h>
#include "data_generation/llama_generator.h"
#include "data_generation/llama_generator_helpers.h"
#include "llama.h"
static constexpr size_t kPromptTokenSlack = 8;
// Minimum tokens to keep when using top-p sampling. Ensures at least one

3
tooling/pipeline/src/data_generation/llama/llama_generator.cc
View File

@@ -11,8 +11,9 @@
#include <stdexcept>
#include <string>
#include <llama.h>
#include "data_model/models.h"
#include "llama.h"
static constexpr uint32_t kMaxContextSize = 32768U;

5
tooling/pipeline/src/data_generation/llama/load.cc
View File

@@ -10,9 +10,10 @@
#include <string>
#include <utility>
#include <ggml-backend.h>
#include <llama.h>
#include "data_generation/llama_generator.h"
#include "ggml-backend.h"
#include "llama.h"
// Maximum batch size for decode operations. Capping the batch prevents
// excessive memory allocation while maintaining inference performance.

10
tooling/pipeline/src/data_generation/mock/deterministic_hash.cc
View File

@@ -14,3 +14,13 @@ size_t MockGenerator::DeterministicHash(const Location& location) {
boost::hash_combine(seed, location.country);
return seed;
}
size_t MockGenerator::DeterministicHash(const Location& location,
const UserPersona& persona,
const Name& name) {
size_t seed = DeterministicHash(location);
boost::hash_combine(seed, persona.name);
boost::hash_combine(seed, name.first_name);
boost::hash_combine(seed, name.last_name);
return seed;
}

29
tooling/pipeline/src/data_generation/mock/generate_user.cc
View File

@@ -1,22 +1,31 @@
/**
* @file data_generation/mock/generate_user.cc
* @brief Generates deterministic mock user profiles by hashing locale values
* into predefined username and bio collections.
* @brief Generates deterministic mock user profiles by hashing city, persona,
* and sampled name into predefined username and bio collections.
*/
#include <functional>
#include <string>
#include <string_view>
#include "data_generation/mock_generator.h"
UserResult MockGenerator::GenerateUser(const std::string& locale) {
const size_t hash = std::hash<std::string>{}(locale);
UserResult MockGenerator::GenerateUser(const EnrichedCity& city,
const UserPersona& persona,
const Name& name) {
const size_t hash = DeterministicHash(city.location, persona, name);
UserResult result;
const std::string_view username = kUsernames[hash % kUsernames.size()];
const std::string_view bio = kBios[hash / kBioHashStride % kBios.size()];
result.username = username;
result.bio = bio;
return result;
const float activity_weight =
static_cast<float>(hash / kActivityWeightHashStride %
kActivityWeightHashRange) /
static_cast<float>(kActivityWeightHashRange);
return {
.first_name = name.first_name,
.last_name = name.last_name,
.gender = name.gender,
.username = std::string(username),
.bio = std::string(bio),
.activity_weight = activity_weight,
};
}

110
tooling/pipeline/src/json_handling/json_loader.cc
View File

@@ -1,110 +0,0 @@
/**
* @file json_handling/json_loader.cc
* @brief Parses curated location JSON input into strongly typed Location
* records with strict field validation and descriptive error reporting.
*/
#include "json_handling/json_loader.h"
#include <format>
#include "services/logging/logger.h"
#include <iostream>
#include <boost/json.hpp>
#include <fstream>
#include <sstream>
#include <stdexcept>
#include <string>
#include <string_view>
static std::string ReadRequiredString(const boost::json::object& object,
const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_string()) {
throw std::runtime_error(
std::format("Missing or invalid string field: {}", key));
}
const std::string_view text = value->as_string();
return std::string(text);
}
static double ReadRequiredNumber(const boost::json::object& object,
const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_number()) {
throw std::runtime_error(
std::format("Missing or invalid numeric field: {}", key));
}
return value->to_number<double>();
}
static std::vector<std::string> ReadRequiredStringArray(
const boost::json::object& object, const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_array()) {
throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key));
}
const auto& array = value->as_array();
std::vector<std::string> items;
items.reserve(array.size());
for (const auto& item : array) {
if (!item.is_string()) {
throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key));
}
items.emplace_back(item.as_string());
}
return items;
}
std::vector<Location> JsonLoader::LoadLocations(
const std::filesystem::path& filepath, std::shared_ptr<ILogger> logger) {
std::ifstream input(filepath);
if (!input.is_open()) {
throw std::runtime_error("Failed to open locations file: " +
filepath.string());
}
std::stringstream buffer;
buffer << input.rdbuf();
const std::string content = buffer.str();
boost::system::error_code error;
boost::json::value root = boost::json::parse(content, error);
if (error) {
throw std::runtime_error("Failed to parse locations JSON: " +
error.message());
}
if (!root.is_array()) {
throw std::runtime_error(
"Invalid locations JSON: root element must be an array");
}
std::vector<Location> locations;
const auto& items = root.as_array();
locations.reserve(items.size());
for (const auto& item : items) {
if (!item.is_object()) {
throw std::runtime_error(
"Invalid locations JSON: each entry must be an object");
}
const auto& object = item.as_object();
locations.push_back(Location{
.city = ReadRequiredString(object, "city"),
.state_province = ReadRequiredString(object, "state_province"),
.iso3166_2 = ReadRequiredString(object, "iso3166_2"),
.country = ReadRequiredString(object, "country"),
.iso3166_1 = ReadRequiredString(object, "iso3166_1"),
.local_languages = ReadRequiredStringArray(object, "local_languages"),
.latitude = ReadRequiredNumber(object, "latitude"),
.longitude = ReadRequiredNumber(object, "longitude"),
});
}
return locations;
}

132
tooling/pipeline/src/main.cc
View File

@@ -18,25 +18,7 @@
#include <string>
#include <thread>
#include "biergarten_pipeline_orchestrator.h"
#include "concurrency/bounded_channel.h"
#include "data_generation/llama_generator.h"
#include "data_generation/mock_generator.h"
#include "data_generation/prompt_formatting/gemma4_jinja_prompt_formatter.h"
#include "data_model/models.h"
#include "llama_backend_state.h"
#include "services/database/export_service.h"
#include "services/database/sqlite_export_service.h"
#include "services/datetime/timer.h"
#include "services/enrichment/enrichment_service.h"
#include "services/enrichment/mock_enrichment.h"
#include "services/enrichment/wikipedia_service.h"
#include "services/logging/log_dispatcher.h"
#include "services/logging/log_entry.h"
#include "services/logging/log_producer.h"
#include "services/logging/logger.h"
#include "services/prompting/prompt_directory.h"
#include "web_client/http_web_client.h"
#include "biergarten_pipeline.h"
namespace di = boost::di;
@@ -53,6 +35,7 @@ int main(const int argc, char** argv) {
std::make_shared<LogProducer>(log_channel);
std::thread log_thread([&log_dispatcher] { log_dispatcher->Run(); });
auto shutdown = [&](const int exit_code) {
log_channel.Close();
log_thread.join();
@@ -66,8 +49,8 @@ int main(const int argc, char** argv) {
const LlamaBackendState llama_backend_state;
#endif
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "STARTING PIPELINE"});
const std::optional<ApplicationOptions> parsed_options =
@@ -89,8 +72,8 @@ int main(const int argc, char** argv) {
prompt_directory = std::make_unique<PromptDirectory>(
options.pipeline.prompt_dir, log_producer);
} catch (const std::exception& dir_error) {
log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Startup,
.message = std::format("Invalid --prompt-dir: {}",
dir_error.what())});
@@ -103,12 +86,34 @@ int main(const int argc, char** argv) {
di::bind<ApplicationOptions>().to(options),
di::bind<std::string>().to(model_path),
di::bind<IExportService>().to<SqliteExportService>(),
di::bind<ICuratedDataService>().to(
[options, &log_producer]() -> std::unique_ptr<ICuratedDataService> {
if (options.generator.use_mocked) {
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Curated data: mock"});
return std::make_unique<MockCuratedDataService>();
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Curated data: JsonLoader"});
return std::make_unique<CuratedJsonDataService>(
CuratedDataFilePaths{
.locations_path = "locations.json",
.personas_path = "personas.json",
.forenames_path = "forenames-by-country.json",
.surnames_path = "surnames-by-country.json",
});
}),
di::bind<IPromptFormatter>().to([options, log_producer] {
if (options.generator.use_mocked) {
{
log_producer->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.phase = PipelinePhase::Startup,
.message = "Prompt formatter: none (mock mode)"});
}
return std::unique_ptr<IPromptFormatter>(nullptr);
@@ -116,7 +121,7 @@ int main(const int argc, char** argv) {
{
log_producer->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.phase = PipelinePhase::Startup,
.message = "Prompt formatter: Gemma4JinjaPromptFormatter"});
}
return std::unique_ptr<IPromptFormatter>(
@@ -125,17 +130,17 @@ int main(const int argc, char** argv) {
di::bind<WebClient>().to([options, log_producer] {
if (options.generator.use_mocked) {
{
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Web client: none (mock mode)"});
}
return std::unique_ptr<WebClient>(nullptr);
}
{
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Web client: HttpWebClient"});
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Web client: HttpWebClient"});
return std::unique_ptr<WebClient>(
std::make_unique<HttpWebClient>(log_producer));
}),
@@ -143,18 +148,17 @@ int main(const int argc, char** argv) {
[options, &log_producer](
const auto& inj) -> std::unique_ptr<IEnrichmentService> {
if (options.generator.use_mocked) {
{
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Enrichment: mock"});
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Enrichment: mock"});
return std::make_unique<MockEnrichmentService>();
}
{
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Enrichment: Wikipedia"});
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Enrichment: Wikipedia"});
return std::make_unique<WikipediaEnrichmentService>(
inj.template create<std::unique_ptr<WebClient>>(),
log_producer);
@@ -163,23 +167,22 @@ int main(const int argc, char** argv) {
[&options, &model_path, &sampling, &prompt_directory,
&log_producer](const auto& inj) -> std::unique_ptr<DataGenerator> {
if (options.generator.use_mocked) {
{
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Generator: mock"});
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = "Generator: mock"});
return std::make_unique<MockGenerator>();
}
{
log_producer->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = std::format(
"Generator: LlamaGenerator | model={} | temp={:.2f} "
"top_p={:.2f} top_k={} n_ctx={} seed={}",
model_path, sampling.temperature, sampling.top_p,
sampling.top_k, sampling.n_ctx, sampling.seed)});
}
log_producer->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = std::format(
"Generator: LlamaGenerator | model={} | "
"temp={:.2f} top_p={:.2f} top_k={} n_ctx={} seed={}",
model_path, sampling.temperature, sampling.top_p,
sampling.top_k, sampling.n_ctx, sampling.seed)});
return std::make_unique<LlamaGenerator>(
options, model_path, log_producer,
inj.template create<std::unique_ptr<IPromptFormatter>>(),
@@ -190,23 +193,22 @@ int main(const int argc, char** argv) {
injector.create<std::unique_ptr<BiergartenPipelineOrchestrator>>();
if (!orchestrator->Run()) {
log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Teardown,
log_producer->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Teardown,
.message = "Pipeline execution failed"});
return shutdown(EXIT_FAILURE);
}
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown,
log_producer->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Teardown,
.message = std::format("Pipeline complete in {} ms",
timer.Elapsed())});
return shutdown(EXIT_SUCCESS);
} catch (const std::exception& exception) {
const LogDTO log_entry{.level = LogLevel::Error,
.phase = PipelinePhase::Teardown,
.message = exception.what()};
.phase = PipelinePhase::Teardown,
.message = exception.what()};
if (log_producer) {
log_producer->Log(log_entry);
} else {
@@ -215,4 +217,4 @@ int main(const int argc, char** argv) {
return shutdown(EXIT_FAILURE);
}
}
}

264
tooling/pipeline/src/services/curated_data/curated_json_data_service.cc Normal file
View File

@@ -0,0 +1,264 @@
/**
* @file json_handling/json_loader.cc
* @brief Parses curated location JSON input into strongly typed Location
* records with strict field validation and descriptive error reporting.
*/
#include "services/curated_data/curated_json_data_service.h"
#include <boost/json.hpp>
#include <format>
#include <fstream>
#include <iostream>
#include <sstream>
#include <stdexcept>
#include <string>
#include <string_view>
#include <utility>
#include <vector>
#include "services/logging/logger.h"
static std::string ReadRequiredString(const boost::json::object& object,
const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_string()) {
throw std::runtime_error(
std::format("Missing or invalid string field: {}", key));
}
const std::string_view text = value->as_string();
return std::string(text);
}
static double ReadRequiredNumber(const boost::json::object& object,
const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_number()) {
throw std::runtime_error(
std::format("Missing or invalid numeric field: {}", key));
}
return value->to_number<double>();
}
static std::vector<std::string> ReadRequiredStringArray(
const boost::json::object& object, const char* key) {
const boost::json::value* value = object.if_contains(key);
if (value == nullptr || !value->is_array()) {
throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key));
}
const auto& array = value->as_array();
std::vector<std::string> items;
items.reserve(array.size());
for (const auto& item : array) {
if (!item.is_string()) {
throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key));
}
items.emplace_back(item.as_string());
}
return items;
}
namespace {
boost::json::value ParseJsonFile(const std::filesystem::path& filepath,
const char* what) {
std::ifstream input(filepath);
if (!input.is_open()) {
throw std::runtime_error(
std::format("Failed to open {} file: {}", what, filepath.string()));
}
std::stringstream buffer;
buffer << input.rdbuf();
boost::system::error_code error;
boost::json::value root = boost::json::parse(buffer.str(), error);
if (error) {
throw std::runtime_error(
std::format("Failed to parse {} JSON: {}", what, error.message()));
}
return root;
}
/**
* @brief Returns the first element of a string array field, falling back to
* `fallback_key` if `key` is missing/empty (some source entries only have a
* "localized" form and no "romanized" form).
*/
std::string ReadFirstOfStringArray(const boost::json::object& object,
const char* key, const char* fallback_key) {
for (const char* candidate_key : {key, fallback_key}) {
const boost::json::value* value = object.if_contains(candidate_key);
if (value == nullptr || !value->is_array()) {
continue;
}
const auto& array = value->as_array();
if (!array.empty() && array.front().is_string()) {
return std::string(array.front().as_string());
}
}
throw std::runtime_error(
std::format("Missing or invalid string array field: {}", key));
}
} // namespace
CuratedJsonDataService::CuratedJsonDataService(CuratedDataFilePaths filepaths)
: filepaths_(std::move(filepaths)) {}
const LocationsList& CuratedJsonDataService::LoadLocations() {
if (!cache_.locations.empty()) {
return cache_.locations;
}
const boost::json::value root =
ParseJsonFile(filepaths_.locations_path, "locations");
if (!root.is_array()) {
throw std::runtime_error(
"Invalid locations JSON: root element must be an array");
}
LocationsList locations;
const auto& items = root.as_array();
locations.reserve(items.size());
for (const auto& item : items) {
if (!item.is_object()) {
throw std::runtime_error(
"Invalid locations JSON: each entry must be an object");
}
const auto& object = item.as_object();
locations.push_back(Location{
.city = ReadRequiredString(object, "city"),
.state_province = ReadRequiredString(object, "state_province"),
.iso3166_2 = ReadRequiredString(object, "iso3166_2"),
.country = ReadRequiredString(object, "country"),
.iso3166_1 = ReadRequiredString(object, "iso3166_1"),
.local_languages = ReadRequiredStringArray(object, "local_languages"),
.latitude = ReadRequiredNumber(object, "latitude"),
.longitude = ReadRequiredNumber(object, "longitude"),
});
}
cache_.locations = std::move(locations);
return cache_.locations;
}
const PersonasList& CuratedJsonDataService::LoadPersonas() {
if (!cache_.personas.empty()) {
return cache_.personas;
}
const boost::json::value root =
ParseJsonFile(filepaths_.personas_path, "personas");
if (!root.is_array()) {
throw std::runtime_error(
"Invalid personas JSON: root element must be an array");
}
PersonasList personas;
const auto& items = root.as_array();
personas.reserve(items.size());
for (const auto& item : items) {
if (!item.is_object()) {
throw std::runtime_error(
"Invalid personas JSON: each entry must be an object");
}
const auto& object = item.as_object();
personas.push_back(UserPersona{
.name = ReadRequiredString(object, "name"),
.description = ReadRequiredString(object, "description"),
.style_affinities = ReadRequiredStringArray(object, "style_affinities"),
});
}
cache_.personas = std::move(personas);
return cache_.personas;
}
const ForenamesByCountryMap& CuratedJsonDataService::LoadForenamesByCountry() {
if (!cache_.forenames_by_country.empty()) {
return cache_.forenames_by_country;
}
const boost::json::value root =
ParseJsonFile(filepaths_.forenames_path, "forenames-by-country");
if (!root.is_object()) {
throw std::runtime_error(
"Invalid forenames-by-country JSON: root element must be an object "
"keyed by ISO 3166-1 country code");
}
ForenamesByCountryMap forenames_by_country;
for (const auto& [country_code, regions] : root.as_object()) {
if (!regions.is_array()) {
continue;
}
ForenameList entries;
for (const auto& region : regions.as_array()) {
if (!region.is_object()) {
continue;
}
const boost::json::value* names = region.as_object().if_contains("names");
if (names == nullptr || !names->is_array()) {
continue;
}
for (const auto& name_value : names->as_array()) {
if (!name_value.is_object()) {
continue;
}
const auto& name_object = name_value.as_object();
entries.emplace_back(ForenameEntry{
.name =
ReadFirstOfStringArray(name_object, "romanized", "localized"),
.gender = ReadRequiredString(name_object, "gender"),
});
}
}
forenames_by_country.emplace(country_code, std::move(entries));
}
cache_.forenames_by_country = std::move(forenames_by_country);
return cache_.forenames_by_country;
}
const SurnamesByCountryMap& CuratedJsonDataService::LoadSurnamesByCountry() {
if (!cache_.surnames_by_country.empty()) {
return cache_.surnames_by_country;
}
const boost::json::value root =
ParseJsonFile(filepaths_.surnames_path, "surnames-by-country");
if (!root.is_object()) {
throw std::runtime_error(
"Invalid surnames-by-country JSON: root element must be an object "
"keyed by ISO 3166-1 country code");
}
SurnamesByCountryMap surnames_by_country;
for (const auto& [country_code, name_entries] : root.as_object()) {
if (!name_entries.is_array()) continue;
SurnameList surnames;
for (const auto& name_value : name_entries.as_array()) {
if (!name_value.is_object()) {
continue;
}
surnames.emplace_back(ReadFirstOfStringArray(name_value.as_object(),
"romanized", "localized"));
}
surnames_by_country.emplace(country_code, std::move(surnames));
}
cache_.surnames_by_country = std::move(surnames_by_country);
return cache_.surnames_by_country;
}

98
tooling/pipeline/src/services/curated_data/mock_curated_data_service.cc Normal file
View File

@@ -0,0 +1,98 @@
/**
* @file services/curated_data/mock_curated_data_service.cc
* @brief Fixed in-memory location, persona, and name dataset for mock mode.
*/
#include "services/curated_data/mock_curated_data_service.h"
MockCuratedDataService::MockCuratedDataService()
: locations_{
Location{.city = "Portland",
.state_province = "Oregon",
.iso3166_2 = "US-OR",
.country = "United States",
.iso3166_1 = "US",
.local_languages = {"en"},
.latitude = 45.5152,
.longitude = -122.6784},
Location{.city = "Munich",
.state_province = "Bavaria",
.iso3166_2 = "DE-BY",
.country = "Germany",
.iso3166_1 = "DE",
.local_languages = {"de"},
.latitude = 48.1351,
.longitude = 11.5820},
Location{.city = "Lyon",
.state_province = "Auvergne-Rhone-Alpes",
.iso3166_2 = "FR-ARA",
.country = "France",
.iso3166_1 = "FR",
.local_languages = {"fr"},
.latitude = 45.7640,
.longitude = 4.8357},
Location{.city = "Brussels",
.state_province = "Brussels-Capital",
.iso3166_2 = "BE-BRU",
.country = "Belgium",
.iso3166_1 = "BE",
.local_languages = {"nl", "fr"},
.latitude = 50.8503,
.longitude = 4.3517},
},
personas_{
UserPersona{.name = "Hophead Explorer",
.description = "Chases hop-forward IPAs and seeks out "
"taprooms with rotating drafts.",
.style_affinities = {"IPA", "Pale Ale"}},
UserPersona{.name = "Lager Traditionalist",
.description = "Prefers clean, balanced lagers and "
"classic pub styles.",
.style_affinities = {"Lager", "Pilsner"}},
UserPersona{.name = "Sour Curious",
.description = "Seeks out wild ales, sours, and "
"barrel-aged experiments.",
.style_affinities = {"Sour", "Wild Ale"}},
},
forenames_by_country_{
{"US",
ForenameList{
ForenameEntry{.name = "James", .gender = "M"},
ForenameEntry{.name = "Mary", .gender = "F"},
}},
{"DE",
ForenameList{
ForenameEntry{.name = "Lukas", .gender = "M"},
ForenameEntry{.name = "Anna", .gender = "F"},
}},
{"FR",
ForenameList{
ForenameEntry{.name = "Lucas", .gender = "M"},
ForenameEntry{.name = "Camille", .gender = "F"},
}},
{"BE",
ForenameList{
ForenameEntry{.name = "Noah", .gender = "M"},
ForenameEntry{.name = "Emma", .gender = "F"},
}},
},
surnames_by_country_{
{"US", SurnameList{"Smith", "Johnson"}},
{"DE", SurnameList{"Muller", "Schmidt"}},
{"FR", SurnameList{"Martin", "Bernard"}},
{"BE", SurnameList{"Peeters", "Janssens"}},
} {}
const LocationsList& MockCuratedDataService::LoadLocations() {
return locations_;
}
const PersonasList& MockCuratedDataService::LoadPersonas() { return personas_; }
const ForenamesByCountryMap& MockCuratedDataService::LoadForenamesByCountry() {
return forenames_by_country_;
}
const SurnamesByCountryMap& MockCuratedDataService::LoadSurnamesByCountry() {
return surnames_by_country_;
}

93
tooling/pipeline/src/services/enrichment/wikipedia/fetch_extract.cc
View File

@@ -1,5 +1,6 @@
/**
* @file wikipedia/fetch_extract.cc
* @brief WikipediaEnrichmentService::FetchExtract() implementation.
*/
#include <boost/json.hpp>
@@ -20,9 +21,10 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (const auto cache_it = this->extract_cache_.find(cache_key);
cache_it != this->extract_cache_.end()) {
if (logger_) {
logger_->Log({.level = LogLevel::Debug,
.phase = PipelinePhase::UserGeneration,
.message = std::format("Wikipedia: Cache hit for {}!", cache_key)});
logger_->Log(
{.level = LogLevel::Debug,
.phase = PipelinePhase::Enrichment,
.message = std::format("Wikipedia: Cache hit for {}!", cache_key)});
}
return cache_it->second;
}
@@ -30,7 +32,8 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
const std::string encoded = this->client_->EncodeURL(cache_key);
const std::string url = std::format(
"https://en.wikipedia.org/w/"
"api.php?action=query&titles={}&prop=extracts&explaintext=1&format=json",
"api.php?action=query&titles={}&prop=extracts&explaintext=1&format="
"json",
encoded);
const std::string body = this->client_->Get(url);
@@ -45,11 +48,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (ec) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format("WikipediaService: JSON parse error for '{}': {}",
std::string(query), ec.message())});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: JSON parse error for '{}': {}",
std::string(query), ec.message())});
}
return {};
}
@@ -58,12 +61,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
const json::object* obj = doc.if_object();
if (obj == nullptr) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message =
std::format("WikipediaService: Expected root object for '{}'",
std::string(query))});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: Expected root object for '{}'",
std::string(query))});
}
return {};
}
@@ -76,12 +78,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if ((pages_ptr == nullptr) || !pages_ptr->is_object()) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message =
std::format("WikipediaService: Missing query.pages for '{}'",
std::string(query))});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: Missing query.pages for '{}'",
std::string(query))});
}
return {};
}
@@ -90,11 +91,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (pages.empty()) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format("WikipediaService: No pages returned for '{}'",
std::string(query))});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: No pages returned for '{}'",
std::string(query))});
}
this->extract_cache_.emplace(cache_key, "");
return {};
@@ -106,12 +107,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if (!page_val.is_object()) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message =
std::format("WikipediaService: Unexpected page format for '{}'",
std::string(query))});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: Unexpected page format for '{}'",
std::string(query))});
}
return {};
}
@@ -121,10 +121,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
// Handle 404/Missing status
if (page.contains("missing")) {
if (logger_) {
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message = std::format("WikipediaService: Page '{}' does not exist",
std::string(query))});
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format("WikipediaService: Page '{}' does not exist",
std::string(query))});
}
this->extract_cache_.emplace(cache_key, "");
return {};
@@ -134,12 +135,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
if ((extract_ptr == nullptr) || !extract_ptr->is_string()) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.message =
std::format("WikipediaService: No extract string found for '{}'",
std::string(query))});
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"WikipediaService: No extract string found for '{}'",
std::string(query))});
}
this->extract_cache_.emplace(cache_key, "");
return {};
@@ -148,10 +148,11 @@ std::string WikipediaEnrichmentService::FetchExtract(std::string_view query) {
// 4. Success
std::string extract(extract_ptr->as_string());
if (logger_) {
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration,
.message = std::format("WikipediaService: Fetched {} chars for '{}'",
extract.size(), std::string(query))});
logger_->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Enrichment,
.message = std::format("WikipediaService: Fetched {} chars for '{}'",
extract.size(), std::string(query))});
}
this->extract_cache_.insert_or_assign(cache_key, extract);

22
tooling/pipeline/src/services/enrichment/wikipedia/get_summary.cc
View File

@@ -1,6 +1,6 @@
/**
* @file wikipedia/get_summary.cc
* @brief WikipediaService::GetLocationContext() implementation.
* @brief WikipediaEnrichmentService::GetLocationContext() implementation.
*/
#include <chrono>
@@ -16,7 +16,7 @@ std::string WikipediaEnrichmentService::GetLocationContext(
if (!this->client_) {
if (logger_) {
logger_->Log({.level = LogLevel::Warn,
.phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::Enrichment,
.message = "Wikipedia client is nullptr."});
}
return {};
@@ -24,13 +24,6 @@ std::string WikipediaEnrichmentService::GetLocationContext(
std::string result;
// std::string region_query(loc.city);
// if (!loc.country.empty()) {
// region_query += loc.state_province,
// region_query += ", ";
// region_query += loc.country;
// }
constexpr std::string_view brewing_query = "brewing";
const std::string location_query =
std::format("{}, {}", loc.city, loc.iso3166_2);
@@ -51,9 +44,10 @@ std::string WikipediaEnrichmentService::GetLocationContext(
append_extract(FetchExtract(beer_query));
if (logger_) {
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::UserGeneration,
.message = std::format("Done fetching for {}. Sleeping for 10 seconds.",
location_query)});
.phase = PipelinePhase::Enrichment,
.message = std::format(
"Done fetching for {}. Sleeping for 10 seconds.",
location_query)});
}
std::this_thread::sleep_for(10s);
@@ -61,9 +55,9 @@ std::string WikipediaEnrichmentService::GetLocationContext(
if (logger_) {
logger_->Log(
{.level = LogLevel::Debug,
.phase = PipelinePhase::UserGeneration,
.phase = PipelinePhase::Enrichment,
.message = std::format("WikipediaService lookup failed for '{}': {}",
location_query, e.what())});
location_query, e.what())});
}
}
return result;

2
tooling/pipeline/src/services/enrichment/wikipedia/wikipedia_service.cc
View File

@@ -1,6 +1,6 @@
/**
* @file services/wikipedia/wikipedia_service.cc
* @brief WikipediaService constructor implementation.
* @brief WikipediaEnrichmentService constructor implementation.
*/
#include "services/enrichment/wikipedia_service.h"

2
tooling/pipeline/src/services/logging/log_dispatcher.cc
View File

@@ -19,6 +19,8 @@ namespace {
switch (phase) {
case PipelinePhase::Startup:
return "Startup";
case PipelinePhase::Enrichment:
return "Enrichment";
case PipelinePhase::UserGeneration:
return "User Generation";
case PipelinePhase::BreweryAndBeerGeneration:

21
tooling/pipeline/src/services/prompt_directory.cc
View File

@@ -44,8 +44,8 @@ PromptDirectory::PromptDirectory(const std::filesystem::path& prompt_dir,
// Scenario 4: directory must be readable (probe with directory_iterator).
std::filesystem::directory_iterator probe(prompt_dir_, ec);
if (ec) {
throw std::runtime_error(
std::format("PromptDirectory: prompt directory is not readable: {} ({})",
throw std::runtime_error(std::format(
"PromptDirectory: prompt directory is not readable: {} ({})",
prompt_dir_.string(), ec.message()));
}
@@ -76,7 +76,7 @@ std::string PromptDirectory::Load(std::string_view key) {
if (!file.is_open()) {
throw std::runtime_error(
std::format("PromptDirectory: prompt file not found for key '{}': {}",
key_str, file_path.string()));
key_str, file_path.string()));
}
std::string content((std::istreambuf_iterator<char>(file)),
@@ -84,15 +84,18 @@ std::string PromptDirectory::Load(std::string_view key) {
file.close();
if (content.empty()) {
throw std::runtime_error(std::format("PromptDirectory: prompt file for key '{}' is empty: {}",
key_str, file_path.string()));
throw std::runtime_error(
std::format("PromptDirectory: prompt file for key '{}' is empty: {}",
key_str, file_path.string()));
}
if (logger_) {
logger_->Log({.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = std::format("[PromptDirectory] Loaded prompt '{}' from '{}' ({} chars)",
key_str, file_path.string(), content.size())});
logger_->Log(
{.level = LogLevel::Info,
.phase = PipelinePhase::Startup,
.message = std::format(
"[PromptDirectory] Loaded prompt '{}' from '{}' ({} chars)",
key_str, file_path.string(), content.size())});
}
cache_.emplace(key_str, content);

1
tooling/pipeline/src/services/sqlite/finalize.cc
View File

@@ -14,6 +14,7 @@ void SqliteExportService::Finalize() {
}
try {
insert_user_stmt_.reset();
insert_brewery_stmt_.reset();
insert_location_stmt_.reset();
if (transaction_open_) {

6
tooling/pipeline/src/services/sqlite/helpers/sqlite_statement_helpers.cc
View File

@@ -33,7 +33,7 @@ void ResetStatement(SqliteStatementHandle& statement) {
}
void Bind(const SqliteStatementHandle& statement,
const BindParam<std::string_view>& param) {
const BoundParam<std::string_view>& param) {
const auto byte_count = param.value.size();
if (byte_count > static_cast<std::size_t>(std::numeric_limits<int>::max())) {
ThrowSqliteError(sqlite3_db_handle(statement.get()), param.action);
@@ -60,7 +60,7 @@ void Bind(const SqliteStatementHandle& statement,
}
void Bind(const SqliteStatementHandle& statement,
const BindParam<double>& param) {
const BoundParam<double>& param) {
if (sqlite3_bind_double(statement.get(), param.index, param.value) !=
SQLITE_OK) {
ThrowSqliteError(sqlite3_db_handle(statement.get()), param.action);
@@ -68,7 +68,7 @@ void Bind(const SqliteStatementHandle& statement,
}
void Bind(const SqliteStatementHandle& statement,
const BindParam<sqlite3_int64>& param) {
const BoundParam<sqlite3_int64>& param) {
if (sqlite3_bind_int64(statement.get(), param.index, param.value) !=
SQLITE_OK) {
ThrowSqliteError(sqlite3_db_handle(statement.get()), param.action);

13
tooling/pipeline/src/services/sqlite/initialize.cc
View File

@@ -18,9 +18,9 @@ std::filesystem::path SqliteExportService::BuildDatabasePath() const {
std::filesystem::path candidate = output_path_ / base_filename;
for (int suffix = 1; std::filesystem::exists(candidate); ++suffix) {
candidate = output_path_ /
std::filesystem::path(std::format("biergarten_seed_{}-{}.sqlite",
run_timestamp_utc_, suffix));
candidate = output_path_ / std::filesystem::path(
std::format("biergarten_seed_{}-{}.sqlite",
run_timestamp_utc_, suffix));
}
return candidate;
@@ -33,6 +33,9 @@ void SqliteExportService::InitializeSchema() const {
sqlite_export_service_internal::ExecSql(
db_handle_, sqlite_export_service_internal::kCreateBreweriesTableSql,
"Failed to create SQLite breweries table");
sqlite_export_service_internal::ExecSql(
db_handle_, sqlite_export_service_internal::kCreateUsersTableSql,
"Failed to create SQLite users table");
}
void SqliteExportService::PrepareStatements() {
@@ -42,6 +45,9 @@ void SqliteExportService::PrepareStatements() {
insert_brewery_stmt_ = sqlite_export_service_internal::PrepareStatement(
db_handle_, sqlite_export_service_internal::kInsertBrewerySql,
"Failed to prepare SQLite brewery insert statement");
insert_user_stmt_ = sqlite_export_service_internal::PrepareStatement(
db_handle_, sqlite_export_service_internal::kInsertUserSql,
"Failed to prepare SQLite user insert statement");
}
void SqliteExportService::RollbackAndCloseNoThrow() noexcept {
@@ -54,6 +60,7 @@ void SqliteExportService::RollbackAndCloseNoThrow() noexcept {
transaction_open_ = false;
}
insert_user_stmt_.reset();
insert_brewery_stmt_.reset();
insert_location_stmt_.reset();
db_handle_.reset();

163
tooling/pipeline/src/services/sqlite/process_record.cc
View File

@@ -1,6 +1,7 @@
/**
* @file services/sqlite/process_record.cc
* @brief SqliteExportService::ProcessRecord() implementation.
* @brief SqliteExportService::ProcessRecord() implementation
* and the shared location-resolution helper.
*/
#include <iomanip>
@@ -29,111 +30,117 @@ std::string SqliteExportService::BuildLocationKey(const Location& location) {
return key_stream.str();
}
uint64_t SqliteExportService::ProcessRecord(const GeneratedBrewery& brewery) {
sqlite3_int64 SqliteExportService::ResolveLocationId(const Location& location) {
const std::string location_key = BuildLocationKey(location);
const auto cached_location = location_cache_.find(location_key);
if (cached_location != location_cache_.end()) {
return cached_location->second;
}
const std::string local_languages_json =
sqlite_export_service_internal::SerializeVector(location.local_languages);
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationCityBindIndex,
.value = location.city,
.action = "Failed to bind SQLite location city"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index =
sqlite_export_service_internal::kLocationStateProvinceBindIndex,
.value = location.state_province,
.action = "Failed to bind SQLite location state/province"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationIso31662BindIndex,
.value = location.iso3166_2,
.action = "Failed to bind SQLite location ISO 3166-2 code"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationCountryBindIndex,
.value = location.country,
.action = "Failed to bind SQLite location country"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationIso31661BindIndex,
.value = location.iso3166_1,
.action = "Failed to bind SQLite location ISO 3166-1 code"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationLanguagesBindIndex,
.value = local_languages_json,
.action = "Failed to bind SQLite location languages"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam{
.index = sqlite_export_service_internal::kLocationLatitudeBindIndex,
.value = location.latitude,
.action = "Failed to bind SQLite location latitude"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BoundParam{
.index = sqlite_export_service_internal::kLocationLongitudeBindIndex,
.value = location.longitude,
.action = "Failed to bind SQLite location longitude"});
sqlite_export_service_internal::StepStatement(
db_handle_, insert_location_stmt_,
"Failed to insert SQLite location row");
const sqlite3_int64 location_id =
sqlite_export_service_internal::LastInsertRowId(db_handle_);
location_cache_.emplace(location_key, location_id);
sqlite_export_service_internal::ResetStatement(insert_location_stmt_);
return location_id;
}
uint64_t SqliteExportService::ProcessRecord(const BreweryRecord& brewery) {
if (db_handle_ == nullptr || !transaction_open_) {
throw std::runtime_error("SQLite export service is not initialized");
}
const std::string location_key = BuildLocationKey(brewery.location);
const auto cached_location = location_cache_.find(location_key);
sqlite3_int64 location_id = 0;
if (cached_location != location_cache_.end()) {
location_id = cached_location->second;
} else {
const std::string local_languages_json =
sqlite_export_service_internal::SerializeVector(
brewery.location.local_languages);
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationCityBindIndex,
.value = brewery.location.city,
.action = "Failed to bind SQLite location city"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index =
sqlite_export_service_internal::kLocationStateProvinceBindIndex,
.value = brewery.location.state_province,
.action = "Failed to bind SQLite location state/province"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationIso31662BindIndex,
.value = brewery.location.iso3166_2,
.action = "Failed to bind SQLite location ISO 3166-2 code"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationCountryBindIndex,
.value = brewery.location.country,
.action = "Failed to bind SQLite location country"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index = sqlite_export_service_internal::kLocationIso31661BindIndex,
.value = brewery.location.iso3166_1,
.action = "Failed to bind SQLite location ISO 3166-1 code"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
.index =
sqlite_export_service_internal::kLocationLanguagesBindIndex,
.value = local_languages_json,
.action = "Failed to bind SQLite location languages"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam{
.index = sqlite_export_service_internal::kLocationLatitudeBindIndex,
.value = brewery.location.latitude,
.action = "Failed to bind SQLite location latitude"});
sqlite_export_service_internal::Bind(
insert_location_stmt_,
sqlite_export_service_internal::BindParam{
.index =
sqlite_export_service_internal::kLocationLongitudeBindIndex,
.value = brewery.location.longitude,
.action = "Failed to bind SQLite location longitude"});
sqlite_export_service_internal::StepStatement(
db_handle_, insert_location_stmt_,
"Failed to insert SQLite location row");
location_id = sqlite_export_service_internal::LastInsertRowId(db_handle_);
location_cache_.emplace(location_key, location_id);
sqlite_export_service_internal::ResetStatement(insert_location_stmt_);
}
const sqlite3_int64 location_id = ResolveLocationId(brewery.location);
sqlite_export_service_internal::Bind(
insert_brewery_stmt_,
sqlite_export_service_internal::BindParam<sqlite3_int64>{
sqlite_export_service_internal::BoundParam<sqlite3_int64>{
.index = sqlite_export_service_internal::kBreweryLocationIdBindIndex,
.value = location_id,
.action = "Failed to bind SQLite brewery location id"});
sqlite_export_service_internal::Bind(
insert_brewery_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kBreweryEnglishNameBindIndex,
.value = brewery.brewery.name_en,
.action = "Failed to bind SQLite brewery English name"});
sqlite_export_service_internal::Bind(
insert_brewery_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::
kBreweryEnglishDescriptionBindIndex,
.value = brewery.brewery.description_en,
.action = "Failed to bind SQLite brewery English description"});
sqlite_export_service_internal::Bind(
insert_brewery_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kBreweryLocalNameBindIndex,
.value = brewery.brewery.name_local,
.action = "Failed to bind SQLite brewery local name"});
sqlite_export_service_internal::Bind(
insert_brewery_stmt_,
sqlite_export_service_internal::BindParam<std::string_view>{
sqlite_export_service_internal::BoundParam<std::string_view>{
.index =
sqlite_export_service_internal::kBreweryLocalDescriptionBindIndex,
.value = brewery.brewery.description_local,

79
tooling/pipeline/src/services/sqlite/process_user_record.cc Normal file
View File

@@ -0,0 +1,79 @@
/**
* @file services/sqlite/process_user_record.cc
* @brief SqliteExportService::ProcessRecord(UserRecord) implementation.
*/
#include <stdexcept>
#include "services/database/sqlite_export_service.h"
#include "services/database/sqlite_export_service_helpers.h"
uint64_t SqliteExportService::ProcessRecord(const UserRecord& user) {
if (db_handle_ == nullptr || !transaction_open_) {
throw std::runtime_error("SQLite export service is not initialized");
}
const sqlite3_int64 location_id = ResolveLocationId(user.location);
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<sqlite3_int64>{
.index = sqlite_export_service_internal::kUserLocationIdBindIndex,
.value = location_id,
.action = "Failed to bind SQLite user location id"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserFirstNameBindIndex,
.value = user.user.first_name,
.action = "Failed to bind SQLite user first name"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserLastNameBindIndex,
.value = user.user.last_name,
.action = "Failed to bind SQLite user last name"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserGenderBindIndex,
.value = user.user.gender,
.action = "Failed to bind SQLite user gender"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserUsernameBindIndex,
.value = user.user.username,
.action = "Failed to bind SQLite user username"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserBioBindIndex,
.value = user.user.bio,
.action = "Failed to bind SQLite user bio"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<double>{
.index = sqlite_export_service_internal::kUserActivityWeightBindIndex,
.value = static_cast<double>(user.user.activity_weight),
.action = "Failed to bind SQLite user activity weight"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserEmailBindIndex,
.value = user.email,
.action = "Failed to bind SQLite user email"});
sqlite_export_service_internal::Bind(
insert_user_stmt_,
sqlite_export_service_internal::BoundParam<std::string_view>{
.index = sqlite_export_service_internal::kUserDateOfBirthBindIndex,
.value = user.date_of_birth,
.action = "Failed to bind SQLite user date of birth"});
sqlite_export_service_internal::StepStatement(
db_handle_, insert_user_stmt_, "Failed to insert SQLite user row");
sqlite_export_service_internal::ResetStatement(insert_user_stmt_);
return sqlite_export_service_internal::LastInsertRowId(db_handle_);
}

17
tooling/pipeline/src/web_client/http_web_client.cc
View File

@@ -48,21 +48,20 @@ std::string HttpWebClient::Get(const std::string& url) {
const httplib::Result result = client.Get(path);
if (!result) {
throw std::runtime_error(std::format(
"[HttpWebClient] Request failed for URL: {} — {}", url,
httplib::to_string(result.error())));
throw std::runtime_error(
std::format("[HttpWebClient] Request failed for URL: {} — {}", url,
httplib::to_string(result.error())));
}
if (result->status < kSuccessMin || result->status >= kSuccessMax) {
if (logger_) {
logger_->Log(
{.level = LogLevel::Error,
.phase = PipelinePhase::UserGeneration,
.message =
std::format("[HttpWebClient] Request failed for URL: {}", url)});
logger_->Log({.level = LogLevel::Error,
.phase = PipelinePhase::Enrichment,
.message = std::format(
"[HttpWebClient] Request failed for URL: {}", url)});
}
throw std::runtime_error(std::format("[HttpWebClient] HTTP {} for URL: {}",
result->status, url));
result->status, url));
}
return result->body;