Data Enrichment

INSERT INTO LOCAL_BUSINESSES (query, category, position, name, address, city, rating, review_count, phone, business_status, price_level, sponsored, status, enriched_at)
SELECT
    q.query,
    q.category,
    p.value:position::INTEGER                          AS position,
    p.value:title::STRING                              AS name,
    p.value:address::STRING                            AS address,
    p.value:city::STRING                               AS city,
    p.value:review_summary:overall_rating::NUMBER(3, 2) AS rating,
    p.value:number_of_reviews::INTEGER                 AS review_count,
    p.value:phone_number::STRING                       AS phone,
    p.value:business_status::STRING                    AS business_status,
    p.value:price_level::STRING                        AS price_level,
    p.value:sponsored::BOOLEAN                         AS sponsored,
    a.status                                           AS status,
    CURRENT_TIMESTAMP()                                AS enriched_at
FROM LOCATION_QUERIES q,
     TABLE(NIMBLE_INTEGRATION.TOOLS.NIMBLE_AGENT_RUN(
         'google_maps_search',
         OBJECT_CONSTRUCT('query', q.query)
     )) a,
     LATERAL FLATTEN(INPUT => a.parsing:entities:SearchResult) p
WHERE a.status = 'success';

SELECT
    p.sku,
    a.parsing:brand::STRING                    AS brand,
    a.parsing:product_title::STRING            AS title,
    a.parsing:brief_product_description::STRING AS description,
    a.parsing:color::STRING                    AS color,
    a.parsing:packaging::STRING                AS packaging,
    a.parsing:web_price::NUMBER(10, 2)         AS price,
    a.parsing:availability::BOOLEAN            AS in_stock,
    a.parsing:average_of_reviews::NUMBER(3, 2) AS rating
FROM   PRODUCTS p,
       TABLE(NIMBLE_INTEGRATION.TOOLS.NIMBLE_AGENT_RUN(
           'amazon_pdp',
           OBJECT_CONSTRUCT('asin', p.amazon_asin)
       )) a
WHERE  p.amazon_asin IS NOT NULL
  AND  a.status = 'success';

CREATE OR REPLACE TASK daily_local_business_discovery
  WAREHOUSE = NIMBLE_AGENT_WH
  SCHEDULE  = 'USING CRON 0 8 * * * America/Los_Angeles'
AS
  INSERT INTO LOCAL_BUSINESSES (query, category, position, name, address, city, rating, review_count, phone, business_status, price_level, sponsored, status, enriched_at)
  SELECT
      q.query,
      q.category,
      p.value:position::INTEGER                          AS position,
      p.value:title::STRING                              AS name,
      p.value:address::STRING                            AS address,
      p.value:city::STRING                               AS city,
      p.value:review_summary:overall_rating::NUMBER(3, 2) AS rating,
      p.value:number_of_reviews::INTEGER                 AS review_count,
      p.value:phone_number::STRING                       AS phone,
      p.value:business_status::STRING                    AS business_status,
      p.value:price_level::STRING                        AS price_level,
      p.value:sponsored::BOOLEAN                         AS sponsored,
      a.status                                           AS status,
      CURRENT_TIMESTAMP()                                AS enriched_at
  FROM LOCATION_QUERIES q,
       TABLE(NIMBLE_INTEGRATION.TOOLS.NIMBLE_AGENT_RUN(
           'google_maps_search',
           OBJECT_CONSTRUCT('query', q.query)
       )) a,
       LATERAL FLATTEN(INPUT => a.parsing:entities:SearchResult) p
  WHERE a.status = 'success';

ALTER TASK daily_local_business_discovery RESUME;

{{ config(
    materialized='incremental',
    unique_key=['query', 'position', 'enriched_at'],
    on_schema_change='append_new_columns'
) }}

SELECT
    q.query,
    q.category,
    p.value:position::INTEGER                          AS position,
    p.value:title::STRING                              AS name,
    p.value:address::STRING                            AS address,
    p.value:review_summary:overall_rating::NUMBER(3, 2) AS rating,
    p.value:number_of_reviews::INTEGER                 AS review_count,
    a.status                                           AS status,
    CURRENT_TIMESTAMP()                                AS enriched_at
FROM {{ ref('location_queries') }} q,
     TABLE(NIMBLE_INTEGRATION.TOOLS.NIMBLE_AGENT_RUN(
         'google_maps_search',
         OBJECT_CONSTRUCT('query', q.query)
     )) a,
     LATERAL FLATTEN(INPUT => a.parsing:entities:SearchResult) p
WHERE a.status = 'success'
{% if is_incremental() %}
  AND CURRENT_TIMESTAMP() > (SELECT MAX(enriched_at) FROM {{ this }})
{% endif %}