# Exemple de Data Management Scorecard
data_management_scorecard:
  organization: "Acme Corp"
  assessment_date: "2024-06-15"
  assessor: "Chief Data Officer"

  knowledge_areas:
    data_governance:
      current_level: 2
      target_level: 4
      priority: "CRITICAL"
      gaps:
        - "Pas de Data Council formel"
        - "Politiques non documentees"
        - "Metriques absentes"
      actions:
        - "Creer Data Governance Council (Q3)"
        - "Nommer Data Owners pour 5 domaines cles (Q3)"
        - "Definir 3 politiques prioritaires (Q4)"

    data_quality:
      current_level: 1
      target_level: 3
      priority: "HIGH"
      gaps:
        - "Aucun profiling systematique"
        - "Pas de DQ rules documentees"
      actions:
        - "Deployer Great Expectations sur 3 pipelines (Q3)"
        - "Definir DQ SLAs pour donnees financieres (Q4)"

    metadata_management:
      current_level: 1
      target_level: 3
      priority: "HIGH"
      gaps:
        - "Pas de data catalog"
        - "Lineage inconnu"
      actions:
        - "POC DataHub ou Atlan (Q3)"
        - "Documenter les 50 tables les plus utilisees (Q4)"

    data_architecture:
      current_level: 2
      target_level: 3
      priority: "MEDIUM"

    data_security:
      current_level: 2
      target_level: 4
      priority: "HIGH"

  overall_maturity: 1.6  # Moyenne des 14 KAs
  target_maturity: 3.2
  estimated_timeline: "24 months"
  estimated_budget: "2.5M EUR"

# DATA GOVERNANCE COUNCIL - Charter Template

## Mission
Superviser et diriger la strategie de gestion des donnees
pour maximiser la valeur business tout en minimisant les risques.

## Composition
- President : CDO (ou sponsor executive)
- Membres permanents :
  - Data Governance Lead (secretaire)
  - Data Owners des 5 domaines principaux
  - CISO (securite)
  - DPO (privacy / RGPD)
  - Head of Data Engineering
  - Representant Compliance
- Invites ponctuels selon les sujets

## Fonctionnement
- Frequence : Mensuelle (1h30 max)
- Quorum : 60% des membres permanents
- Decisions : Majorite simple, droit de veto du CDO

## Responsabilites
1. Valider les politiques de donnees
2. Arbitrer les conflits de propriete des donnees
3. Approuver les investissements data > 50K EUR
4. Reviser les KPIs de data quality trimestriellement
5. Prioriser les initiatives data governance
6. Escalader les risques data au Comex si necessaire

## Agenda type
1. Revue des KPIs (15 min)
2. Points d'escalade des Data Stewards (20 min)
3. Decisions en attente (30 min)
4. Nouveaux sujets / initiatives (20 min)
5. Actions et prochaines etapes (5 min)

-- Exemple simplifie de matching par regles
-- Etape 1 : Standardisation
CREATE VIEW v_standardized_customers AS
SELECT
    source_system,
    customer_id,
    UPPER(TRIM(first_name)) AS first_name_std,
    UPPER(TRIM(last_name)) AS last_name_std,
    REGEXP_REPLACE(phone, '[^0-9]', '') AS phone_std,
    LOWER(TRIM(email)) AS email_std,
    SOUNDEX(last_name) AS last_name_soundex
FROM raw_customers;

-- Etape 2 : Matching (identifier les paires candidates)
CREATE TABLE match_candidates AS
SELECT
    a.source_system AS source_a,
    a.customer_id AS id_a,
    b.source_system AS source_b,
    b.customer_id AS id_b,
    -- Score de matching
    CASE WHEN a.email_std = b.email_std AND a.email_std IS NOT NULL
         THEN 40 ELSE 0 END
    + CASE WHEN a.phone_std = b.phone_std AND a.phone_std IS NOT NULL
           THEN 30 ELSE 0 END
    + CASE WHEN a.last_name_std = b.last_name_std THEN 15 ELSE
           CASE WHEN a.last_name_soundex = b.last_name_soundex
                THEN 8 ELSE 0 END END
    + CASE WHEN a.first_name_std = b.first_name_std THEN 15 ELSE 0 END
    AS match_score
FROM v_standardized_customers a
JOIN v_standardized_customers b
    ON a.source_system < b.source_system  -- eviter auto-match
    AND (a.email_std = b.email_std
         OR a.phone_std = b.phone_std
         OR (a.last_name_soundex = b.last_name_soundex
             AND a.first_name_std = b.first_name_std))
