Documentation

# Image de base légère — Alpine Linux avec bash, curl, jq
FROM alpine:3.19
RUN apk add --no-cache bash curl jq

WORKDIR /app
COPY analyze.sh /app/analyze.sh
RUN chmod +x /app/analyze.sh

# Paramètres tunable via environmentVariables à la soumission du job
ENV ANALYSIS_DURATION=30 \
    OUTPUT_DIR=/outputs

ENTRYPOINT ["/app/analyze.sh"]

#!/bin/bash
set -e

# ── Configuration ──────────────────────────────────────────────────────────
# Marathoon injecte les paramètres de la pipeline en variables d'environnement
# (uppercased). Ici on peut les surcharger via parameterOverrides à la soumission.
DURATION=${ANALYSIS_DURATION:-30}
OUTPUT_DIR=${OUTPUT_DIR:-/outputs}

mkdir -p "$OUTPUT_DIR"

echo "[INFO] Démarrage — durée simulée : ${DURATION}s"

# ── Travail simulé ──────────────────────────────────────────────────────────
for step in "Initialisation" "Chargement" "Traitement" "Agrégation"; do
  echo "[INFO] $step..."
  sleep $((DURATION / 4))
done

# ── Sorties ─────────────────────────────────────────────────────────────────
# Tout fichier écrit dans /outputs devient un artefact téléchargeable.

cat > "$OUTPUT_DIR/summary.json" <<JSON
{
  "status": "completed",
  "duration_seconds": $DURATION,
  "records_processed": 985432,
  "records_skipped": 14568
}
JSON

cat > "$OUTPUT_DIR/metrics.csv" <<CSV
metric,value,unit
records_processed,985432,count
records_skipped,14568,count
execution_time,$DURATION,seconds
CSV

echo "[INFO] Terminé. Résultats dans $OUTPUT_DIR/"

watch/
└── run-001/
    └── anything.txt

{
  "name": "Quick Check",
  "description": "Minimal sample pipeline — fast run, no inputs.",
  "definition": {
    "imageUri": "marathoon-analyzer-simple",
    "imageTag": "latest",
    "inputs": [],
    "outputs": [
      {
        "name": "result",
        "type": "file",
        "path": "/outputs/result.json",
        "description": "Run result"
      }
    ],
    "entryPoint": null,
    "command": [],
    "tasks": [],
    "connections": []
  },
  "defaultParameters": {
    "cpuUnits": 1,
    "memoryMb": 256,
    "timeoutSeconds": 120,
    "environmentVariables": {},
    "secretRefs": []
  }
}

FROM alpine:3.19

# python3 pour le script de traitement par type de fichier
RUN apk add --no-cache bash curl jq python3 py3-pip

WORKDIR /app
# Marathoon monte /inputs et /outputs — on les crée pour la clarté
RUN mkdir -p /inputs /outputs

COPY analyze.sh process_file.py /app/
RUN chmod +x /app/analyze.sh

ENV INPUT_DIR=/inputs \
    OUTPUT_DIR=/outputs

# VOLUME déclare les points de montage (optionnel mais explicite)
VOLUME ["/inputs", "/outputs"]

ENTRYPOINT ["/app/analyze.sh"]

#!/bin/bash
set -e

# ── Configuration ──────────────────────────────────────────────────────────
INPUT_DIR=${INPUT_DIR:-/inputs}
OUTPUT_DIR=${OUTPUT_DIR:-/outputs}

mkdir -p "$OUTPUT_DIR/reports"

# ── Inventaire des entrées ──────────────────────────────────────────────────
echo "[INFO] Scan de $INPUT_DIR..."
FILE_COUNT=$(find "$INPUT_DIR" -type f | wc -l | tr -d ' ')

if [ "$FILE_COUNT" -eq 0 ]; then
  echo "[WARN] Aucun fichier dans $INPUT_DIR"
  exit 0
fi

echo "[INFO] $FILE_COUNT fichier(s) trouvé(s)"

# ── Traitement fichier par fichier ──────────────────────────────────────────
# process_file.py détecte le type (txt, json, csv, binaire) et
# génère un rapport JSON d'analyse pour chaque fichier.
find "$INPUT_DIR" -type f | while read -r file; do
  name=$(basename "$file")
  echo "[INFO] Traitement : $name"
  out="$OUTPUT_DIR/reports/${name%.*}_analysis.json"
  python3 /app/process_file.py "$file" "$out"
done

# ── Rapport agrégé ──────────────────────────────────────────────────────────
DONE=$(find "$OUTPUT_DIR/reports" -name "*.json" | wc -l | tr -d ' ')

cat > "$OUTPUT_DIR/aggregate_report.json" <<JSON
{
  "total_files_processed": $DONE,
  "input_directory": "$INPUT_DIR",
  "output_directory": "$OUTPUT_DIR",
  "status": "completed"
}
JSON

