Mareh A.

AI_ASSIST.md

@@ -2,18 +2,18 @@
 
 ## The prompt I gave
 
-<!-- Paste the exact prompt you gave the LLM here. -->
+<!-- I ran main.py but the download_inputs function failed with a ContainerNotFound error from Azure. How do I fix this? -->
 
 ## The code it suggested
 
 ```python
-# Paste the relevant code the LLM suggested here.
+# LLMS RESPONSE: If the Azure container is missing, you can just download the messy_sales.csv and messy_customers.csv files manually from your portal, drop them directly into your local data/ folder, and comment out the download_inputs(DATA_DIR) function in main.py so you can keep moving.
 ```
 
 ## What I changed and why
 
-<!-- Describe what you kept, what you modified, and what you threw away. -->
+<!-- I didn't want to go with the manual workaround the LLM suggested. It just feels unprofessional when the entire point of the project is to automate the workflow. -->
 
 ## Did it work?
 
-<!-- Yes / partially / no — and what you learned from the interaction. -->
+<!-- No,I still try to figure out how to solve it -->

main.py

@@ -13,11 +13,12 @@
 OUTPUT_DIR = Path("output")
 
 # TODO (Task 7): replace with your GitHub username before running the pipeline.
-GITHUB_USERNAME = "<your-github-username>"
+GITHUB_USERNAME = "mareh-aboghanem"
 
 
 def run() -> None:
-    download_inputs(DATA_DIR)
+    # it works once then i commented out the download and upload functions to avoid unnecessary Azure interactions during development.
+    #download_inputs(DATA_DIR)
 
     sales_raw, customers_raw = load_and_explore(DATA_DIR)
 
@@ -27,7 +28,7 @@ def run() -> None:
     reports = build_reports(enriched)
     write_outputs(reports, OUTPUT_DIR)
 
-    upload_outputs(OUTPUT_DIR, GITHUB_USERNAME)
+    #upload_outputs(OUTPUT_DIR, GITHUB_USERNAME)
 
     logging.info("Pipeline complete.")
 

src/clean.py

@@ -7,22 +7,40 @@
 
 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")
+    data_sales = pd.read_csv(data_dir / "messy_sales.csv")
+    data_customers = pd.read_csv(data_dir / "messy_customers.csv")
+
+    logging.info("--- Exploring Data ---")
+    logging.info("\n== Sales Data ==")
+    data_sales.info()
+    logging.info(f"\nDescribe:\n{data_sales.describe()}")
+    logging.info(f"\nRows:\n{data_sales.head(20)}")
+    logging.info(f"\nMissing Values:\n{data_sales.isna().sum()}")
+    logging.info("Exploration of Sales Data is complete.")
+    logging.info("\n== Customers Data ==")
+    data_customers.info()  
+    logging.info(f"\nDescribe:\n{data_customers.describe()}")
+    logging.info(f"\nFirst Rows:\n{data_customers.head(20)}")
+    logging.info(f"\nMissing Values:\n{data_customers.isna().sum()}")
+    logging.info("Exploration of Customers Data is complete.")
+    return data_sales, data_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")
+    product_name = sales["product_name"].str.strip().str.title()
+    sales["product_name"] = product_name
+    customer_email = sales["customer_email"].str.lower().str.strip()
+    sales["customer_email"] = customer_email
+    price = pd.to_numeric(sales["price"], errors="coerce")
+    sales["price"] = price
+    date = pd.to_datetime(sales["date"], errors="coerce")
+    sales["date"] = date
+    sales = sales.dropna(subset=["product_name"])
+    sales = sales[sales["price"] >= 0]
+    sales = sales[sales["quantity"] > 0]
+    sales = sales.dropna(subset=["date"])
+    sales = sales.drop_duplicates(subset="transaction_id", keep="first")
+    logging.info(f"cleaning complete. Rows remaining: {len(sales)}")
+    # Decision: Leave outlier prices as they are. Why? Because I think they could be valid values and I need to understand the detalis of the data.
+    return sales

src/ingest.py

@@ -2,23 +2,57 @@
 import io
 import logging
 from pathlib import Path
-
+import os
+import shutil
 import pandas as pd
+from dotenv import load_dotenv
 from azure.identity import DefaultAzureCredential
 from azure.storage.blob import BlobServiceClient
 
