Baraah A.

AI_ASSIST.md

@@ -3,17 +3,34 @@
 ## The prompt I gave
 
 <!-- Paste the exact prompt you gave the LLM here. -->
-
+I needed help implementing the final part of Task 7 in my Pandas + Azure data pipeline assignment.
 ## The code it suggested
+local_file = output_dir / "customer_summary.parquet"
+
+blob = container.get_blob_client("customer_summary.parquet")
+downloaded = blob.download_blob().readall()
+
+remote_df = pd.read_parquet(io.BytesIO(downloaded))
+local_df = pd.read_parquet(local_file)
 
+assert len(remote_df) == len(local_df), "Row count mismatch!"
+
+logging.info("Round-trip verification passed ✔")
+logging.info(f"Uploaded {len(parquet_files)} files to {container_name}")
 ```python
 # Paste the relevant code the LLM suggested here.
 ```
 
 ## What I changed and why
 
 <!-- Describe what you kept, what you modified, and what you threw away. -->
+I did not change the main logic because the round-trip verification approach was already correct.
+I kept the assert statement because it is a simple and effective way to validate data consistency.
+I kept logging as it helps track successful execution in the pipeline.
 
 ## Did it work?
 
 <!-- Yes / partially / no — and what you learned from the interaction. -->
+The code worked successfully and uploaded files to Azure and downloaded customer_summary.parquet back from Blob Storage , also verified that the row counts matched without any errors
+
+I learned that the key part of this step is not just uploading files, but ensuring data integrity between local and cloud storage.

main.py

@@ -13,7 +13,7 @@
 OUTPUT_DIR = Path("output")
 
 # TODO (Task 7): replace with your GitHub username before running the pipeline.
-GITHUB_USERNAME = "<your-github-username>"
+GITHUB_USERNAME = "thebaraah"
 
 
 def run() -> None:

src/clean.py

@@ -10,19 +10,65 @@ def load_and_explore(data_dir: Path) -> tuple[pd.DataFrame, pd.DataFrame]:
     # 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).
+    sales_path = data_dir / "messy_sales.csv"
+    customers_path = data_dir / "messy_customers.csv"
+
+    sales = pd.read_csv(sales_path)
+    customers = pd.read_csv(customers_path)
+
+    logging.info("=== SALES INFO ===")
+    logging.info(sales.info())
+
+    logging.info("=== CUSTOMERS INFO ===")
+    logging.info(customers.info())
+
+    logging.info("=== SALES MISSING VALUES ===")
+    logging.info(sales.isna().sum())
+
+    logging.info("=== CUSTOMERS MISSING VALUES ===")
+    logging.info(customers.isna().sum())
+
+    logging.info("=== SALES SUMMARY ===")
+    logging.info(sales.describe())
+
+    logging.info("=== CUSTOMERS SAMPLE ===")
+    logging.info(sales.head(20))
+
+    return sales, customers
     raise NotImplementedError("Task 2: implement load_and_explore")
 
 
 def clean_sales(sales: pd.DataFrame) -> pd.DataFrame:
     """Task 3: Clean the sales DataFrame using vectorized Pandas operations."""
+    df = sales.copy()
     # TODO: Normalize product_name with .str.strip().str.title().
+    df["product_name"] = df["product_name"].str.strip().str.title()
+
     # TODO: Normalize customer_email with .str.lower().str.strip().
+    df["customer_email"] = df["customer_email"].str.lower().str.strip()
     # TODO: Convert price to numeric with pd.to_numeric(errors="coerce").
+    df["price"] = pd.to_numeric(df["price"], errors="coerce")
     # TODO: Parse date with pd.to_datetime(errors="coerce").
+    df["date"] = pd.to_datetime(df["date"], errors="coerce")
     # TODO: Drop rows where product_name is missing.
+    df = df[df["product_name"].notna()]
+    df = df[df["price"].notna()]
+    df = df[df["price"] >= 0]
+    df = df[df["quantity"] > 0]
+    df = df[df["date"].notna()]
+
     # TODO: Drop rows where price is negative.
+    df = df.drop_duplicates(subset="transaction_id", keep="first")
     # 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.
+    q99 = df["price"].quantile(0.99)
+    df["price"] = df["price"].clip(upper=q99)
+
+    logging.info(f"Cleaned sales rows: {len(df)}")
+
+    return df
+
     raise NotImplementedError("Task 3: implement clean_sales")

src/ingest.py

@@ -15,20 +15,70 @@
 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.
+     #Create Azure client
+    credential = DefaultAzureCredential()
+    service = BlobServiceClient(
+        account_url=ACCOUNT_URL,
+        credential=credential
+    )
     # TODO: Get a container client for SOURCE_CONTAINER.
