diff --git a/AI_ASSIST.md b/AI_ASSIST.md index a50dbac..196bf96 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -2,18 +2,21 @@ ## The prompt I gave - + ## The code it suggested ```python -# Paste the relevant code the LLM suggested here. +sales["date"] = pd.to_datetime( + sales["date"], + errors="coerce" +) ``` ## What I changed and why - + ## Did it work? - + diff --git a/main.py b/main.py index 9f17efc..9a076c4 100644 --- a/main.py +++ b/main.py @@ -13,7 +13,7 @@ OUTPUT_DIR = Path("output") # TODO (Task 7): replace with your GitHub username before running the pipeline. -GITHUB_USERNAME = "" +GITHUB_USERNAME = "mohammedalfakih-dev" def run() -> None: diff --git a/src/clean.py b/src/clean.py index b4036fd..9be3b21 100644 --- a/src/clean.py +++ b/src/clean.py @@ -6,23 +6,58 @@ 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 INFO ===") + sales.info() + + logging.info("=== SALES DESCRIBE ===") + logging.info("\n%s", sales.describe(include="all")) + + logging.info("=== SALES HEAD ===") + logging.info("\n%s", sales.head(20)) + + logging.info("=== SALES MISSING VALUES ===") + logging.info("\n%s", sales.isna().sum()) + + logging.info("=== CUSTOMERS INFO ===") + customers.info() + + logging.info("=== CUSTOMERS DESCRIBE ===") + logging.info("\n%s", customers.describe(include="all")) + + logging.info("=== CUSTOMERS HEAD ===") + logging.info("\n%s", customers.head(20)) + + logging.info("=== CUSTOMERS MISSING VALUES ===") + logging.info("\n%s", customers.isna().sum()) + + 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 = sales.copy() + + 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["product_name"] != "") + & (sales["price"] >= 0) + & (sales["quantity"] != 0) + & sales["date"].notna() + ] + + sales = sales.drop_duplicates( + subset="transaction_id", + keep="first", + ) + + logging.info("Cleaned sales rows: %s", len(sales)) + + return sales diff --git a/src/ingest.py b/src/ingest.py index 01fe28f..d149554 100644 --- a/src/ingest.py +++ b/src/ingest.py @@ -14,21 +14,68 @@ 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") + data_dir.mkdir(exist_ok=True) + + credential = DefaultAzureCredential() + service = BlobServiceClient( + account_url=ACCOUNT_URL, + credential=credential, + ) + + container = service.get_container_client(SOURCE_CONTAINER) + + for filename in FILES: + blob = container.get_blob_client(filename) + + with open(data_dir / filename, "wb") as f: + f.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) + + try: + container.create_container() + logging.info("Created container %s", container_name) + except Exception: + logging.info("Container %s already exists", container_name) + + parquet_files = list(output_dir.glob("*.parquet")) + + for path in parquet_files: + blob = container.get_blob_client(path.name) + + with open(path, "rb") as f: + blob.upload_blob(f, overwrite=True) + + logging.info("Uploaded %s", path.name) + + local_customer_summary = pd.read_parquet( + output_dir / "customer_summary.parquet" + ) + + downloaded_bytes = container.get_blob_client( + "customer_summary.parquet" + ).download_blob().readall() + + remote_customer_summary = pd.read_parquet( + io.BytesIO(downloaded_bytes) + ) + + assert len(local_customer_summary) == len(remote_customer_summary) + + logging.info( + "Verified customer_summary.parquet row count: %s rows", + len(local_customer_summary), + ) diff --git a/src/report.py b/src/report.py index a002b7b..bfc4a99 100644 --- a/src/report.py +++ b/src/report.py @@ -2,30 +2,108 @@ 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. - # TODO: Build loyalty_analysis: group by loyalty_tier, columns loyalty_tier/avg_spent/customer_count. - raise NotImplementedError("Task 5: implement build_reports") + 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() + ) + + 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_sorted = reports["category_performance"].sort_values( + "total_revenue", + ascending=False, + ) + + ax = category_sorted.plot( + kind="bar", + x="category", + y="total_revenue", + title="Revenue by category", + ) + + fig = ax.get_figure() + + fig.savefig( + output_dir / "category_revenue.png", + bbox_inches="tight", + ) + + plt.close(fig) + + logging.info("Output files written successfully") diff --git a/src/transform.py b/src/transform.py index 18f82e5..00d2565 100644 --- a/src/transform.py +++ b/src/transform.py @@ -6,8 +6,20 @@ 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() + + enriched = sales.merge( + customers, + on="customer_email", + how="inner", + ) + + enriched["is_high_value"] = enriched["price"] * enriched["quantity"] >= 150 + + logging.info("Rows after customer join: %s", len(enriched)) + + return enriched