bambulab-store-tracker/main.py

import argparse
from datetime import datetime
import logging
import sqlite3
import time

import requests


def check_sqlite(conn: sqlite3.Connection):
    filament_table = '''
    CREATE TABLE IF NOT EXISTS filaments (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        -- internal bambulab id
        store_id INTEGER UNIQUE NOT NULL,
        -- region where the filament is stored
        region TEXT NOT NULL,
        -- filament title (e.g. PLA Basic, PLA Matte, ...)
        title TEXT NOT NULL,
        -- filament vendor (e.g. Bambu Lab, Bambu Lab EU, Bambu Lab US)
        vendor TEXT NOT NULL,
        -- filament type (e.g. PLA Basic, PLA Matte, ...). unlike `store_id`, this is the same across all languages
        type TEXT NOT NULL,
        -- date when this filament got created / published
        created INTEGER NOT NULL
    )
    '''
    filament_store_id_index = '''
    CREATE INDEX IF NOT EXISTS filaments_store_id_idx ON filaments (store_id)
    '''

    filament_variant_tables = '''
    CREATE TABLE IF NOT EXISTS filament_variants (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        -- internal bambulab id
        store_id INTEGER UNIQUE NOT NULL,
        -- in most cases the type of the filament (e.g. Filament with spool, Refill, ...). might be something other with special products, e.g. a filament name in starter packs
        option1 TEXT,
        -- in most cases the actual filament weight. might be something other with special products, e.g. a filament name in starter packs
        option2 TEXT,
        -- in most cases the filament color. might be something other with special products, e.g. a filament name in starter packs
        option3 TEXT,
        -- identifier between different languages. might be null if the variant contains multiple filaments (e.g. this is the case with starter packs)
        sku TEXT,
        -- grams of the filament (+ the spool if applicable). might be null if the variant contains multiple filaments (e.g. this is the case with starter packs)
        grams REAL,
        -- date when this variant got created / published
        created INTEGER NOT NULL,
        -- relation to parent filament
        filament_id INTEGER NOT NULL,
        FOREIGN KEY (filament_id) REFERENCES filaments (id)
    )
    '''
    filament_variant_store_id_index = '''
    CREATE INDEX IF NOT EXISTS filament_variants_store_id_idx ON filament_variants (store_id)
    '''

    measurement_table = '''
    CREATE TABLE IF NOT EXISTS measurements (
        id INTEGER PRIMARY KEY AUTOINCREMENT,
        -- timestamp of a measurement
        timestamp INTEGER NOT NULL
    )
    '''
    measurement_timestamp_index = '''
    CREATE INDEX IF NOT EXISTS measurements_timestamp_idx ON measurements (timestamp)
    '''

    availability_table = '''
    CREATE TABLE IF NOT EXISTS availability (
        -- 0 if not available, 1 if available
        available INTEGER NOT NULL,
        measurement_id INTEGER NOT NULL,
        filament_variant_id INTEGER NOT NULL,
        FOREIGN KEY (measurement_id) REFERENCES measurements (id),
        FOREIGN KEY (filament_variant_id) REFERENCES filament_variants (id)
    )
    '''

    price_table = '''
    CREATE TABLE IF NOT EXISTS prices (
        price REAL NOT NULL,
        measurement_id INTEGER NOT NULL,
        filament_variant_id INTEGER NOT NULL,
        FOREIGN KEY (measurement_id) REFERENCES measurements (id),
        FOREIGN KEY (filament_variant_id) REFERENCES filament_variants (id)
    )
    '''

    conn.execute(filament_table)
    conn.execute(filament_store_id_index)
    conn.execute(filament_variant_tables)
    conn.execute(filament_variant_store_id_index)
    conn.execute(measurement_table)
    conn.execute(measurement_timestamp_index)
    conn.execute(availability_table)
    conn.execute(price_table)

    conn.commit()


