Управление тестовыми данными в DevOps-пайплайнах: синхронизация, маскирование и версионирование

# data-classification-schema.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: data-classification
data:
  classification-rules.json: |
    {
      "classifications": {
        "PII": {
          "level": "чувствительный",
          "patterns": [
            "\\b[А-Я]{1}[а-я]+\\s[А-Я]{1}[а-я]+\\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": ["имя", "email", "снилс", "телефон", "адрес"],
          "handling": {
            "storage": "зашифрованный",
            "transit": "требуется-tls",
            "retention": "30-дней",
            "masking": "обязательно"
          }
        },
        "PHI": {
          "level": "высокочувствительный",
          "fields": ["диагноз", "медицинская_карта", "рецепт"],
          "compliance": ["HIPAA"],
          "handling": {
            "storage": "шифрование-в-покое",
            "access": "с-аудитом",
            "masking": "токенизация"
          }
        },
        "Финансовый": {
          "level": "чувствительный",
          "patterns": [
            "\\b\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}[\\s-]?\\d{4}\\b",
            "\\b\\d{3,4}\\b"
          ],
          "fields": ["кредитная_карта", "банковский_счет", "номер_маршрутизации"],
          "compliance": ["PCI-DSS"],
          "handling": {
            "storage": "токенизированный",
            "masking": "шифрование-с-сохранением-формата"
          }
        }
      }
    }

# 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 ЗапросТестовыхДанных:
    окружение: str
    размер_набора: str  # маленький, средний, большой
    свежесть_данных: str  # реальное-время, ежедневно, еженедельно
    требования_соответствия: List[str]
    начальное_значение: int

class ПоставщикТестовыхДанных:
    def __init__(self, путь_конфигурации: str):
        with open(путь_конфигурации, 'r') as f:
            self.config = json.load(f)
        self.faker = Faker('ru_RU')
        self.s3 = boto3.client('s3')

    def подготовить_набор_данных(self, запрос: ЗапросТестовыхДанных) -> Dict[str, Any]:
        """Основной рабочий процесс подготовки"""
        набор_данных = {
            'metadata': self._создать_метаданные(запрос),
            'data': {}
        }

        # Определить источники данных на основе требований
        if 'реальное-время' in запрос.свежесть_данных:
            набор_данных['data'] = self._извлечь_подмножество_продакшена(запрос)
        else:
            набор_данных['data'] = self._создать_синтетические_данные(запрос)

        # Применить трансформации соответствия
        if запрос.требования_соответствия:
            набор_данных['data'] = self._применить_маскирование_соответствия(
                набор_данных['data'],
                запрос.требования_соответствия
            )

        # Версионировать и сохранить набор данных
        dataset_id = self._версионировать_набор_данных(набор_данных)

        # Развернуть в целевое окружение
        self._развернуть_в_окружении(dataset_id, запрос.окружение)

        return {
            'dataset_id': dataset_id,
            'окружение': запрос.окружение,
            'статус': 'подготовлен',
            'timestamp': datetime.utcnow().isoformat()
        }

    def _извлечь_подмножество_продакшена(self, запрос: ЗапросТестовыхДанных) -> Dict:
        """Извлечь и создать подмножество производственных данных"""
        conn = psycopg2.connect(
            host=self.config['production_db']['хост'],
            port=self.config['production_db']['порт'],
            database=self.config['production_db']['база_данных'],
            user=self.config['production_db']['пользователь'],
            password=self.config['production_db']['пароль']
        )

        конфиг_подмножества = self._получить_конфиг_подмножества(запрос.размер_набора)

        query = f"""
        WITH выборка_пользователей AS (
            SELECT * FROM пользователи
            TABLESAMPLE BERNOULLI ({конфиг_подмножества['процент_выборки']})
            WHERE дата_создания > NOW() - INTERVAL '{конфиг_подмножества['временное_окно']}'
            LIMIT {конфиг_подмножества['макс_записей']}
        ),
        связанные_заказы AS (
            SELECT o.* FROM заказы o
            INNER JOIN выборка_пользователей u ON o.id_пользователя = u.id
        ),
        связанные_транзакции AS (
            SELECT t.* FROM транзакции t
            INNER JOIN связанные_заказы o ON t.id_заказа = o.id
        )
        SELECT
            json_build_object(
                'пользователи', (SELECT json_agg(u) FROM выборка_пользователей u),
                'заказы', (SELECT json_agg(o) FROM связанные_заказы o),
                'транзакции', (SELECT json_agg(t) FROM связанные_транзакции t)
            ) as dataset
        """

        cursor = conn.cursor()
        cursor.execute(query)
        результат = cursor.fetchone()[0]

        conn.close()
        return результат

    def _создать_синтетические_данные(self, запрос: ЗапросТестовыхДанных) -> Dict:
        """Создать синтетические тестовые данные"""
        Faker.seed(запрос.начальное_значение)
        random.seed(запрос.начальное_значение)

        конфиг_набора = self._получить_конфиг_набора(запрос.размер_набора)

        пользователи = []
        for _ in range(конфиг_набора['количество_пользователей']):
            пользователь = {
                'id': self.faker.uuid4(),
                'имя': self.faker.name(),
                'email': self.faker.email(),
                'телефон': self.faker.phone_number(),
                'адрес': {
                    'улица': self.faker.street_address(),
                    'город': self.faker.city(),
                    'область': self.faker.region(),
                    'индекс': self.faker.postcode()
                },
                'дата_создания': self.faker.date_time_between(
                    start_date='-1y',
                    end_date='now'
                ).isoformat()
            }
            пользователи.append(пользователь)

        заказы = []
        for пользователь in пользователи[:int(len(пользователи) * 0.7)]:  # 70% пользователей имеют заказы
            количество_заказов = random.randint(1, 5)
            for _ in range(количество_заказов):
                заказ = {
                    'id': self.faker.uuid4(),
                    'id_пользователя': пользователь['id'],
                    'сумма': round(random.uniform(10, 500), 2),
                    'статус': random.choice(['ожидание', 'завершен', 'отменен']),
                    'дата_создания': self.faker.date_time_between(
                        start_date=пользователь['дата_создания'],
                        end_date='now'
                    ).isoformat()
                }
                заказы.append(заказ)

        return {
            'пользователи': пользователи,
            'заказы': заказы,
            'создано': datetime.utcnow().isoformat(),
            'начальное_значение': запрос.начальное_значение
        }

# 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 МеханизмМаскирования:
    def __init__(self, config: Dict[str, Any]):
        self.config = config
        self.fernet = Fernet(config['ключ_шифрования'].encode())
        self.хранилище_токенов = {}

    def маскировать_набор_данных(self, данные: Dict, правила: List[Dict]) -> Dict:
        """Применить правила маскирования к набору данных"""
        маскированные_данные = {}

        for имя_таблицы, записи in данные.items():
            маскированные_записи = []
            for запись in записи:
                маскированная_запись = self._маскировать_запись(запись, правила)
                маскированные_записи.append(маскированная_запись)
            маскированные_данные[имя_таблицы] = маскированные_записи

        return маскированные_данные

    def _маскировать_запись(self, запись: Dict, правила: List[Dict]) -> Dict:
        """Применить правила маскирования к отдельной записи"""
        маскированная = запись.copy()

        for поле, значение in запись.items():
            for правило in правила:
                if self._поле_соответствует_правилу(поле, значение, правило):
                    маскированная[поле] = self._применить_технику_маскирования(
                        значение,
                        правило['техника'],
                        правило.get('параметры', {})
                    )
                    break

        return маскированная

    def _применить_технику_маскирования(self, значение: Any, техника: str, params: Dict) -> Any:
        """Применить конкретную технику маскирования"""
        if техника == 'хеш':
            соль = params.get('соль', 'стандартная-соль')
            return hashlib.sha256(f"{значение}{соль}".encode()).hexdigest()[:len(str(значение))]

        elif техника == 'токенизация':
            if значение not in self.хранилище_токенов:
                токен = secrets.token_urlsafe(32)
                self.хранилище_токенов[значение] = токен
                self._сохранить_сопоставление_токенов(значение, токен)
            return self.хранилище_токенов[значение]

        elif техника == 'сохранение_формата':
            return self._шифрование_с_сохранением_формата(значение, params)

        elif техника == 'частичное':
            символ_маски = params.get('символ_маски', '*')
            видимые_символы = params.get('видимые_символы', 4)
            if len(str(значение)) > видимые_символы:
                return str(значение)[:видимые_символы] + символ_маски * (len(str(значение)) - видимые_символы)
            return значение

        elif техника == 'перемешать':
            import random
            символы = list(str(значение))
            random.shuffle(символы)
            return ''.join(символы)

        elif техника == 'сдвиг_даты':
            дни_сдвига = params.get('дни_сдвига', 30)
            if isinstance(значение, str):
                dt = datetime.fromisoformat(значение)
                сдвинуто = dt.timestamp() + (дни_сдвига * 86400)
                return datetime.fromtimestamp(сдвинуто).isoformat()
            return значение

        elif техника == 'редактирование':
            return params.get('замена', '[УДАЛЕНО]')

        return значение

    def _шифрование_с_сохранением_формата(self, значение: str, params: Dict) -> str:
        """Реализовать шифрование с сохранением формата"""
        # Сохранить формат при шифровании
        if re.match(r'\d{4}-\d{4}-\d{4}-\d{4}', значение):  # Кредитная карта
            зашифрованное = self.fernet.encrypt(значение.encode()).decode()
            # Генерировать вывод с сохранением формата
            хеш_знач = hashlib.md5(зашифрованное.encode()).hexdigest()
            return f"{хеш_знач[:4]}-{хеш_знач[4:8]}-{хеш_знач[8:12]}-{хеш_знач[12:16]}"

        elif re.match(r'\d{3}-\d{2}-\d{4}', значение):  # СНИЛС
            зашифрованное = self.fernet.encrypt(значение.encode()).decode()
            хеш_знач = hashlib.md5(зашифрованное.encode()).hexdigest()
            цифры = ''.join(filter(str.isdigit, хеш_знач))[:9]
            return f"{цифры[:3]}-{цифры[3:5]}-{цифры[5:9]}"

        return значение

# gdpr-compliance/pipeline.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: gdpr-compliance-rules
data:
  compliance-config.yaml: |
    gdpr:
      категории_данных:
        персональные_данные:
          - имя
          - email
          - телефон
          - адрес
          - ip_адрес
        специальные_категории:
          - раса_этническая_принадлежность
          - политические_взгляды
          - религиозные_убеждения
          - данные_о_здоровье
          - биометрические_данные

      правила_обработки:
        право_на_удаление:
          период_хранения: 30
          метод_удаления: "безопасное_стирание"

        минимизация_данных:
          поля_для_исключения:
            - ненужные_метаданные
            - внутренние_идентификаторы
            - системные_временные_метки

        псевдонимизация:
          техника: "токенизация"
          обратимая: true
          хранение_ключей: "hsm"

      требования_тестовых_данных:
        симуляция_согласия: true
        аудит_логирование: обязательно
        шифрование_в_покое: требуется
        трансграничная_передача: запрещена

# 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 ГенераторСинтетическихДанных:
    def __init__(self, путь_исходных_данных: str, config: Dict[str, Any]):
        self.исходные_данные = pd.read_csv(путь_исходных_данных)
        self.config = config
        self.статистические_свойства = {}

    def анализировать_распределение_источника(self) -> Dict[str, Any]:
        """Анализ статистических свойств исходных данных"""
        свойства = {}

        for колонка in self.исходные_данные.columns:
            данные_кол = self.исходные_данные[колонка]

            if pd.api.types.is_numeric_dtype(данные_кол):
                свойства[колонка] = {
                    'тип': 'числовой',
                    'среднее': данные_кол.mean(),
                    'стд_откл': данные_кол.std(),
                    'мин': данные_кол.min(),
                    'макс': данные_кол.max(),
                    'распределение': self._подобрать_распределение(данные_кол),
                    'корреляции': self._вычислить_корреляции(колонка)
                }
            elif pd.api.types.is_categorical_dtype(данные_кол) or данные_кол.dtype == 'object':
                свойства[колонка] = {
                    'тип': 'категориальный',
                    'категории': данные_кол.value_counts().to_dict(),
                    'вероятности': данные_кол.value_counts(normalize=True).to_dict()
                }
            elif pd.api.types.is_datetime64_any_dtype(данные_кол):
                свойства[колонка] = {
                    'тип': 'дата_время',
                    'мин': данные_кол.min(),
                    'макс': данные_кол.max(),
                    'частота': pd.infer_freq(данные_кол)
                }

        self.статистические_свойства = свойства
        return свойства

    def генерировать_синтетический_набор(self, кол_записей: int) -> pd.DataFrame:
        """Генерация синтетического набора с сохранением статистических свойств"""
        if not self.статистические_свойства:
            self.анализировать_распределение_источника()

        синтетические_данные = {}

        # Генерация базовых колонок
        for колонка, свойства in self.статистические_свойства.items():
            if свойства['тип'] == 'числовой':
                синтетические_данные[колонка] = self._генерировать_числовую_колонку(
                    свойства, кол_записей
                )
            elif свойства['тип'] == 'категориальный':
                синтетические_данные[колонка] = self._генерировать_категориальную_колонку(
                    свойства, кол_записей
                )
            elif свойства['тип'] == 'дата_время':
                синтетические_данные[колонка] = self._генерировать_колонку_дат(
                    свойства, кол_записей
                )

        df = pd.DataFrame(синтетические_данные)

        # Применение корреляций
        df = self._применить_корреляции(df)

        # Валидация статистического сходства
        результаты_валидации = self._валидировать_синтетические_данные(df)

        return df

    def _генерировать_числовую_колонку(self, свойства: Dict, кол_записей: int) -> np.ndarray:
        """Генерация числовой колонки на основе распределения"""
        имя_распр = свойства['распределение']['название']
        параметры_распр = свойства['распределение']['параметры']

        if имя_распр == 'нормальное':
            данные = np.random.normal(
                параметры_распр['loc'],
                параметры_распр['масштаб'],
                кол_записей
            )
        elif имя_распр == 'экспоненциальное':
            данные = np.random.exponential(
                параметры_распр['масштаб'],
                кол_записей
            )
        elif имя_распр == 'равномерное':
            данные = np.random.uniform(
                свойства['мин'],
                свойства['макс'],
                кол_записей
            )
        else:
            # Откат к нормальному распределению
            данные = np.random.normal(
                свойства['среднее'],
                свойства['стд_откл'],
                кол_записей
            )

        # Обрезка до исходных границ
        данные = np.clip(данные, свойства['мин'], свойства['макс'])

        return данные

    def _подобрать_распределение(self, данные: pd.Series) -> Dict[str, Any]:
        """Подобрать статистическое распределение к данным"""
        распределения = ['norm', 'expon', 'uniform', 'gamma', 'beta']
        лучшее_распр = None
        лучшие_параметры = None
        лучшая_ks_стат = float('inf')

        for имя_распр in распределения:
            try:
                распр = getattr(stats, имя_распр)
                параметры = распр.fit(данные.dropna())
                ks_стат, _ = stats.kstest(данные.dropna(), lambda x: распр.cdf(x, *параметры))

                if ks_стат < лучшая_ks_стат:
                    лучшая_ks_стат = ks_стат
                    лучшее_распр = имя_распр
                    лучшие_параметры = параметры
            except:
                continue

        return {
            'название': лучшее_распр,
            'параметры': dict(zip(['loc', 'масштаб'], лучшие_параметры[:2])),
            'ks_статистика': лучшая_ks_стат
        }

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

class ТабличнаяGAN:
    def __init__(self, размерность_входа: int, латентная_размерность: int = 100):
        self.размерность_входа = размерность_входа
        self.латентная_размерность = латентная_размерность
        self.генератор = self._построить_генератор()
        self.дискриминатор = self._построить_дискриминатор()
        self.gan = self._построить_gan()

    def _построить_генератор(self) -> keras.Model:
        """Построение сети генератора"""
        модель = keras.Sequential([
            layers.Dense(256, activation='relu', input_dim=self.латентная_размерность),
            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.размерность_входа, activation='tanh')
        ])
        return модель

    def _построить_дискриминатор(self) -> keras.Model:
        """Построение сети дискриминатора"""
        модель = keras.Sequential([
            layers.Dense(1024, activation='relu', input_dim=self.размерность_входа),
            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')
        ])

        модель.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy',
            metrics=['accuracy']
        )
        return модель

    def _построить_gan(self) -> keras.Model:
        """Объединение генератора и дискриминатора"""
        self.дискриминатор.trainable = False

        вход_gan = keras.Input(shape=(self.латентная_размерность,))
        сгенерированное = self.генератор(вход_gan)
        выход_gan = self.дискриминатор(сгенерированное)

        модель = keras.Model(вход_gan, выход_gan)
        модель.compile(
            optimizer=keras.optimizers.Adam(learning_rate=0.0002, beta_1=0.5),
            loss='binary_crossentropy'
        )
        return модель

    def обучить(self, реальные_данные: np.ndarray, эпохи: int = 1000, размер_батча: int = 32):
        """Обучение GAN"""
        for эпоха in range(эпохи):
            # Обучение дискриминатора
            idx = np.random.randint(0, реальные_данные.shape[0], размер_батча)
            реальный_батч = реальные_данные[idx]

            шум = np.random.normal(0, 1, (размер_батча, self.латентная_размерность))
            сгенерированный_батч = self.генератор.predict(шум, verbose=0)

            потеря_д_реальная = self.дискриминатор.train_on_batch(
                реальный_батч,
                np.ones((размер_батча, 1))
            )
            потеря_д_фальшивая = self.дискриминатор.train_on_batch(
                сгенерированный_батч,
                np.zeros((размер_батча, 1))
            )
            потеря_д = 0.5 * np.add(потеря_д_реальная, потеря_д_фальшивая)

            # Обучение генератора
            шум = np.random.normal(0, 1, (размер_батча, self.латентная_размерность))
            потеря_г = self.gan.train_on_batch(
                шум,
                np.ones((размер_батча, 1))
            )

            if эпоха % 100 == 0:
                print(f"Эпоха {эпоха}, Потеря Д: {потеря_д[0]:.4f}, Потеря Г: {потеря_г:.4f}")

    def генерировать_синтетические_данные(self, кол_образцов: int) -> np.ndarray:
        """Генерация синтетических образцов"""
        шум = np.random.normal(0, 1, (кол_образцов, self.латентная_размерность))
        return self.генератор.predict(шум)

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

