#!/usr/bin/env python3 """Machine Learning Model Pipeline Builder and Security Check Script. This script creates a practical, vendor-neutral scaffold for a secure ML pipeline. It focuses on reproducibility, validation gates, artifact tracking, and safe promotion. What it does: - Validates input paths and required configuration - Verifies dataset snapshots and schema expectations - Simulates a controlled training run with logged metadata - Evaluates model metrics against promotion thresholds - Writes a simple manifest for traceability - Optionally prepares a deployment decision without performing deployment What it does NOT do: - It does not train a real model by default - It does not connect to cloud services - It does not read secrets or credentials - It does not deploy to production Use this as a starting point for a real pipeline in your CI/CD or orchestration system. """ from __future__ import annotations import argparse import csv import hashlib import json import os import random import statistics import sys from dataclasses import dataclass, asdict from datetime import datetime, timezone from pathlib import Path from typing import Dict, List, Optional, Sequence, Tuple @dataclass class PipelineConfig: raw_data: Path output_dir: Path required_columns: List[str] label_column: str evaluation_threshold: float random_seed: int simulate_training: bool allow_deploy: bool @dataclass class DatasetSummary: path: str row_count: int column_count: int checksum_sha256: str headers: List[str] @dataclass class EvaluationResult: metric_name: str metric_value: float threshold: float passed: bool @dataclass class Manifest: timestamp_utc: str raw_dataset: DatasetSummary required_columns_present: bool validation_notes: List[str] training_notes: List[str] evaluation: EvaluationResult promotion_allowed: bool def parse_args() -> PipelineConfig: parser = argparse.ArgumentParser( description="Build and verify a secure, reproducible ML pipeline scaffold." ) parser.add_argument( "--raw-data", required=True, help="Path to a CSV dataset snapshot used for validation and training.", ) parser.add_argument( "--output-dir", required=True, help="Directory where manifests and results will be written.", ) parser.add_argument( "--required-columns", required=True, help="Comma-separated list of columns that must exist in the dataset.", ) parser.add_argument( "--label-column", required=True, help="Name of the label/target column.", ) parser.add_argument( "--evaluation-threshold", type=float, default=0.80, help="Minimum acceptable evaluation metric for promotion.", ) parser.add_argument( "--random-seed", type=int, default=42, help="Random seed used for deterministic behavior in the simulation.", ) parser.add_argument( "--simulate-training", action="store_true", help="Run a safe simulated training flow instead of a real model training job.", ) parser.add_argument( "--allow-deploy", action="store_true", help="Allow the script to mark promotion as permitted if all checks pass.", ) args = parser.parse_args() raw_data = Path(args.raw_data).expanduser().resolve() output_dir = Path(args.output_dir).expanduser().resolve() required_columns = [c.strip() for c in args.required_columns.split(",") if c.strip()] if not required_columns: raise ValueError("At least one required column must be provided.") if args.label_column.strip() == "": raise ValueError("Label column must not be empty.") if args.evaluation_threshold < 0 or args.evaluation_threshold > 1: raise ValueError("evaluation-threshold must be between 0 and 1.") return PipelineConfig( raw_data=raw_data, output_dir=output_dir, required_columns=required_columns, label_column=args.label_column.strip(), evaluation_threshold=args.evaluation_threshold, random_seed=args.random_seed, simulate_training=args.simulate_training, allow_deploy=args.allow_deploy, ) def ensure_safe_paths(config: PipelineConfig) -> None: if not config.raw_data.exists(): raise FileNotFoundError(f"Raw data file not found: {config.raw_data}") if not config.raw_data.is_file(): raise ValueError(f"Raw data path is not a file: {config.raw_data}") config.output_dir.mkdir(parents=True, exist_ok=True) if not config.output_dir.is_dir(): raise ValueError(f"Output path is not a directory: {config.output_dir}") def sha256_file(path: Path) -> str: digest = hashlib.sha256() with path.open("rb") as f: for chunk in iter(lambda: f.read(8192), b""): digest.update(chunk) return digest.hexdigest() def load_csv(path: Path) -> Tuple[List[str], List[Dict[str, str]]]: with path.open("r", newline="", encoding="utf-8") as f: reader = csv.DictReader(f) headers = reader.fieldnames or [] rows = list(reader) return headers, rows def summarize_dataset(path: Path) -> DatasetSummary: headers, rows = load_csv(path) return DatasetSummary( path=str(path), row_count=len(rows), column_count=len(headers), checksum_sha256=sha256_file(path), headers=headers, ) def validate_dataset(summary: DatasetSummary, required_columns: Sequence[str], label_column: str) -> List[str]: notes: List[str] = [] missing = [c for c in required_columns if c not in summary.headers] if missing: raise ValueError(f"Missing required columns: {', '.join(missing)}") if label_column not in summary.headers: raise ValueError(f"Label column '{label_column}' not found in dataset.") if summary.row_count <= 0: raise ValueError("Dataset contains no rows.") notes.append("Schema validation passed.") notes.append("Label column present.") notes.append("Dataset snapshot is non-empty.") return notes def simulated_training(config: PipelineConfig, summary: DatasetSummary) -> Dict[str, object]: random.seed(config.random_seed) training_score = round(0.70 + random.random() * 0.25, 4) return { "mode": "simulated", "dataset_checksum": summary.checksum_sha256, "random_seed": config.random_seed, "hyperparameters": { "algorithm": "placeholder", "learning_rate": 0.01, "max_depth": 5, }, "training_score": training_score, "notes": [ "This is a safe simulation only.", "Replace with a real training job in a controlled environment.", "Record pinned dependencies, runtime details, and artifact versioning in production.", ], } def evaluate_model(training_artifact: Dict[str, object], threshold: float) -> EvaluationResult: score = float(training_artifact.get("training_score", 0.0)) passed = score >= threshold return EvaluationResult( metric_name="primary_score", metric_value=score, threshold=threshold, passed=passed, ) def write_json(path: Path, payload: object) -> None: with path.open("w", encoding="utf-8") as f: json.dump(payload, f, indent=2, sort_keys=True) f.write("\n") def main() -> int: try: config = parse_args() ensure_safe_paths(config) dataset_summary = summarize_dataset(config.raw_data) validation_notes = validate_dataset(dataset_summary, config.required_columns, config.label_column) if config.simulate_training: training_artifact = simulated_training(config, dataset_summary) training_notes = list(training_artifact["notes"]) # type: ignore[index] else: training_artifact = { "mode": "not-executed", "notes": [ "No training was executed.", "Use --simulate-training for a safe demonstration run.", ], } training_notes = list(training_artifact["notes"]) # type: ignore[index] evaluation = evaluate_model(training_artifact, config.evaluation_threshold) promotion_allowed = evaluation.passed and config.allow_deploy manifest = Manifest( timestamp_utc=datetime.now(timezone.utc).isoformat(), raw_dataset=dataset_summary, required_columns_present=True, validation_notes=validation_notes, training_notes=training_notes, evaluation=evaluation, promotion_allowed=promotion_allowed, ) manifest_path = config.output_dir / "pipeline-manifest.json" training_path = config.output_dir / "training-artifact.json" evaluation_path = config.output_dir / "evaluation-result.json" write_json(manifest_path, asdict(manifest)) write_json(training_path, training_artifact) write_json(evaluation_path, asdict(evaluation)) print("Pipeline completed successfully.") print(f"Manifest written to: {manifest_path}") print(f"Training artifact written to: {training_path}") print(f"Evaluation result written to: {evaluation_path}") print(f"Promotion allowed: {promotion_allowed}") if not config.allow_deploy: print("Deployment was not attempted. Use --allow-deploy only in a controlled release process.") return 0 except Exception as exc: print(f"ERROR: {exc}", file=sys.stderr) return 1 if __name__ == "__main__": raise SystemExit(main())