Gestión de Datos de Prueba en Pipelines DevOps: Sincronización, Enmascaramiento y Versionado

# data-classification-schema.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: data-classification
data:
  classification-rules.json: |
    {
      "classifications": {
        "PII": {
          "level": "sensible",
          "patterns": [
            "\\b[A-Z]{1}[a-z]+\\s[A-Z]{1}[a-z]+\\b",
            "\\b\\d{3}-\\d{2}-\\d{4}\\b",
            "\\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\\.[A-Z|a-z]{2,}\\b"
          ],
          "fields": ["nombre", "email", "ssn", "telefono", "direccion"],
          "handling": {
            "storage": "encriptado",
            "transit": "tls-requerido",
            "retention": "30-dias",
            "masking": "requerido"
          }
        },
        "PHI": {
          "level": "altamente-sensible",
          "fields": ["diagnostico", "historial_medico", "prescripcion"],
          "compliance": ["HIPAA"],
          "handling": {
            "storage": "encriptado-en-reposo",
            "access": "auditado",
            "masking": "tokenizacion"
          }
        },
        "Financiero": {
          "level": "sensible",
          "patterns": [
            "\\b\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}\\b",
            "\\b\\d{3,4}\\b"
          ],
          "fields": ["tarjeta_credito", "cuenta_bancaria", "numero_ruta"],
          "compliance": ["PCI-DSS"],
          "handling": {
            "storage": "tokenizado",
            "masking": "encriptacion-formato-preservado"
          }
        }
      }
    }

# test-data-provisioning/provisioner.py
import json
import hashlib
import random
from datetime import datetime, timedelta
from typing import Dict, List, Any
import boto3
import psycopg2
from faker import Faker
from dataclasses import dataclass

@dataclass
class SolicitudDatosPrueba:
    entorno: str
    tamano_dataset: str  # pequeño, mediano, grande
    frescura_datos: str  # tiempo-real, diario, semanal
    requisitos_cumplimiento: List[str]
    valor_semilla: int

