diff --git a/AI_ASSIST.md b/AI_ASSIST.md index a50dbac..4a29c40 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -2,18 +2,25 @@ ## The prompt I gave - - + i am trying to switch from print() to info.logging on python but its not showing anything, i am working on a different file than main and logger is configured there ## The code it suggested ```python -# Paste the relevant code the LLM suggested here. -``` +import logging + +# __name__ evaluates to the module name (e.g., 'helper'), +# which correctly links it as a child of the root logger. +logger = logging.getLogger(__name__) + +def do_something(): + # Note: It's 'logging.info()', or 'logger.info()' if using an instance + logger.info("This is an info message from the sub-file!")``` ## What I changed and why - + ## Did it work? - + diff --git a/main.py b/main.py index 9f17efc..b49334c 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 = "" def run() -> None: diff --git a/src/clean.py b/src/clean.py index b4036fd..1296cb3 100644 --- a/src/clean.py +++ b/src/clean.py @@ -1,28 +1,63 @@ """Tasks 2 and 3: Explore and clean the raw DataFrames.""" -import logging -from pathlib import Path +from pathlib import Path import pandas as pd - +import logging +logger = logging.getLogger(__name__) 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") + messyS = pd.read_csv(data_dir / "messy_sales.csv") + messyC = pd.read_csv(data_dir / "messy_customers.csv") + logger.info("--- messy sales info ---") + messyS.info() + logger.info("--- describe ---") + logger.info(messyS.describe()) + logger.info("--- head ---") + logger.info(messyS.head(20)) + logger.info("--- missing values ---") + logger.info(messyS.isna().sum()) + logger.info("--- messy customers info ---") + messyC.info() + logger.info("--- describe ---") + logger.info(messyC.describe()) + logger.info("--- head ---") + logger.info(messyC.head(20)) + logger.info("--- missing values ---") + logger.info(messyC.isna().sum()) + return messyS, messyC 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(). + sales["product_name"] = sales["product_name"].str.strip().str.title() # TODO: Normalize customer_email with .str.lower().str.strip(). + sales["customer_email"] = sales["customer_email"].str.lower().str.strip() # TODO: Convert price to numeric with pd.to_numeric(errors="coerce"). + sales["price"] = pd.to_numeric(sales["price"], errors="coerce") # TODO: Parse date with pd.to_datetime(errors="coerce"). + sales["date"] = pd.to_datetime(sales["date"], errors="coerce") # TODO: Drop rows where product_name is missing. + sales = sales.dropna(subset=["product_name"]) # TODO: Drop rows where price is negative. + sales = sales[sales["price"] >= 0] # TODO: Drop rows where quantity is zero. + sales = sales[sales["quantity"] != 0] # TODO: Drop rows where date is NaT (invalid after parsing). + sales = sales.dropna(subset=["date"]) # TODO: Remove duplicate transactions: .drop_duplicates(subset="transaction_id", keep="first"). + sales = sales.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_std = sales.groupby("product_name")["price"].transform("std") + product_mean = sales.groupby("product_name")["price"].transform("mean") + sales["is_price_outlier"] = sales["price"] > (product_mean + 3 * product_std) + # since these are sales transactions, we want to keep outliers (they may be real sales of expensive items) + # but flag them for later just in case for future analysts who may want to filter them out. This way we preserve the data but keep it tracked. + return sales + + +load_and_explore(Path("./data")) diff --git a/src/ingest.py b/src/ingest.py index 01fe28f..2bc9d4a 100644 --- a/src/ingest.py +++ b/src/ingest.py @@ -1,11 +1,10 @@ """Task 1: Download inputs from Azure. Task 7: Upload outputs back to Azure.""" -import io + import logging from pathlib import Path - -import pandas as pd from azure.identity import DefaultAzureCredential from azure.storage.blob import BlobServiceClient +logger = logging.getLogger(__name__) ACCOUNT_URL = "https://sthyfstudentsdemo.blob.core.windows.net" SOURCE_CONTAINER = "week4-inputs" @@ -14,11 +13,31 @@ 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") + logger.info("Initializing Azure credentials...") + credential = DefaultAzureCredential() + service = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential) + container = service.get_container_client(SOURCE_CONTAINER) + + data_dir.mkdir(parents=True, exist_ok=True) + logger.info(f"Target directory verified at: {data_dir.resolve()}") + + for name in FILES: + logger.info(f"Attempting to download {name}...") + blob = container.get_blob_client(name) + + file_path = data_dir / name + with open(file_path, "wb") as f: + f.write(blob.download_blob().readall()) + + logger.info("Downloaded %s to %s", name, file_path) + logger.info(f"Successfully downloaded: {name}") + + +if __name__ == "__main__": + logger.info("Script started...") + target_directory = Path("./data") + download_inputs(target_directory) + logger.info("Script finished.") def upload_outputs(output_dir: Path, github_username: str) -> None: @@ -30,5 +49,4 @@ def upload_outputs(output_dir: Path, github_username: str) -> None: # 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") + # TODO: Log the container name and number of files uploaded. \ No newline at end of file diff --git a/src/report.py b/src/report.py index a002b7b..d92b153 100644 --- a/src/report.py +++ b/src/report.py @@ -1,31 +1,123 @@ """Tasks 5 and 6: Build report tables and write outputs.""" + 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") + # Add ISO week number + enriched["week"] = ( + enriched["date"] + .dt.isocalendar() + .week + .astype(int) + ) + + # Weekly revenue report + weekly_revenue = ( + enriched.groupby(["week", "region"]) + .agg( + total_revenue=("price", "sum"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) + + # Customer summary report + customer_summary = ( + enriched.groupby("customer_email") + .agg( + customer_name=("customer_name", "first"), + region=("region", "first"), + loyalty_tier=("loyalty_tier", "first"), + total_spent=("price", "sum"), + avg_order=("price", "mean"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) + + # Category performance report + category_performance = ( + enriched.groupby("category") + .agg( + total_revenue=("price", "sum"), + order_count=("transaction_id", "count"), + ) + .reset_index() + ) -def write_outputs(reports: dict[str, pd.DataFrame], output_dir: Path) -> None: + # Loyalty analysis report + loyalty_analysis = ( + enriched.groupby("loyalty_tier") + .agg( + avg_spent=("price", "mean"), + customer_count=("customer_email", "nunique"), + ) + .reset_index() + ) + + 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") + + # Create output folder + output_dir.mkdir(parents=True, exist_ok=True) + + # Write report tables + 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, + ) + + # Sort category performance table + category_sorted = reports["category_performance"].sort_values( + by="total_revenue", + ascending=False, + ) + + # Create sanity-check chart + category_sorted.plot( + kind="bar", + x="category", + y="total_revenue", + title="Revenue by category", + ) + + # Save chart + plt.savefig( + output_dir / "category_revenue.png", + bbox_inches="tight", + ) + + plt.close() + + logging.info("Reports written to %s", output_dir) \ No newline at end of file diff --git a/src/transform.py b/src/transform.py index 18f82e5..1a4dffc 100644 --- a/src/transform.py +++ b/src/transform.py @@ -1,13 +1,15 @@ """Task 4: Join customer data and add derived columns.""" -import logging - import pandas as pd 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(). + 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. + merged = pd.merge(sales, customers, on="customer_email", how="inner") # TODO: Add a vectorized boolean column is_high_value: True where price * quantity >= 150. + merged["is_high_value"] = merged["price"] * merged["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") + return merged