diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index 51fcbfd..983de9e 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -15,7 +15,7 @@ name: CI on: push: - branches: ["TODO-replace-with-main"] + branches: ["main"] pull_request: jobs: @@ -29,10 +29,20 @@ jobs: - name: Install dependencies run: pip install -r requirements.txt - name: Lint - run: echo "TODO implement this step" + run: ruff check src - name: Format - run: echo "TODO implement this step" + run: ruff format --check src - name: Test - run: echo "TODO implement this step" + run: pytest -q - name: Build image - run: echo "TODO implement this step" + run: docker build -t hannahwn-pipeline:${{ github.sha }} . + - name: Azure login + uses: azure/login@v2 + with: + creds: ${{ secrets.AZURE_CREDENTIALS }} + - name: ACR login + run: az acr login --name hyfregistry + - name: Push image + run: | + docker tag hannahwn-pipeline:${{ github.sha }} hyfregistry.azurecr.io/hannahwn-pipeline:${{ github.sha }} + docker push hyfregistry.azurecr.io/hannahwn-pipeline:${{ github.sha }} diff --git a/AI_ASSIST.md b/AI_ASSIST.md index 70463b0..4a14ed8 100644 --- a/AI_ASSIST.md +++ b/AI_ASSIST.md @@ -10,9 +10,22 @@ TODO: paste your prompt here. +how to solve mitigate this issue, please refer to the troubleshooting guidelines here at https://aka.ms/azsdk/python/identity/defaultazurecredential/troubleshoot. +=================================================================== warnings summary =================================================================== +tests/test_pipeline.py: 12 warnings + C:\Users\Gebruiker\c55-data-week-5\venv\Lib\site-packages\msal\token_cache.py:293: DeprecationWarning: Use list(search(...)) instead to explicitly geta list. + warnings.warn( + +-- Docs: https://docs.pytest.org/en/stable/how-to/capture-warnings.html +=============================================================== short test summary info ================================================================ +FAILED tests/test_pipeline.py::TestDownloadInputs::test_downloads_files - azure.core.exceptions.ClientAuthenticationError: DefaultAzureCredential failedto retrieve a token from the included credentials. +================================================= 1 failed, 10 passed, 12 warnings in 95.82s (0:01:35) ================================================= +(venv) + ## The code or suggestion it returned +the same Azure auth issue.Still the same Azure auth issue. Your token has expired and needs refreshing. Run this: ```python # TODO: paste the AI-generated code here @@ -23,3 +36,5 @@ TODO: paste your prompt here. TODO: describe your review and any changes you made. + +There is no change , i still experience problems with loggin in in azure diff --git a/Dockerfile b/Dockerfile index d665b11..fed3002 100644 --- a/Dockerfile +++ b/Dockerfile @@ -10,15 +10,18 @@ # Replace each TODO comment with the correct Dockerfile instruction. # TODO: set the base image -FROM TODO +FROM python:3.11-slim WORKDIR /app # TODO: copy requirements.txt (before source — this keeps the install layer cached) +COPY requirements.txt . # TODO: install dependencies +RUN pip install --no-cache-dir -r requirements.txt # TODO: copy source code +COPY src/ ./src/ # TODO: set the command that runs when the container starts -CMD ["TODO"] +CMD ["python", "-m", "src.pipeline"] diff --git a/assets/azureportalproblem.png b/assets/azureportalproblem.png new file mode 100644 index 0000000..0de4a15 Binary files /dev/null and b/assets/azureportalproblem.png differ diff --git a/data/messy_customers.csv b/data/messy_customers.csv new file mode 100644 index 0000000..31b2e1f --- /dev/null +++ b/data/messy_customers.csv @@ -0,0 +1,38 @@ +customer_email,customer_name,region,signup_date,loyalty_tier +alice@example.com,Alice van den Berg,NL,2023-11-15,Gold +Bob@Company.COM,Bob De Smet,BE,2023-12-01,Silver +charlie@work.org,Charlie Müller,DE,2024-01-10,Bronze +dave@email.com,,NL,2024-01-15,Silver +eve@startup.io,Eve Jansen,NL,2024-02-01,Gold +frank@corp.com,Frank Dubois,FR,2024-02-14,Bronze +Grace@University.EDU,Grace van Dijk,NL,2024-03-01,Silver +henry@business.com,Henry Peeters,BE,2024-03-10,Bronze +ivan@email.com,Ivan Schneider,DE,2024-03-15,Silver +jenny@work.org,Jenny Laurent,FR,2024-01-20,Gold +karl@startup.io,Karl Bakker,NL,2024-02-28,Bronze +lena@mail.nl,Lena de Vries,NL,2024-04-01,Silver +Marco@Business.DE,Marco Weber,DE,2024-04-10,Gold +nina@university.edu,,NL,2024-04-15,Bronze +oliver@corp.com,Oliver Martin,FR,2024-05-01,Silver +Paula@Startup.IO,Paula Visser,NL,2024-05-10,Bronze +quinn@work.org,Quinn Claes,BE,2024-05-15,Silver +rachel@email.com,Rachel Schmitt,DE,not_a_date,Gold +simon@company.com,Simon Leroy,FR,2024-06-01,Bronze +tina@mail.nl,Tina Meijer,NL,2024-06-05,Silver +uwe@business.de,Uwe Fischer,DE,2024-06-10,Gold +vera@mail.nl,Vera Willems,BE,2024-06-15,Bronze +wendy@startup.io,Wendy van Leeuwen,NL,2024-01-05,Silver +xander@corp.com,Xander Moreau,FR,2024-02-20,Gold +Yara@University.EDU,Yara Hendriks,NL,2024-03-25,Bronze +zach@email.com,Zach Bauer,DE,2024-04-05,Silver +orphan_customer1@mail.nl,Dirk Janssen,NL,2024-01-01,Gold +orphan_customer2@business.de,Petra Hofmann,DE,2024-02-15,Bronze +orphan_customer3@work.fr,Louis Petit,FR,2024-03-20,Silver +orphan_customer4@mail.be,,BE,invalid_date,Gold +sam@company.com,Sam de Groot,NL,2024-05-20,Silver +lisa@email.com,Lisa Maes,BE,2024-06-01,Bronze +tom@corp.com,Tom Bernard,FR,2024-04-18,Gold +emma@startup.io,Emma van Houten,NL,2024-03-12,Silver +jan@mail.nl,Jan de Boer,NL,2024-02-08,Bronze +sophie@business.de,Sophie Klein,DE,2024-05-25,Gold +orphan_customer5@university.edu,Maria Garcia,FR,2024-06-20,Silver diff --git a/requirements.txt b/requirements.txt index 42299cf..4e5c564 100644 --- a/requirements.txt +++ b/requirements.txt @@ -11,3 +11,11 @@ # ruff== # # Add your pinned dependencies below: + +pandas==2.2.2 +azure-identity==1.15.0 +azure-storage-blob==12.14.1 +python-dotenv==1.0.0 +ruff==0.0.241 +pytest==7.3.1 + diff --git a/src/clean.py b/src/clean.py new file mode 100644 index 0000000..368a163 --- /dev/null +++ b/src/clean.py @@ -0,0 +1,51 @@ +"""Tasks 2 and 3: Explore and clean the raw DataFrames.""" +import logging +from pathlib import Path + +import pandas as pd + + +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(). + sales_df = pd.read_csv(data_dir / "messy_sales.csv") + customers_df = pd.read_csv(data_dir / "messy_customers.csv") + # TODO: For each DataFrame call .info(), .describe(), .head(20), and .isna().sum(). + logging.info("Sales DataFrame info:\n%s", sales_df.info()) + logging.info("Sales DataFrame description:\n%s", sales_df.describe()) + logging.info("Sales DataFrame head:\n%s", sales_df.head(20)) + logging.info("Sales DataFrame null counts:\n%s", sales_df.isna().sum()) + logging.info("Customers DataFrame info:\n%s", customers_df.info()) + logging.info("Customers DataFrame description:\n%s", customers_df.describe()) + logging.info("Customers DataFrame head:\n%s", customers_df.head(20)) + logging.info("Customers DataFrame null counts:\n%s", customers_df.isna().sum()) + # TODO: Log what you discover (e.g. which columns have nulls, any suspicious values). + logging.info("Sales DataFrame has %d rows and %d columns", sales_df.shape[0], sales_df.shape[1]) + logging.info("Customers DataFrame has %d rows and %d columns", customers_df.shape[0], customers_df.shape[1]) + return sales_df, customers_df + + +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. + # For simplicity, we'll leave outlier prices as they are, but in a real scenario, we might want to investigate them further or apply business rules to handle them. + return sales + \ No newline at end of file diff --git a/src/ingest.py b/src/ingest.py new file mode 100644 index 0000000..b5de28d --- /dev/null +++ b/src/ingest.py @@ -0,0 +1,69 @@ +"""Task 1: Download inputs from Azure. Task 7: Upload outputs back to Azure.""" +import io +import logging +import os +from dotenv import load_dotenv +from pathlib import Path + +import pandas as pd +from azure.identity import DefaultAzureCredential +from azure.storage.blob import BlobServiceClient + +load_dotenv() + +ACCOUNT_URL = os.getenv("ACCOUNT_URL", "https://c55data.blob.core.windows.net") +SOURCE_CONTAINER = os.getenv("SOURCE_CONTAINER","week4-inputs") +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. + credential = DefaultAzureCredential() + blob_service_client = BlobServiceClient(account_url=ACCOUNT_URL, credential=credential) + # TODO: Get a container client for SOURCE_CONTAINER. + container_client = blob_service_client.get_container_client(SOURCE_CONTAINER) + # TODO: For each filename in FILES, download the blob and write it to data_dir/. + data_dir.mkdir(parents=True, exist_ok=True) + + for name in FILES: + blob = container_client.get_blob_client(name) + with open(f"{data_dir}/{name}", "wb") as f: + f.write(blob.download_blob().readall()) + + # TODO: Log a message for each downloaded file. + for filename in FILES: + 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. + account_url = ACCOUNT_URL + credential = DefaultAzureCredential() + blob_service_client = BlobServiceClient(account_url=account_url, credential=credential) + container_client = blob_service_client.get_container_client(container_name) + if not container_client.exists(): + container_client.create_container() + parquet_files = list(output_dir.glob("*.parquet")) + for parquet_file in parquet_files: + blob_client = container_client.get_blob_client(parquet_file.name) + with open(parquet_file, "rb") as f: + blob_client.upload_blob(f, overwrite=True) + logging.info("Uploaded %d files to container %s", len(parquet_files), container_name) + # Verify round-trip for customer_summary.parquet + local_file = output_dir / "customer_summary.parquet" + blob_client = container_client.get_blob_client(local_file.name) + downloaded_data = blob_client.download_blob().readall() + downloaded_df = pd.read_parquet(io.BytesIO(downloaded_data)) + local_df = pd.read_parquet(local_file) + assert len(downloaded_df) == len(local_df), "Row count mismatch after round-trip" + logging.info("Round-trip verification successful for customer_summary.parquet") + diff --git a/src/pipeline.py b/src/pipeline.py index 1cd17a4..a3c1675 100644 --- a/src/pipeline.py +++ b/src/pipeline.py @@ -9,7 +9,14 @@ """ import logging +import os from pathlib import Path +DATA_DIR = Path("data") + +from src.ingest import download_inputs, upload_outputs +from src.clean import load_and_explore, clean_sales +from src.transform import join_customers +from src.report import build_reports, write_outputs logging.basicConfig(level=logging.INFO, format="%(levelname)s %(message)s") logger = logging.getLogger(__name__) @@ -24,7 +31,12 @@ def get_config() -> dict: Raise RuntimeError with a clear message if a required variable is missing. """ - raise NotImplementedError("Task 5: read API_KEY and OUTPUT_DIR from the environment") + api_key = os.getenv("API_KEY") + if api_key is None: + raise RuntimeError("Missing required environment variable: API_KEY") + output_dir = os.getenv("OUTPUT_DIR", "output") + return {"api_key": api_key, "output_dir": output_dir} + def fetch_data(api_key: str) -> list[dict]: @@ -34,7 +46,17 @@ def fetch_data(api_key: str) -> list[dict]: Return a list of at least one dict representing a record. In a real pipeline you would call requests.get(...) here. """ - raise NotImplementedError("Task 1: return at least one sample record") + mock_record = { + "transaction_id": "12345", + "customer_email": "h@gmail.com", + "date": "2024-01-01", + "region": "North", + "category": "Widgets", + "quantity": 10, + "price": 9.99, + } + return [mock_record] + def save_results(records: list[dict], output_dir: Path) -> None: @@ -44,7 +66,11 @@ def save_results(records: list[dict], output_dir: Path) -> None: Create output_dir if it does not exist. Log the number of records written. """ - raise NotImplementedError("Task 1: write records to output_dir/results.txt") + output_dir.mkdir(parents=True, exist_ok=True) + with open(output_dir / "results.txt", "w") as f: + for record in records: + f.write(f"{record}\n") + logger.info(f"Written {len(records)} records to {output_dir}/results.txt") def run() -> None: @@ -55,6 +81,22 @@ def run() -> None: save_results(records, output_dir) logger.info("pipeline complete") + download_inputs(DATA_DIR) + sales_raw, customers_raw = load_and_explore(DATA_DIR) + + sales_clean = clean_sales(sales_raw) + enriched = join_customers(sales_clean, customers_raw) + + reports = build_reports(enriched) + write_outputs(reports, OUTPUT_DIR) + + upload_outputs(OUTPUT_DIR, GITHUB_USERNAME) + + logging.info("Pipeline complete.") + if __name__ == "__main__": run() + + + diff --git a/src/report.py b/src/report.py new file mode 100644 index 0000000..95e664b --- /dev/null +++ b/src/report.py @@ -0,0 +1,77 @@ +"""Tasks 5 and 6: Build report tables and write outputs.""" +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. + enriched["week"] = enriched["date"].dt.isocalendar().week + # TODO: Build weekly_revenue: group by week and region, columns week/region/total_revenue/order_count. + enriched["revenue"] = enriched["price"] * enriched["quantity"] + weekly_revenue = enriched.groupby(["week", "region"]).agg( + total_revenue=pd.NamedAgg(column="revenue", aggfunc="sum"), + order_count=pd.NamedAgg(column="transaction_id", aggfunc="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 = enriched.groupby("customer_email").agg( + customer_name=pd.NamedAgg(column="customer_name", aggfunc="first"), + region=pd.NamedAgg(column="region", aggfunc="first"), + loyalty_tier=pd.NamedAgg(column="loyalty_tier", aggfunc="first"), + total_spent=pd.NamedAgg(column="revenue", aggfunc="sum"), + avg_order=pd.NamedAgg(column="revenue", aggfunc="mean"), + order_count=pd.NamedAgg(column="transaction_id", aggfunc="count"), + ).reset_index() + # TODO: Build category_performance: group by category, columns category/total_revenue/order_count. + category_performance = enriched.groupby("category").agg( + total_revenue=pd.NamedAgg(column="revenue", aggfunc="sum"), + order_count=pd.NamedAgg(column="transaction_id", aggfunc="count"), + ).reset_index() + # TODO: Build loyalty_analysis: group by loyalty_tier, columns loyalty_tier/avg_spent/customer_count. + loyalty_analysis = enriched.groupby("loyalty_tier").agg( + avg_spent=pd.NamedAgg(column="revenue", aggfunc="mean"), + customer_count=pd.NamedAgg(column="customer_email", aggfunc="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. + reports["weekly_revenue"].to_csv(output_dir / "weekly_revenue.csv", index=False) + # TODO: Write reports["customer_summary"] to customer_summary.parquet with index=False. + reports["customer_summary"].to_parquet(output_dir / "customer_summary.parquet", index=False) + # TODO: Write reports["category_performance"] to category_performance.csv with index=False. + reports["category_performance"].to_csv(output_dir / "category_performance.csv", index=False) + # TODO: Sort category_performance by total_revenue descending. + reports["category_performance"] = reports["category_performance"].sort_values("total_revenue", ascending=False) + # 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. + import matplotlib + matplotlib.use("Agg") + import matplotlib.pyplot as plt + plt.figure(figsize=(10, 6)) + plt.bar(reports["category_performance"]["category"], reports["category_performance"]["total_revenue"]) + plt.xlabel("Category") + plt.ylabel("Total Revenue") + plt.title("Total Revenue by Category") + plt.xticks(rotation=45) + plt.tight_layout() + plt.savefig(output_dir / "category_revenue.png", bbox_inches="tight") + logging.info(f"Reports written to {output_dir}.") + \ No newline at end of file diff --git a/src/transform.py b/src/transform.py new file mode 100644 index 0000000..3c403a4 --- /dev/null +++ b/src/transform.py @@ -0,0 +1,18 @@ +"""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_df = pd.merge(sales, customers, on="customer_email", how="inner") + # TODO: Add a vectorized boolean column is_high_value: True where price * quantity >= 150. + merged_df["is_high_value"] = merged_df["price"] * merged_df["quantity"] >= 150 + # TODO: (Optional hands-on) Try a left join instead and inspect rows where customer_name is NaN. + return merged_df + diff --git a/tests/test_pipeline.py b/tests/test_pipeline.py index 73029e3..addf521 100644 --- a/tests/test_pipeline.py +++ b/tests/test_pipeline.py @@ -3,6 +3,7 @@ import pytest from src.pipeline import fetch_data, get_config, save_results +from src.ingest import download_inputs, upload_outputs class TestGetConfig: @@ -59,3 +60,12 @@ def test_file_contains_records(self, tmp_path): save_results([{"id": 1}, {"id": 2}], tmp_path) content = (tmp_path / "results.txt").read_text() assert len(content.strip().splitlines()) >= 2 + + +class TestDownloadInputs: + def test_downloads_files(self, tmp_path): + download_inputs(tmp_path) + for filename in ["messy_sales.csv", "messy_customers.csv"]: + assert (tmp_path / filename).exists() + + \ No newline at end of file