class AprovisionadorDatosPrueba:
    def __init__(self, ruta_config: str):
        with open(ruta_config, 'r') as f:
            self.config = json.load(f)
        self.faker = Faker('es_ES')
        self.s3 = boto3.client('s3')

    def aprovisionar_dataset(self, solicitud: SolicitudDatosPrueba) -> Dict[str, Any]:
        """Flujo principal de aprovisionamiento"""
        dataset = {
            'metadata': self._generar_metadata(solicitud),
            'data': {}
        }

        # Determinar fuentes de datos basadas en requisitos
        if 'tiempo-real' in solicitud.frescura_datos:
            dataset['data'] = self._extraer_subconjunto_produccion(solicitud)
        else:
            dataset['data'] = self._generar_datos_sinteticos(solicitud)

        # Aplicar transformaciones de cumplimiento
        if solicitud.requisitos_cumplimiento:
            dataset['data'] = self._aplicar_enmascaramiento_cumplimiento(
                dataset['data'],
                solicitud.requisitos_cumplimiento
            )

        # Versionar y almacenar el dataset
        dataset_id = self._versionar_dataset(dataset)

        # Aprovisionar al entorno objetivo
        self._desplegar_a_entorno(dataset_id, solicitud.entorno)

        return {
            'dataset_id': dataset_id,
            'entorno': solicitud.entorno,
            'estado': 'aprovisionado',
            'timestamp': datetime.utcnow().isoformat()
        }

    def _extraer_subconjunto_produccion(self, solicitud: SolicitudDatosPrueba) -> Dict:
        """Extraer y crear subconjunto de datos de producción"""
        conn = psycopg2.connect(
            host=self.config['base_datos_prod']['host'],
            port=self.config['base_datos_prod']['puerto'],
            database=self.config['base_datos_prod']['base_datos'],
            user=self.config['base_datos_prod']['usuario'],
            password=self.config['base_datos_prod']['contraseña']
        )

        config_subconjunto = self._obtener_config_subconjunto(solicitud.tamano_dataset)

        query = f"""
        WITH usuarios_muestreados AS (
            SELECT * FROM usuarios
            TABLESAMPLE BERNOULLI ({config_subconjunto['porcentaje_muestreo']})
            WHERE fecha_creacion > NOW() - INTERVAL '{config_subconjunto['ventana_tiempo']}'
            LIMIT {config_subconjunto['max_registros']}
        ),
        ordenes_relacionadas AS (
            SELECT o.* FROM ordenes o
            INNER JOIN usuarios_muestreados u ON o.id_usuario = u.id
        ),
        transacciones_relacionadas AS (
            SELECT t.* FROM transacciones t
            INNER JOIN ordenes_relacionadas o ON t.id_orden = o.id
        )
        SELECT
            json_build_object(
                'usuarios', (SELECT json_agg(u) FROM usuarios_muestreados u),
                'ordenes', (SELECT json_agg(o) FROM ordenes_relacionadas o),
                'transacciones', (SELECT json_agg(t) FROM transacciones_relacionadas t)
            ) as dataset
        """

        cursor = conn.cursor()
        cursor.execute(query)
        resultado = cursor.fetchone()[0]

        conn.close()
        return resultado

    def _generar_datos_sinteticos(self, solicitud: SolicitudDatosPrueba) -> Dict:
        """Generar datos de prueba sintéticos"""
        Faker.seed(solicitud.valor_semilla)
        random.seed(solicitud.valor_semilla)

        config_dataset = self._obtener_config_dataset(solicitud.tamano_dataset)

        usuarios = []
        for _ in range(config_dataset['cantidad_usuarios']):
            usuario = {
                'id': self.faker.uuid4(),
                'nombre': self.faker.name(),
                'email': self.faker.email(),
                'telefono': self.faker.phone_number(),
                'direccion': {
                    'calle': self.faker.street_address(),
                    'ciudad': self.faker.city(),
                    'provincia': self.faker.state(),
                    'codigo_postal': self.faker.postcode()
                },
                'fecha_creacion': self.faker.date_time_between(
                    start_date='-1y',
                    end_date='now'
                ).isoformat()
            }
            usuarios.append(usuario)

        ordenes = []
        for usuario in usuarios[:int(len(usuarios) * 0.7)]:  # 70% de usuarios tienen órdenes
            cantidad_ordenes = random.randint(1, 5)
            for _ in range(cantidad_ordenes):
                orden = {
                    'id': self.faker.uuid4(),
                    'id_usuario': usuario['id'],
                    'total': round(random.uniform(10, 500), 2),
                    'estado': random.choice(['pendiente', 'completado', 'cancelado']),
                    'fecha_creacion': self.faker.date_time_between(
                        start_date=usuario['fecha_creacion'],
                        end_date='now'
                    ).isoformat()
                }
                ordenes.append(orden)

        return {
            'usuarios': usuarios,
            'ordenes': ordenes,
            'generado_en': datetime.utcnow().isoformat(),
            'semilla': solicitud.valor_semilla
        }

# masking-engine/masker.py
import re
import hashlib
import secrets
from typing import Any, Dict, List
from cryptography.fernet import Fernet
from datetime import datetime

