Pretix Products Parsing (#35)

An example implementation of parsing pretix data

Author: artcz

intbot/core/analysis/products.py

@@ -0,0 +1,119 @@
+"""
+Parse Products from PretixData for further joins and analysis in other places.
+"""
+
+from decimal import Decimal
+from typing import ClassVar, Iterable
+
+import polars as pl
+from core.models import PretixData
+from pydantic import BaseModel, model_validator
+
+
+class LocalisedFieldsMixin:
+    # Marking as ClassVar here is important. It doens't work without it :)
+    _localised_fields: ClassVar[Iterable[str]] = ()
+
+    @model_validator(mode="before")
+    @classmethod
+    def extract(cls, values):
+        for field in cls._localised_fields:
+            if isinstance(values[field], dict) and "en" in values[field]:
+                values[field] = values[field]["en"]
+                continue
+
+        return values
+
+
+class ProductVariation(LocalisedFieldsMixin, BaseModel):
+    id: int
+    value: str
+    description: str | dict
+    price: Decimal
+
+    _localised_fields = ["value", "description"]
+
+
+class Product(LocalisedFieldsMixin, BaseModel):
+    id: int
+    name: str
+    description: str | dict
+    variations: list[ProductVariation]
+    default_price: Decimal | None = None
+
+    _localised_fields = ["name", "description"]
+
+
+class FlatProductDescription(BaseModel):
+    product_id: int
+    variation_id: int | None
+    product_name: str
+    type: str
+    variant: str
+    price: Decimal | None
+
+
+def get_latest_products_data() -> PretixData:
+    qs = PretixData.objects.filter(resource=PretixData.PretixResources.products)
+    return qs.latest("created_at")
+
+
+def parse_latest_products_to_objects(pretix_data: PretixData) -> list[Product]:
+    data = pretix_data.content
+    products = [Product.model_validate(entry) for entry in data]
+    return products
+
+
+def flat_product_data(products: list[Product]) -> pl.DataFrame:
+    """
+    Returns a polars data frame with flat description of available Products.
+    Products hold nested `ProductVariation`s; flatten them so every variation
+    (or base product) becomes one row in a DataFrame.
+    """
+    rows = []
+    for p in products:
+        if p.variations:
+            for v in p.variations:
+                if "Late" in v.value:
+                    type_ = "Late"
+                    name = v.value.replace("Late", "").strip()
+                else:
+                    type_ = "Regular"
+                    name = v.value
+
+                if "Combined" in name:
+                    name = "Combined"
+
+                rows.append(
+                    FlatProductDescription(
+                        product_id=p.id,
+                        variation_id=v.id,
+                        product_name=p.name,
+                        type=type_,
+                        variant=name,
+                        price=v.price,
+                    )
+                )
+        else:
+            rows.append(
+                FlatProductDescription(
+                    product_id=p.id,
+                    variation_id=None,
+                    product_name=p.name,
+                    variant=p.name,
+                    type="Late" if "Late" in p.name else "Regular",
+                    price=p.default_price,
+                )
+            )
+
+    return pl.DataFrame(rows)
+
+
+def latest_flat_product_data() -> pl.DataFrame:
+    """
+    Thin wrapper on getting latest information from the database, and
+    converting into a polars data frame
+    """
+    pretix_data = get_latest_products_data()
+    products = parse_latest_products_to_objects(pretix_data)
+    return flat_product_data(products)

intbot/core/bot/main.py

@@ -1,4 +1,6 @@
 import discord
+from asgiref.sync import sync_to_async
+from core.analysis.products import latest_flat_product_data
 from core.models import DiscordMessage, InboxItem
 from discord.ext import commands, tasks
 from django.conf import settings
@@ -199,6 +201,25 @@ async def until(ctx):
     await ctx.send(f"{delta.days} days left until the conference")
 
 
+@bot.command()
+@commands.has_role("EP2025")
+async def products(ctx):
+    """
+    Returns pretix Products but only if called by a EP2025 volunteer on a test channel.
+
+    This is an example of implementation of a command that has some basic
+    access control (in this case works only on certain channels).
+    """
+
+    if ctx.channel.id != settings.DISCORD_TEST_CHANNEL_ID:
+        await ctx.send("This command only works on certain channels")
+        return
+
+    data = await sync_to_async(latest_flat_product_data)()
+
+    await ctx.send(f"```{str(data)}```")
+
+
 def run_bot():
     bot_token = settings.DISCORD_BOT_TOKEN
     bot.run(bot_token)

intbot/core/views.py

@@ -1,9 +1,39 @@
+from core.analysis.products import latest_flat_product_data
 from django.conf import settings
+from django.contrib.auth.decorators import login_required
 from django.template.response import TemplateResponse
 from django.utils import timezone
 
 
 def days_until(request):
     delta = settings.CONFERENCE_START - timezone.now()
 
-    return TemplateResponse(request, "days_until.html", {"days_until": delta.days})
+    return TemplateResponse(
+        request,
+        "days_until.html",
+        {
+            "days_until": delta.days,
+        },
+    )
+
+
+@login_required
+def products(request):
+    """
+    For now this is just an example of the implementation.
+
+    Table with products is not that useful, but it's the easiest one to
+    implement first as proof of concept
+    """
+    df = latest_flat_product_data()
+    columns = df.columns
+    rows = df.rows()
+
+    return TemplateResponse(
+        request,
+        "table.html",
+        {
+            "columns": columns,
+            "rows": rows,
+        },
+    )

intbot/intbot/urls.py

@@ -4,7 +4,7 @@
     internal_webhook_endpoint,
     zammad_webhook_endpoint,
 )
-from core.views import days_until
+from core.views import days_until, products
 from django.contrib import admin
 from django.urls import path
 
@@ -17,4 +17,5 @@
     path("webhook/zammad/", zammad_webhook_endpoint),
     # Public Pages
     path("days-until/", days_until),
+    path("products/", products),
 ]

intbot/templates/table.html

@@ -0,0 +1,36 @@
+
+<html>
+    <body>
+        <style type="text/css">
+body { background-color: #ffe; color: #151f38; font-family: sans-serif;}
+h1 { font-size: 3em; text-align: center; margin-top: 3em; }
+thead { background: #aaa; }
+table { font-family: monospace; width: 100%}
+table td { padding: .5em 1em; }
+table tr:nth-child(2n) td { background: #eee; }
+table tr:hover td { background: #fee; }
+        </style>
+        <div style="margin: 3em">
+
+            <h1>Flat Products Table</h1>
+            <p style="text-align: center">This is not particularly useful now, but it's here as a proof of concept :)</p>
+
+            <table>
+                <thead>
+                    {% for column in columns %}
+                    <td>{{ column }}</td>
+                    {% endfor %}
+                </thead>
+
+                {% for row in rows %}
+                <tr>
+                    {% for cell in row %}
+                    <td>{{ cell }}</td>
+                    {% endfor %}
+                </tr>
+                {% endfor %}
+            </table>
+
+        </div>
+    </body>
+</html>