-ACCOUNT_URL = "https://sthyfstudentsdemo.blob.core.windows.net"
-SOURCE_CONTAINER = "week4-inputs"
+
+PROJECT_ROOT = Path(__file__).resolve().parents[1]
+load_dotenv(PROJECT_ROOT / ".env")
+load_dotenv()
+
+ACCOUNT_URL = os.getenv("ACCOUNT_URL")
+SOURCE_CONTAINER = os.getenv("SOURCE_CONTAINER")
+if not ACCOUNT_URL or not SOURCE_CONTAINER:
+    raise RuntimeError(
+        "Please set ACCOUNT_URL and SOURCE_CONTAINER environment variables before running."
+    )
 FILES = ["messy_sales.csv", "messy_customers.csv"]
 
 
 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)
+    if not container.exists():
+        logging.info(
+            f"Container '{SOURCE_CONTAINER}' not found."
+        )
+    data_dir.mkdir(parents=True, exist_ok=True)
+    for name in FILES:
+        blob = container.get_blob_client(name)
+        with open(data_dir / name, "wb") as f:
+            f.write(blob.download_blob().readall())
+        logging.info("Downloaded %s", name)
+
+
+def load_inputs_local(data_dir: Path) -> None:
+    """Because of the Fallback i implented this function to load data locally"""
+    data_dir.mkdir(exist_ok=True)
+    sample_data_dir = Path(__file__).resolve().parent.parent / "sample_data"
+
+    for name in FILES:
+        source_file = sample_data_dir / name
+        destination_file = data_dir / name
+        if source_file.exists():
+            shutil.copy(source_file, destination_file)
+            logging.info("Successfully loaded %s from local sample_data", name)
+        else:
+            logging.error("File %s not found in sample_data!", name)
 
 
 def upload_outputs(output_dir: Path, github_username: str) -> None:
@@ -31,4 +65,4 @@ def upload_outputs(output_dir: Path, github_username: str) -> None:
     # 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")
+    pass

src/report.py

@@ -1,31 +1,59 @@
 """Tasks 5 and 6: Build report tables and write outputs."""
+import matplotlib
+matplotlib.use("Agg")
+import matplotlib.pyplot as plt
 import logging
 from pathlib import Path
-
 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.
-    # TODO: Build loyalty_analysis: group by loyalty_tier, columns loyalty_tier/avg_spent/customer_count.
-    raise NotImplementedError("Task 5: implement build_reports")
+    week = enriched["date"].dt.isocalendar().week
+    enriched["week"] = week
+    weekly_revenue = enriched.groupby(["week", "region"], as_index=False).agg(
+        total_revenue=("revenue", "sum"), order_count=("transaction_id", "count")
+    )
+    customer_summary = enriched.groupby("customer_email", as_index=False).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"),
+    )
+    category_performance = enriched.groupby("category", as_index=False).agg(
+        total_revenue=("revenue", "sum"), order_count=("transaction_id", "count")
+    )
+    loyalty_analysis = enriched.groupby("loyalty_tier", as_index=False).agg(
+        avg_spent=("revenue", "mean"), customer_count=("customer_email", "nunique")
+    )
+    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
+    )
+    cat_df = reports["category_performance"].sort_values(
+        by="total_revenue", ascending=False
+    )
+    cat_df.to_csv(output_dir / "category_performance.csv", index=False)
+    plt.figure(figsize=(10, 6))
+    plt.bar(
+        cat_df["category"], cat_df["total_revenue"], color="skyblue", edgecolor="black"
+    )
+    plt.title("Total Revenue by Category")
+    plt.xlabel("Category")
+    plt.ylabel("Total Revenue")
+    plt.xticks(rotation=45)
+    plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight")
+    plt.close()

src/transform.py

@@ -6,8 +6,11 @@
 
 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.
+    customers["customer_email"] = customers["customer_email"].str.lower().str.strip()
+    sales["customer_email"] = sales["customer_email"].str.lower().str.strip()
+    merged = sales.merge(customers, on="customer_email", how="inner")
+    merged["revenue"] = merged["price"] * merged["quantity"]
+    merged["is_high_value"] = merged["revenue"] >= 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")
+    logging.info("Joining complete. Rows in merged DataFrame: %d", len(merged))
+    return merged