WHERE -- Score minimum pour etre candidat
    CASE WHEN a.email_std = b.email_std THEN 40 ELSE 0 END
    + CASE WHEN a.phone_std = b.phone_std THEN 30 ELSE 0 END
    + CASE WHEN a.last_name_std = b.last_name_std THEN 15 ELSE 0 END
    + CASE WHEN a.first_name_std = b.first_name_std THEN 15 ELSE 0 END
    >= 70;  -- Seuil de matching

-- Etape 3 : Survivorship Rules (quel attribut garder de quelle source)
-- Rgle : email du CRM, adresse de l'ERP, SIRET du Legal
CREATE TABLE golden_customers AS
SELECT
    m.golden_id,
    crm.email,           -- CRM = autorite pour l'email
    erp.address_line1,   -- ERP = autorite pour l'adresse
    erp.address_city,
    erp.address_zip,
    legal.siret,         -- Legal = autorite pour le SIRET
    COALESCE(crm.first_name, erp.first_name, legal.first_name) AS first_name,
    COALESCE(crm.last_name, erp.last_name, legal.last_name) AS last_name,
    COALESCE(crm.phone, erp.phone) AS phone,
    GREATEST(crm.updated_at, erp.updated_at, legal.updated_at) AS last_updated
FROM master_mapping m
LEFT JOIN crm_customers crm ON m.crm_id = crm.customer_id
LEFT JOIN erp_customers erp ON m.erp_id = erp.customer_id
LEFT JOIN legal_entities legal ON m.legal_id = legal.entity_id;

# models/marts/finance/schema.yml
version: 2

models:
  - name: monthly_revenue
    description: |
      Chiffre d'affaires mensuel par business unit, calcule
      apres exclusion des retours et annulations.
      Source: ERP Oracle (orders) + CRM Salesforce (opportunities)
      Rafraichissement: quotidien a 06:00 UTC
    meta:
      owner: "finance-team@company.com"
      domain: "Finance"
      classification: "Confidential"
      sla: "disponible avant 08:00 UTC"
      tier: "Tier 1 - Critical"
    columns:
      - name: business_unit_id
        description: "Identifiant unique de la business unit (FK vers dim_business_unit)"
        tests:
          - not_null
          - relationships:
              to: ref('dim_business_unit')
              field: business_unit_id

      - name: month_key
        description: "Cle du mois au format YYYYMM"
        tests:
          - not_null

      - name: revenue_eur
        description: |
          Chiffre d'affaires net en EUR, apres:
          - Conversion devise au taux du jour de transaction
          - Exclusion des retours (status = 'returned')
          - Exclusion des annulations (status = 'cancelled')
          - Hors taxes (TVA exclue)
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
              max_value: 1000000000

      - name: order_count
        description: "Nombre de commandes validees pour le mois"

      - name: avg_order_value_eur
        description: "Valeur moyenne par commande = revenue_eur / order_count"

# DATA PROTECTION IMPACT ASSESSMENT (DPIA)

## 1. Description du traitement
- **Nom du projet :** Systeme de scoring credit automatise
- **Responsable du traitement :** BankCo SA
- **DPO consulte :** Oui - Marie Dupont (marie.dupont@bankco.fr)
- **Date :** 2024-06-15

## 2. Finalite du traitement
Evaluer automatiquement la solvabilite des demandeurs de credit
immobilier en combinant des donnees bancaires internes et des
donnees externes (Banque de France, bureaux de credit).

## 3. Donnees traitees
| Categorie          | Donnees                        | Sensibilite   |
|-------------------|-------------------------------|---------------|
| Identite          | Nom, prenom, date naissance   | Personnel     |
| Financier         | Revenus, patrimoine, dettes   | Confidentiel  |
| Bancaire          | Historique comptes, incidents  | Confidentiel  |
| Scoring externe   | Score Banque de France         | Confidentiel  |
| Professionnel     | Employeur, anciennete          | Personnel     |

## 4. Base legale
- Article 6(1)(b) : Execution d'un contrat (demande de credit)
- Article 6(1)(f) : Interet legitime (evaluation du risque)

## 5. Evaluation des risques
| Risque                        | Probabilite | Impact | Niveau |
|-------------------------------|-------------|--------|--------|
| Discrimination algorithmique  | Moyen       | Eleve  | ELEVE  |
| Fuite de donnees financieres  | Faible      | Eleve  | MOYEN  |
| Scoring errone (faux positif) | Moyen       | Moyen  | MOYEN  |
| Non-exercice du droit d'acces | Faible      | Moyen  | FAIBLE |

## 6. Mesures d'attenuation
- **Discrimination :** Audit trimestriel du modele par variable
  protegee (genre, age, origine). Tests de fairness (equalized odds)