class MotorEnmascaramiento:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.fernet = Fernet(config['clave_encriptacion'].encode())
        self.boveda_tokens = {}

    def enmascarar_dataset(self, datos: Dict, reglas: List[Dict]) -> Dict:
        """Aplicar reglas de enmascaramiento al dataset"""
        datos_enmascarados = {}

        for nombre_tabla, registros in datos.items():
            registros_enmascarados = []
            for registro in registros:
                registro_enmascarado = self._enmascarar_registro(registro, reglas)
                registros_enmascarados.append(registro_enmascarado)
            datos_enmascarados[nombre_tabla] = registros_enmascarados

        return datos_enmascarados

    def _enmascarar_registro(self, registro: Dict, reglas: List[Dict]) -> Dict:
        """Aplicar reglas de enmascaramiento a registro individual"""
        enmascarado = registro.copy()

        for campo, valor in registro.items():
            for regla in reglas:
                if self._campo_coincide_regla(campo, valor, regla):
                    enmascarado[campo] = self._aplicar_tecnica_enmascaramiento(
                        valor,
                        regla['tecnica'],
                        regla.get('parametros', {})
                    )
                    break

        return enmascarado

    def _aplicar_tecnica_enmascaramiento(self, valor: Any, tecnica: str, params: Dict) -> Any:
        """Aplicar técnica específica de enmascaramiento"""
        if tecnica == 'hash':
            sal = params.get('sal', 'sal-predeterminada')
            return hashlib.sha256(f"{valor}{sal}".encode()).hexdigest()[:len(str(valor))]

        elif tecnica == 'tokenizar':
            if valor not in self.boveda_tokens:
                token = secrets.token_urlsafe(32)
                self.boveda_tokens[valor] = token
                self._persistir_mapeo_token(valor, token)
            return self.boveda_tokens[valor]

        elif tecnica == 'formato_preservado':
            return self._encriptacion_formato_preservado(valor, params)

        elif tecnica == 'parcial':
            char_mascara = params.get('char_mascara', '*')
            chars_visibles = params.get('chars_visibles', 4)
            if len(str(valor)) > chars_visibles:
                return str(valor)[:chars_visibles] + char_mascara * (len(str(valor)) - chars_visibles)
            return valor

        elif tecnica == 'mezclar':
            import random
            chars = list(str(valor))
            random.shuffle(chars)
            return ''.join(chars)

        elif tecnica == 'desplazar_fecha':
            dias_desplazamiento = params.get('dias_desplazamiento', 30)
            if isinstance(valor, str):
                dt = datetime.fromisoformat(valor)
                desplazado = dt.timestamp() + (dias_desplazamiento * 86400)
                return datetime.fromtimestamp(desplazado).isoformat()
            return valor

        elif tecnica == 'censurar':
            return params.get('reemplazo', '[CENSURADO]')

        return valor

    def _encriptacion_formato_preservado(self, valor: str, params: Dict) -> str:
        """Implementar encriptación preservando formato"""
        # Preservar formato mientras se encripta
        if re.match(r'\d{4}-\d{4}-\d{4}-\d{4}', valor):  # Tarjeta de crédito
            encriptado = self.fernet.encrypt(valor.encode()).decode()
            # Generar salida preservando formato
            hash_val = hashlib.md5(encriptado.encode()).hexdigest()
            return f"{hash_val[:4]}-{hash_val[4:8]}-{hash_val[8:12]}-{hash_val[12:16]}"

        elif re.match(r'\d{3}-\d{2}-\d{4}', valor):  # SSN
            encriptado = self.fernet.encrypt(valor.encode()).decode()
            hash_val = hashlib.md5(encriptado.encode()).hexdigest()
            nums = ''.join(filter(str.isdigit, hash_val))[:9]
            return f"{nums[:3]}-{nums[3:5]}-{nums[5:9]}"

        return valor

# gdpr-compliance/pipeline.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: gdpr-compliance-rules
data:
  compliance-config.yaml: |
    gdpr:
      categorias_datos:
        datos_personales:
          - nombre
          - email
          - telefono
          - direccion
          - direccion_ip
        categorias_especiales:
          - raza_etnia
          - opiniones_politicas
          - creencias_religiosas
          - datos_salud
          - datos_biometricos

      reglas_procesamiento:
        derecho_supresion:
          periodo_retencion: 30
          metodo_eliminacion: "borrado_seguro"

        minimizacion_datos:
          campos_a_excluir:
            - metadata_innecesaria
            - ids_internos
            - timestamps_sistema

        pseudonimizacion:
          tecnica: "tokenizacion"
          reversible: true
          almacenamiento_claves: "hsm"

      requisitos_datos_prueba:
        simulacion_consentimiento: true
        registro_auditoria: obligatorio
        encriptacion_reposo: requerido
        transferencia_transfronteriza: prohibido

# synthetic-generator/generator.py
import numpy as np
import pandas as pd
from sklearn.preprocessing import StandardScaler
from sklearn.decomposition import PCA
from scipy import stats
import tensorflow as tf
from typing import Tuple, Dict, Any

