# Creer la structure du projet
mkdir -p data-contracts/{contracts,schemas,tests,scripts,ci}

# Structure finale :
data-contracts/
├── contracts/              # Contrats YAML par domaine
│   ├── checkout/
│   │   └── orders.yaml
│   └── customer/
│       └── customers.yaml
├── schemas/                # JSON Schemas de validation
│   └── contract-schema.json
├── tests/                  # Tests de validation
│   ├── test_contracts.py
│   └── test_data_quality.py
├── scripts/                # Scripts utilitaires
│   ├── validate.py
│   └── breaking_change_detector.py
├── ci/                     # Configuration CI/CD
│   └── github-actions.yml
├── pyproject.toml
└── README.md

dataContractSpecification: 0.9.3
id: urn:datacontract:checkout:orders:v2
info:
  title: Orders Data Contract
  version: 2.1.0
  status: active
  description: |
    Contrat de donnees pour les commandes du domaine Checkout.
    Ce contrat definit les garanties fournies par l'equipe
    Checkout aux consommateurs (Finance, Analytics, Marketing).
  owner: team-checkout
  contact:
    name: Marie Dupont
    email: checkout-data@company.com
    slack: "#checkout-data-support"
    oncall: "https://pagerduty.com/team-checkout"

servers:
  production:
    type: snowflake
    account: company.eu-west-1
    database: ANALYTICS
    schema: CHECKOUT_V2
  staging:
    type: snowflake
    account: company-staging.eu-west-1
    database: ANALYTICS_STAGING
    schema: CHECKOUT_V2

models:
  orders:
    description: Table des commandes validees et traitees
    type: table
    fields:
      order_id:
        type: string
        format: uuid
        required: true
        unique: true
        pii: false
        description: Identifiant unique UUID v4 de la commande
        example: "550e8400-e29b-41d4-a716-446655440000"

      customer_id:
        type: string
        required: true
        pii: true
        classification: confidential
        description: Identifiant du client (PII - masquer hors Finance)

      order_date:
        type: timestamp
        required: true
        description: Date et heure UTC de la commande

      total_amount:
        type: decimal
        precision: 10
        scale: 2
        required: true
        minimum: 0
        maximum: 999999.99
        description: Montant total TTC en euros

      currency:
        type: string
        required: true
        pattern: "^[A-Z]{3}$"
        default: "EUR"
        description: Code devise ISO 4217

      status:
        type: string
        required: true
        enum:
          - pending
          - confirmed
          - processing
          - shipped
          - delivered
          - cancelled
          - refunded
        description: Statut actuel de la commande

      channel:
        type: string
        required: true
        enum: [web, mobile_app, marketplace, store]
        description: Canal d'acquisition de la commande

      updated_at:
        type: timestamp
        required: true
        description: Derniere mise a jour (utilise pour le CDC)

quality:
  type: SodaCL
  specification:
    checks for orders:
      - row_count > 0
      - duplicate_count(order_id) = 0
      - missing_percent(customer_id) = 0
      - missing_percent(order_date) = 0
      - invalid_percent(status) = 0
      - invalid_percent(channel) = 0
      - avg(total_amount) between 15 and 500
      - max(total_amount) < 1000000
      - freshness(updated_at) < 2h
      - schema:
          name: Validate schema
          fail:
            when required column missing:
              - order_id
              - customer_id
              - order_date
              - total_amount
              - status
            when wrong type:
              order_id: varchar
              total_amount: decimal

sla:
  freshness:
    target: 2 hours
    critical: 4 hours
  availability:
    target: 99.9%
    measurement_window: 30 days
  latency:
    p95: 30 seconds
    p99: 60 seconds
  update_frequency: every 15 minutes
  incident_response:
    severity_1: 30 minutes
    severity_2: 4 hours

terms:
  usage:
    - internal-analytics
    - financial-reporting
    - marketing-with-consent
  limitations: |
    - Donnees PII (customer_id) masquees hors du domaine Finance
    - Pas d'export vers des systemes externes sans approbation DPO
    - Retention maximale de 7 ans (obligation legale)
  billing: chargeback-by-query-volume

changelog:
  - version: 2.1.0
    date: 2025-06-01
    changes: "Ajout du champ 'channel' (non-breaking)"
  - version: 2.0.0
    date: 2025-03-15
    changes: "Breaking: renommage 'amount' → 'total_amount', ajout 'currency'"
  - version: 1.0.0
    date: 2024-09-01
    changes: "Version initiale du contrat"

