Pavel Tisner

AI_ASSIST.md

@@ -2,18 +2,33 @@
 
 ## The prompt I gave
 
-<!-- Paste the exact prompt you gave the LLM here. -->
+I asked an LLM to help implement a Week 4 Pandas pipeline for the MessyCorp assignment. The tasks were to download CSV inputs from Azure Blob Storage, explore the raw data, clean sales data with vectorized Pandas operations, join customer data, build report tables with groupby named aggregations, and write CSV/Parquet/chart outputs.
 
 ## The code it suggested
 
-```python
-# Paste the relevant code the LLM suggested here.
-```
+The LLM suggested using:
+
+- `DefaultAzureCredential` and `BlobServiceClient` for downloading input files from Azure.
+- `pd.read_csv()` for loading `messy_sales.csv` and `messy_customers.csv`.
+- `.str.strip().str.title()` to normalize product names.
+- `.str.lower().str.strip()` to normalize customer emails.
+- `pd.to_numeric(..., errors="coerce")` for price conversion.
+- `pd.to_datetime(..., errors="coerce")` for date parsing.
+- Boolean filters for removing bad rows.
+- `.drop_duplicates(subset="transaction_id", keep="first")` for duplicate transactions.
+- `merge(..., how="inner")` for joining sales and customers.
+- A vectorized `is_high_value` column based on `price * quantity >= 150`.
+- `groupby().agg(...)` with named aggregations for report tables.
+- `to_csv()`, `to_parquet()`, and `plt.savefig()` for outputs.
 
 ## What I changed and why
 
-<!-- Describe what you kept, what you modified, and what you threw away. -->
+I followed the assignment comments closely and implemented the code step by step. I left out Task 7 upload functionality for now because it is extra credit. I kept the outlier price values unchanged and added a comment explaining that decision, because the assignment does not define a business rule for clipping or removing outlier prices.
+
+I also used `matplotlib.use("Agg")` before importing `pyplot` so the chart can be generated in headless environments.
 
 ## Did it work?
 
-<!-- Yes / partially / no — and what you learned from the interaction. -->
+Yes. Tasks 1–6 worked successfully. The pipeline downloads the input CSV files from Azure, explores the raw data, cleans the sales data, joins it with customer data, builds the report tables, and writes the CSV, Parquet, and PNG chart outputs.
+
+The main thing I learned was how Pandas replaces manual row-by-row loops with vectorized operations, boolean filters, joins, and groupby aggregations.

main.py

@@ -12,8 +12,7 @@
 DATA_DIR = Path("data")
 OUTPUT_DIR = Path("output")
 
-# TODO (Task 7): replace with your GitHub username before running the pipeline.
-GITHUB_USERNAME = "<your-github-username>"
+GITHUB_USERNAME = "pavel-tisner"
 
 
 def run() -> None:
@@ -26,7 +25,7 @@ def run() -> None:
 
     reports = build_reports(enriched)
     write_outputs(reports, OUTPUT_DIR)
-
+    
     upload_outputs(OUTPUT_DIR, GITHUB_USERNAME)
 
     logging.info("Pipeline complete.")

src/clean.py

@@ -7,22 +7,42 @@
 
 def load_and_explore(data_dir: Path) -> tuple[pd.DataFrame, pd.DataFrame]:
     """Task 2: Load both CSV files and explore their contents before cleaning."""
-    # TODO: Read messy_sales.csv and messy_customers.csv with pd.read_csv().
-    # TODO: For each DataFrame call .info(), .describe(), .head(20), and .isna().sum().
-    # TODO: Log what you discover (e.g. which columns have nulls, any suspicious values).
-    raise NotImplementedError("Task 2: implement load_and_explore")
+    sales = pd.read_csv(data_dir / "messy_sales.csv")
+    customers = pd.read_csv(data_dir / "messy_customers.csv")
+
+    logging.info("Sales DataFrame info:")
+    sales.info()
+
+    logging.info("Customers DataFrame info:")
+    customers.info()
+
+    logging.info("Sales describe:\n%s", sales.describe())
+    logging.info("Customers describe:\n%s", customers.describe())
+
+    logging.info("Sales first 20 rows:\n%s", sales.head(20))
+    logging.info("Customers first 20 rows:\n%s", customers.head(20))
+
+    logging.info("Sales missing values:\n%s", sales.isna().sum())
+    logging.info("Customers missing values:\n%s", customers.isna().sum())
+
+    logging.info("Exploration complete. Check missing values, dtypes, and numeric outliers above.")
+
+    return sales, customers
 
 
 def clean_sales(sales: pd.DataFrame) -> pd.DataFrame:
     """Task 3: Clean the sales DataFrame using vectorized Pandas operations."""