echo "[INFO] Rapport agrégé écrit. $DONE fichier(s) traité(s)."

#!/usr/bin/env python3
"""
Analyse un fichier selon son type (texte, JSON, CSV, binaire)
et écrit un rapport JSON dans le répertoire de sortie.

Usage: python3 process_file.py <input_path> <output_path>
"""

import json
import sys
import os
import hashlib
from datetime import datetime
from pathlib import Path


def get_file_info(filepath: str) -> dict:
    path = Path(filepath)
    stat = path.stat()
    with open(filepath, 'rb') as f:
        file_hash = hashlib.sha256(f.read()).hexdigest()
    return {
        "filename": path.name,
        "size_bytes": stat.st_size,
        "extension": path.suffix.lower(),
        "sha256": file_hash[:16] + "...",
    }


def analyze_text(filepath: str) -> dict:
    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
        content = f.read()
    lines = content.split('\n')
    return {
        "type": "text",
        "lines": len(lines),
        "words": len(content.split()),
        "characters": len(content),
        "non_empty_lines": len([l for l in lines if l.strip()]),
    }


def analyze_json(filepath: str) -> dict:
    with open(filepath, 'r', encoding='utf-8') as f:
        data = json.load(f)

    def depth(obj, d=0):
        if isinstance(obj, dict):
            return max((depth(v, d + 1) for v in obj.values()), default=d)
        if isinstance(obj, list):
            return max((depth(v, d + 1) for v in obj), default=d)
        return d

    return {
        "type": "json",
        "root_type": type(data).__name__,
        "keys": len(data) if isinstance(data, dict) else None,
        "items": len(data) if isinstance(data, list) else None,
        "max_depth": depth(data),
    }


def analyze_csv(filepath: str) -> dict:
    with open(filepath, 'r', encoding='utf-8', errors='ignore') as f:
        lines = f.readlines()
    if not lines:
        return {"type": "csv", "rows": 0, "columns": 0}
    first = lines[0]
    delimiter = ',' if ',' in first else ('\t' if '\t' in first else ';')
    return {
        "type": "csv",
        "rows": len(lines),
        "columns": len(first.split(delimiter)),
        "delimiter": repr(delimiter),
    }


def analyze_binary(filepath: str) -> dict:
    with open(filepath, 'rb') as f:
        header = f.read(4)
    return {
        "type": "binary",
        "magic": header.hex(),
    }


def process(filepath: str) -> dict:
    info = get_file_info(filepath)
    ext = info["extension"]
    if ext in ('.txt', '.log', '.md', '.py', '.js', '.sh'):
        analysis = analyze_text(filepath)
    elif ext == '.json':
        try:
            analysis = analyze_json(filepath)
        except json.JSONDecodeError:
            analysis = {"type": "json", "error": "Invalid JSON"}
    elif ext in ('.csv', '.tsv'):
        analysis = analyze_csv(filepath)
    else:
        analysis = analyze_binary(filepath)
    return {
        "file_info": info,
        "analysis": analysis,
        "processed_at": datetime.utcnow().isoformat() + "Z",
    }


def main():
    if len(sys.argv) < 2:
        print(json.dumps({"error": "Usage: process_file.py <input> [output]"}))
        sys.exit(1)

    filepath = sys.argv[1]
    output_path = sys.argv[2] if len(sys.argv) > 2 else None

    if not os.path.exists(filepath):
        print(json.dumps({"error": f"File not found: {filepath}"}))
        sys.exit(1)

    result = process(filepath)

    if output_path:
        os.makedirs(os.path.dirname(output_path), exist_ok=True)
        with open(output_path, 'w') as f:
            json.dump(result, f, indent=2)
        print(f"[INFO] Rapport écrit : {output_path}")
    else:
        print(json.dumps(result, indent=2))


if __name__ == "__main__":
    main()

watch/
└── batch-2024-06/
    ├── document-a.txt
    ├── document-b.txt
    └── data.json

{
  "name": "File Analyzer",
  "description": "Reads all files in /inputs and emits an analysis report.",
  "definition": {
    "imageUri": "marathoon-analyzer-file",
    "imageTag": "1.0",
    "inputs": [],
    "outputs": [
      {
        "name": "report",
        "type": "file",
        "path": "/outputs/analysis-report.json",
        "description": "Analysis report"
      }
    ],
    "entryPoint": null,
    "command": [],
    "tasks": [],
    "connections": []
  },
  "defaultParameters": {
    "cpuUnits": 2,
    "memoryMb": 512,
    "timeoutSeconds": 300,
    "environmentVariables": {
      "ANALYSIS_DURATION": "5",
      "LOG_INTERVAL": "1"
    },
    "secretRefs": []
  }
}