{
  "$schema": "https://json-schema.org/draft/2020-12/schema",
  "title": "Data Contract Schema",
  "type": "object",
  "required": [
    "dataContractSpecification", "id", "info",
    "models", "quality", "sla"
  ],
  "properties": {
    "dataContractSpecification": {
      "type": "string",
      "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$"
    },
    "id": {
      "type": "string",
      "pattern": "^urn:datacontract:"
    },
    "info": {
      "type": "object",
      "required": ["title", "version", "owner", "contact"],
      "properties": {
        "title": { "type": "string", "minLength": 5 },
        "version": {
          "type": "string",
          "pattern": "^[0-9]+\\.[0-9]+\\.[0-9]+$"
        },
        "status": {
          "type": "string",
          "enum": ["draft", "active", "deprecated", "retired"]
        },
        "owner": { "type": "string" },
        "contact": {
          "type": "object",
          "required": ["email"],
          "properties": {
            "name": { "type": "string" },
            "email": { "type": "string", "format": "email" },
            "slack": { "type": "string" }
          }
        }
      }
    },
    "sla": {
      "type": "object",
      "required": ["freshness", "availability"]
    }
  }
}

"""
Validation des Data Contracts
- Validation du schema YAML
- Detection de breaking changes
- Verification des SLAs
"""
import yaml
import json
import sys
from pathlib import Path
from jsonschema import validate, ValidationError
from deepdiff import DeepDiff

class ContractValidator:
    def __init__(self, schema_path: str):
        with open(schema_path) as f:
            self.schema = json.load(f)

    def validate_contract(self, contract_path: str) -> dict:
        """Valider un contrat contre le schema."""
        with open(contract_path) as f:
            contract = yaml.safe_load(f)

        errors = []
        warnings = []

        # 1. Validation du schema
        try:
            validate(instance=contract, schema=self.schema)
        except ValidationError as e:
            errors.append(f"Schema invalide: {e.message}")

        # 2. Verification des champs requis dans les modeles
        for model_name, model in contract.get('models', {}).items():
            for field_name, field in model.get('fields', {}).items():
                if field.get('pii') and not field.get('classification'):
                    warnings.append(
                        f"Champ PII '{field_name}' dans '{model_name}' "
                        f"sans classification de securite"
                    )
                if field.get('required') and field.get('type') is None:
                    errors.append(
                        f"Champ requis '{field_name}' sans type defini"
                    )

        # 3. Verification du SLA
        sla = contract.get('sla', {})
        if 'freshness' not in sla:
            errors.append("SLA de fraicheur manquant")
        if 'availability' not in sla:
            errors.append("SLA de disponibilite manquant")

        return {
            "valid": len(errors) == 0,
            "errors": errors,
            "warnings": warnings,
            "contract": contract
        }

    def detect_breaking_changes(
        self, old_path: str, new_path: str
    ) -> dict:
        """Detecter les breaking changes entre deux versions."""
        with open(old_path) as f:
            old = yaml.safe_load(f)
        with open(new_path) as f:
            new = yaml.safe_load(f)

        breaking = []
        non_breaking = []

        old_models = old.get('models', {})
        new_models = new.get('models', {})

        for model_name in old_models:
            if model_name not in new_models:
                breaking.append(
                    f"BREAKING: Modele '{model_name}' supprime"
                )
                continue

            old_fields = old_models[model_name].get('fields', {})
            new_fields = new_models[model_name].get('fields', {})

            # Champs supprimes
            for field in old_fields:
                if field not in new_fields:
                    breaking.append(
                        f"BREAKING: Champ '{field}' supprime "
                        f"de '{model_name}'"
                    )

            # Champs modifies
            for field in old_fields:
                if field in new_fields:
                    old_type = old_fields[field].get('type')
                    new_type = new_fields[field].get('type')
                    if old_type != new_type:
                        breaking.append(
                            f"BREAKING: Type de '{field}' change "
                            f"de '{old_type}' a '{new_type}'"
                        )

                    # Nouveau required
                    was_required = old_fields[field].get('required', False)
                    is_required = new_fields[field].get('required', False)
                    if not was_required and is_required:
                        breaking.append(
                            f"BREAKING: '{field}' est devenu required"
                        )

            # Nouveaux champs (non-breaking)
            for field in new_fields:
                if field not in old_fields:
                    non_breaking.append(
                        f"NON-BREAKING: Nouveau champ '{field}' "
                        f"dans '{model_name}'"
                    )

        return {
            "has_breaking_changes": len(breaking) > 0,
            "breaking_changes": breaking,
            "non_breaking_changes": non_breaking
        }