-    # TODO: Normalize product_name with .str.strip().str.title().
-    # TODO: Normalize customer_email with .str.lower().str.strip().
-    # TODO: Convert price to numeric with pd.to_numeric(errors="coerce").
-    # TODO: Parse date with pd.to_datetime(errors="coerce").
-    # TODO: Drop rows where product_name is missing.
-    # TODO: Drop rows where price is negative.
-    # TODO: Drop rows where quantity is zero.
-    # TODO: Drop rows where date is NaT (invalid after parsing).
-    # TODO: Remove duplicate transactions: .drop_duplicates(subset="transaction_id", keep="first").
-    # TODO: Decide what to do with outlier prices (clip, flag, or leave) and add a comment explaining why.
-    raise NotImplementedError("Task 3: implement clean_sales")
+    sales["product_name"] = sales["product_name"].str.strip().str.title()
+    sales["customer_email"] = sales["customer_email"].str.lower().str.strip()
+    sales["price"] = pd.to_numeric(sales["price"], errors="coerce")
+    sales["date"] = pd.to_datetime(sales["date"], errors="coerce")
+    sales = sales[sales["product_name"].notna()]
+    sales = sales[sales["price"] >= 0]
+    sales = sales[sales["quantity"] != 0]
+    sales = sales[sales["date"].notna()]
+    sales = sales.drop_duplicates(subset="transaction_id", keep="first")
+
+    # Outlier prices are left unchanged because the assignment does not define a business rule for capping prices. Keeping them preserves the original transaction value for reporting.
+    logging.info("Cleaned sales rows: %s", len(sales))
+
+    return sales

src/ingest.py

@@ -14,21 +14,49 @@
 
 def download_inputs(data_dir: Path) -> None:
     """Task 1: Download input CSV files from Azure Blob Storage."""
-    # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL.
-    # TODO: Get a container client for SOURCE_CONTAINER.
-    # TODO: For each filename in FILES, download the blob and write it to data_dir/<filename>.
-    # TODO: Log a message for each downloaded file.
-    raise NotImplementedError("Task 1: implement download_inputs")
+    credential = DefaultAzureCredential()
+    service = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential)
+    container = service.get_container_client(SOURCE_CONTAINER)
+
+    data_dir.mkdir(exist_ok=True)
+
+    for filename in FILES:
+        blob = container.get_blob_client(filename)
+        output_path = data_dir / filename
+        with output_path.open("wb") as file:
+            file.write(blob.download_blob().readall())
+
+        logging.info("Downloaded %s", filename)
 
 
 def upload_outputs(output_dir: Path, github_username: str) -> None:
     """Task 7 (extra credit): Upload Parquet outputs to Azure and verify the round-trip."""
     container_name = f"week4-{github_username}"
 
-    # EXTRA CREDIT — implement this after Tasks 2–6 are working.
-    # TODO: Create a BlobServiceClient using DefaultAzureCredential and ACCOUNT_URL.
-    # TODO: Get (or create) the container named container_name.
-    # TODO: Upload every .parquet file in output_dir to the container.
-    # TODO: Download customer_summary.parquet back and assert its row count matches the local file.
-    # TODO: Log the container name and number of files uploaded.
-    raise NotImplementedError("Task 7: implement upload_outputs")
+    credential = DefaultAzureCredential()
+    service = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential)
+
+    container = service.get_container_client(container_name)
+    if not container.exists():
+        container.create_container()
+
+    parquet_files = list(output_dir.glob("*.parquet"))
+    for file_path in parquet_files:
+        blob = container.get_blob_client(file_path.name)
+        with file_path.open("rb") as file:
+            blob.upload_blob(file, overwrite=True)
+
+        logging.info("Uploaded %s", file_path.name)
+
+    local_customer_summary = pd.read_parquet(output_dir / "customer_summary.parquet")
+    blob = container.get_blob_client("customer_summary.parquet")
+    downloaded_bytes = blob.download_blob().readall()
+    downloaded_customer_summary = pd.read_parquet(io.BytesIO(downloaded_bytes))
+
+    assert len(downloaded_customer_summary) == len(local_customer_summary)
+
+    logging.info(
+        "Uploaded %s parquet file(s) to container %s",
+        len(parquet_files),
+        container_name,
+    )

src/report.py

@@ -2,30 +2,102 @@
 import logging
 from pathlib import Path
 
+import matplotlib
+
+matplotlib.use("Agg")
+
+import matplotlib.pyplot as plt
 import pandas as pd
 
 
 def build_reports(enriched: pd.DataFrame) -> dict[str, pd.DataFrame]:
     """Task 5: Build four summary tables using groupby and named aggregations."""
