From 7b98c67168ffbe81e4f43def2f997dfa86456c11 Mon Sep 17 00:00:00 2001 From: Mareh Date: Fri, 29 May 2026 00:38:17 +0200 Subject: [PATCH] week4 --- AI_ASSIST.md | 8 +++--- main.py | 7 +++--- src/clean.py | 48 ++++++++++++++++++++++++------------ src/ingest.py | 52 ++++++++++++++++++++++++++++++++------- src/report.py | 64 ++++++++++++++++++++++++++++++++++-------------- src/transform.py | 11 ++++++--- 6 files changed, 137 insertions(+), 53 deletions(-) diff --git a/AI_ASSIST.md b/AI_ASSIST.md index a50dbac..19cbcd6 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -2,18 +2,18 @@ ## The prompt I gave - + ## 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 - + ## Did it work? - + diff --git a/main.py b/main.py index 9f17efc..9db8622 100644 --- a/main.py +++ b/main.py @@ -13,11 +13,12 @@ OUTPUT_DIR = Path("output") # TODO (Task 7): replace with your GitHub username before running the pipeline. -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.") diff --git a/src/clean.py b/src/clean.py index b4036fd..508ecf1 100644 --- a/src/clean.py +++ b/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 diff --git a/src/ingest.py b/src/ingest.py index 01fe28f..5359c15 100644 --- a/src/ingest.py +++ b/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/. - # 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 diff --git a/src/report.py b/src/report.py index a002b7b..c144664 100644 --- a/src/report.py +++ b/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() diff --git a/src/transform.py b/src/transform.py index 18f82e5..51f41e9 100644 --- a/src/transform.py +++ b/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