02 - ML Pipelines with CI/CD: GitHub Actions and Docker

  +------------------+     +------------------+     +------------------+
  |  DATA INGESTION  |---->|  PREPROCESSING   |---->|    TRAINING      |
  |                  |     |                  |     |                  |
  | - Pull DVC data  |     | - Data cleaning  |     | - Train model    |
  | - Validation     |     | - Feature eng.   |     | - Log metrics    |
  | - Schema check   |     | - Train/val/test  |     | - Log params     |
  |                  |     |   split          |     | - Save artifacts |
  +------------------+     +------------------+     +------------------+
         |                                                   |
         | (trigger: new data                                |
         |  or schedule)                                     v
         |                                          +------------------+
         |                                          |   EVALUATION     |
         |                                          |                  |
         |                                          | - Metrics        |
         |                                          | - Compare with   |
         |                                          |   production     |
         |                                          | - Gate: thresholds|
         |                                          +------------------+
         |                                                   |
         |                                    (if metrics > threshold)
         |                                                   v
  +------------------+     +------------------+     +------------------+
  |   MONITORING     |<----|   SMOKE TEST     |<----|   DEPLOYMENT     |
  |                  |     |                  |     |                  |
  | - Health check   |     | - Test endpoint  |     | - Push registry  |
  | - Drift detect.  |     | - Sample predict |     | - Stage/Prod     |
  | - Alert          |     | - Latency check  |     | - Rollback ready |
  | - Trigger retrain|     |                  |     |                  |
  +------------------+     +------------------+     +------------------+

# ============================================
# Stage 1: Builder - install dependencies
# ============================================
FROM python:3.11-slim AS builder

WORKDIR /build