-    # TODO: Add a week column using .dt.isocalendar().week.
-    # TODO: Build weekly_revenue: group by week and region, columns week/region/total_revenue/order_count.
-    # TODO: Build customer_summary: group by customer_email, columns customer_email/customer_name/
-    #       region/loyalty_tier/total_spent/avg_order/order_count.
-    #       Use ("customer_name", "first") to pick the constant-per-group string columns.
-    # TODO: Build category_performance: group by category, columns category/total_revenue/order_count.
+    enriched = enriched.copy()
+    enriched["revenue"] = enriched["price"] * enriched["quantity"]
+
+    enriched["week"] = enriched["date"].dt.isocalendar().week
+
+    weekly_revenue = (
+        enriched.groupby(["week", "region"])
+        .agg(
+            total_revenue=("revenue", "sum"),
+            order_count=("transaction_id", "count"),
+        )
+        .reset_index()
+    )
+
+    customer_summary = (
+        enriched.groupby("customer_email")
+        .agg(
+            customer_name=("customer_name", "first"),
+            region=("region", "first"),
+            loyalty_tier=("loyalty_tier", "first"),
+            total_spent=("revenue", "sum"),
+            avg_order=("revenue", "mean"),
+            order_count=("transaction_id", "count"),
+        )
+        .reset_index()
+    )
+
+    category_performance = (
+        enriched.groupby("category")
+        .agg(
+            total_revenue=("revenue", "sum"),
+            order_count=("transaction_id", "count"),
+        )
+        .reset_index()
+    )
+
     # TODO: Build loyalty_analysis: group by loyalty_tier, columns loyalty_tier/avg_spent/customer_count.
-    raise NotImplementedError("Task 5: implement build_reports")
+
+    loyalty_analysis = (
+        enriched.groupby("loyalty_tier")
+        .agg(
+            avg_spent=("revenue", "mean"),
+            customer_count=("customer_email", "nunique"),
+        )
+        .reset_index()
+    )
+
+    logging.info("Built report tables.")
+
+    return {
+        "weekly_revenue": weekly_revenue,
+        "customer_summary": customer_summary,
+        "category_performance": category_performance,
+        "loyalty_analysis": loyalty_analysis,
+    }
 
 
 def write_outputs(reports: dict[str, pd.DataFrame], output_dir: Path) -> None:
     """Task 6: Write report tables to CSV/Parquet and save a bar chart."""
     output_dir.mkdir(exist_ok=True)
 
-    # TODO: Write reports["weekly_revenue"] to weekly_revenue.csv with index=False.
-    # TODO: Write reports["customer_summary"] to customer_summary.parquet with index=False.
-    # TODO: Write reports["category_performance"] to category_performance.csv with index=False.
-    # TODO: Sort category_performance by total_revenue descending.
-    # TODO: Plot a bar chart (x="category", y="total_revenue") and save to category_revenue.png
-    #       using plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight").
-    #       Use matplotlib.use("Agg") before importing pyplot for headless environments.
-    raise NotImplementedError("Task 6: implement write_outputs")
+    reports["weekly_revenue"].to_csv(output_dir / "weekly_revenue.csv", index=False)
+
+    reports["customer_summary"].to_parquet(
+        output_dir / "customer_summary.parquet",
+        index=False,
+    )
+
+    reports["category_performance"].to_csv(
+        output_dir / "category_performance.csv",
+        index=False,
+    )
+
+    category_performance = reports["category_performance"].sort_values(
+        "total_revenue",
+        ascending=False,
+    )
+
+    category_performance.plot(
+        kind="bar",
+        x="category",
+        y="total_revenue",
+        title="Revenue by category",
+    )
+
+    plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight")
+    plt.close()
+
+    logging.info("Wrote outputs to %s", output_dir)

src/transform.py

@@ -6,8 +6,21 @@
 
 def join_customers(sales: pd.DataFrame, customers: pd.DataFrame) -> pd.DataFrame:
     """Task 4: Normalize join keys, merge, and add a derived boolean flag."""
-    # TODO: Normalize customer_email in both DataFrames with .str.lower().str.strip().
-    # TODO: Merge sales with customers on customer_email using an inner join.
-    # TODO: Add a vectorized boolean column is_high_value: True where price * quantity >= 150.
-    # TODO: (Optional hands-on) Try a left join instead and inspect rows where customer_name is NaN.
-    raise NotImplementedError("Task 4: implement join_customers")
+    sales = sales.copy()
+    customers = customers.copy()
+
+    sales["customer_email"] = sales["customer_email"].str.lower().str.strip()
+    customers["customer_email"] = customers["customer_email"].str.lower().str.strip()
+
+    orphan_check = sales.merge(customers, on="customer_email", how="left")
+    orphan_orders = orphan_check[orphan_check["customer_name"].isna()]
+
+    logging.info("Orphan orders after left join: %s", len(orphan_orders))
+
+    enriched = sales.merge(customers, on="customer_email", how="inner")
+
+    enriched["is_high_value"] = enriched["price"] * enriched["quantity"] >= 150
+
+    logging.info("Rows after inner join: %s", len(enriched))
+
+    return enriched