Automated Market Intelligence: Flaschenpost Data Pipeline

Business Case: In highly competitive e-commerce markets (like e-grocery), daily competitor price information is critical for your own margins. This project automates data collection for over 5,000 products to detect price trends and inventory gaps (out-of-stock) early.

Technologies: Python, SQL (SQLite/Postgres), API Reverse Engineering, Flask (Dashboarding).

Flask Articles

Step 1: Process Analysis & Data Strategy

Instead of parsing HTML (which is error-prone), I analyzed the internal APIs of the Flaschenpost mobile app. This allowed structured access to master data without generating the server load of a classical crawler. The pipeline monitors categories such as:

  • obst-gemuese
  • kuehlregal/frisch-fertig
  • kuehlregal/fleisch-fisch-veggie

Step 2: Network Analysis with Firefox

To understand the exact format of data returned by the Flaschenpost website, I used the Firefox network analysis tool. Particularly interesting was the URL https://www.flaschenpost.de/data/[CATEGORY]/23, which returned a JSON list of products.

By analyzing the network requests, I could determine how the data is structured and which endpoints I need to use for scraping. This allowed me to access the required information in a targeted manner and extract it efficiently.

Step 3: Relational Data Model & SQL Analysis

To enable long-term price analysis, the JSON data was transferred into a relational schema (3rd normal form). This separates product master data (products) from variable article data (articles — e.g., packaging sizes) and enables complex queries.

Data model:

  • product: Static master data (name, brand, description).
  • articles: Pricing factors (package price, price per liter, availability).
  • allergenes / details: Normalized detail tables.

After import, SQL enables deeper analyses that are technically more complex than pure Excel evaluations — for example, finding price outliers within a brand category:

-- Example analysis: average price per brand with deviation
SELECT 
    p.brandName,
    COUNT(a.id) as article_count,
    ROUND(AVG(a.price), 2) as average_price,
    MAX(a.price) as max_price
FROM product p
JOIN articles a ON p.id = a.productId
GROUP BY p.brandName
HAVING COUNT(a.id) > 5
ORDER BY average_price DESC;

Schema implementation:

c.execute("""
    CREATE TABLE IF NOT EXISTS product (
        id INTEGER PRIMARY KEY,
        alcoholInfo TEXT,
        alphabeticSort INTEGER,
        backgroundColor TEXT,
        brandId INTEGER,
        brandName TEXT,
        brandWebShopUrl TEXT,
        categoryId INTEGER,
        colorProfileId INTEGER,
        descriptionText TEXT,
        ghsInfo TEXT,
        ...
    )
""")
conn.commit()

Step 4: Data Extraction and Database Integration

After identifying the categories and analyzing the data structure, I began scraping the Flaschenpost website. This process occurred in several steps, focusing on parsing the HTML pages and extracting the JSON data to feed the information into the SQLite database.

Parsing HTML Pages

To extract the required data, I used BeautifulSoup to parse the HTML pages of the Flaschenpost website. By navigating through the DOM (Document Object Model), I was able to identify the relevant sections containing the URLs I needed for scraping.

Extracting JSON Data

After identifying the relevant HTML pages, I extracted the JSON data they contained. This was done using Python code that traversed the HTML structure and extracted the JSON information. I oriented myself by the patterns and tags on the pages to find and extract the right data.

The actual data extraction happens in the scrape_data(value) function. This function takes a category (value) as input and constructs a URL to retrieve data from the corresponding page. Once the data is received, it is saved in JSON format. The filename is created by replacing slashes (/) in the category with dashes (-). The downloaded JSON files are stored in the previously created folder.

To speed up the scraping process, I used ThreadPoolExecutor from the concurrent.futures library. This enables parallel scraping of multiple categories simultaneously, reducing the overall runtime of the scraping operation. Each category in the value_list is passed to the scrape_data function to download the data.

This approach makes it possible to efficiently extract large amounts of data from the website and store it in a structured format for further processing and analysis.

timestamp = int(time.time())
folder_path = f"data/{timestamp}"
os.makedirs(folder_path, exist_ok=True)

def scrape_data(value):
    url = f"https://www.flaschenpost.de/data/{value}/23"
    response = requests.get(url)
    data = response.json()
    value = value.replace("/", "-")
    file_path = os.path.join(folder_path, f"{value}.json")
    with open(file_path, "w", encoding="utf-8") as file:
        json.dump(data, file, indent=4)

with concurrent.futures.ThreadPoolExecutor() as executor:
    executor.map(scrape_data, value_list)

Step 5: Downloading Product Images