def cmd(conn: sqlite3.Connection):
    timestamp = int(time.time())
    logging.info('scraping at %d', timestamp)

    try:
        global_products = requests.request('GET', 'https://store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        us_products = requests.request('GET', 'https://us.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        eu_products = requests.request('GET', 'https://eu.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        uk_products = requests.request('GET', 'https://uk.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        au_products = requests.request('GET', 'https://au.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        ca_products = requests.request('GET', 'https://ca.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
        jp_products = requests.request('GET', 'https://jp.store.bambulab.com/collections/bambu-lab-3d-printer-filament/products.json').json()
    except requests.exceptions.RequestException as e:
        logging.error(f'request error: {e}')
        return

    all_products = {
        'global': global_products['products'],
        'us': us_products['products'],
        'eu': eu_products['products'],
        'uk': uk_products['products'],
        'au': au_products['products'],
        'ca': ca_products['products'],
        'jp': jp_products['products']
    }

    cur = conn.cursor()

    availability = {}
    prices = {}

    all_filament_ids: list[tuple[int, int]] = cur.execute('SELECT id, store_id FROM filaments').fetchall()
    all_filament_ids: dict[int, int] = {ids[1]: ids[0] for ids in all_filament_ids}
    all_variant_ids: list[tuple[int, int]] = cur.execute('SELECT id, store_id FROM filament_variants').fetchall()
    all_variant_ids: dict[int, int] = {ids[1]: ids[0] for ids in all_variant_ids}

    for region, products in all_products.items():
        for product in products:
            # get the internal id for the filament or insert it if it does not exist
            if (filament_id := all_filament_ids.get(product['id'])) is None:
                cur.execute('INSERT INTO filaments (store_id, region, title, vendor, type, created) VALUES (?, ?, ?, ?, ?, ?)', (
                    product['id'],
                    region,
                    product['title'],
                    product['vendor'],
                    product['product_type'],
                    int(time.mktime(datetime.fromisoformat(product['created_at']).utctimetuple()))
                ))
                filament_id = cur.lastrowid
                all_filament_ids[product['id']] = filament_id

            for variant in product['variants']:
                # get the internal id for the filament variant or insert it if it does not exist
                if (filament_variant_id := all_variant_ids.get(variant['id'])) is None:
                    cur.execute('INSERT INTO filament_variants (store_id, option1, option2, option3, sku, grams, created, filament_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)', (
                        variant['id'],
                        variant['option1'],
                        variant['option2'],
                        variant['option3'],
                        variant['sku'] or None,
                        variant['grams'] or None,
                        int(time.mktime(datetime.fromisoformat(variant['created_at']).utctimetuple())),
                        filament_id,
                    ))
                    filament_variant_id = cur.lastrowid
                    all_variant_ids[variant['id']] = filament_variant_id

                availability[filament_variant_id] = int(variant['available'])
                prices[filament_variant_id] = float(variant['price'])

    cur.execute('INSERT INTO measurements (timestamp) VALUES (?)', (timestamp,))
    measurement_id = cur.lastrowid

    last_availability_changes = cur.execute('SELECT DISTINCT filament_variant_id, last_value(available) over (ORDER BY filament_variant_id) FROM availability').fetchall()
    last_price_changes = cur.execute('SELECT DISTINCT filament_variant_id, last_value(price) over (ORDER BY filament_variant_id) FROM prices').fetchall()

    for row in last_availability_changes:
        if row[0] not in availability:
            continue
        if availability[row[0]] == row[1]:
            del availability[row[0]]

    for row in last_price_changes:
        if row[0] not in prices:
            continue
        if prices[row[0]] == row[1]:
            del prices[row[0]]

    if availability:
        logging.info('found %d availability changes', len(availability))
    if prices:
        logging.info('found %d prices changes', len(prices))

    cur.executemany('INSERT INTO availability (available, measurement_id, filament_variant_id) VALUES (?, ?, ?)', [(available, measurement_id, id) for id, available in availability.items()])
    cur.executemany('INSERT INTO prices (price, measurement_id, filament_variant_id) VALUES (?, ?, ?)', [(price, measurement_id, id) for id, price in prices.items()])

    conn.commit()


if __name__ == '__main__':
    # --- cli parser --- #
    parser = argparse.ArgumentParser()
    parser.add_argument('--db', help='Path to the sqlite file where the stocks should be saved in', required=True)

    # --- cli input --- #
    args = parser.parse_args()

    logging.basicConfig(format='%(levelname)s [%(asctime)s] - %(message)s', datefmt='%Y-%m-%d %H:%M:%S', level=logging.INFO)

    conn = sqlite3.connect(args.db)
    check_sqlite(conn)

    cmd(conn)