+    container_client = service.get_container_client(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")
+    data_dir.mkdir(parents=True, exist_ok=True)
+    for filename in FILES:
+        blob = container_client.get_blob_client(filename)
+
+        file_path = data_dir / filename
+
+        with open(file_path, "wb") as f:
+            f.write(blob.download_blob().readall())
+
+        logging.info(f"Downloaded {filename} -> {file_path}")
+
 
 
 def upload_outputs(output_dir: Path, github_username: str) -> None:
     """Task 7 (extra credit): Upload Parquet outputs to Azure and verify the round-trip."""
+    logging.info("Starting upload to Azure...")
     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.
+
+    credential = DefaultAzureCredential()
+    service = BlobServiceClient(
+        account_url=ACCOUNT_URL,
+        credential=credential
+    )
     # TODO: Get (or create) the container named container_name.
+    container_client = service.get_container_client(container_name)
+    try:
+        container_client.create_container()
+    except Exception as e:
+        logging.warning(f"Container might already exist: {e}")
+
     # TODO: Upload every .parquet file in output_dir to the container.
+    parquet_files = list(output_dir.glob("*.parquet"))
+
+    for file_path in parquet_files:
+        blob = container_client.get_blob_client(file_path.name)
+
+        with open(file_path, "rb") as data:
+            blob.upload_blob(data, overwrite=True)
+
+        logging.info(f"Uploaded {file_path.name}")
     # 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")
+    local_file = output_dir / "customer_summary.parquet"
+
+    blob = container_client.get_blob_client("customer_summary.parquet")
+    downloaded = blob.download_blob().readall()
+
+    remote_df = pd.read_parquet(io.BytesIO(downloaded))
+    local_df = pd.read_parquet(local_file)
+
+    assert len(remote_df) == len(local_df), "Row count mismatch!"
+
+    logging.info("Round-trip verification passed ✔")
+    logging.info(f"Uploaded {len(parquet_files)} files to {container_name}")

src/report.py

@@ -3,29 +3,107 @@
 from pathlib import Path
 
 import pandas as pd
+import matplotlib
+matplotlib.use("Agg")
+
+import matplotlib.pyplot as plt
 
 
 def build_reports(enriched: pd.DataFrame) -> dict[str, pd.DataFrame]:
     """Task 5: Build four summary tables using groupby and named aggregations."""
+    df = enriched.copy()
+
     # TODO: Add a week column using .dt.isocalendar().week.
+    df["week"] = df["date"].dt.isocalendar().week
+
     # TODO: Build weekly_revenue: group by week and region, columns week/region/total_revenue/order_count.
+    weekly_revenue = (
+        df.groupby(["week", "region"])
+        .agg(
+            total_revenue=("price", lambda x: (x * df.loc[x.index, "quantity"]).sum()),
+            order_count=("transaction_id", "count")
+        )
+        .reset_index()
+    )
+
     # 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.
+    customer_summary = (
+        df.groupby("customer_email")
+        .agg(
+            customer_name=("customer_name", "first"),
+            region=("region", "first"),
+            loyalty_tier=("loyalty_tier", "first"),
+            total_spent=("price", lambda x: (x * df.loc[x.index, "quantity"]).sum()),
+            avg_order=("price", lambda x: (x * df.loc[x.index, "quantity"]).mean()),
+            order_count=("transaction_id", "count")
+        )
+        .reset_index()
+    )
     # TODO: Build category_performance: group by category, columns category/total_revenue/order_count.
+    category_performance = (
+        df.groupby("category")
+        .agg(
+            total_revenue=("price", lambda x: (x * df.loc[x.index, "quantity"]).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 = (
+        df.groupby("loyalty_tier")
+        .agg(
+            avg_spent=("price", lambda x: (x * df.loc[x.index, "quantity"]).mean()),
+            customer_count=("customer_email", "nunique")
+        )
+        .reset_index()
+    )
+
+    logging.info("Reports built successfully")
+
+    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.
+    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
+    )
     # TODO: Write reports["customer_summary"] to customer_summary.parquet with index=False.
+    cat = reports["category_performance"].sort_values(
+        "total_revenue",
+        ascending=False
+    )
+
     # TODO: Write reports["category_performance"] to category_performance.csv with index=False.
+    plt.figure()
+    cat.plot(
+        kind="bar",
+        x="category",
+        y="total_revenue",
+        title="Revenue by category"
+    )
     # TODO: Sort category_performance by total_revenue descending.
+    plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight")
+
+    logging.info(f"Outputs written to {output_dir}")
     # 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")

src/transform.py

@@ -6,8 +6,18 @@
 
 def join_customers(sales: pd.DataFrame, customers: pd.DataFrame) -> pd.DataFrame:
     """Task 4: Normalize join keys, merge, and add a derived boolean flag."""
+    sales = sales.copy()
+    customers = customers.copy()
     # TODO: Normalize customer_email in both DataFrames with .str.lower().str.strip().
+    sales["customer_email"] = sales["customer_email"].str.lower().str.strip()
+    customers["customer_email"] = customers["customer_email"].str.lower().str.strip()
     # TODO: Merge sales with customers on customer_email using an inner join.
+    df = sales.merge(customers, on="customer_email", how="inner")
     # 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.
+    df["is_high_value"] = (df["price"] * df["quantity"]) >= 150
+
+    logging.info(f"Merged rows: {len(df)}")
+
+    return df
     raise NotImplementedError("Task 4: implement join_customers")