- **Fuite :** Chiffrement AES-256 au repos, TLS 1.3 en transit.
  Acces limite aux equipes credit (RBAC)
- **Scoring errone :** Droit a l'explication (Art. 22).
  Revue humaine obligatoire pour les refus
- **Droit d'acces :** API self-service pour telecharger son scoring

## 7. Conclusion
Risque residuel : ACCEPTABLE avec les mesures ci-dessus.
Prochaine revue : dans 12 mois ou si changement du modele.

-- Exemple de controle SOX : reconciliation journal entries
-- Ce check tourne chaque nuit apres le chargement du DWH

-- Controle 1 : Total du DWH = Total du GL (General Ledger)
SELECT
    CASE
        WHEN ABS(dwh.total_amount - gl.total_amount) < 0.01
        THEN 'PASS'
        ELSE 'FAIL - ECART: ' ||
             CAST(dwh.total_amount - gl.total_amount AS VARCHAR)
    END AS sox_control_result,
    dwh.total_amount AS dwh_total,
    gl.total_amount AS gl_total,
    CURRENT_TIMESTAMP AS check_timestamp
FROM (
    SELECT SUM(amount) AS total_amount
    FROM dwh.fact_journal_entries
    WHERE posting_date = CURRENT_DATE - 1
) dwh
CROSS JOIN (
    SELECT SUM(amount) AS total_amount
    FROM erp.gl_journal_entries
    WHERE posting_date = CURRENT_DATE - 1
) gl;

-- Controle 2 : Audit trail - toute modification est loggee
CREATE TABLE audit.financial_data_changes (
    change_id       BIGINT IDENTITY PRIMARY KEY,
    table_name      VARCHAR(100) NOT NULL,
    record_id       BIGINT NOT NULL,
    field_name      VARCHAR(100) NOT NULL,
    old_value       VARCHAR(500),
    new_value       VARCHAR(500),
    changed_by      VARCHAR(100) NOT NULL,
    changed_at      TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
    change_reason   VARCHAR(500) NOT NULL,
    approved_by     VARCHAR(100),  -- Segregation of duties
    approved_at     TIMESTAMP
    -- Cette table est en INSERT ONLY : pas de UPDATE ni DELETE
);