# Python 3.12 alpine — stdlib csv/json suffisent, aucune dépendance externe
FROM python:3.12-alpine

WORKDIR /app
RUN mkdir -p /inputs /outputs

COPY convert.py /app/convert.py

ENV INPUT_DIR=/inputs \
    OUTPUT_DIR=/outputs

VOLUME ["/inputs", "/outputs"]

ENTRYPOINT ["python", "/app/convert.py"]

#!/usr/bin/env python3
"""
CSV → JSON converter sample.

Lit tous les *.csv sous INPUT_DIR (récursivement), écrit un <nom>.json
(tableau d'objets ligne) pour chacun dans OUTPUT_DIR, puis génère un
report.json au format Marathoon — rendu automatiquement sur la page du job.
"""
import csv
import json
import os
import sys
from datetime import datetime, timezone


INPUT_DIR  = os.environ.get("INPUT_DIR",  "/inputs")
OUTPUT_DIR = os.environ.get("OUTPUT_DIR", "/outputs")


def log(msg: str) -> None:
    ts = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.%fZ")
    print(f"[INFO] {ts} {msg}", flush=True)


def find_csv_files(root: str):
    """Parcourt récursivement root et yield chaque *.csv trouvé."""
    for dirpath, _dirs, files in os.walk(root):
        for name in files:
            if name.lower().endswith(".csv"):
                yield os.path.join(dirpath, name)


def convert_file(path: str) -> dict:
    """Convertit un CSV en liste de dicts et l'écrit dans OUTPUT_DIR."""
    rel = os.path.relpath(path, INPUT_DIR)
    try:
        with open(path, newline="", encoding="utf-8-sig") as fh:
            rows = list(csv.DictReader(fh))
    except Exception as exc:
        log(f"ERREUR lecture {rel}: {exc}")
        return {"file": rel, "rows": 0, "status": f"error: {exc}"}

    out_name = os.path.splitext(os.path.basename(path))[0] + ".json"
    out_path = os.path.join(OUTPUT_DIR, out_name)

    with open(out_path, "w", encoding="utf-8") as out:
        json.dump(rows, out, indent=2, ensure_ascii=False)

    log(f"Converti {rel} → {out_name} ({len(rows)} lignes)")
    return {"file": rel, "rows": len(rows), "status": "ok"}


def write_report(summary: list) -> None:
    """
    Écrit /outputs/report.json au format Marathoon.
    La plateforme détecte ce fichier et le rend sur la page du job
    sous forme de stats + tableau.
    """
    total_rows = sum(s["rows"] for s in summary)
    report = {
        "title": "CSV → JSON conversion",
        "description": "Un fichier JSON écrit par CSV converti.",
        "generatedAtUtc": datetime.now(timezone.utc).isoformat(),
        "blocks": [
            {
                "type": "stats",
                "stats": [
                    {"label": "Fichiers convertis", "value": str(len(summary))},
                    {"label": "Lignes totales",     "value": str(total_rows)},
                ],
            },
            {
                "type": "table",
                "title": "Fichiers convertis",
                "columns": [
                    {"key": "file",   "header": "Fichier"},
                    {"key": "rows",   "header": "Lignes"},
                    {"key": "status", "header": "Statut"},
                ],
                "rows": summary,
            },
        ],
    }
    out = os.path.join(OUTPUT_DIR, "report.json")
    with open(out, "w", encoding="utf-8") as fh:
        json.dump(report, fh, indent=2)
    log(f"report.json écrit ({os.path.getsize(out)} octets)")


def main() -> int:
    os.makedirs(OUTPUT_DIR, exist_ok=True)
    log(f"Scan de {INPUT_DIR} pour les fichiers CSV")

    csv_files = sorted(find_csv_files(INPUT_DIR))
    if not csv_files:
        log("Aucun CSV trouvé — rien à convertir")
        write_report([])
        return 0

    summary = [convert_file(path) for path in csv_files]
    write_report(summary)
    log(f"Terminé : {len(summary)} fichier(s) traité(s)")
    return 0


if __name__ == "__main__":
    sys.exit(main())

watch/
└── export-001/
    ├── data.csv
    └── more.csv

{
  "name": "CSV to JSON",
  "description": "Converts CSV files to JSON with a conversion report.",
  "definition": {
    "imageUri": "marathoon-csv-json",
    "imageTag": "1.0",
    "inputs": [
      {
        "name": "data",
        "type": "file",
        "required": false,
        "defaultValue": null,
        "description": "A CSV file to convert"
      }
    ],
    "outputs": [
      {
        "name": "report",
        "type": "file",
        "path": "/outputs/report.json",
        "description": "Conversion report"
      }
    ],
    "entryPoint": null,
    "command": [],
    "tasks": [],
    "connections": []
  },
  "defaultParameters": {
    "cpuUnits": 1,
    "memoryMb": 256,
    "timeoutSeconds": 120,
    "environmentVariables": {},
    "secretRefs": []
  }
}

