Mohamad Bader AA

AI_ASSIST.md

@@ -2,18 +2,25 @@
 
 ## The prompt I gave
 
-<!-- Paste the exact prompt you gave the LLM here. -->
-
+ 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
 
-<!-- Describe what you kept, what you modified, and what you threw away. -->
+<!-- i kept logger = logging.getlogger(__name__)
+and i threw away all the rest of the stuff around it because the response was far too long and llm listed all options which i dont need in this instance -->
 
 ## Did it work?
 
-<!-- Yes / partially / no — and what you learned from the interaction. -->
+<!-- Yes / i learned to use logger in a different file than original and how to link config  from main to the rest of the project -->

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 = "<noneeeed>"
 
 
 def run() -> None:

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"))

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/<filename>.
-    # 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.

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)

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