combine sounds and plato deals, use jinja for email

apps/vinyl/src/assets.py

@@ -6,8 +6,9 @@ from types import SimpleNamespace
 import polars as pl
 import structlog
 from dagster_polars.patito import patito_model_to_dagster_type
+from jinja2 import Environment, FileSystemLoader
 from models import Deal
-from partitions import multi_partitions_def
+from partitions import daily_partitions_def, multi_partitions_def
 from plato.fetch import scrape_plato
 from plato.parse import parse as parse_plato
 from shared.utils import get_partition_keys, load_partitions, parse_partition_keys
@@ -163,7 +164,7 @@ def works(context: dg.AssetExecutionContext) -> pl.DataFrame | None:
     automation_condition=dg.AutomationCondition.eager(),
 )
 def new_deals(
-    context: dg.AssetExecutionContext, partitions: dict[str, pl.LazyFrame | None]
+    context: dg.AssetExecutionContext, partitions: dict[str, pl.LazyFrame]
 ) -> Iterator[dg.Output[Deal.DataFrame]]:
     """Fetch new deals from all sources."""
     ic()
@@ -203,50 +204,47 @@ def new_deals(
 
 @dg.asset(
     io_manager_key="polars_parquet_io_manager",
-    partitions_def=multi_partitions_def,
-    ins={"df": dg.AssetIn(key=new_deals.key)},
+    partitions_def=daily_partitions_def,
+    metadata={
+        "partition_by": ["date", "source", "release"],
+    },
+    ins={"partitions": dg.AssetIn(key=new_deals.key)},
     output_required=False,
     automation_condition=dg.AutomationCondition.eager(),
 )
 def good_deals(
-    context: dg.AssetExecutionContext, email_service: EmailService, df: pl.LazyFrame
+    context: dg.AssetExecutionContext,
+    email_service: EmailService,
+    partitions: dict[str, pl.LazyFrame],
 ) -> Iterator[dg.Output[Deal.DataFrame]]:
-    filtered_df = df.filter(pl.col("price") <= 25).collect()
-    num_rows = filtered_df.height
-    if not num_rows:
-        context.log.info("No good deals found!")
+    parsed_partition_keys = parse_partition_keys(context, "partitions")
+    ic(parsed_partition_keys)
+
+    df = pl.concat(partitions.values(), how="vertical_relaxed").collect()
+
+    counts = dict(df.group_by("source").len().iter_rows())
+    logger.info(f"Processing new deals ({df.height}x).", counts=counts)
+
+    filtered_df = df.filter(pl.col("price") <= 25)
+    if filtered_df.is_empty():
+        logger.info("No good deals found!")
         return
 
-    context.log.info(f"Good deals found ({num_rows}x)!")
+    logger.info(f"Good deals found ({filtered_df.height}x)!", counts=counts)
     yield dg.Output(Deal.DataFrame(filtered_df))
 
-    lines = []
-    lines.append(
-        """
-    <div style="font-family: Arial, sans-serif; background-color: #f4f4f4; padding: 20px;">
-      <h2 style="color: #333;">🎶 New Music Releases</h2>
-    """
-    )
+    # Prepare data for email
+    deals: dict[str, list[SimpleNamespace]] = {}
+    for source in filtered_df.select("source").unique().to_series():
+        group_df = filtered_df.filter(pl.col("source") == source)
+        deals[source] = [
+            SimpleNamespace(**row) for row in group_df.head(10).iter_rows(named=True)
+        ]
 
-    # Each item
-    for data in filtered_df.head(10).iter_rows(named=True):
-        row = SimpleNamespace(**data)
-        lines.append(
-            f"""
-      <div style="background-color: #fff; margin-bottom: 20px; padding: 15px; border-radius: 8px; box-shadow: 0 2px 5px rgba(0,0,0,0.05);">
-        <a href="https://www.platomania.nl{row.url}" style="text-decoration: none; color: #1a73e8;">
-          <h3 style="margin: 0 0 10px;">🆕 {row.artist} - {row.title}</h3>
-        </a>
-        <ul style="list-style: none; padding-left: 0; margin: 0;">
-          <li><strong>Artist:</strong> {row.artist}</li>
-          <li><strong>Title:</strong> {row.title}</li>
-          <li><strong>Price:</strong> €{row.price}</li>
-          <li><strong>Release Date:</strong> {row.release}</li>
-        </ul>
-      </div>
-    """
-        )
+    # Render HTML from Jinja template
+    env = Environment(loader=FileSystemLoader(".."))
+    template = env.get_template("email.html")
+    html_content = template.render(deals=deals)
 
-    # Email footer
-    lines.append("</div>")
-    email_service.send_email("\n".join(lines))
+    # Send the email
+    email_service.send_email(html_content)