class GeneradorDatosSinteticos:
    def __init__(self, ruta_datos_origen: str, config: Dict[str, Any]):
        self.datos_origen = pd.read_csv(ruta_datos_origen)
        self.config = config
        self.propiedades_estadisticas = {}

    def analizar_distribucion_origen(self) -> Dict[str, Any]:
        """Analizar propiedades estadísticas de datos origen"""
        propiedades = {}

        for columna in self.datos_origen.columns:
            datos_col = self.datos_origen[columna]

            if pd.api.types.is_numeric_dtype(datos_col):
                propiedades[columna] = {
                    'tipo': 'numerico',
                    'media': datos_col.mean(),
                    'desv_std': datos_col.std(),
                    'min': datos_col.min(),
                    'max': datos_col.max(),
                    'distribucion': self._ajustar_distribucion(datos_col),
                    'correlaciones': self._calcular_correlaciones(columna)
                }
            elif pd.api.types.is_categorical_dtype(datos_col) or datos_col.dtype == 'object':
                propiedades[columna] = {
                    'tipo': 'categorico',
                    'categorias': datos_col.value_counts().to_dict(),
                    'probabilidades': datos_col.value_counts(normalize=True).to_dict()
                }
            elif pd.api.types.is_datetime64_any_dtype(datos_col):
                propiedades[columna] = {
                    'tipo': 'fecha',
                    'min': datos_col.min(),
                    'max': datos_col.max(),
                    'frecuencia': pd.infer_freq(datos_col)
                }

        self.propiedades_estadisticas = propiedades
        return propiedades

    def generar_dataset_sintetico(self, num_registros: int) -> pd.DataFrame:
        """Generar dataset sintético manteniendo propiedades estadísticas"""
        if not self.propiedades_estadisticas:
            self.analizar_distribucion_origen()

        datos_sinteticos = {}

        # Generar columnas base
        for columna, props in self.propiedades_estadisticas.items():
            if props['tipo'] == 'numerico':
                datos_sinteticos[columna] = self._generar_columna_numerica(
                    props, num_registros
                )
            elif props['tipo'] == 'categorico':
                datos_sinteticos[columna] = self._generar_columna_categorica(
                    props, num_registros
                )
            elif props['tipo'] == 'fecha':
                datos_sinteticos[columna] = self._generar_columna_fecha(
                    props, num_registros
                )

        df = pd.DataFrame(datos_sinteticos)

        # Aplicar correlaciones
        df = self._aplicar_correlaciones(df)

        # Validar similitud estadística
        resultados_validacion = self._validar_datos_sinteticos(df)

        return df

    def _generar_columna_numerica(self, props: Dict, num_registros: int) -> np.ndarray:
        """Generar columna numérica basada en distribución"""
        nombre_dist = props['distribucion']['nombre']
        params_dist = props['distribucion']['parametros']

        if nombre_dist == 'normal':
            datos = np.random.normal(
                params_dist['loc'],
                params_dist['escala'],
                num_registros
            )
        elif nombre_dist == 'exponencial':
            datos = np.random.exponential(
                params_dist['escala'],
                num_registros
            )
        elif nombre_dist == 'uniforme':
            datos = np.random.uniform(
                props['min'],
                props['max'],
                num_registros
            )
        else:
            # Respaldo a distribución normal
            datos = np.random.normal(
                props['media'],
                props['desv_std'],
                num_registros
            )

        # Ajustar a límites originales
        datos = np.clip(datos, props['min'], props['max'])

        return datos

    def _ajustar_distribucion(self, datos: pd.Series) -> Dict[str, Any]:
        """Ajustar distribución estadística a datos"""
        distribuciones = ['normal', 'exponencial', 'uniforme', 'gamma', 'beta']
        mejor_dist = None
        mejores_params = None
        mejor_ks_stat = float('inf')

        for nombre_dist in distribuciones:
            try:
                dist = getattr(stats, nombre_dist)
                params = dist.fit(datos.dropna())
                ks_stat, _ = stats.kstest(datos.dropna(), lambda x: dist.cdf(x, *params))

                if ks_stat < mejor_ks_stat:
                    mejor_ks_stat = ks_stat
                    mejor_dist = nombre_dist
                    mejores_params = params
            except:
                continue

        return {
            'nombre': mejor_dist,
            'parametros': dict(zip(['loc', 'escala'], mejores_params[:2])),
            'estadistico_ks': mejor_ks_stat
        }

# gan-generator/tabular_gan.py
import tensorflow as tf
from tensorflow import keras
from tensorflow.keras import layers
import numpy as np