pipeline {
    agent any

    parameters {
        choice(name: 'ОКРУЖЕНИЕ',
               choices: ['dev', 'qa', 'staging', 'performance'],
               description: 'Целевое окружение')
        choice(name: 'ИСТОЧНИК_ДАННЫХ',
               choices: ['подмножество-продакшн', 'синтетический', 'гибрид'],
               description: 'Стратегия источника данных')
        choice(name: 'РАЗМЕР_НАБОРА',
               choices: ['маленький', 'средний', 'большой', 'xlarge'],
               description: 'Размер набора данных')
        multiChoice(name: 'ТРЕБОВАНИЯ_СООТВЕТСТВИЯ',
                   choices: ['GDPR', 'CCPA', 'HIPAA', 'PCI-DSS'],
                   description: 'Требования соответствия')
    }

    environment {
        VAULT_ADDR = 'https://vault.example.com'
        DATA_LAKE_BUCKET = 's3://озеро-тестовых-данных'
    }

    stages {
        stage('Инициализация пайплайна тестовых данных') {
            steps {
                script {
                    // Инициализация конфигурации
                    конфигТестовыхДанных = [
                        окружение: params.ОКРУЖЕНИЕ,
                        источникДанных: params.ИСТОЧНИК_ДАННЫХ,
                        размерНабора: params.РАЗМЕР_НАБОРА,
                        соответствие: params.ТРЕБОВАНИЯ_СООТВЕТСТВИЯ,
                        временнаяМетка: new Date().format('yyyy-MM-dd-HH-mm-ss')
                    ]

                    // Генерация уникального ID набора данных
                    конфигТестовыхДанных.datasetId = генерироватьIdНабора(конфигТестовыхДанных)
                }
            }
        }

        stage('Получение исходных данных') {
            when {
                expression { params.ИСТОЧНИК_ДАННЫХ != 'синтетический' }
            }
            steps {
                script {
                    withCredentials([
                        usernamePassword(
                            credentialsId: 'production-db-readonly',
                            usernameVariable: 'DB_USER',
                            passwordVariable: 'DB_PASS'
                        )
                    ]) {
                        sh """
                            python3 scripts/извлечь_подмножество_продакшна.py \
                                --host ${PROD_DB_HOST} \
                                --user ${DB_USER} \
                                --password ${DB_PASS} \
                                --размер ${params.РАЗМЕР_НАБОРА} \
                                --вывод /tmp/сырые-данные.json
                        """
                    }
                }
            }
        }

        stage('Генерация синтетических данных') {
            when {
                expression { params.ИСТОЧНИК_ДАННЫХ in ['синтетический', 'гибрид'] }
            }
            steps {
                script {
                    sh """
                        python3 scripts/генерировать_синтетические_данные.py \
                            --конфиг configs/synthetic-${params.ОКРУЖЕНИЕ}.yaml \
                            --размер ${params.РАЗМЕР_НАБОРА} \
                            --семя ${BUILD_NUMBER} \
                            --вывод /tmp/синтетические-данные.json
                    """
                }
            }
        }

        stage('Применение маскирования данных') {
            steps {
                script {
                    def правилаМаскирования = загрузитьПравилаМаскирования(params.ТРЕБОВАНИЯ_СООТВЕТСТВИЯ)

                    sh """
                        python3 scripts/применить_маскирование_данных.py \
                            --вход /tmp/сырые-данные.json \
                            --правила ${правилаМаскирования} \
                            --соответствие ${params.ТРЕБОВАНИЯ_СООТВЕТСТВИЯ.join(',')} \
                            --вывод /tmp/маскированные-данные.json
                    """
                }
            }
        }

        stage('Валидация качества данных') {
            steps {
                script {
                    sh """
                        python3 scripts/валидировать_тестовые_данные.py \
                            --данные /tmp/маскированные-данные.json \
                            --схема schemas/${params.ОКРУЖЕНИЕ}-schema.json \
                            --правила validation-rules.yaml \
                            --отчет /tmp/отчет-валидации.html
                    """

                    publishHTML(target: [
                        allowMissing: false,
                        alwaysLinkToLastBuild: true,
                        keepAll: true,
                        reportDir: '/tmp',
                        reportFiles: 'отчет-валидации.html',
                        reportName: 'Отчет валидации данных'
                    ])
                }
            }
        }

        stage('Версионирование и сохранение набора данных') {
            steps {
                script {
                    sh """
                        # Сжатие и шифрование набора данных
                        tar -czf /tmp/dataset-${конфигТестовыхДанных.datasetId}.tar.gz \
                            /tmp/маскированные-данные.json

                        # Загрузка в S3 с версионированием
                        aws s3 cp /tmp/dataset-${конфигТестовыхДанных.datasetId}.tar.gz \
                            ${DATA_LAKE_BUCKET}/${params.ОКРУЖЕНИЕ}/ \
                            --server-side-encryption aws:kms \
                            --metadata "build=${BUILD_NUMBER},environment=${params.ОКРУЖЕНИЕ}"

                        # Регистрация в каталоге данных
                        python3 scripts/зарегистрировать_набор.py \
                            --dataset-id ${конфигТестовыхДанных.datasetId} \
                            --расположение ${DATA_LAKE_BUCKET}/${params.ОКРУЖЕНИЕ}/ \
                            --metadata '${groovy.json.JsonOutput.toJson(конфигТестовыхДанных)}'
                    """
                }
            }
        }

        stage('Развертывание в тестовое окружение') {
            steps {
                script {
                    parallel(
                        'База данных': {
                            sh """
                                python3 scripts/загрузить_в_базу.py \
                                    --dataset /tmp/маскированные-данные.json \
                                    --цель ${params.ОКРУЖЕНИЕ} \
                                    --строка-подключения \${${params.ОКРУЖЕНИЕ.toUpperCase()}_DB_URL}
                            """
                        },
                        'Кэш': {
                            sh """
                                python3 scripts/загрузить_в_кэш.py \
                                    --dataset /tmp/маскированные-данные.json \
                                    --redis-host \${${params.ОКРУЖЕНИЕ.toUpperCase()}_REDIS_HOST} \
                                    --ttl 3600
                            """
                        },
                        'Файловая система': {
                            sh """
                                kubectl cp /tmp/маскированные-данные.json \
                                    ${params.ОКРУЖЕНИЕ}/test-data-pod:/data/test-data.json
                            """
                        }
                    )
                }
            }
        }

        stage('Запуск тестов верификации данных') {
            steps {
                script {
                    sh """
                        pytest tests/data_verification/ \
                            --environment ${params.ОКРУЖЕНИЕ} \
                            --dataset-id ${конфигТестовыхДанных.datasetId} \
                            --junitxml=test-results.xml
                    """
                }
            }
        }
    }

    post {
        always {
            junit 'test-results.xml'
            cleanWs()
        }
        success {
            emailext(
                subject: "Тестовые данные подготовлены: ${конфигТестовыхДанных.datasetId}",
                body: """
                    Тестовые данные успешно подготовлены для ${params.ОКРУЖЕНИЕ}

                    ID набора: ${конфигТестовыхДанных.datasetId}
                    Размер: ${params.РАЗМЕР_НАБОРА}
                    Источник: ${params.ИСТОЧНИК_ДАННЫХ}
                    Соответствие: ${params.ТРЕБОВАНИЯ_СООТВЕТСТВИЯ}

                    Доступ к набору данных:
                    ${env.BUILD_URL}
                """,
                to: 'qa-team@example.com'
            )
        }
        failure {
            sh """
                # Откат частичных развертываний
                python3 scripts/откатить_тестовые_данные.py \
                    --environment ${params.ОКРУЖЕНИЕ} \
                    --dataset-id ${конфигТестовыхДанных.datasetId}
            """
        }
    }
}