This step covers downloading images linked to the products in the database. Downloading images can be useful for providing a visual representation of products in an application or website.

First, we extract information about the images to be downloaded from the database. We use a SQL query to retrieve the necessary data:

import requests
import json
import time
import os
import concurrent.futures
import sqlite3
from datetime import datetime

conn = sqlite3.connect('main.db')
c = conn.cursor()
c.execute("SELECT id, image, imageArticleId FROM product;")
rows = c.fetchall()
conn.close()

# create folder if not exists images
if not os.path.exists('images'):
    os.makedirs('images')

# download images from database concurrently
def download_image(row):
    url = row[1]
    imageArticleId = row[2]
    # check if image already exists
    if os.path.exists(f'images/{imageArticleId}.png'):
        print(f'Image {imageArticleId} already exists')
        return
    else:
        r = requests.get(url)
        with open(f'images/{imageArticleId}.png', 'wb') as f:
            f.write(r.content)
        print(f'Image {imageArticleId} downloaded')
        return

# download images concurrently
with concurrent.futures.ThreadPoolExecutor() as executor:
    executor.map(download_image, rows)

print('All images downloaded')

Step 6: Displaying Data via Flask

Flask is a popular Python web framework that allows us to create web applications quickly and easily. We use Flask to create a user interface through which users can retrieve data.

In this Flask application, we define a route (/) that handles both GET and POST requests. When a user sends a POST request (e.g., by filling out and submitting a form), the user query is sent to the database and the results are displayed on the same page.

The query selects the product name, article identifier, product identifier, price, crossed-out price, offer price, unit price, and offer unit price from the articles and product tables. It calculates the discount percentage for each product and filters for products with the tag “Tagesangebot” or “TOP-ANGEBOT” that are available. The results are sorted by discount percentage in descending order.

SELECT DISTINCT p.name,
                a.imageArticleId,
                a.id,
                a.product_id,
                a.price,
                a.crossedPrice,
                ROUND(((a.price - a.offerPrice) / a.price) * 100, 2) AS discountPercentage,
                a.offerPrice,
                a.pricePerUnit,
                a.offerPricePerUnit
FROM articles a
JOIN product p ON a.product_id = p.id
WHERE a.onTopOfferText IN ("Tagesangebot", "TOP-ANGEBOT")
AND a.isAvailable = 1
ORDER BY discountPercentage DESC
from flask import Flask, render_template, request
import sqlite3

app = Flask(__name__, static_folder="images")
DATABASE = 'main.db'

def query_database(query):
    conn = sqlite3.connect(DATABASE)
    cursor = conn.cursor()
    cursor.execute(query)
    data = cursor.fetchall()
    conn.close()
    return data, [description[0] for description in cursor.description]

@app.route('/', methods=['GET', 'POST'])
def index():
    headers = []
    rows = []

    if request.method == 'POST':
        user_query = request.form['query']
        results, headers = query_database(user_query)
        rows = results

    return render_template('index.html', headers=headers, rows=rows)

if __name__ == '__main__':
    app.run(debug=True)

Bonus Feature: Adding Items to the Cart

To add items to the shopping cart, a request is sent to the corresponding API of the Flaschenpost web shop. A JSON data structure is used to define the positions of the items to be added. This JSON object specifies the article ID, the origin of the article, and the desired quantity. After the data structure is created, a POST request is sent to the API endpoint URL https://www.flaschenpost.de/webshop-cart-api/api/v1/addCartPositions, passing the required cookies and headers. The server response to this request contains information about the updated cart, such as the included items and their total price.

json_data = {
    'positions': [
        {
            'articleId': row[2],
            'articleOrigin': 1,
            'quantity': 1,
        },
    ],
    'cartVersion': 40,
}

response = requests.post(
    'https://www.flaschenpost.de/webshop-cart-api/api/v1/addCartPositions',
    cookies=cookies,
    headers=headers,
    json=json_data,
)

Possible Extensions

  • Price Comparison and Price Tracking: By regularly recording prices for products, you can identify trends in pricing behavior, track price fluctuations, and conduct competitive analyses. This could enable setting up price alerts to notify users of price drops or increases.

  • Personalized Product Recommendations: Based on previous purchases, personalized product recommendations can be generated.

  • Inventory Tracking and Notifications: By monitoring the inventory of specific products, you can send notifications when a product is back in stock or when stock levels are low.

  • Subscription Shopping: No feature currently exists to repeat a previous purchase. Now the shopping cart can be automatically filled and an order automated.

  • LLM – Recipe Suggestions: By using LLMs, a dish can be assembled based on the data in the database. Example: “Create a cart with a healthy and affordable lunch.”