class GANTabular:
    def __init__(self, dim_entrada: int, dim_latente: int = 100):
        self.dim_entrada = dim_entrada
        self.dim_latente = dim_latente
        self.generador = self._construir_generador()
        self.discriminador = self._construir_discriminador()
        self.gan = self._construir_gan()

    def _construir_generador(self) -> keras.Model:
        """Construir red generadora"""
        modelo = keras.Sequential([
            layers.Dense(256, activation='relu', input_dim=self.dim_latente),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(512, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(1024, activation='relu'),
            layers.BatchNormalization(),
            layers.Dropout(0.3),
            layers.Dense(self.dim_entrada, activation='tanh')
        ])
        return modelo

    def _construir_discriminador(self) -> keras.Model:
        """Construir red discriminadora"""
        modelo = keras.Sequential([
            layers.Dense(1024, activation='relu', input_dim=self.dim_entrada),
            layers.Dropout(0.3),
            layers.Dense(512, activation='relu'),
            layers.Dropout(0.3),
            layers.Dense(256, activation='relu'),
            layers.Dropout(0.3),
            layers.Dense(1, activation='sigmoid')
        ])

        modelo.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        return modelo

    def _construir_gan(self) -> keras.Model:
        """Combinar generador y discriminador"""
        self.discriminador.trainable = False

        entrada_gan = keras.Input(shape=(self.dim_latente,))
        generado = self.generador(entrada_gan)
        salida_gan = self.discriminador(generado)

        modelo = keras.Model(entrada_gan, salida_gan)
        modelo.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy'
        )
        return modelo

    def entrenar(self, datos_reales: np.ndarray, epocas: int = 1000, tam_lote: int = 32):
        """Entrenar la GAN"""
        for epoca in range(epocas):
            # Entrenar discriminador
            idx = np.random.randint(0, datos_reales.shape[0], tam_lote)
            lote_real = datos_reales[idx]

            ruido = np.random.normal(0, 1, (tam_lote, self.dim_latente))
            lote_generado = self.generador.predict(ruido, verbose=0)

            perdida_d_real = self.discriminador.train_on_batch(
                lote_real,
                np.ones((tam_lote, 1))
            )
            perdida_d_falso = self.discriminador.train_on_batch(
                lote_generado,
                np.zeros((tam_lote, 1))
            )
            perdida_d = 0.5 * np.add(perdida_d_real, perdida_d_falso)

            # Entrenar generador
            ruido = np.random.normal(0, 1, (tam_lote, self.dim_latente))
            perdida_g = self.gan.train_on_batch(
                ruido,
                np.ones((tam_lote, 1))
            )

            if epoca % 100 == 0:
                print(f"Época {epoca}, Pérdida D: {perdida_d[0]:.4f}, Pérdida G: {perdida_g:.4f}")

    def generar_datos_sinteticos(self, num_muestras: int) -> np.ndarray:
        """Generar muestras sintéticas"""
        ruido = np.random.normal(0, 1, (num_muestras, self.dim_latente))
        return self.generador.predict(ruido)

// Jenkinsfile
@Library('test-data-lib') _