# .gitlab-ci.yml
stages:
  - подготовка
  - извлечение
  - трансформация
  - загрузка
  - валидация

variables:
  ВЕРСИЯ_ПАЙПЛАЙНА_ДАННЫХ: "2.1.0"
  ОБРАЗ_PYTHON: "python:3.9-slim"

.шаблон-тестовых-данных:
  image: ${ОБРАЗ_PYTHON}
  before_script:
    - pip install -r requirements.txt
    - export DATASET_ID=$(date +%Y%m%d-%H%M%S)-${CI_PIPELINE_ID}

подготовка:окружения:
  stage: подготовка
  script:
    - echo "Подготовка окружения тестовых данных для ${CI_ENVIRONMENT_NAME}"
    - |
      python3 scripts/подготовить_окружение.py \
        --окружение ${CI_ENVIRONMENT_NAME} \
        --очистить-предыдущие ${ОЧИСТИТЬ_ПРЕДЫДУЩИЕ_ДАННЫЕ}

извлечение:подмножество-продакшн:
  stage: извлечение
  only:
    variables:
      - $ИСТОЧНИК_ДАННЫХ == "продакшн"
  script:
    - |
      python3 scripts/извлечь_подмножество.py \
        --источник ${URL_БД_ПРОДАКШН} \
        --размер ${РАЗМЕР_НАБОРА} \
        --фильтры config/фильтры-извлечения.yaml \
        --вывод артефакты/сырые-данные.json
  artifacts:
    paths:
      - артефакты/сырые-данные.json
    expire_in: 1 hour