# Install build tools required for native dependencies
RUN apt-get update && apt-get install -y --no-install-recommends \
    build-essential \
    gcc \
    g++ \
    && rm -rf /var/lib/apt/lists/*

# Copy and install dependencies into a virtual environment
COPY requirements.txt .
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
RUN pip install --no-cache-dir --upgrade pip && \
    pip install --no-cache-dir -r requirements.txt

# ============================================
# Stage 2: Training - runs the training job
# ============================================
FROM python:3.11-slim AS trainer

WORKDIR /app

# Copy virtual environment from builder stage
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Copy source code
COPY src/ ./src/
COPY config/ ./config/
COPY train.py .
COPY evaluate.py .

ENTRYPOINT ["python", "train.py"]

# ============================================
# Stage 3: Serving - production API
# ============================================
FROM python:3.11-slim AS serving

WORKDIR /app

# Non-root user for security
RUN useradd --create-home appuser

# Copy virtual environment from builder stage
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# Copy only the code needed for serving
COPY src/serving/ ./src/serving/
COPY src/preprocessing/ ./src/preprocessing/

# Health check
HEALTHCHECK --interval=30s --timeout=5s --retries=3 \
    CMD python -c "import urllib.request; urllib.request.urlopen('http://localhost:8000/health')"

USER appuser
EXPOSE 8000

ENTRYPOINT ["uvicorn", "src.serving.app:app", "--host", "0.0.0.0", "--port", "8000"]

# Data and models (managed by DVC, not Docker)
data/
models/
*.pkl
*.h5
*.pt

# Development environment
.venv/
__pycache__/
*.pyc
.pytest_cache/
.mypy_cache/

# Git and CI
.git/
.github/
.dvc/cache/

# IDE and editor
.vscode/
.idea/
*.swp

# Documentation
docs/
*.md
LICENSE

# Base image with CUDA runtime
FROM nvidia/cuda:12.1.1-runtime-ubuntu22.04 AS gpu-trainer

WORKDIR /app

RUN apt-get update && apt-get install -y --no-install-recommends \
    python3.11 \
    python3.11-venv \
    python3-pip \
    && rm -rf /var/lib/apt/lists/*

RUN python3.11 -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"

# PyTorch with CUDA dependencies
COPY requirements-gpu.txt .
RUN pip install --no-cache-dir -r requirements-gpu.txt

COPY src/ ./src/
COPY train.py .

# CUDA environment variables
ENV NVIDIA_VISIBLE_DEVICES=all
ENV NVIDIA_DRIVER_CAPABILITIES=compute,utility

ENTRYPOINT ["python", "train.py"]

# Build GPU image
docker build -f Dockerfile.gpu -t ml-trainer:gpu .

# Run with GPU access
docker run --gpus all \
    -v $(pwd)/data:/app/data \
    -v $(pwd)/models:/app/models \
    ml-trainer:gpu \
    --config config/training.yaml

name: ML Pipeline - Train, Evaluate, Deploy

on:
  # Trigger on push to main (code or config changes)
  push:
    branches: [main]
    paths:
      - 'src/**'
      - 'config/**'
      - 'requirements.txt'
      - 'train.py'
      - 'evaluate.py'

  # Scheduled trigger for periodic retraining
  schedule:
    - cron: '0 6 * * 1'  # Every Monday at 06:00 UTC

  # Manual trigger with parameters
  workflow_dispatch:
    inputs:
      force_deploy:
        description: 'Force deployment even if metrics do not improve'
        required: false
        default: 'false'
        type: choice
        options:
          - 'true'
          - 'false'
      training_config:
        description: 'Training configuration file'
        required: false
        default: 'config/training.yaml'

env:
  PYTHON_VERSION: '3.11'
  DOCKER_REGISTRY: ghcr.io
  IMAGE_NAME: ${{ github.repository }}/ml-model
  MLFLOW_TRACKING_URI: ${{ secrets.MLFLOW_TRACKING_URI }}

jobs:
  # ============================================
  # Job 1: Data Validation
  # ============================================
  data-validation:
    name: Validate Data Quality
    runs-on: ubuntu-latest
    outputs:
      data_valid: ${{ steps.validate.outputs.valid }}

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: 'pip'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Setup DVC
        uses: iterative/setup-dvc@v2

      - name: Pull data from DVC
        run: dvc pull
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: Validate data quality
        id: validate
        run: |
          python -m src.data.validate_data \
            --data-path data/raw/reviews.csv \
            --schema-path config/data_schema.yaml
          echo "valid=true" >> $GITHUB_OUTPUT

  # ============================================
  # Job 2: Model Training
  # ============================================
  training:
    name: Train Model
    needs: data-validation
    if: needs.data-validation.outputs.data_valid == 'true'
    runs-on: ubuntu-latest
    outputs:
      model_version: ${{ steps.train.outputs.model_version }}
      run_id: ${{ steps.train.outputs.run_id }}

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: 'pip'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Setup DVC and pull data
        run: |
          pip install dvc[s3]
          dvc pull
        env:
          AWS_ACCESS_KEY_ID: ${{ secrets.AWS_ACCESS_KEY_ID }}
          AWS_SECRET_ACCESS_KEY: ${{ secrets.AWS_SECRET_ACCESS_KEY }}

      - name: Train model
        id: train
        run: |
          python train.py \
            --config ${{ github.event.inputs.training_config || 'config/training.yaml' }} \
            --output-dir models/
          echo "model_version=$(cat models/version.txt)" >> $GITHUB_OUTPUT
          echo "run_id=$(cat models/run_id.txt)" >> $GITHUB_OUTPUT
        env:
          MLFLOW_TRACKING_URI: ${{ env.MLFLOW_TRACKING_URI }}
          MLFLOW_TRACKING_USERNAME: ${{ secrets.MLFLOW_USERNAME }}
          MLFLOW_TRACKING_PASSWORD: ${{ secrets.MLFLOW_PASSWORD }}

      - name: Upload model artifact
        uses: actions/upload-artifact@v4
        with:
          name: trained-model
          path: models/
          retention-days: 30

  # ============================================
  # Job 3: Model Evaluation (Gate)
  # ============================================
  evaluation:
    name: Evaluate Model
    needs: training
    runs-on: ubuntu-latest
    outputs:
      metrics_pass: ${{ steps.evaluate.outputs.metrics_pass }}
      accuracy: ${{ steps.evaluate.outputs.accuracy }}
      auc_roc: ${{ steps.evaluate.outputs.auc_roc }}

    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: ${{ env.PYTHON_VERSION }}
          cache: 'pip'

      - name: Install dependencies
        run: pip install -r requirements.txt

      - name: Download trained model
        uses: actions/download-artifact@v4
        with:
          name: trained-model
          path: models/

      - name: Run evaluation and metric gate
        id: evaluate
        run: |
          python evaluate.py \
            --model-path models/ \
            --data-path data/processed/test.csv \
            --output-path evaluation/metrics.json
          # Parse metrics and set gate
          python -c "
          import json, sys
          with open('evaluation/metrics.json') as f:
              m = json.load(f)
          passed = m['auc_roc'] >= 0.82 and m['accuracy'] >= 0.80
          print(f'AUC-ROC: {m[\"auc_roc\"]:.4f} | Accuracy: {m[\"accuracy\"]:.4f}')
          print(f'Gate: {\"PASS\" if passed else \"FAIL\"}')
          with open('$GITHUB_OUTPUT', 'a') as out:
              out.write(f'metrics_pass={str(passed).lower()}\n')
              out.write(f'accuracy={m[\"accuracy\"]}\n')
              out.write(f'auc_roc={m[\"auc_roc\"]}\n')
          sys.exit(0 if passed else 1)
          "

  # ============================================
  # Job 4: Build and Push Docker Image
  # ============================================
  build-push:
    name: Build and Push Image
    needs: evaluation
    if: needs.evaluation.outputs.metrics_pass == 'true'
    runs-on: ubuntu-latest
    permissions:
      contents: read
      packages: write
    outputs:
      image_tag: ${{ steps.meta.outputs.tags }}

    steps:
      - uses: actions/checkout@v4

      - name: Download model artifact
        uses: actions/download-artifact@v4
        with:
          name: trained-model
          path: models/

      - name: Log in to GitHub Container Registry
        uses: docker/login-action@v3
        with:
          registry: ${{ env.DOCKER_REGISTRY }}
          username: ${{ github.actor }}
          password: ${{ secrets.GITHUB_TOKEN }}

      - name: Extract Docker metadata
        id: meta
        uses: docker/metadata-action@v5
        with:
          images: ${{ env.DOCKER_REGISTRY }}/${{ env.IMAGE_NAME }}
          tags: |
            type=sha,prefix=sha-
            type=raw,value=latest,enable=${{ github.ref == 'refs/heads/main' }}

      - name: Build and push Docker image
        uses: docker/build-push-action@v5
        with:
          context: .
          file: Dockerfile
          target: serving
          push: true
          tags: ${{ steps.meta.outputs.tags }}
          labels: ${{ steps.meta.outputs.labels }}
          cache-from: type=gha
          cache-to: type=gha,mode=max

  # ============================================
  # Job 5: Deploy to Staging
  # ============================================
  deploy-staging:
    name: Deploy to Staging
    needs: build-push
    runs-on: ubuntu-latest
    environment: staging

    steps:
      - uses: actions/checkout@v4

      - name: Deploy to staging environment
        run: |
          echo "Deploying ${{ needs.build-push.outputs.image_tag }} to staging"
          # kubectl set image or docker service update
          kubectl set image deployment/ml-model-staging \
            ml-api=${{ needs.build-push.outputs.image_tag }} \
            --namespace staging
          kubectl rollout status deployment/ml-model-staging --namespace staging

      - name: Run smoke tests on staging
        run: |
          python tests/smoke/test_api.py \
            --base-url ${{ secrets.STAGING_URL }}

sentiment-classifier/
  src/
    data/
      __init__.py
      preprocessing.py     # Text cleaning and transformation
      validate_data.py     # Schema and quality validation
    models/
      __init__.py
      trainer.py           # Classifier training
    serving/
      __init__.py
      app.py               # FastAPI application
      schemas.py           # Pydantic schemas
    monitoring/
      __init__.py
      health.py            # Health checks
  tests/
    test_preprocessing.py  # Unit test preprocessing
    test_trainer.py        # Unit test training
    test_api.py            # Integration test API
  config/
    training.yaml          # Training configuration
    thresholds.yaml        # Metric thresholds
    data_schema.yaml       # Data validation schema
  train.py                 # Training entrypoint
  evaluate.py              # Evaluation entrypoint
  Dockerfile               # Multi-stage build
  requirements.txt         # Python dependencies
  .github/
    workflows/
      ml-pipeline.yml      # CI/CD pipeline
  .dvc/                    # DVC configuration
  dvc.yaml                 # DVC pipeline
  dvc.lock                 # DVC lock file

"""Main training script for the sentiment classifier."""
import argparse
import yaml
import mlflow
import mlflow.sklearn
from pathlib import Path
from datetime import datetime

from src.data.preprocessing import load_and_preprocess_data, split_dataset
from src.models.trainer import create_pipeline, train_model


def parse_args():
    """Parse command-line arguments."""
    parser = argparse.ArgumentParser(
        description="Train sentiment classifier"
    )
    parser.add_argument(
        "--config",
        type=str,
        default="config/training.yaml",
        help="Path to the configuration file"
    )
    parser.add_argument(
        "--output-dir",
        type=str,
        default="models/",
        help="Directory where the model is saved"
    )
    parser.add_argument(
        "--experiment-name",
        type=str,
        default="sentiment-classifier",
        help="Name of the MLflow experiment"
    )
    return parser.parse_args()


def main():
    """Runs the full training pipeline."""
    args = parse_args()

    # 1. Load configuration
    with open(args.config) as f:
        config = yaml.safe_load(f)

    # 2. Setup MLflow
    mlflow.set_experiment(args.experiment_name)

    with mlflow.start_run(run_name=f"train-{datetime.now().strftime('%Y%m%d-%H%M%S')}") as run:
        # 3. Log parameters
        mlflow.log_params(config["model"])
        mlflow.log_param("data_path", config["data"]["train_path"])
        mlflow.log_param("test_size", config["data"]["test_size"])

        # 4. Load and preprocess data
        print("[1/5] Loading and preprocessing data...")
        X, y = load_and_preprocess_data(config["data"]["train_path"])

        # 5. Split dataset
        print("[2/5] Train/validation split...")
        X_train, X_val, y_train, y_val = split_dataset(
            X, y,
            test_size=config["data"]["test_size"],
            random_state=config["data"]["random_state"]
        )
        mlflow.log_param("train_size", len(X_train))
        mlflow.log_param("val_size", len(X_val))

        # 6. Create preprocessing + model pipeline
        print("[3/5] Creating ML pipeline...")
        pipeline = create_pipeline(config["model"])

        # 7. Training
        print("[4/5] Training in progress...")
        trained_pipeline = train_model(pipeline, X_train, y_train)

        # 8. Evaluation on validation set
        print("[5/5] Evaluation...")
        from src.models.trainer import evaluate_model
        metrics = evaluate_model(trained_pipeline, X_val, y_val)

        # 9. Log metrics to MLflow
        mlflow.log_metrics(metrics)

        # 10. Save model
        output_dir = Path(args.output_dir)
        output_dir.mkdir(parents=True, exist_ok=True)

        import joblib
        model_path = output_dir / "model.pkl"
        joblib.dump(trained_pipeline, model_path)

        # Log model to MLflow with signature
        from mlflow.models.signature import infer_signature
        signature = infer_signature(X_train[:5], trained_pipeline.predict(X_train[:5]))
        mlflow.sklearn.log_model(
            trained_pipeline,
            "model",
            signature=signature,
            registered_model_name="sentiment-classifier"
        )

        # Save version and run_id for CI/CD
        (output_dir / "version.txt").write_text(run.info.run_id[:8])
        (output_dir / "run_id.txt").write_text(run.info.run_id)

        print(f"\nTraining completed!")
        print(f"  Run ID: {run.info.run_id}")
        for name, value in metrics.items():
            print(f"  {name}: {value:.4f}")


if __name__ == "__main__":
    main()

"""Module for preprocessing text data."""
import re
import pandas as pd
from typing import Tuple
from sklearn.model_selection import train_test_split


def clean_text(text: str) -> str:
    """Cleans a single text by removing HTML, special characters and extra spaces."""
    if not isinstance(text, str):
        return ""
    # Remove HTML tags
    text = re.sub(r'<[^>]+>', '', text)
    # Remove URLs
    text = re.sub(r'http\S+|www\.\S+', '', text)
    # Remove special characters (keep letters, numbers, spaces)
    text = re.sub(r'[^a-zA-Z0-9\s]', ' ', text)
    # Normalize spaces
    text = re.sub(r'\s+', ' ', text).strip()
    return text.lower()


def load_and_preprocess_data(data_path: str) -> Tuple[pd.Series, pd.Series]:
    """Loads and preprocesses the review dataset."""
    df = pd.read_csv(data_path)

    # Validate required columns
    required_cols = ["review_text", "sentiment"]
    missing = [c for c in required_cols if c not in df.columns]
    if missing:
        raise ValueError(f"Missing columns in dataset: {missing}")

    # Remove rows with missing values
    df = df.dropna(subset=required_cols)

    # Clean texts
    df["clean_text"] = df["review_text"].apply(clean_text)

    # Remove empty texts after cleaning
    df = df[df["clean_text"].str.len() > 0]

    return df["clean_text"], df["sentiment"]


def split_dataset(
    X: pd.Series,
    y: pd.Series,
    test_size: float = 0.2,
    random_state: int = 42
) -> Tuple[pd.Series, pd.Series, pd.Series, pd.Series]:
    """Stratified split of the dataset."""
    return train_test_split(
        X, y,
        test_size=test_size,
        random_state=random_state,
        stratify=y
    )

"""Module for creating, training and evaluating the model."""
from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import (
    accuracy_score, f1_score, precision_score,
    recall_score, roc_auc_score
)
from typing import Dict, Any
import pandas as pd


MODELS = {
    "logistic_regression": LogisticRegression,
    "random_forest": RandomForestClassifier,
}


def create_pipeline(model_config: Dict[str, Any]) -> Pipeline:
    """Creates a sklearn pipeline with TF-IDF + classifier."""
    algorithm = model_config.get("algorithm", "logistic_regression")
    model_class = MODELS.get(algorithm)

    if model_class is None:
        raise ValueError(
            f"Unsupported algorithm: {algorithm}. "
            f"Supported: {list(MODELS.keys())}"
        )

    # Model-specific parameters
    model_params = {
        k: v for k, v in model_config.items()
        if k not in ("algorithm", "tfidf")
    }

    # TF-IDF parameters
    tfidf_params = model_config.get("tfidf", {})

    return Pipeline([
        ("tfidf", TfidfVectorizer(
            max_features=tfidf_params.get("max_features", 10000),
            ngram_range=tuple(tfidf_params.get("ngram_range", [1, 2])),
            min_df=tfidf_params.get("min_df", 2),
            max_df=tfidf_params.get("max_df", 0.95),
        )),
        ("classifier", model_class(**model_params)),
    ])


def train_model(
    pipeline: Pipeline,
    X_train: pd.Series,
    y_train: pd.Series
) -> Pipeline:
    """Trains the pipeline on the training set."""
    pipeline.fit(X_train, y_train)
    return pipeline


def evaluate_model(
    pipeline: Pipeline,
    X_test: pd.Series,
    y_test: pd.Series
) -> Dict[str, float]:
    """Evaluates the model and returns all metrics."""
    y_pred = pipeline.predict(X_test)

    metrics = {
        "accuracy": accuracy_score(y_test, y_pred),
        "f1_score": f1_score(y_test, y_pred, average="weighted"),
        "precision": precision_score(y_test, y_pred, average="weighted"),
        "recall": recall_score(y_test, y_pred, average="weighted"),
    }

    # AUC-ROC only for binary classification
    if len(set(y_test)) == 2:
        y_proba = pipeline.predict_proba(X_test)[:, 1]
        metrics["auc_roc"] = roc_auc_score(y_test, y_proba)

    return metrics

"""Model evaluation script with threshold comparison."""
import argparse
import json
import yaml
import joblib
import pandas as pd
from pathlib import Path
from src.data.preprocessing import clean_text
from src.models.trainer import evaluate_model


def parse_args():
    parser = argparse.ArgumentParser(description="Evaluate trained model")
    parser.add_argument("--model-path", required=True, help="Path to the .pkl model")
    parser.add_argument("--test-data", required=True, help="Path to the test data")
    parser.add_argument("--thresholds", required=True, help="Path to the YAML thresholds file")
    parser.add_argument("--output", required=True, help="Path for the JSON report")
    return parser.parse_args()


def main():
    args = parse_args()

    # 1. Load model and data
    pipeline = joblib.load(args.model_path)
    df = pd.read_csv(args.test_data)
    df["clean_text"] = df["review_text"].apply(clean_text)
    X_test = df["clean_text"]
    y_test = df["sentiment"]

    # 2. Evaluate
    metrics = evaluate_model(pipeline, X_test, y_test)

    # 3. Compare with thresholds
    with open(args.thresholds) as f:
        thresholds = yaml.safe_load(f)["thresholds"]

    all_pass = True
    results = {}
    for metric_name, threshold_value in thresholds.items():
        actual = metrics.get(metric_name, 0.0)
        passed = actual >= threshold_value
        if not passed:
            all_pass = False
        results[metric_name] = {
            "value": actual,
            "threshold": threshold_value,
            "pass": passed
        }

    # 4. Generate report
    report = {
        **metrics,
        "thresholds": thresholds,
        "details": results,
        "pass": all_pass
    }

    output_path = Path(args.output)
    output_path.parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        json.dump(report, f, indent=2)

    # 5. Print results
    print("\n=== Evaluation Report ===")
    for name, detail in results.items():
        status = "PASS" if detail["pass"] else "FAIL"
        print(f"  {name}: {detail['value']:.4f} (threshold: {detail['threshold']}) [{status}]")
    print(f"\nOverall: {'PASS' if all_pass else 'FAIL'}")

    # Non-zero exit code if the metrics do not pass
    if not all_pass:
        print("\nWARNING: Metrics do not reach the minimum thresholds!")
        # We do not use exit(1) because the workflow reads the output


if __name__ == "__main__":
    main()

# Training pipeline configuration
data:
  train_path: "data/raw/reviews.csv"
  test_size: 0.2
  random_state: 42

model:
  algorithm: "logistic_regression"
  max_iter: 1000
  C: 1.0
  random_state: 42
  tfidf:
    max_features: 15000
    ngram_range: [1, 2]
    min_df: 3
    max_df: 0.9

mlflow:
  experiment_name: "sentiment-classifier"
  registered_model_name: "sentiment-classifier"

# Minimum thresholds to approve the deployment
thresholds:
  accuracy: 0.85
  f1_score: 0.83
  precision: 0.80
  recall: 0.80
  auc_roc: 0.90

# Comparison with the production model
comparison:
  # The new model must improve by at least 0.5%
  min_improvement: 0.005
  # Metrics on which the improvement is required
  compare_metrics:
    - f1_score
    - auc_roc

# Core ML
scikit-learn==1.4.0
pandas==2.2.0
numpy==1.26.3

# NLP preprocessing
nltk==3.8.1

# Experiment tracking
mlflow==2.10.0

# Model serving
fastapi==0.109.0
uvicorn==0.27.0
pydantic==2.5.3

# Data validation
pandera==0.18.0
great-expectations==0.18.8

# Data versioning
dvc[s3]==3.42.0

# Configuration
pyyaml==6.0.1
python-dotenv==1.0.1

# Serialization
joblib==1.3.2

# Testing
pytest==7.4.4
httpx==0.26.0

# validate_data.py
# Data quality checks using Great Expectations
# Executed as first step in the CI/CD pipeline

import json
import sys
import logging
from pathlib import Path
import pandas as pd
import great_expectations as gx
from great_expectations.core.batch import BatchRequest

logger = logging.getLogger(__name__)


def validate_training_data(
    data_path: str,
    output_path: str = "data/raw/validation_summary.json"
) -> bool:
    """
    Validates training data against a set of predefined expectations.

    Returns:
        True if all expectations pass, False otherwise.
    """
    df = pd.read_csv(data_path)
    context = gx.get_context()

    # Create a pandas datasource
    datasource = context.sources.add_pandas("training_data")
    data_asset = datasource.add_dataframe_asset("reviews")
    batch_request = data_asset.build_batch_request(dataframe=df)

    # Define expectations
    suite = context.add_expectation_suite("training_data_suite")

    validator = context.get_validator(
        batch_request=batch_request,
        expectation_suite=suite
    )

    # ---- Core expectations ----
    # 1. Minimum number of rows
    validator.expect_table_row_count_to_be_between(min_value=10_000)

    # 2. Required columns must exist
    for col in ["text", "label", "timestamp"]:
        validator.expect_column_to_exist(col)

    # 3. No nulls in critical columns
    validator.expect_column_values_to_not_be_null("text")
    validator.expect_column_values_to_not_be_null("label")

    # 4. Label values within expected range (binary classification)
    validator.expect_column_values_to_be_in_set("label", [0, 1])

    # 5. Class balance check: neither class exceeds 90%
    validator.expect_column_mean_to_be_between("label", min_value=0.1, max_value=0.9)

    # 6. Text column: reasonable length
    validator.expect_column_value_lengths_to_be_between(
        "text", min_value=5, max_value=5000
    )

    # 7. No duplicate rows
    validator.expect_compound_columns_to_be_unique(["text"])

    # ---- Run validation ----
    results = validator.validate()

    summary = {
        "all_checks_passed": results.success,
        "total_checks": len(results.results),
        "passed_checks": sum(1 for r in results.results if r.success),
        "failed_checks": sum(1 for r in results.results if not r.success),
        "row_count": int(df.shape[0]),
        "column_count": int(df.shape[1]),
        "failures": [
            {
                "expectation": str(r.expectation_config.expectation_type),
                "column": r.expectation_config.kwargs.get("column", "N/A")
            }
            for r in results.results if not r.success
        ]
    }

    Path(output_path).parent.mkdir(parents=True, exist_ok=True)
    with open(output_path, "w") as f:
        json.dump(summary, f, indent=2)

    if not results.success:
        logger.error(f"Data validation FAILED: {summary['failed_checks']} checks failed")
        for failure in summary["failures"]:
            logger.error(f"  - {failure['expectation']} on column '{failure['column']}'")
    else:
        logger.info(f"Data validation PASSED: {summary['passed_checks']}/{summary['total_checks']} checks")

    return results.success


if __name__ == "__main__":
    import argparse
    parser = argparse.ArgumentParser()
    parser.add_argument("--data-path", required=True)
    parser.add_argument("--output-path", default="data/raw/validation_summary.json")
    args = parser.parse_args()

    passed = validate_training_data(args.data_path, args.output_path)
    sys.exit(0 if passed else 1)

stages:
  prepare:
    cmd: python -m src.data.preprocessing --config config/training.yaml
    deps:
      - src/data/preprocessing.py
      - config/training.yaml
      - data/raw/reviews.csv
    outs:
      - data/processed/train.csv
      - data/processed/test.csv

  train:
    cmd: python train.py --config config/training.yaml
    deps:
      - train.py
      - src/models/trainer.py
      - data/processed/train.csv
      - config/training.yaml
    outs:
      - models/model.pkl
    metrics:
      - metrics/train_metrics.json:
          cache: false

  evaluate:
    cmd: >-
      python evaluate.py
      --model-path models/model.pkl
      --test-data data/processed/test.csv
      --thresholds config/thresholds.yaml
      --output metrics/eval_metrics.json
    deps:
      - evaluate.py
      - models/model.pkl
      - data/processed/test.csv
      - config/thresholds.yaml
    metrics:
      - metrics/eval_metrics.json:
          cache: false

# Initialize DVC
dvc init

# Configure remote storage (S3 in this example)
dvc remote add -d myremote s3://my-ml-data-bucket/sentiment-classifier
dvc remote modify myremote region eu-west-1

# Track the dataset
dvc add data/raw/reviews.csv

# Commit the DVC files to Git
git add data/raw/reviews.csv.dvc data/raw/.gitignore dvc.yaml dvc.lock
git commit -m "feat: add DVC tracking for training data"

# Push the data to S3
dvc push

"""Script for promoting the model in the MLflow registry."""
import mlflow
from mlflow.tracking import MlflowClient


def promote_model(
    model_name: str,
    run_id: str,
    target_stage: str = "Production"
) -> None:
    """Promotes a model to Production in the registry."""
    client = MlflowClient()

    # Find the model version associated with the run
    model_versions = client.search_model_versions(
        f"name='{model_name}'"
    )

    target_version = None
    for mv in model_versions:
        if mv.run_id == run_id:
            target_version = mv.version
            break

    if target_version is None:
        raise ValueError(
            f"No model found for run_id={run_id}"
        )

    # Archive the model currently in Production
    for mv in model_versions:
        if mv.current_stage == "Production":
            client.transition_model_version_stage(
                name=model_name,
                version=mv.version,
                stage="Archived",
                archive_existing_versions=False
            )
            print(f"Archived model v{mv.version} (previous Production)")

    # Promote the new model
    client.transition_model_version_stage(
        name=model_name,
        version=target_version,
        stage=target_stage
    )
    print(f"Promoted model v{target_version} to {target_stage}")


def load_production_model(model_name: str):
    """Loads the model currently in Production."""
    model_uri = f"models:/{model_name}/Production"
    return mlflow.sklearn.load_model(model_uri)

import mlflow
from mlflow.models.signature import ModelSignature
from mlflow.types.schema import Schema, ColSpec

# Define the explicit signature
input_schema = Schema([
    ColSpec("string", "text")
])
output_schema = Schema([
    ColSpec("long", "prediction")
])
signature = ModelSignature(
    inputs=input_schema,
    outputs=output_schema
)

# Example input for documentation
input_example = {
    "text": "This product is excellent, highly recommended!"
}

# Register the model with signature and example
mlflow.sklearn.log_model(
    sk_model=trained_pipeline,
    artifact_path="model",
    signature=signature,
    input_example=input_example,
    registered_model_name="sentiment-classifier"
)

"""Unit tests for the preprocessing module."""
import pytest
import pandas as pd
from src.data.preprocessing import clean_text, load_and_preprocess_data


class TestCleanText:
    """Tests for the clean_text function."""

    def test_removes_html_tags(self):
        assert clean_text("<p>Hello</p>") == "hello"

    def test_removes_urls(self):
        assert clean_text("Visit http://example.com for info") == "visit for info"

    def test_removes_special_characters(self):
        assert clean_text("Hello!!! World???") == "hello world"

    def test_normalizes_whitespace(self):
        assert clean_text("Hello     World") == "hello world"

    def test_lowercases(self):
        assert clean_text("HELLO WORLD") == "hello world"

    def test_empty_string(self):
        assert clean_text("") == ""

    def test_none_input(self):
        assert clean_text(None) == ""

    def test_numeric_preserved(self):
        assert clean_text("Rating 5 out of 10") == "rating 5 out of 10"


class TestLoadAndPreprocess:
    """Tests for loading and preprocessing the data."""

    def test_missing_columns_raises(self, tmp_path):
        """Checks that missing columns raise an error."""
        df = pd.DataFrame({"wrong_col": ["text"]})
        csv_path = tmp_path / "test.csv"
        df.to_csv(csv_path, index=False)

        with pytest.raises(ValueError, match="Missing columns"):
            load_and_preprocess_data(str(csv_path))

    def test_drops_na_rows(self, tmp_path):
        """Checks that rows with NA values are removed."""
        df = pd.DataFrame({
            "review_text": ["Good product", None, "Bad product"],
            "sentiment": [1, 0, 0]
        })
        csv_path = tmp_path / "test.csv"
        df.to_csv(csv_path, index=False)

        X, y = load_and_preprocess_data(str(csv_path))
        assert len(X) == 2

    def test_output_types(self, tmp_path):
        """Checks the output types."""
        df = pd.DataFrame({
            "review_text": ["Great product", "Terrible service"],
            "sentiment": [1, 0]
        })
        csv_path = tmp_path / "test.csv"
        df.to_csv(csv_path, index=False)

        X, y = load_and_preprocess_data(str(csv_path))
        assert isinstance(X, pd.Series)
        assert isinstance(y, pd.Series)

"""Integration tests for training and evaluating the model."""
import pytest
import pandas as pd
from src.models.trainer import create_pipeline, train_model, evaluate_model


@pytest.fixture
def sample_data():
    """Creates a synthetic test dataset."""
    texts = [
        "amazing product love it", "terrible waste of money",
        "great quality recommended", "horrible experience never again",
        "excellent value for price", "poor quality disappointed",
        "best purchase ever made", "worst product i bought",
        "fantastic results happy", "awful terrible regret buying",
    ] * 10  # Repeated to have enough data

    sentiments = [1, 0, 1, 0, 1, 0, 1, 0, 1, 0] * 10
    return pd.Series(texts), pd.Series(sentiments)


class TestPipeline:
    """Tests for the training pipeline."""

    def test_create_pipeline_logistic(self):
        """Checks pipeline creation with LogisticRegression."""
        config = {"algorithm": "logistic_regression", "max_iter": 100}
        pipeline = create_pipeline(config)
        assert len(pipeline.steps) == 2

    def test_create_pipeline_invalid_algorithm(self):
        """Checks the error raised for an unsupported algorithm."""
        with pytest.raises(ValueError, match="Unsupported"):
            create_pipeline({"algorithm": "invalid_algo"})

    def test_train_and_evaluate(self, sample_data):
        """End-to-end test: training + evaluation."""
        X, y = sample_data
        config = {
            "algorithm": "logistic_regression",
            "max_iter": 200,
            "random_state": 42
        }

        pipeline = create_pipeline(config)
        trained = train_model(pipeline, X[:80], y[:80])
        metrics = evaluate_model(trained, X[80:], y[80:])

        assert "accuracy" in metrics
        assert "f1_score" in metrics
        assert 0.0 <= metrics["accuracy"] <= 1.0
        assert 0.0 <= metrics["f1_score"] <= 1.0

    def test_model_deterministic(self, sample_data):
        """Checks that training is deterministic with a fixed seed."""
        X, y = sample_data
        config = {
            "algorithm": "logistic_regression",
            "max_iter": 200,
            "random_state": 42
        }

        p1 = train_model(create_pipeline(config), X[:80], y[:80])
        p2 = train_model(create_pipeline(config), X[:80], y[:80])

        m1 = evaluate_model(p1, X[80:], y[80:])
        m2 = evaluate_model(p2, X[80:], y[80:])

        assert m1["accuracy"] == m2["accuracy"]

# test_api.py
# Smoke tests run immediately after deployment
# Verify the API is working correctly before traffic routing

import json
import sys
import time
import argparse
import requests
import logging

logger = logging.getLogger(__name__)


def run_smoke_tests(base_url: str, timeout: int = 30) -> bool:
    """
    Runs smoke tests against the deployed API.

    Args:
        base_url: Base URL of the deployed API (e.g., https://api.example.com)
        timeout: Max wait time in seconds for health check

    Returns:
        True if all tests pass, False otherwise
    """
    results = {}

    # ---- Test 1: Health check ----
    print("Test 1: Health check...")
    deadline = time.time() + timeout
    while time.time() < deadline:
        try:
            resp = requests.get(f"{base_url}/health", timeout=5)
            if resp.status_code == 200:
                health = resp.json()
                assert health.get("status") == "healthy", f"Unexpected health: {health}"
                results["health_check"] = True
                print(f"  PASS - model_version: {health.get('model_version', 'unknown')}")
                break
        except Exception:
            time.sleep(2)
    else:
        results["health_check"] = False
        print("  FAIL - health check timed out")

    # ---- Test 2: Single prediction ----
    print("Test 2: Single prediction...")
    try:
        payload = {
            "text": "This product is absolutely fantastic, I love it!",
            "max_length": 128
        }
        start = time.time()
        resp = requests.post(f"{base_url}/predict", json=payload, timeout=10)
        latency_ms = (time.time() - start) * 1000

        assert resp.status_code == 200, f"HTTP {resp.status_code}: {resp.text}"
        result = resp.json()
        assert "prediction" in result, "Missing 'prediction' field"
        assert "confidence" in result, "Missing 'confidence' field"
        assert 0.0 <= result["confidence"] <= 1.0, "Confidence out of range [0,1]"
        assert latency_ms < 500, f"Latency too high: {latency_ms:.0f}ms (max 500ms)"

        results["single_prediction"] = True
        print(f"  PASS - prediction: {result['prediction']}, latency: {latency_ms:.0f}ms")
    except Exception as e:
        results["single_prediction"] = False
        print(f"  FAIL - {e}")

    # ---- Test 3: Batch prediction ----
    print("Test 3: Batch prediction...")
    try:
        batch_payload = {
            "texts": [
                "Excellent quality and fast shipping!",
                "Terrible experience, never buying again.",
                "Average product, nothing special."
            ]
        }
        resp = requests.post(f"{base_url}/predict/batch", json=batch_payload, timeout=15)
        assert resp.status_code == 200
        batch_result = resp.json()
        assert len(batch_result["predictions"]) == 3
        for pred in batch_result["predictions"]:
            assert 0.0 <= pred["confidence"] <= 1.0

        results["batch_prediction"] = True
        print(f"  PASS - {len(batch_result['predictions'])} predictions returned")
    except Exception as e:
        results["batch_prediction"] = False
        print(f"  FAIL - {e}")

    # ---- Test 4: Edge cases ----
    print("Test 4: Edge cases...")
    try:
        # Empty text should return validation error, not 500
        resp = requests.post(f"{base_url}/predict", json={"text": ""}, timeout=5)
        assert resp.status_code == 422, f"Expected 422 for empty text, got {resp.status_code}"

        # Very long text - should not crash
        long_text = "word " * 2000
        resp = requests.post(f"{base_url}/predict", json={"text": long_text}, timeout=10)
        assert resp.status_code in [200, 422], f"Unexpected status: {resp.status_code}"

        results["edge_cases"] = True
        print("  PASS")
    except Exception as e:
        results["edge_cases"] = False
        print(f"  FAIL - {e}")

    # ---- Summary ----
    all_passed = all(results.values())
    passed = sum(results.values())
    total = len(results)
    print(f"\nSmoke test results: {passed}/{total} passed")
    if not all_passed:
        failed = [k for k, v in results.items() if not v]
        print(f"Failed tests: {', '.join(failed)}")

    return all_passed


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--base-url", required=True)
    parser.add_argument("--timeout", type=int, default=30)
    args = parser.parse_args()

    passed = run_smoke_tests(args.base_url, args.timeout)
    sys.exit(0 if passed else 1)

"""Smoke tests for the FastAPI serving API."""
import pytest
from httpx import AsyncClient, ASGITransport
from src.serving.app import app


@pytest.fixture
def client():
    """HTTP client for testing the API."""
    transport = ASGITransport(app=app)
    return AsyncClient(transport=transport, base_url="http://test")


@pytest.mark.asyncio
async def test_health_endpoint(client):
    """Checks that the /health endpoint returns 200."""
    response = await client.get("/health")
    assert response.status_code == 200
    data = response.json()
    assert data["status"] == "healthy"


@pytest.mark.asyncio
async def test_predict_positive(client):
    """Checks the prediction for a positive text."""
    response = await client.post(
        "/predict",
        json={"text": "This product is amazing, I love it!"}
    )
    assert response.status_code == 200
    data = response.json()
    assert "prediction" in data
    assert "confidence" in data
    assert data["confidence"] > 0.0


@pytest.mark.asyncio
async def test_predict_empty_text(client):
    """Checks the error returned for an empty text."""
    response = await client.post(
        "/predict",
        json={"text": ""}
    )
    assert response.status_code == 422


@pytest.mark.asyncio
async def test_predict_batch(client):
    """Checks batch prediction."""
    response = await client.post(
        "/predict/batch",
        json={"texts": [
            "Great product",
            "Terrible experience"
        ]}
    )
    assert response.status_code == 200
    data = response.json()
    assert len(data["predictions"]) == 2

  # Add this job before training in the workflow
  tests:
    name: Run Tests
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4

      - name: Setup Python
        uses: actions/setup-python@v5
        with:
          python-version: '3.11'
          cache: 'pip'

      - name: Install dependencies
        run: |
          pip install -r requirements.txt
          pip install pytest-asyncio pytest-cov

      - name: Run unit tests
        run: pytest tests/ -v --cov=src --cov-report=xml

      - name: Upload coverage
        uses: codecov/codecov-action@v4
        with:
          file: coverage.xml

"""Serving API for the sentiment classifier."""
import os
import time
import joblib
from pathlib import Path
from contextlib import asynccontextmanager
from fastapi import FastAPI, HTTPException
from src.serving.schemas import (
    PredictionRequest, PredictionResponse,
    BatchPredictionRequest, BatchPredictionResponse,
    HealthResponse
)
from src.data.preprocessing import clean_text


# Global variables for the model
model_pipeline = None
model_version = None


@asynccontextmanager
async def lifespan(app: FastAPI):
    """Loads the model when the application starts."""
    global model_pipeline, model_version

    model_path = os.getenv("MODEL_PATH", "models/model.pkl")
    if not Path(model_path).exists():
        raise RuntimeError(f"Model not found: {model_path}")

    model_pipeline = joblib.load(model_path)
    model_version = os.getenv("MODEL_VERSION", "unknown")
    print(f"Model loaded: v{model_version}")

    yield

    model_pipeline = None


app = FastAPI(
    title="Sentiment Classifier API",
    version="1.0.0",
    lifespan=lifespan
)


@app.get("/health", response_model=HealthResponse)
async def health_check():
    """Health check of the API."""
    return HealthResponse(
        status="healthy" if model_pipeline is not None else "unhealthy",
        model_version=model_version or "not loaded"
    )


@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    """Single prediction."""
    if model_pipeline is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    start = time.time()
    cleaned = clean_text(request.text)

    if not cleaned:
        raise HTTPException(status_code=422, detail="Text is empty after cleaning")

    prediction = model_pipeline.predict([cleaned])[0]
    probabilities = model_pipeline.predict_proba([cleaned])[0]
    confidence = float(max(probabilities))
    latency_ms = (time.time() - start) * 1000

    return PredictionResponse(
        prediction=int(prediction),
        confidence=confidence,
        label="positive" if prediction == 1 else "negative",
        latency_ms=round(latency_ms, 2)
    )


@app.post("/predict/batch", response_model=BatchPredictionResponse)
async def predict_batch(request: BatchPredictionRequest):
    """Batch prediction on multiple texts."""
    if model_pipeline is None:
        raise HTTPException(status_code=503, detail="Model not loaded")

    start = time.time()
    cleaned_texts = [clean_text(t) for t in request.texts]
    predictions = model_pipeline.predict(cleaned_texts)
    probabilities = model_pipeline.predict_proba(cleaned_texts)
    latency_ms = (time.time() - start) * 1000

    results = []
    for i, text in enumerate(request.texts):
        results.append(PredictionResponse(
            prediction=int(predictions[i]),
            confidence=float(max(probabilities[i])),
            label="positive" if predictions[i] == 1 else "negative",
            latency_ms=0
        ))

    return BatchPredictionResponse(
        predictions=results,
        total_latency_ms=round(latency_ms, 2)
    )

"""Pydantic schemas for the serving API."""
from pydantic import BaseModel, Field
from typing import List


class PredictionRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000,
                      description="Review text")


class PredictionResponse(BaseModel):
    prediction: int = Field(..., description="0=negative, 1=positive")
    confidence: float = Field(..., ge=0.0, le=1.0,
                              description="Prediction confidence")
    label: str = Field(..., description="Readable label")
    latency_ms: float = Field(..., description="Latency in milliseconds")


class BatchPredictionRequest(BaseModel):
    texts: List[str] = Field(..., min_length=1, max_length=100,
                             description="List of texts")


class BatchPredictionResponse(BaseModel):
    predictions: List[PredictionResponse]
    total_latency_ms: float


class HealthResponse(BaseModel):
    status: str
    model_version: str

"""Post-deployment monitoring for the ML model."""
import time
import logging
from datetime import datetime, timedelta
from collections import deque
from typing import Dict, Optional
from dataclasses import dataclass, field


@dataclass
class PredictionLog:
    """Log of a single prediction."""
    timestamp: datetime
    input_text: str
    prediction: int
    confidence: float
    latency_ms: float


class ModelMonitor:
    """Monitors the performance of the model in production."""

    def __init__(
        self,
        window_size: int = 1000,
        min_confidence_threshold: float = 0.6,
        max_latency_ms: float = 500.0,
        drift_check_interval: int = 100
    ):
        self.window_size = window_size
        self.min_confidence = min_confidence_threshold
        self.max_latency = max_latency_ms
        self.drift_check_interval = drift_check_interval
        self.predictions: deque = deque(maxlen=window_size)
        self.alert_callbacks = []
        self.prediction_count = 0
        self.logger = logging.getLogger("model_monitor")

    def log_prediction(self, log: PredictionLog) -> None:
        """Records a prediction and checks the metrics."""
        self.predictions.append(log)
        self.prediction_count += 1

        # Latency check
        if log.latency_ms > self.max_latency:
            self._alert(
                "HIGH_LATENCY",
                f"Latency {log.latency_ms:.0f}ms exceeds threshold {self.max_latency}ms"
            )

        # Low confidence check
        if log.confidence < self.min_confidence:
            self._alert(
                "LOW_CONFIDENCE",
                f"Confidence {log.confidence:.2f} below threshold {self.min_confidence}"
            )

        # Periodic drift check
        if self.prediction_count % self.drift_check_interval == 0:
            self._check_distribution_drift()

    def get_metrics(self) -> Dict:
        """Returns the current metrics of the window."""
        if not self.predictions:
            return {"status": "no_data"}

        recent = list(self.predictions)
        confidences = [p.confidence for p in recent]
        latencies = [p.latency_ms for p in recent]
        predictions = [p.prediction for p in recent]

        positive_rate = sum(1 for p in predictions if p == 1) / len(predictions)

        return {
            "total_predictions": self.prediction_count,
            "window_size": len(recent),
            "avg_confidence": sum(confidences) / len(confidences),
            "min_confidence": min(confidences),
            "avg_latency_ms": sum(latencies) / len(latencies),
            "p95_latency_ms": sorted(latencies)[int(len(latencies) * 0.95)],
            "positive_rate": positive_rate,
            "low_confidence_pct": sum(
                1 for c in confidences if c < self.min_confidence
            ) / len(confidences),
        }

    def _check_distribution_drift(self) -> None:
        """Checks whether the distribution of predictions has changed."""
        if len(self.predictions) < self.window_size:
            return

        recent = list(self.predictions)
        half = len(recent) // 2
        first_half = [p.prediction for p in recent[:half]]
        second_half = [p.prediction for p in recent[half:]]

        rate_first = sum(first_half) / len(first_half)
        rate_second = sum(second_half) / len(second_half)

        # If the distribution changes by more than 15%, signal drift
        if abs(rate_first - rate_second) > 0.15:
            self._alert(
                "DISTRIBUTION_DRIFT",
                f"Positive rate changed: {rate_first:.2f} -> {rate_second:.2f}"
            )

    def _alert(self, alert_type: str, message: str) -> None:
        """Sends an alert."""
        self.logger.warning(f"[{alert_type}] {message}")
        for callback in self.alert_callbacks:
            callback(alert_type, message)

# In your GitHub Actions workflow: maximize caching
- name: Setup Python with full pip cache
  uses: actions/setup-python@v5
  with:
    python-version: "3.11"
    cache: 'pip'
    cache-dependency-path: |
      requirements.txt
      requirements-dev.txt

# DVC cache: avoid re-downloading unchanged data
- name: Cache DVC data
  uses: actions/cache@v4
  with:
    path: .dvc/cache
    key: dvc-${{ hashFiles('dvc.lock') }}
    restore-keys: |
      dvc-

# Docker BuildKit cache (GHA cache backend)
- name: Build with BuildKit cache
  uses: docker/build-push-action@v5
  with:
    cache-from: type=gha
    cache-to: type=gha,mode=max
    # 'max' mode caches every layer, not just final

# Only trigger full training when relevant files change
on:
  push:
    paths:
      - 'src/training/**'
      - 'src/data/**'
      - 'config/model_config.yaml'
      - 'requirements.txt'
    # Ignore: docs/, README.md, tests/ (unless you want test-triggered training)

jobs:
  check-changes:
    runs-on: ubuntu-latest
    outputs:
      training_needed: ${{ steps.filter.outputs.training }}
    steps:
      - uses: dorny/paths-filter@v3
        id: filter
        with:
          filters: |
            training:
              - 'src/training/**'
              - 'config/model_config.yaml'
              - 'data/**'

  training:
    needs: check-changes
    if: needs.check-changes.outputs.training_needed == 'true'
    runs-on: ubuntu-latest
    # ... training steps

# 1. On the GPU machine, download the runner
mkdir actions-runner && cd actions-runner
curl -o actions-runner.tar.gz -L \
  https://github.com/actions/runner/releases/download/v2.311.0/actions-runner-linux-x64-2.311.0.tar.gz
tar xzf actions-runner.tar.gz

# 2. Configure the runner
./config.sh --url https://github.com/YOUR_ORG/YOUR_REPO \
  --token YOUR_TOKEN \
  --labels gpu,cuda12,ml-training

# 3. Install as a service
sudo ./svc.sh install
sudo ./svc.sh start

  training:
    name: Train Model (GPU)
    runs-on: [self-hosted, gpu, cuda12]
    # The job runs on the machine with the GPU
    steps:
      - uses: actions/checkout@v4
      - name: Train with GPU
        run: |
          nvidia-smi  # Check that the GPU is available
          python train.py --config config/training-gpu.yaml --device cuda