pipeline {
    agent any

    parameters {
        choice(name: 'ENTORNO',
               choices: ['dev', 'qa', 'staging', 'performance'],
               description: 'Entorno objetivo')
        choice(name: 'FUENTE_DATOS',
               choices: ['subconjunto-produccion', 'sintetico', 'hibrido'],
               description: 'Estrategia de fuente de datos')
        choice(name: 'TAMAÑO_DATASET',
               choices: ['pequeño', 'mediano', 'grande', 'xlarge'],
               description: 'Tamaño del dataset')
        multiChoice(name: 'REQUISITOS_CUMPLIMIENTO',
                   choices: ['GDPR', 'CCPA', 'HIPAA', 'PCI-DSS'],
                   description: 'Requisitos de cumplimiento')
    }

    environment {
        VAULT_ADDR = 'https://vault.ejemplo.com'
        LAGO_DATOS_BUCKET = 's3://lago-datos-prueba'
    }

    stages {
        stage('Inicializar Pipeline Datos Prueba') {
            steps {
                script {
                    // Inicializar configuración
                    configDatosPrueba = [
                        entorno: params.ENTORNO,
                        fuenteDatos: params.FUENTE_DATOS,
                        tamañoDataset: params.TAMAÑO_DATASET,
                        cumplimiento: params.REQUISITOS_CUMPLIMIENTO,
                        timestamp: new Date().format('yyyy-MM-dd-HH-mm-ss')
                    ]

                    // Generar ID único del dataset
                    configDatosPrueba.datasetId = generarIdDataset(configDatosPrueba)
                }
            }
        }

        stage('Obtener Datos Origen') {
            when {
                expression { params.FUENTE_DATOS != 'sintetico' }
            }
            steps {
                script {
                    withCredentials([
                        usernamePassword(
                            credentialsId: 'bd-prod-solo-lectura',
                            usernameVariable: 'DB_USUARIO',
                            passwordVariable: 'DB_PASS'
                        )
                    ]) {
                        sh """
                            python3 scripts/extraer_subconjunto_produccion.py \
                                --host ${PROD_DB_HOST} \
                                --usuario ${DB_USUARIO} \
                                --password ${DB_PASS} \
                                --tamaño ${params.TAMAÑO_DATASET} \
                                --salida /tmp/datos-crudos.json
                        """
                    }
                }
            }
        }

        stage('Generar Datos Sintéticos') {
            when {
                expression { params.FUENTE_DATOS in ['sintetico', 'hibrido'] }
            }
            steps {
                script {
                    sh """
                        python3 scripts/generar_datos_sinteticos.py \
                            --config configs/sintetico-${params.ENTORNO}.yaml \
                            --tamaño ${params.TAMAÑO_DATASET} \
                            --semilla ${BUILD_NUMBER} \
                            --salida /tmp/datos-sinteticos.json
                    """
                }
            }
        }

        stage('Aplicar Enmascaramiento') {
            steps {
                script {
                    def reglasEnmascaramiento = cargarReglasEnmascaramiento(params.REQUISITOS_CUMPLIMIENTO)

                    sh """
                        python3 scripts/aplicar_enmascaramiento_datos.py \
                            --entrada /tmp/datos-crudos.json \
                            --reglas ${reglasEnmascaramiento} \
                            --cumplimiento ${params.REQUISITOS_CUMPLIMIENTO.join(',')} \
                            --salida /tmp/datos-enmascarados.json
                    """
                }
            }
        }

        stage('Validar Calidad de Datos') {
            steps {
                script {
                    sh """
                        python3 scripts/validar_datos_prueba.py \
                            --datos /tmp/datos-enmascarados.json \
                            --esquema esquemas/${params.ENTORNO}-esquema.json \
                            --reglas reglas-validacion.yaml \
                            --reporte /tmp/reporte-validacion.html
                    """

                    publishHTML(target: [
                        allowMissing: false,
                        alwaysLinkToLastBuild: true,
                        keepAll: true,
                        reportDir: '/tmp',
                        reportFiles: 'reporte-validacion.html',
                        reportName: 'Reporte Validación Datos'
                    ])
                }
            }
        }

        stage('Versionar y Almacenar Dataset') {
            steps {
                script {
                    sh """
                        # Comprimir y encriptar dataset
                        tar -czf /tmp/dataset-${configDatosPrueba.datasetId}.tar.gz \
                            /tmp/datos-enmascarados.json

                        # Subir a S3 con versionado
                        aws s3 cp /tmp/dataset-${configDatosPrueba.datasetId}.tar.gz \
                            ${LAGO_DATOS_BUCKET}/${params.ENTORNO}/ \
                            --server-side-encryption aws:kms \
                            --metadata "build=${BUILD_NUMBER},entorno=${params.ENTORNO}"

                        # Registrar en catálogo de datos
                        python3 scripts/registrar_dataset.py \
                            --dataset-id ${configDatosPrueba.datasetId} \
                            --ubicacion ${LAGO_DATOS_BUCKET}/${params.ENTORNO}/ \
                            --metadata '${groovy.json.JsonOutput.toJson(configDatosPrueba)}'
                    """
                }
            }
        }

        stage('Desplegar a Entorno de Prueba') {
            steps {
                script {
                    parallel(
                        'Base de Datos': {
                            sh """
                                python3 scripts/cargar_a_base_datos.py \
                                    --dataset /tmp/datos-enmascarados.json \
                                    --objetivo ${params.ENTORNO} \
                                    --cadena-conexion \${${params.ENTORNO.toUpperCase()}_DB_URL}
                            """
                        },
                        'Cache': {
                            sh """
                                python3 scripts/cargar_a_cache.py \
                                    --dataset /tmp/datos-enmascarados.json \
                                    --redis-host \${${params.ENTORNO.toUpperCase()}_REDIS_HOST} \
                                    --ttl 3600
                            """
                        },
                        'Sistema de Archivos': {
                            sh """
                                kubectl cp /tmp/datos-enmascarados.json \
                                    ${params.ENTORNO}/pod-datos-prueba:/data/datos-prueba.json
                            """
                        }
                    )
                }
            }
        }

        stage('Ejecutar Pruebas de Verificación') {
            steps {
                script {
                    sh """
                        pytest tests/verificacion_datos/ \
                            --entorno ${params.ENTORNO} \
                            --dataset-id ${configDatosPrueba.datasetId} \
                            --junitxml=resultados-pruebas.xml
                    """
                }
            }
        }
    }

    post {
        always {
            junit 'resultados-pruebas.xml'
            cleanWs()
        }
        success {
            emailext(
                subject: "Datos de Prueba Aprovisionados: ${configDatosPrueba.datasetId}",
                body: """
                    Datos de prueba aprovisionados exitosamente para ${params.ENTORNO}

                    ID Dataset: ${configDatosPrueba.datasetId}
                    Tamaño: ${params.TAMAÑO_DATASET}
                    Fuente: ${params.FUENTE_DATOS}
                    Cumplimiento: ${params.REQUISITOS_CUMPLIMIENTO}

                    Acceder al dataset:
                    ${env.BUILD_URL}
                """,
                to: 'equipo-qa@ejemplo.com'
            )
        }
        failure {
            sh """
                # Revertir cualquier despliegue parcial
                python3 scripts/revertir_datos_prueba.py \
                    --entorno ${params.ENTORNO} \
                    --dataset-id ${configDatosPrueba.datasetId}
            """
        }
    }
}