генерация:синтетические-данные:
  stage: извлечение
  only:
    variables:
      - $ИСТОЧНИК_ДАННЫХ == "синтетический"
  script:
    - |
      python3 scripts/синтетический_генератор.py \
        --модель models/модель-генерации-данных.pkl \
        --размер ${РАЗМЕР_НАБОРА} \
        --конфиг config/конфиг-синтетический.yaml \
        --вывод артефакты/синтетические-данные.json
  artifacts:
    paths:
      - артефакты/синтетические-данные.json
    expire_in: 1 hour

трансформация:применить-маскирование:
  stage: трансформация
  dependencies:
    - извлечение:подмножество-продакшн
    - генерация:синтетические-данные
  script:
    - |
      python3 scripts/маскировщик_данных.py \
        --вход артефакты/*.json \
        --правила config/правила-маскирования-${УРОВЕНЬ_СООТВЕТСТВИЯ}.yaml \
        --вывод артефакты/маскированные-данные.json \
        --лог-аудита артефакты/аудит-маскирования.log
  artifacts:
    paths:
      - артефакты/маскированные-данные.json
      - артефакты/аудит-маскирования.log
    expire_in: 1 day

загрузка:в-базу-данных:
  stage: загрузка
  dependencies:
    - трансформация:применить-маскирование
  script:
    - |
      python3 scripts/загрузчик_базы_данных.py \
        --данные артефакты/маскированные-данные.json \
        --цель ${CI_ENVIRONMENT_NAME} \
        --подключение ${СТРОКА_ПОДКЛЮЧЕНИЯ_ТЕСТОВОЙ_БД} \
        --размер-батча 1000

валидация:качество-данных:
  stage: валидация
  dependencies:
    - загрузка:в-базу-данных
  script:
    - |
      python3 scripts/валидатор_данных.py \
        --окружение ${CI_ENVIRONMENT_NAME} \
        --проверки config/проверки-валидации.yaml \
        --отчет артефакты/отчет-валидации.html
  artifacts:
    reports:
      junit: артефакты/результаты-валидации.xml
    paths:
      - артефакты/отчет-валидации.html
    expire_in: 30 days

# security/data-security-framework.yaml
apiVersion: v1
kind: ConfigMap
metadata:
  name: test-data-security
data:
  security-controls.yaml: |
    шифрование:
      в_покое:
        алгоритм: "AES-256-GCM"
        ротация_ключей: "30-дней"
        хранилище_ключей: "HSM"

      в_передаче:
        протокол: "TLS 1.3"
        валидация_сертификатов: "требуется"
        взаимный_tls: "включен"

    контроль_доступа:
      аутентификация:
        метод: "oauth2"
        требуется_mfa: true
        таймаут_сессии: "30-минут"

      авторизация:
        модель: "RBAC"
        роли:
          - имя: "админ-тестовых-данных"
            разрешения: ["создать", "читать", "обновить", "удалить", "маскировать"]
          - имя: "пользователь-тестовых-данных"
            разрешения: ["читать"]
          - имя: "аудитор-соответствия"
            разрешения: ["читать", "аудит"]

      аудит_логирование:
        включено: true
        хранение: "7-лет"
        неизменяемое_хранилище: true
        события:
          - доступ_к_данным
          - модификация_данных
          - операции_маскирования
          - нарушения_соответствия

    жизненный_цикл_данных:
      хранение:
        тестовые_данные: "30-дней"
        маскированные_данные: "90-дней"
        синтетические_данные: "неограниченно"

      удаление:
        метод: "криптографическое-стирание"
        верификация: "требуется"
        генерация_сертификата: 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 МониторСоответствия:
    def __init__(self):
        self.db = psycopg2.connect(
            host="localhost",
            database="compliance_db",
            user="monitor",
            password="secure_password"
        )
        self.cache = redis.Redis(host='localhost', port=6379)

    def получить_метрики_соответствия(self) -> dict:
        """Сбор метрик соответствия"""
        метрики = {
            'gdpr': self._проверить_соответствие_gdpr(),
            'pci_dss': self._проверить_соответствие_pci(),
            'hipaa': self._проверить_соответствие_hipaa(),
            'качество_данных': self._проверить_качество_данных(),
            'последнее_обновление': datetime.utcnow().isoformat()
        }
        return метрики

    def _проверить_соответствие_gdpr(self) -> dict:
        cursor = self.db.cursor()
        cursor.execute("""
            SELECT
                COUNT(*) FILTER (WHERE is_masked = true) as маскированные_записи,
                COUNT(*) as всего_записей,
                COUNT(DISTINCT dataset_id) as наборы_данных,
                MAX(created_at) as последняя_обработка
            FROM test_data_records
            WHERE contains_pii = true
        """)
        результат = cursor.fetchone()

        return {
            'соответствует': результат[0] == результат[1],
            'процент_маскированных': (результат[0] / результат[1] * 100) if результат[1] > 0 else 100,
            'всего_наборов': результат[2],
            'последняя_обработка': результат[3].isoformat() if результат[3] else None
        }

@app.route('/api/compliance/status')
def статус_соответствия():
    монитор = МониторСоответствия()
    return jsonify(монитор.получить_метрики_соответствия())

@app.route('/api/compliance/audit-trail/<dataset_id>')
def аудит_трейл(dataset_id):
    монитор = МониторСоответствия()
    трейл = монитор.получить_аудит_трейл(dataset_id)
    return jsonify(трейл)

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