import logging import random import time import mysql.connector import requests # Outputs info/warnings to console so the user can monitor progress. logging.basicConfig(level=logging.INFO, format='[%(levelname)s] %(message)s') # Bazaarvoice API keys (required for accessing review statistics) BAZAARVOICE_PASSKEY = "h2au8voubpqmikormkh7b8t86" DISPLAY_CODE = "0236-en_us" # Database connection settings DB_CONFIG = { "host": "104.236.70.14", "user": 'dev', "charset": 'utf8mb4', "password": '.p0H@Xb.](dfHGg.', "autocommit": True, } # Source/target tables SOURCE_SKU_QUERY = """ SELECT DISTINCT SKU FROM Salsify.MainData WHERE SKU IS NOT NULL AND SKU <> '' AND Status IN ('Active', 'Liquidation') """ TARGET_TABLE = "Reviews.Shopify" def fetch(url, delay_range=(0.2, 0.5)): """ Makes a GET request with a random delay to avoid rate limiting. Returns a response object or None if request fails. """ time.sleep(random.uniform(*delay_range)) # Anti-spam delay try: resp = requests.get(url, timeout=8) except Exception as e: logging.warning(f"Request failed {url}: {e}") return None if resp.status_code != 200: logging.warning(f"Status {resp.status_code} for {url}") return None return resp def get_product_id_by_sku(sku: str): """ Looks up a product on vigoindustries.com using a SKU. Returns (product_id, product_handle). If SKU not found → returns (None, None). """ sku = str(sku).strip() if not sku: return None, None # Shopify Search Suggest API url = ( "https://vigoindustries.com/search/suggest.json" f"?q={sku}" "&resources[type]=product" "&resources[options][fields]=title,product_type,variants.sku" ) resp = fetch(url) if not resp: return None, None try: data = resp.json() except Exception: logging.warning(f"Failed to parse JSON for SKU={sku}") return None, None # Extract product list products = ( data.get("resources", {}) .get("results", {}) .get("products", []) ) # Try to match exact SKU in product variants for p in products: pid = p.get("id") handle = p.get("handle") variants = p.get("variants", []) for v in variants: v_sku = str(v.get("sku", "")).strip() if v_sku.upper() == sku.upper(): return (str(pid) if pid else None), handle if products: p = products[0] return (str(p.get("id")) if p.get("id") else None), p.get("handle") logging.warning(f"No product found for SKU={sku}") return None, None def fetch_bazaarvoice_stats(product_id: str): """ Requests review statistics from Bazaarvoice API. Returns: count → number of reviews avg → average rating (float) If something fails → returns (0, None). """ if not product_id: return 0, None url = ( "https://api.bazaarvoice.com/data/reviews.json" f"?filter=productid:eq:{product_id}" f"&limit=1&offset=0" f"&include=products" f"&stats=reviews" f"&passkey={BAZAARVOICE_PASSKEY}" f"&apiversion=5.5" f"&displaycode={DISPLAY_CODE}" ) resp = fetch(url) if not resp: return 0, None try: data = resp.json() except Exception: return 0, None # Extract review count try: count = int(data.get("TotalResults")) except Exception: count = 0 avg = None try: products = data.get("Includes", {}).get("Products", {}) if products: first_prod = next(iter(products.values())) stats = first_prod.get("ReviewStatistics", {}) or {} avg_value = stats.get("AverageOverallRating") or stats.get("AverageRating") if isinstance(avg_value, (int, float)): avg = float(avg_value) except Exception: avg = None return count, avg def process_sku(sku: str): """ Main function that: 1) Finds product_id via Vigo API 2) Fetches review statistics from Bazaarvoice 3) Returns (sku, product_id, handle, review_count, avg_rating) """ product_id, handle = get_product_id_by_sku(sku) if not product_id: logging.info(f"SKU {sku}: product_id NOT FOUND") return sku, None, None, 0, None count, avg = fetch_bazaarvoice_stats(product_id) logging.info( f"SKU {sku} → product_id={product_id}, handle={handle}, " f"reviews={count}, avg={avg}" ) return sku, product_id, handle, count, avg def get_db_connection(): """Opens a new MySQL connection.""" conn = mysql.connector.connect(**DB_CONFIG) conn.autocommit = True return conn def ensure_target_table(conn): """ Creates the target table if it does not exist. Stores review count/average per SKU with last update timestamp. """ ddl = f""" CREATE TABLE IF NOT EXISTS {TARGET_TABLE} ( SKU VARCHAR(100) PRIMARY KEY, ProductId VARCHAR(64), Handle VARCHAR(255), ReviewCount INT DEFAULT 0, AverageRating DECIMAL(4,2) NULL, UpdatedAt TIMESTAMP DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP ) CHARACTER SET utf8mb4; """ cur = conn.cursor() cur.execute(ddl) cur.close() def fetch_skus_from_db(conn): """Reads SKU list from Salsify.MainData.""" cur = conn.cursor(dictionary=True) cur.execute(SOURCE_SKU_QUERY) rows = cur.fetchall() cur.close() return [str(row["SKU"]).strip() for row in rows if str(row["SKU"]).strip()] def upsert_results(conn, rows): """Writes results back into the target table.""" if not rows: return insert_sql = f""" INSERT INTO {TARGET_TABLE} (SKU, ProductId, Handle, ReviewCount, AverageRating) VALUES (%s, %s, %s, %s, %s) ON DUPLICATE KEY UPDATE ProductId = VALUES(ProductId), Handle = VALUES(Handle), ReviewCount = VALUES(ReviewCount), AverageRating = VALUES(AverageRating), UpdatedAt = CURRENT_TIMESTAMP; """ cur = conn.cursor() cur.executemany(insert_sql, rows) cur.close() def main(): """ Main program flow: 1) Read SKUs from DB 2) Process each SKU 3) Upsert review stats back into DB table """ conn = None try: conn = get_db_connection() ensure_target_table(conn) sku_list = fetch_skus_from_db(conn) if not sku_list: logging.warning("No SKUs found in source table.") return logging.info(f"Loaded {len(sku_list)} SKUs from DB.") rows_for_db = [] total = len(sku_list) for idx, sku in enumerate(sku_list, start=1): sku, product_id, handle, count, avg = process_sku(sku) rows_for_db.append((sku, product_id, handle, count, avg)) if idx % 50 == 0 or idx == total: logging.info(f"Completed {idx}/{total} SKUs") upsert_results(conn, rows_for_db) logging.info(f"Upserted {len(rows_for_db)} rows into {TARGET_TABLE}.") finally: if conn is not None and conn.is_connected(): conn.close() if __name__ == "__main__": main()