# .gitlab-ci.yml
stages:
  - preparar
  - extraer
  - transformar
  - cargar
  - validar

variables:
  VERSION_PIPELINE_DATOS: "2.1.0"
  IMAGEN_PYTHON: "python:3.9-slim"

.plantilla-datos-prueba:
  image: ${IMAGEN_PYTHON}
  before_script:
    - pip install -r requirements.txt
    - export DATASET_ID=$(date +%Y%m%d-%H%M%S)-${CI_PIPELINE_ID}

preparar:entorno:
  stage: preparar
  script:
    - echo "Preparando entorno de datos de prueba para ${CI_ENVIRONMENT_NAME}"
    - |
      python3 scripts/preparar_entorno.py \
        --entorno ${CI_ENVIRONMENT_NAME} \
        --limpiar-anterior ${LIMPIAR_DATOS_ANTERIORES}

extraer:subconjunto-produccion:
  stage: extraer
  only:
    variables:
      - $FUENTE_DATOS == "produccion"
  script:
    - |
      python3 scripts/extraer_subconjunto.py \
        --origen ${URL_BD_PRODUCCION} \
        --tamaño ${TAMAÑO_DATASET} \
        --filtros config/filtros-extraccion.yaml \
        --salida artefactos/datos-crudos.json
  artifacts:
    paths:
      - artefactos/datos-crudos.json
    expire_in: 1 hour

generar:datos-sinteticos:
  stage: extraer
  only:
    variables:
      - $FUENTE_DATOS == "sintetico"
  script:
    - |
      python3 scripts/generador_sintetico.py \
        --modelo modelos/modelo-generacion-datos.pkl \
        --tamaño ${TAMAÑO_DATASET} \
        --config config/config-sintetico.yaml \
        --salida artefactos/datos-sinteticos.json
  artifacts:
    paths:
      - artefactos/datos-sinteticos.json
    expire_in: 1 hour