if __name__ == "__main__":
    validator = ContractValidator("schemas/contract-schema.json")

    # Valider tous les contrats
    contracts_dir = Path("contracts")
    all_valid = True

    for contract_file in contracts_dir.rglob("*.yaml"):
        print(f"\nValidation: {contract_file}")
        result = validator.validate_contract(str(contract_file))

        if result["valid"]:
            print(f"  ✓ Contrat valide")
        else:
            print(f"  ✗ Contrat invalide")
            all_valid = False
            for err in result["errors"]:
                print(f"    ERREUR: {err}")

        for warn in result["warnings"]:
            print(f"    ATTENTION: {warn}")

    sys.exit(0 if all_valid else 1)

name: Data Contract CI/CD

on:
  pull_request:
    paths:
      - 'contracts/**'
      - 'schemas/**'

jobs:
  validate-contracts:
    runs-on: ubuntu-latest
    steps:
      - uses: actions/checkout@v4
        with:
          fetch-depth: 0  # Historique complet pour diff

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

      - name: Install dependencies
        run: |
          pip install pyyaml jsonschema deepdiff soda-core

      - name: Validate contract schemas
        run: python scripts/validate.py

      - name: Detect breaking changes
        run: |
          # Comparer avec la branche main
          git diff --name-only origin/main -- contracts/ | while read file; do
            if git show origin/main:$file > /tmp/old_contract.yaml 2>/dev/null; then
              echo "Checking breaking changes: $file"
              python -c "
          from scripts.validate import ContractValidator
          v = ContractValidator('schemas/contract-schema.json')
          result = v.detect_breaking_changes('/tmp/old_contract.yaml', '$file')
          if result['has_breaking_changes']:
              print('BREAKING CHANGES DETECTED:')
              for bc in result['breaking_changes']:
                  print(f'  - {bc}')
              exit(1)
          else:
              print('No breaking changes.')
              for nbc in result['non_breaking_changes']:
                  print(f'  - {nbc}')
          "
            fi
          done

      - name: Run data quality checks
        if: github.event.pull_request.labels.*.name == 'test-data'
        run: |
          # Verifier la qualite des donnees avec SodaCL
          soda scan -d production -c soda/configuration.yml \
            contracts/checkout/orders.yaml

      - name: Comment PR with results
        uses: actions/github-script@v7
        with:
          script: |
            github.rest.issues.createComment({
              issue_number: context.issue.number,
              owner: context.repo.owner,
              repo: context.repo.repo,
              body: '## Data Contract Validation\n\n' +
                    '✅ Schema validation passed\n' +
                    '✅ No breaking changes detected\n' +
                    '✅ SLA requirements met'
            })

"""Tests de validation des Data Contracts."""
import pytest
import yaml
from pathlib import Path
from scripts.validate import ContractValidator

SCHEMA_PATH = "schemas/contract-schema.json"
CONTRACTS_DIR = Path("contracts")

@pytest.fixture
def validator():
    return ContractValidator(SCHEMA_PATH)

def get_all_contracts():
    """Decouvrir tous les contrats YAML."""
    return list(CONTRACTS_DIR.rglob("*.yaml"))

@pytest.mark.parametrize("contract_path", get_all_contracts())
def test_contract_schema_valid(validator, contract_path):
    """Chaque contrat doit etre valide selon le schema."""
    result = validator.validate_contract(str(contract_path))
    assert result["valid"], \
        f"Contrat invalide: {result['errors']}"

@pytest.mark.parametrize("contract_path", get_all_contracts())
def test_contract_has_owner(contract_path):
    """Chaque contrat doit avoir un owner defini."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)
    owner = contract.get("info", {}).get("owner")
    assert owner, f"Owner manquant dans {contract_path}"

@pytest.mark.parametrize("contract_path", get_all_contracts())
def test_contract_has_sla(contract_path):
    """Chaque contrat doit definir des SLAs."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)
    sla = contract.get("sla", {})
    assert "freshness" in sla, "SLA fraicheur manquant"
    assert "availability" in sla, "SLA disponibilite manquant"