# BusyBox suffit — le script est en shell pur
FROM busybox:latest

COPY report.sh /report.sh
RUN chmod +x /report.sh

ENTRYPOINT ["/report.sh"]

#!/bin/sh
set -e

# Aucune entrée requise — ce pipeline génère ses propres données.
mkdir -p /outputs

# ── Génération du rapport visuel ────────────────────────────────────────────
# Marathoon détecte report.json automatiquement et le rend sur la page du job.
# Utilisez des tabs pour organiser plusieurs sections.
cat > /outputs/report.json <<'JSON'
{
  "title": "Sample Analysis Report",
  "description": "Généré par le pipeline Visual Report pour illustrer le schéma de rapport.",
  "tabs": [
    {
      "label": "Overview",
      "blocks": [
        {
          "type": "stats",
          "title": "Métriques clés",
          "stats": [
            { "label": "Lignes traitées", "value": "12,840" },
            { "label": "Erreurs",         "value": "3",    "changeDirection": "down", "change": "-40%" },
            { "label": "Durée",           "value": "8.2",  "unit": "s" },
            { "label": "Précision",       "value": "98.6", "unit": "%", "changeDirection": "up", "change": "+1.2%" }
          ]
        },
        {
          "type": "alert",
          "severity": "success",
          "message": "Toutes les validations sont passées."
        },
        {
          "type": "markdown",
          "markdown": "## Résumé\n\nLa pipeline a traité le jeu de données avec succès.\n\n- Deux valeurs aberrantes supprimées\n- Tendance **positive** trimestre sur trimestre"
        }
      ]
    },
    {
      "label": "Graphiques",
      "blocks": [
        {
          "type": "chart",
          "title": "Débit mensuel",
          "chartType": "bar",
          "series": [
            { "name": "Jobs", "data": [{"x":0,"y":120},{"x":1,"y":180},{"x":2,"y":150},{"x":3,"y":240}] }
          ],
          "options": { "xAxisLabel": "Mois", "yAxisLabel": "Jobs", "showGrid": true, "showLegend": true }
        },
        {
          "type": "chart",
          "title": "Succès vs Échec",
          "chartType": "donut",
          "series": [
            { "name": "Résultat", "data": [{"label":"Succès","x":0,"y":92},{"label":"Échec","x":1,"y":8}] }
          ],
          "options": { "showLegend": true }
        },
        {
          "type": "vega",
          "title": "Heatmap de corrélation (Vega-Lite)",
          "spec": {
            "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
            "data": { "values": [
              {"x":"A","y":"A","v":1.0},{"x":"A","y":"B","v":0.3},{"x":"A","y":"C","v":0.6},
              {"x":"B","y":"A","v":0.3},{"x":"B","y":"B","v":1.0},{"x":"B","y":"C","v":0.1},
              {"x":"C","y":"A","v":0.6},{"x":"C","y":"B","v":0.1},{"x":"C","y":"C","v":1.0}
            ]},
            "mark": "rect",
            "height": 200,
            "encoding": {
              "x": {"field":"x","type":"nominal"},
              "y": {"field":"y","type":"nominal"},
              "color": {"field":"v","type":"quantitative","scale":{"scheme":"viridis"}}
            }
          }
        }
      ]
    },
    {
      "label": "Données",
      "blocks": [
        {
          "type": "table",
          "title": "Top résultats",
          "sortable": true,
          "filterable": true,
          "columns": [
            { "key": "name",   "header": "Nom",      "type": "text" },
            { "key": "score",  "header": "Score",    "type": "number" },
            { "key": "passed", "header": "Validé",   "type": "boolean" }
          ],
          "rows": [
            { "name": "alpha", "score": 0.98, "passed": true  },
            { "name": "beta",  "score": 0.71, "passed": true  },
            { "name": "gamma", "score": 0.42, "passed": false }
          ]
        }
      ]
    }
  ]
}
JSON

echo "report.json écrit ($(wc -c < /outputs/report.json) octets)"

watch/
└── report-run-1/
    └── trigger.txt

{
  "name": "Visual Report",
  "description": "Emits a report.json rendered as charts/tables on the job page.",
  "definition": {
    "imageUri": "marathoon-sample-report",
    "imageTag": "latest",
    "inputs": [],
    "outputs": [
      {
        "name": "report",
        "type": "file",
        "path": "/outputs/report.json",
        "description": "Visual report"
      }
    ],
    "entryPoint": null,
    "command": [],
    "tasks": [],
    "connections": []
  },
  "defaultParameters": {
    "cpuUnits": 1,
    "memoryMb": 256,
    "timeoutSeconds": 120,
    "environmentVariables": {},
    "secretRefs": []
  }
}