transformar:aplicar-enmascaramiento:
  stage: transformar
  dependencies:
    - extraer:subconjunto-produccion
    - generar:datos-sinteticos
  script:
    - |
      python3 scripts/enmascarador_datos.py \
        --entrada artefactos/*.json \
        --reglas config/reglas-enmascaramiento-${NIVEL_CUMPLIMIENTO}.yaml \
        --salida artefactos/datos-enmascarados.json \
        --log-auditoria artefactos/auditoria-enmascaramiento.log
  artifacts:
    paths:
      - artefactos/datos-enmascarados.json
      - artefactos/auditoria-enmascaramiento.log
    expire_in: 1 day

cargar:a-base-datos:
  stage: cargar
  dependencies:
    - transformar:aplicar-enmascaramiento
  script:
    - |
      python3 scripts/cargador_base_datos.py \
        --datos artefactos/datos-enmascarados.json \
        --objetivo ${CI_ENVIRONMENT_NAME} \
        --conexion ${CADENA_CONEXION_BD_PRUEBA} \
        --tamaño-lote 1000

validar:calidad-datos:
  stage: validar
  dependencies:
    - cargar:a-base-datos
  script:
    - |
      python3 scripts/validador_datos.py \
        --entorno ${CI_ENVIRONMENT_NAME} \
        --verificaciones config/verificaciones-validacion.yaml \
        --reporte artefactos/reporte-validacion.html
  artifacts:
    reports:
      junit: artefactos/resultados-validacion.xml
    paths:
      - artefactos/reporte-validacion.html
    expire_in: 30 days

# security/data-security-framework.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: test-data-security
data:
  security-controls.yaml: |
    encriptacion:
      en_reposo:
        algoritmo: "AES-256-GCM"
        rotacion_claves: "30-dias"
        almacenamiento_claves: "HSM"

      en_transito:
        protocolo: "TLS 1.3"
        validacion_certificado: "requerido"
        tls_mutuo: "habilitado"

    control_acceso:
      autenticacion:
        metodo: "oauth2"
        mfa_requerido: true
        timeout_sesion: "30-minutos"

      autorizacion:
        modelo: "RBAC"
        roles:
          - nombre: "admin-datos-prueba"
            permisos: ["crear", "leer", "actualizar", "eliminar", "enmascarar"]
          - nombre: "usuario-datos-prueba"
            permisos: ["leer"]
          - nombre: "auditor-cumplimiento"
            permisos: ["leer", "auditar"]

      registro_auditoria:
        habilitado: true
        retencion: "7-años"
        almacenamiento_inmutable: true
        eventos:
          - acceso_datos
          - modificacion_datos
          - operaciones_enmascaramiento
          - violaciones_cumplimiento

    ciclo_vida_datos:
      retencion:
        datos_prueba: "30-dias"
        datos_enmascarados: "90-dias"
        datos_sinteticos: "ilimitado"

      eliminacion:
        metodo: "borrado-criptografico"
        verificacion: "requerido"
        generacion_certificado: true

# compliance-monitoring/dashboard.py
from flask import Flask, jsonify, render_template
from datetime import datetime, timedelta
import psycopg2
import redis

app = Flask(__name__)

class MonitorCumplimiento:
    def __init__(self):
        self.db = psycopg2.connect(
            host="localhost",
            database="bd_cumplimiento",
            user="monitor",
            password="contraseña_segura"
        )
        self.cache = redis.Redis(host='localhost', port=6379)

    def obtener_metricas_cumplimiento(self) -> dict:
        """Recolectar métricas de cumplimiento"""
        metricas = {
            'gdpr': self._verificar_cumplimiento_gdpr(),
            'pci_dss': self._verificar_cumplimiento_pci(),
            'hipaa': self._verificar_cumplimiento_hipaa(),
            'calidad_datos': self._verificar_calidad_datos(),
            'ultima_actualizacion': datetime.utcnow().isoformat()
        }
        return metricas

    def _verificar_cumplimiento_gdpr(self) -> dict:
        cursor = self.db.cursor()
        cursor.execute("""
            SELECT
                COUNT(*) FILTER (WHERE esta_enmascarado = true) as registros_enmascarados,
                COUNT(*) as total_registros,
                COUNT(DISTINCT id_dataset) as datasets,
                MAX(fecha_creacion) as ultimo_procesamiento
            FROM registros_datos_prueba
            WHERE contiene_pii = true
        """)
        resultado = cursor.fetchone()

        return {
            'cumple': resultado[0] == resultado[1],
            'porcentaje_enmascarado': (resultado[0] / resultado[1] * 100) if resultado[1] > 0 else 100,
            'total_datasets': resultado[2],
            'ultimo_procesamiento': resultado[3].isoformat() if resultado[3] else None
        }

@app.route('/api/cumplimiento/estado')
def estado_cumplimiento():
    monitor = MonitorCumplimiento()
    return jsonify(monitor.obtener_metricas_cumplimiento())

@app.route('/api/cumplimiento/rastro-auditoria/<id_dataset>')
def rastro_auditoria(id_dataset):
    monitor = MonitorCumplimiento()
    rastro = monitor.obtener_rastro_auditoria(id_dataset)
    return jsonify(rastro)

if __name__ == '__main__':
    app.run(debug=False, port=5000)