@pytest.mark.parametrize("contract_path", get_all_contracts())
def test_pii_fields_classified(contract_path):
    """Les champs PII doivent avoir une classification."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)
    for model_name, model in contract.get("models", {}).items():
        for field_name, field in model.get("fields", {}).items():
            if field.get("pii"):
                assert field.get("classification"), \
                    f"Champ PII '{field_name}' dans " \
                    f"'{model_name}' sans classification"

@pytest.mark.parametrize("contract_path", get_all_contracts())
def test_semver_format(contract_path):
    """La version doit suivre le format SemVer."""
    with open(contract_path) as f:
        contract = yaml.safe_load(f)
    version = contract.get("info", {}).get("version", "")
    parts = version.split(".")
    assert len(parts) == 3, \
        f"Version '{version}' non conforme SemVer"
    for part in parts:
        assert part.isdigit(), \
            f"Version '{version}' non conforme SemVer"

-- models/staging/stg_aws_costs.sql
WITH raw_cur AS (
    SELECT * FROM {{ source('aws_billing', 'cost_usage_report') }}
),

normalized AS (
    SELECT
        -- Identifiants FOCUS
        bill_payer_account_id    AS billing_account_id,
        line_item_usage_account_id AS sub_account_id,

        -- Periodes
        DATE(line_item_usage_start_date) AS usage_date,
        line_item_usage_start_date AS charge_period_start,
        line_item_usage_end_date   AS charge_period_end,

        -- Classification
        line_item_product_code   AS service_name,
        product_product_family   AS service_category,
        line_item_resource_id    AS resource_id,
        product_region           AS region,

        -- Couts
        line_item_unblended_cost AS billed_cost,
        COALESCE(
            reservation_effective_cost,
            savings_plan_savings_plan_effective_cost,
            line_item_unblended_cost
        ) AS effective_cost,

        -- Tags
        resource_tags_user_team        AS tag_team,
        resource_tags_user_project     AS tag_project,
        resource_tags_user_environment AS tag_environment,
        resource_tags_user_cost_center AS tag_cost_center,

        -- Meta
        line_item_line_item_type AS charge_type
    FROM raw_cur
    WHERE line_item_line_item_type NOT IN ('Credit', 'Refund')
)

SELECT * FROM normalized

-- models/marts/mart_cost_by_team.sql
WITH daily_costs AS (
    SELECT
        usage_date,
        COALESCE(tag_team, 'untagged') AS team,
        COALESCE(tag_project, 'unknown') AS project,
        COALESCE(tag_environment, 'unknown') AS environment,
        service_category,
        SUM(billed_cost)    AS total_billed_cost,
        SUM(effective_cost) AS total_effective_cost,
        COUNT(DISTINCT resource_id) AS resource_count
    FROM {{ ref('stg_aws_costs') }}
    GROUP BY 1, 2, 3, 4, 5
),

with_moving_avg AS (
    SELECT
        *,
        AVG(total_effective_cost) OVER (
            PARTITION BY team, service_category
            ORDER BY usage_date
            ROWS BETWEEN 13 PRECEDING AND CURRENT ROW
        ) AS cost_14d_avg,

        STDDEV(total_effective_cost) OVER (
            PARTITION BY team, service_category
            ORDER BY usage_date
            ROWS BETWEEN 13 PRECEDING AND CURRENT ROW
        ) AS cost_14d_stddev,

        -- Variation jour sur jour
        LAG(total_effective_cost) OVER (
            PARTITION BY team, service_category
            ORDER BY usage_date
        ) AS prev_day_cost
    FROM daily_costs
),

final AS (
    SELECT
        *,
        -- Score d'anomalie
        CASE
            WHEN cost_14d_stddev > 0 THEN
                (total_effective_cost - cost_14d_avg) / cost_14d_stddev
            ELSE 0
        END AS anomaly_score,

        -- Variation en pourcentage
        CASE
            WHEN prev_day_cost > 0 THEN
                ROUND((total_effective_cost - prev_day_cost)
                  / prev_day_cost * 100, 2)
            ELSE 0
        END AS day_over_day_pct,

        -- Flag anomalie
        CASE
            WHEN cost_14d_stddev > 0
            AND (total_effective_cost - cost_14d_avg)
                / cost_14d_stddev > 2.5
            THEN TRUE
            ELSE FALSE
        END AS is_anomaly
    FROM with_moving_avg
)

SELECT * FROM final

-- models/marts/mart_optimization_recommendations.sql
WITH idle_resources AS (
    -- Ressources avec cout mais utilisation faible
    SELECT
        resource_id,
        service_name,
        tag_team,
        SUM(effective_cost) AS monthly_cost,
        'IDLE_RESOURCE' AS recommendation_type,
        'Ressource potentiellement inutilisee - '
        || 'verifier l''utilisation' AS recommendation,
        SUM(effective_cost) * 0.9 AS estimated_savings
    FROM {{ ref('stg_aws_costs') }}
    WHERE usage_date >= CURRENT_DATE - INTERVAL '30 days'
    AND service_category = 'Compute'
    GROUP BY 1, 2, 3
    HAVING SUM(effective_cost) > 100
    AND COUNT(DISTINCT usage_date) < 5
),

untagged_resources AS (
    -- Ressources sans tags obligatoires
    SELECT
        resource_id,
        service_name,
        'unknown' AS tag_team,
        SUM(effective_cost) AS monthly_cost,
        'MISSING_TAGS' AS recommendation_type,
        'Tags obligatoires manquants - '
        || 'ajouter team, project, environment' AS recommendation,
        0 AS estimated_savings
    FROM {{ ref('stg_aws_costs') }}
    WHERE usage_date >= CURRENT_DATE - INTERVAL '30 days'
    AND (tag_team IS NULL OR tag_project IS NULL)
    GROUP BY 1, 2
    HAVING SUM(effective_cost) > 50
),

spot_candidates AS (
    -- Instances On-Demand qui pourraient etre Spot
    SELECT
        resource_id,
        service_name,
        COALESCE(tag_team, 'unknown') AS tag_team,
        SUM(effective_cost) AS monthly_cost,
        'SPOT_CANDIDATE' AS recommendation_type,
        'Instance On-Demand eligible Spot - '
        || 'economie potentielle de 60-90%' AS recommendation,
        SUM(effective_cost) * 0.7 AS estimated_savings
    FROM {{ ref('stg_aws_costs') }}
    WHERE usage_date >= CURRENT_DATE - INTERVAL '30 days'
    AND charge_type = 'Usage'
    AND service_name = 'AmazonEC2'
    AND tag_environment IN ('development', 'staging')
    GROUP BY 1, 2, 3
    HAVING SUM(effective_cost) > 200
)

SELECT * FROM idle_resources
UNION ALL
SELECT * FROM untagged_resources
UNION ALL
SELECT * FROM spot_candidates
ORDER BY estimated_savings DESC

import requests
import json
from datetime import datetime

SLACK_WEBHOOK_URL = "https://hooks.slack.com/services/T.../B.../..."

def send_finops_alert(anomalies_df, recommendations_df):
    """
    Envoie un resume quotidien FinOps sur Slack
    avec anomalies et recommandations.
    """
    # Section anomalies
    anomaly_text = ""
    critical = anomalies_df[anomalies_df['anomaly_score'] > 3.0]
    if len(critical) > 0:
        anomaly_text = "*Anomalies critiques detectees :*\n"
        for _, row in critical.head(5).iterrows():
            anomaly_text += (
                f"  - {row['team']} / {row['service_category']}: "
                f"${row['total_effective_cost']:.0f} "
                f"(+{row['day_over_day_pct']:.0f}% vs veille, "
                f"score: {row['anomaly_score']:.1f})\n"
            )

    # Section recommandations
    top_savings = recommendations_df.head(5)
    reco_text = "*Top 5 recommandations d'optimisation :*\n"
    total_savings = 0
    for _, row in top_savings.iterrows():
        reco_text += (
            f"  - [{row['recommendation_type']}] "
            f"{row['recommendation']} "
            f"(${row['estimated_savings']:.0f}/mois)\n"
        )
        total_savings += row['estimated_savings']

    # Construction du message Slack
    payload = {
        "blocks": [
            {
                "type": "header",
                "text": {
                    "type": "plain_text",
                    "text": f"FinOps Daily Report - "
                            f"{datetime.now():%d/%m/%Y}"
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": anomaly_text if anomaly_text
                            else "*Aucune anomalie detectee*"
                }
            },
            {"type": "divider"},
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": reco_text
                }
            },
            {
                "type": "section",
                "text": {
                    "type": "mrkdwn",
                    "text": f"*Economie potentielle totale : "
                            f"${total_savings:,.0f}/mois*"
                }
            }
        ]
    }

    response = requests.post(
        SLACK_WEBHOOK_URL,
        json=payload,
        headers={"Content-Type": "application/json"}
    )
    return response.status_code == 200