-- Activity Schema : la structure de base
CREATE TABLE activity_stream (
    -- Identite
    activity_id     BIGINT PRIMARY KEY,      -- ID unique de l'activite
    entity_id       VARCHAR(100) NOT NULL,    -- ID de l'entite (customer, product...)
    entity_type     VARCHAR(50) NOT NULL,     -- 'customer', 'product', 'employee'

    -- Activite
    activity        VARCHAR(100) NOT NULL,    -- 'placed_order', 'contacted_support'
    ts              TIMESTAMP NOT NULL,       -- Quand l'activite s'est produite

    -- Attributs flexibles (JSON ou colonnes dediees)
    feature_1       VARCHAR(500),             -- Attribut contextuel 1
    feature_2       VARCHAR(500),             -- Attribut contextuel 2
    feature_3       VARCHAR(500),             -- Attribut contextuel 3
    revenue_impact  DECIMAL(15,2),            -- Impact financier (si applicable)

    -- Metadata
    source_system   VARCHAR(50),
    loaded_at       TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- Exemples de donnees
INSERT INTO activity_stream VALUES
-- Client 123 passe une commande
(1, 'cust_123', 'customer', 'placed_order',
 '2024-01-15 10:30:00', 'order_456', 'electronics', 'online', 299.99, 'ecommerce', NOW()),
-- Client 123 contacte le support
(2, 'cust_123', 'customer', 'contacted_support',
 '2024-01-16 14:00:00', 'ticket_789', 'complaint', 'email', NULL, 'zendesk', NOW()),
-- Client 123 retourne un produit
(3, 'cust_123', 'customer', 'returned_product',
 '2024-01-18 09:00:00', 'return_012', 'defective', NULL, -299.99, 'ecommerce', NOW());

-- Requete : timeline complete d'un client
SELECT activity, ts, feature_1, revenue_impact
FROM activity_stream
WHERE entity_id = 'cust_123'
ORDER BY ts;

-- Requete : customers qui ont commande puis contacte le support dans les 7 jours
SELECT
    a.entity_id,
    a.ts AS order_date,
    b.ts AS support_date,
    DATEDIFF(day, a.ts, b.ts) AS days_between
FROM activity_stream a
JOIN activity_stream b
    ON a.entity_id = b.entity_id
    AND a.activity = 'placed_order'
    AND b.activity = 'contacted_support'
    AND b.ts BETWEEN a.ts AND a.ts + INTERVAL '7 days';

# Pipeline ETL traditionnel - pseudo-code
class ETLPipeline:
    def __init__(self, source_config, target_config):
        self.source = SourceConnector(source_config)
        self.target = TargetConnector(target_config)
        self.staging_db = StagingDatabase()

    def extract(self):
        """Phase 1: Extraction des sources"""
        raw_data = self.source.read_incremental(
            since=self.get_last_watermark()
        )
        # Donnees brutes dans staging temporaire
        self.staging_db.write('staging_raw', raw_data)
        return raw_data

    def transform(self, raw_data):
        """Phase 2: Transformation sur serveur ETL"""
        # Nettoyage
        cleaned = self.clean_nulls(raw_data)
        cleaned = self.standardize_dates(cleaned)
        cleaned = self.validate_business_rules(cleaned)

        # Enrichissement
        enriched = self.lookup_reference_data(cleaned)
        enriched = self.apply_business_logic(enriched)

        # Conformite
        conformed = self.map_to_target_schema(enriched)
        conformed = self.mask_pii_columns(conformed)  # RGPD

        # Qualite
        self.run_quality_checks(conformed)
        return conformed

    def load(self, transformed_data):
        """Phase 3: Chargement dans le DWH"""
        self.target.bulk_insert(
            table='dim_customer',
            data=transformed_data,
            mode='merge'  # SCD Type 2
        )
        self.update_watermark()

    def run(self):
        raw = self.extract()
        transformed = self.transform(raw)
        self.load(transformed)

-- models/staging/stg_orders.sql (dbt model)
-- Phase "Transform" executee DANS le warehouse

WITH source AS (
    -- Donnees brutes chargees par Fivetran/Airbyte
    SELECT * FROM {{ source('raw', 'orders') }}
),

cleaned AS (
    SELECT
        id AS order_id,
        TRIM(LOWER(customer_email)) AS customer_email,
        CAST(order_date AS DATE) AS order_date,
        CAST(amount AS DECIMAL(12,2)) AS order_amount,
        UPPER(currency) AS currency,
        -- Nettoyage des statuts
        CASE
            WHEN status IN ('cancelled', 'canceled', 'annule')
                THEN 'CANCELLED'
            WHEN status IN ('completed', 'done', 'termine')
                THEN 'COMPLETED'
            WHEN status IN ('pending', 'en_attente')
                THEN 'PENDING'
            ELSE 'UNKNOWN'
        END AS order_status,
        _fivetran_synced AS loaded_at
    FROM source
    WHERE id IS NOT NULL
      AND order_date >= '2020-01-01'  -- Filtre qualite
),

validated AS (
    SELECT
        *,
        -- Tests de qualite inline
        CASE WHEN order_amount < 0 THEN TRUE ELSE FALSE END AS is_suspicious
    FROM cleaned
)

SELECT * FROM validated

-- schema.yml pour les tests dbt
-- version: 2
-- models:
--   - name: stg_orders
--     columns:
--       - name: order_id
--         tests: [unique, not_null]
--       - name: order_amount
--         tests: [not_null, positive_values]

import apache_beam as beam
from apache_beam.options.pipeline_options import PipelineOptions

def compute_user_metrics(events):
    """Logique metier UNIQUE - utilisee en batch ET streaming"""
    return (
        events
        | 'ParseEvent' >> beam.Map(parse_event)
        | 'FilterValid' >> beam.Filter(is_valid_event)
        | 'KeyByUser' >> beam.Map(lambda e: (e['user_id'], e))
        | 'WindowEvents' >> beam.WindowInto(
            beam.window.FixedWindows(3600)  # fenetres de 1h
        )
        | 'GroupByUser' >> beam.GroupByKey()
        | 'ComputeMetrics' >> beam.Map(calculate_metrics)
        | 'FormatOutput' >> beam.Map(format_for_serving)
    )

# MODE BATCH - Spark Runner
batch_options = PipelineOptions([
    '--runner=SparkRunner',
    '--spark_master_url=spark://master:7077',
    '--input=gs://data/events/2024/**/*.json'
])
with beam.Pipeline(options=batch_options) as p:
    events = p | beam.io.ReadFromText('gs://data/events/')
    compute_user_metrics(events) | beam.io.WriteToBigQuery('batch_views')

# MODE STREAMING - Flink Runner
stream_options = PipelineOptions([
    '--runner=FlinkRunner',
    '--streaming',
    '--flink_master=flink-jobmanager:8081'
])
with beam.Pipeline(options=stream_options) as p:
    events = p | beam.io.ReadFromKafka(
        consumer_config={'bootstrap.servers': 'kafka:9092'},
        topics=['user_events']
    )
    compute_user_metrics(events) | beam.io.WriteToBigQuery('realtime_views')

from pyspark.sql import SparkSession
from pyspark.sql.functions import current_timestamp, lit, input_file_name
from delta.tables import DeltaTable

spark = SparkSession.builder \
    .appName("Bronze Ingestion") \
    .config("spark.sql.extensions",
            "io.delta.sql.DeltaSparkSessionExtension") \
    .getOrCreate()

# ========================================
# BRONZE: Ingestion brute depuis les sources
# ========================================
def ingest_to_bronze(source_path, bronze_table, source_name):
    """Charge les donnees brutes dans la couche Bronze"""

    # Lecture des donnees brutes (JSON, CSV, Parquet...)
    raw_df = spark.read \
        .option("mergeSchema", "true") \
        .json(source_path)

    # Ajout des metadata d'ingestion
    bronze_df = raw_df \
        .withColumn("_bronze_loaded_at", current_timestamp()) \
        .withColumn("_bronze_source", lit(source_name)) \
        .withColumn("_bronze_file", input_file_name()) \
        .withColumn("_bronze_batch_id", lit(get_batch_id()))

    # Ecriture append-only dans Delta Lake
    bronze_df.write \
        .format("delta") \
        .mode("append") \
        .option("mergeSchema", "true") \
        .saveAsTable(f"bronze.{bronze_table}")

    print(f"Bronze: {bronze_df.count()} rows ingested into {bronze_table}")

# Exemples d'ingestion
ingest_to_bronze("s3://raw/orders/2024-01-15/", "orders", "ERP_SAP")
ingest_to_bronze("s3://raw/customers/", "customers", "CRM_Salesforce")
ingest_to_bronze("s3://raw/events/", "clickstream", "Web_Analytics")

-- models/silver/slv_orders.sql
-- Transformation Bronze → Silver pour les commandes

{{
    config(
        materialized='incremental',
        unique_key='order_id',
        incremental_strategy='merge',
        file_format='delta',
        schema='silver'
    )
}}

WITH bronze_orders AS (
    SELECT
        *,
        ROW_NUMBER() OVER (
            PARTITION BY id
            ORDER BY _bronze_loaded_at DESC
        ) AS _row_rank
    FROM {{ source('bronze', 'orders') }}
    {% if is_incremental() %}
    WHERE _bronze_loaded_at > (
        SELECT MAX(_silver_loaded_at) FROM {{ this }}
    )
    {% endif %}
),

-- Deduplication : garder la version la plus recente
deduplicated AS (
    SELECT * FROM bronze_orders WHERE _row_rank = 1
),

-- Nettoyage et typage
cleaned AS (
    SELECT
        -- Cle metier
        CAST(id AS BIGINT) AS order_id,

        -- Nettoyage client
        CAST(customer_id AS BIGINT) AS customer_id,
        TRIM(LOWER(customer_email)) AS customer_email,

        -- Typage des dates
        TRY_CAST(order_date AS TIMESTAMP) AS order_date,
        TRY_CAST(shipped_date AS TIMESTAMP) AS shipped_date,

        -- Montants
        ROUND(CAST(amount AS DECIMAL(12,2)), 2) AS order_amount,
        UPPER(TRIM(COALESCE(currency, 'EUR'))) AS currency,

        -- Standardisation des statuts
        CASE
            WHEN LOWER(status) IN ('cancelled','canceled','annule')
                THEN 'CANCELLED'
            WHEN LOWER(status) IN ('completed','done','livre')
                THEN 'COMPLETED'
            WHEN LOWER(status) IN ('shipped','expedie','en_transit')
                THEN 'SHIPPED'
            WHEN LOWER(status) IN ('pending','en_attente','new')
                THEN 'PENDING'
            ELSE 'UNKNOWN'
        END AS order_status,

        -- Metadata
        _bronze_loaded_at,
        _bronze_source,
        CURRENT_TIMESTAMP() AS _silver_loaded_at

    FROM deduplicated
    WHERE id IS NOT NULL
      AND TRY_CAST(amount AS DECIMAL(12,2)) IS NOT NULL
      AND TRY_CAST(order_date AS TIMESTAMP) IS NOT NULL
),

-- Validation metier
validated AS (
    SELECT
        *,
        CASE
            WHEN order_amount <= 0 THEN 'INVALID_AMOUNT'
            WHEN order_date > CURRENT_DATE() THEN 'FUTURE_DATE'
            WHEN customer_id IS NULL THEN 'MISSING_CUSTOMER'
            ELSE 'VALID'
        END AS _validation_status
    FROM cleaned
)

SELECT * FROM validated
-- On garde les lignes invalides mais flaggees pour audit

-- models/gold/gold_daily_revenue.sql
-- Gold layer: KPI quotidien de revenu par pays et categorie

{{
    config(
        materialized='table',
        file_format='delta',
        schema='gold',
        tags=['daily', 'finance', 'executive']
    )
}}

WITH orders AS (
    SELECT * FROM {{ ref('slv_orders') }}
    WHERE _validation_status = 'VALID'
      AND order_status = 'COMPLETED'
),

customers AS (
    SELECT * FROM {{ ref('slv_customers') }}
),

products AS (
    SELECT * FROM {{ ref('slv_products') }}
),

order_lines AS (
    SELECT * FROM {{ ref('slv_order_lines') }}
),

-- Jointure enrichie
enriched AS (
    SELECT
        o.order_id,
        o.order_date,
        c.country AS customer_country,
        c.segment AS customer_segment,
        p.category AS product_category,
        p.brand AS product_brand,
        ol.quantity,
        ol.unit_price,
        ol.quantity * ol.unit_price AS line_revenue,
        o.currency
    FROM orders o
    INNER JOIN customers c ON o.customer_id = c.customer_id
    INNER JOIN order_lines ol ON o.order_id = ol.order_id
    INNER JOIN products p ON ol.product_id = p.product_id
),

-- Aggregation Gold
daily_revenue AS (
    SELECT
        DATE_TRUNC('day', order_date) AS revenue_date,
        customer_country,
        customer_segment,
        product_category,
        COUNT(DISTINCT order_id) AS total_orders,
        SUM(quantity) AS total_items_sold,
        SUM(line_revenue) AS gross_revenue,
        AVG(line_revenue) AS avg_order_value,
        COUNT(DISTINCT customer_country) AS countries_served,
        CURRENT_TIMESTAMP() AS _gold_refreshed_at
    FROM enriched
    GROUP BY 1, 2, 3, 4
)

SELECT * FROM daily_revenue

-- Ce modele Gold est directement consomme par:
-- - Le dashboard Tableau "Executive Revenue"
-- - L'API interne /api/v2/revenue
-- - Le rapport hebdomadaire CFO

// Bronze: Donnee brute telle que recue de l'API
{
  "id": "ord_20240115_78432",
  "cust_id": "C-9912",
  "cust_email": " Marie.Dupont@Gmail.COM ",
  "date": "2024-01-15T14:22:33+01:00",
  "shipped": null,
  "amt": "149.99",           // String au lieu de number!
  "currency": "eur",         // Minuscule, pas standardise
  "status": "termine",       // Terme francais, pas normalise
  "items": [
    {"sku": "SKU-A1234", "qty": "2", "price": "49.99"},
    {"sku": "SKU-B5678", "qty": "1", "price": "50.01"}
  ],
  "_bronze_loaded_at": "2024-01-15T13:25:00Z",
  "_bronze_source": "ERP_API_v2"
}

// Silver: Donnee nettoyee, typee, standardisee
{
  "order_id": 78432,
  "customer_id": 9912,
  "customer_email": "marie.dupont@gmail.com",   // Trim + lowercase
  "order_date": "2024-01-15T14:22:33",          // ISO format
  "shipped_date": null,
  "order_amount": 149.99,                        // DECIMAL
  "currency": "EUR",                             // Uppercase standard
  "order_status": "COMPLETED",                   // Normalise
  "_validation_status": "VALID",
  "_bronze_loaded_at": "2024-01-15T13:25:00Z",
  "_silver_loaded_at": "2024-01-15T14:00:00Z"
}

// Gold: Aggregation business-ready
{
  "revenue_date": "2024-01-15",
  "customer_country": "France",
  "customer_segment": "Premium",
  "product_category": "Electronics",
  "total_orders": 1247,
  "total_items_sold": 3891,
  "gross_revenue": 298450.67,
  "avg_order_value": 239.33,
  "_gold_refreshed_at": "2024-01-16T06:00:00Z"
}

# data-products/orders/datacontract.yaml
apiVersion: datacontract/v1
kind: DataProduct
metadata:
  name: dp-orders
  domain: commerce
  owner: team-commerce@company.com
  tier: tier-1  # Critique business

spec:
  description: |
    Commandes validees et enrichies du domaine Commerce.
    Source de verite pour toute analyse liee aux ventes.

  schema:
    type: delta_table
    location: s3://data-products/commerce/orders/
    fields:
      - name: order_id
        type: bigint
        description: Identifiant unique de la commande
        constraints: [not_null, unique]
      - name: customer_id
        type: bigint
        description: FK vers le data product dp-customers
      - name: order_date
        type: timestamp
        description: Date et heure de la commande (UTC)
      - name: total_amount
        type: decimal(12,2)
        constraints: [not_null, positive]
      - name: status
        type: string
        enum: [PENDING, SHIPPED, COMPLETED, CANCELLED]

  sla:
    freshness: "< 1 hour"
    availability: "99.9%"
    completeness: "> 99.5%"
    refresh_schedule: "every 30 minutes"

  quality:
    tests:
      - unique: order_id
      - not_null: [order_id, customer_id, order_date, total_amount]
      - accepted_values:
          column: status
          values: [PENDING, SHIPPED, COMPLETED, CANCELLED]
      - relationship:
          column: customer_id
          to: dp-customers.customer_id

  access:
    classification: confidential
    pii_columns: []  # PII masquee dans ce product
    consumers:
      - team-finance (read)
      - team-marketing (read, filtered)
      - team-analytics (read)

# Knowledge Graph - Relations entre entites de donnees
@prefix data: <https://company.com/data/> .
@prefix schema: <https://company.com/schema/> .
@prefix quality: <https://company.com/quality/> .

# Table et ses colonnes
data:table_orders
    schema:database "snowflake_prod" ;
    schema:schema "gold" ;
    schema:owner "team-finance" ;
    schema:contains data:col_order_id, data:col_customer_id ;
    schema:refreshFrequency "hourly" ;
    quality:completeness 0.995 ;
    quality:lastChecked "2024-01-15T06:00:00Z" .

# Relations entre tables
data:col_customer_id
    schema:type "BIGINT" ;
    schema:foreignKey data:table_customers.customer_id ;
    schema:piiClassification "none" .

# Lineage : d'ou viennent les donnees ?
data:table_orders
    schema:derivedFrom data:bronze_orders ;
    schema:transformedBy data:dbt_model_slv_orders ;
    schema:consumedBy data:dashboard_revenue, data:ml_model_churn .

# Usage : qui utilise quoi ?
data:dashboard_revenue
    schema:owner "CFO Office" ;
    schema:queries data:table_orders ;
    schema:avgDailyQueries 450 ;
    schema:lastAccessed "2024-01-15T14:32:00Z" .

-- L'utilisateur ecrit UNE requete SQL
-- Le moteur de virtualisation la distribue aux 3 sources

SELECT
    c.customer_name,             -- Source: PostgreSQL (CRM)
    o.total_orders,              -- Source: Snowflake (DWH)
    s.satisfaction_score,        -- Source: API Salesforce
    m.churn_probability          -- Source: ML Model API
FROM virtual.customers c
JOIN virtual.order_summary o ON c.id = o.customer_id
JOIN virtual.satisfaction s ON c.id = s.customer_id
JOIN virtual.ml_predictions m ON c.id = m.customer_id
WHERE c.country = 'FR'
  AND o.total_orders > 10;

-- Le moteur de virtualisation :
-- 1. Envoie "SELECT name FROM postgres.crm.customers WHERE country='FR'"
-- 2. Envoie "SELECT * FROM snowflake.gold.order_summary"
-- 3. Appelle l'API Salesforce /satisfaction?country=FR
-- 4. Appelle l'API ML /predictions
-- 5. Fait la jointure en memoire et retourne le resultat

# Cube.js - Definition semantique centralisee
cube('Orders', {
  sql: `SELECT * FROM gold.fact_orders`,

  measures: {
    count: {
      type: 'count',
      description: 'Nombre total de commandes'
    },
    totalRevenue: {
      type: 'sum',
      sql: 'total_amount',
      description: 'Revenu brut total (hors taxes, hors annulations)',
      filters: [{ sql: `${CUBE}.status != 'CANCELLED'` }],
      format: 'currency'
    },
    avgOrderValue: {
      type: 'avg',
      sql: 'total_amount',
      description: 'Panier moyen (commandes non annulees)',
      filters: [{ sql: `${CUBE}.status != 'CANCELLED'` }],
      format: 'currency'
    },
    conversionRate: {
      type: 'number',
      sql: `${count} / NULLIF(${Sessions.count}, 0) * 100`,
      description: 'Taux de conversion sessions -> commandes',
      format: 'percent'
    }
  },

  dimensions: {
    status: { sql: 'status', type: 'string' },
    orderDate: { sql: 'order_date', type: 'time' },
    country: { sql: 'customer_country', type: 'string' }
  }
});

// Maintenant TOUS les outils (Tableau, PowerBI, Metabase, API)
// utilisent la MEME definition de "totalRevenue"

// ===========================================
// Modele E-Commerce - dbdiagram.io (DBML)
// ===========================================

// Configuration du projet
Project ecommerce {
  database_type: 'PostgreSQL'
  Note: 'Modele e-commerce pour la plateforme ShopFast'
}

// =====================
// TABLES DIMENSION
// =====================

Table dim_customers {
  customer_id bigint [pk, increment]
  email varchar(255) [unique, not null]
  first_name varchar(100) [not null]
  last_name varchar(100) [not null]
  phone varchar(20)
  country_code char(2) [not null, default: 'FR']
  segment varchar(20) [note: 'Premium, Standard, New']
  created_at timestamp [default: `now()`]
  updated_at timestamp
  is_active boolean [default: true]

  Note: 'Table dimension clients - SCD Type 2'
}

Table dim_products {
  product_id bigint [pk, increment]
  sku varchar(50) [unique, not null]
  name varchar(255) [not null]
  description text
  category_id int [ref: > dim_categories.category_id]
  brand varchar(100)
  unit_price decimal(10,2) [not null]
  cost_price decimal(10,2)
  weight_kg decimal(6,2)
  is_active boolean [default: true]
  created_at timestamp [default: `now()`]

  indexes {
    sku [unique]
    (category_id, is_active) [name: 'idx_product_category']
  }
}

Table dim_categories {
  category_id int [pk, increment]
  name varchar(100) [not null]
  parent_id int [ref: > dim_categories.category_id]
  level int [note: '1=root, 2=sub, 3=leaf']
  path varchar(500) [note: 'Materialized path: /Electronics/Phones/']
}

Table dim_date {
  date_key int [pk, note: 'Format YYYYMMDD']
  full_date date [unique, not null]
  day_of_week int
  day_name varchar(10)
  month int
  month_name varchar(10)
  quarter int
  year int
  is_weekend boolean
  is_holiday boolean
}

// =====================
// TABLES DE FAITS
// =====================

Table fact_orders {
  order_id bigint [pk, increment]
  customer_id bigint [not null, ref: > dim_customers.customer_id]
  order_date_key int [not null, ref: > dim_date.date_key]
  status varchar(20) [not null, note: 'PENDING|SHIPPED|COMPLETED|CANCELLED']
  total_amount decimal(12,2) [not null]
  discount_amount decimal(10,2) [default: 0]
  shipping_amount decimal(8,2) [default: 0]
  tax_amount decimal(10,2)
  payment_method varchar(30)
  shipping_country char(2)
  created_at timestamp [default: `now()`]

  indexes {
    customer_id
    order_date_key
    (status, order_date_key) [name: 'idx_status_date']
  }
}

Table fact_order_lines {
  line_id bigint [pk, increment]
  order_id bigint [not null, ref: > fact_orders.order_id]
  product_id bigint [not null, ref: > dim_products.product_id]
  quantity int [not null]
  unit_price decimal(10,2) [not null]
  line_total decimal(12,2) [not null, note: 'quantity * unit_price']
  discount_pct decimal(5,2) [default: 0]
}

// =====================
// ENUM (optionnel)
// =====================

Enum order_status {
  PENDING
  SHIPPED
  COMPLETED
  CANCELLED
}

# models/gold/schema.yml
version: 2

models:
  - name: gold_daily_revenue
    description: |
      KPI quotidien de revenu agrege par pays et categorie.
      **Owner**: Finance Team
      **SLA**: Rafraichi avant 6h UTC chaque jour
      **Consumers**: Dashboard CFO, API Revenue
    columns:
      - name: revenue_date
        description: Date du jour agrege (UTC)
        tests: [not_null, unique]
      - name: customer_country
        description: Code pays ISO 3166-1 alpha-2
        tests: [not_null]
      - name: gross_revenue
        description: |
          Revenu brut = SUM(quantity * unit_price)
          Exclut les commandes CANCELLED.
          Devise: EUR (converti au taux du jour)
        tests:
          - not_null
          - dbt_utils.accepted_range:
              min_value: 0
      - name: total_orders
        description: Nombre de commandes COMPLETED distinctes
        tests: [not_null]

  - name: gold_customer_360
    description: |
      Vue 360 du client combinant profil, commandes,
      satisfaction et predictions ML.
    meta:
      owner: team-analytics
      tier: tier-1
      pii: false  # PII masquee
    columns:
      - name: customer_id
        description: Identifiant unique client
        tests: [unique, not_null]
      - name: lifetime_value
        description: CLV predite sur 12 mois (modele ML v3.2)

# Generer la documentation
$ dbt docs generate

# Servir le site de documentation en local
$ dbt docs serve --port 8080

# Le site inclut automatiquement :
# - Description de chaque modele
# - Schema des colonnes avec tests
# - Lineage DAG interactif (qui depend de qui)
# - Source freshness
# - Tags et metadata custom