## 資料清冊範本

| 序號 | 資料來源名稱 | 類型 | 位置 | 擁有者 | 筆數（估計） | 更新頻率 | 備註 |
|------|-------------|------|------|--------|-------------|---------|------|
| 1 | CUSTOMER_MASTER | Table | Oracle/PROD | 客戶部 | 500萬 | 即時 | 主檔 |
| 2 | TRANSACTION_LOG | Table | Oracle/PROD | 交易部 | 2億 | 即時 | 需歸檔 |
| 3 | daily_report.csv | File | /data/reports | 營運部 | 每日1千 | 每日 | T+1 |

-- 取得 Table 結構資訊（Oracle 範例）
SELECT 
    column_name,
    data_type,
    data_length,
    nullable,
    data_default
FROM all_tab_columns
WHERE table_name = 'CUSTOMER_MASTER'
ORDER BY column_id;

-- 取得 Table 統計資訊
SELECT 
    table_name,
    num_rows,
    avg_row_len,
    last_analyzed
FROM all_tables
WHERE table_name = 'CUSTOMER_MASTER';

# 客戶主檔 File Layout (customer_master.dat)
# 編碼: Big5
# 記錄長度: 200 bytes

欄位名稱        起始位置  長度  類型      說明
-----------    --------  ----  ------    ----------------
CUST_ID        1         10    AN        客戶編號
CUST_NAME      11        40    AN        客戶姓名
CUST_TYPE      51        2     N         客戶類型
ID_NO          53        10    AN        身分證字號
BIRTH_DATE     63        8     N         生日(YYYYMMDD)
CREATE_DATE    71        8     N         建立日期
FILLER         79        122   AN        保留欄位

-- 查詢 Primary Key（Oracle）
SELECT 
    cols.column_name,
    cons.constraint_name
FROM all_constraints cons
JOIN all_cons_columns cols 
    ON cons.constraint_name = cols.constraint_name
WHERE cons.constraint_type = 'P'
    AND cons.table_name = 'CUSTOMER_MASTER';

-- 查詢 Foreign Key 關聯
SELECT 
    a.table_name AS child_table,
    a.column_name AS child_column,
    c_pk.table_name AS parent_table,
    b.column_name AS parent_column
FROM all_cons_columns a
JOIN all_constraints c 
    ON a.constraint_name = c.constraint_name
JOIN all_constraints c_pk 
    ON c.r_constraint_name = c_pk.constraint_name
JOIN all_cons_columns b 
    ON c_pk.constraint_name = b.constraint_name
WHERE c.constraint_type = 'R'
    AND a.table_name = 'ACCOUNT';

-- 資料 Profiling 完整腳本範例
-- 1. 基本統計
SELECT 
    'CUSTOMER_MASTER' AS table_name,
    COUNT(*) AS total_rows,
    COUNT(DISTINCT cust_id) AS unique_cust_id,
    COUNT(*) - COUNT(cust_name) AS null_cust_name,
    MIN(create_date) AS min_create_date,
    MAX(create_date) AS max_create_date
FROM customer_master;

-- 2. 欄位值分布
SELECT 
    cust_type,
    COUNT(*) AS cnt,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER(), 2) AS pct
FROM customer_master
GROUP BY cust_type
ORDER BY cnt DESC;

-- 3. 重複 Key 檢測
SELECT cust_id, COUNT(*) AS dup_count
FROM customer_master
GROUP BY cust_id
HAVING COUNT(*) > 1;

-- 4. 參照完整性檢測
SELECT a.acct_no, a.cust_id
FROM account a
LEFT JOIN customer_master c ON a.cust_id = c.cust_id
WHERE c.cust_id IS NULL;

## 資料品質檢測報告

### 檢測日期：2026-02-02
### 檢測範圍：CUSTOMER_MASTER

| 檢測項目 | 結果 | 筆數 | 比例 | 嚴重度 |
|---------|------|------|------|--------|
| 總筆數 | - | 5,234,567 | 100% | - |
| CUST_ID 空值 | ✅ PASS | 0 | 0% | - |
| CUST_NAME 空值 | ⚠️ WARN | 123 | 0.002% | 中 |
| CUST_TYPE 異常值 | 🔴 FAIL | 45 | 0.001% | 高 |
| 重複 CUST_ID | ✅ PASS | 0 | 0% | - |

### 需處理項目
1. CUST_NAME 空值 123 筆 → 建議補值或設預設值
2. CUST_TYPE 值 'X' 未在代碼表中定義 → 需確認對應規則

-- 1. 直接對應（型態轉換）
-- 舊：VARCHAR2(10) → 新：BIGINT
SELECT CAST(cust_id AS BIGINT) AS id FROM old_customer;

-- 2. 日期格式轉換
-- 舊：YYYYMMDD (CHAR) → 新：TIMESTAMP
SELECT 
    TO_TIMESTAMP(create_dt, 'YYYYMMDD') AS created_at
FROM old_customer;

-- 3. Code 對應轉換
SELECT 
    CASE cust_type
        WHEN '1' THEN 'INDIVIDUAL'
        WHEN '2' THEN 'CORPORATE'
        WHEN '3' THEN 'VIP'
        ELSE 'UNKNOWN'
    END AS customer_type
FROM old_customer;

-- 4. 地址拆分（使用 REGEXP）
SELECT 
    REGEXP_SUBSTR(addr, '^(.{2}[市縣])', 1, 1) AS city,
    REGEXP_SUBSTR(addr, '[市縣](.+?[區鄉鎮市])', 1, 1, NULL, 1) AS district,
    REGEXP_SUBSTR(addr, '[區鄉鎮市](.+)$', 1, 1, NULL, 1) AS street
FROM old_customer;

-- 5. 欄位合併
SELECT 
    CONCAT(last_name, first_name) AS full_name
FROM old_employee;

-- 建立代碼對應表
CREATE TABLE migration_code_mapping (
    id              BIGINT PRIMARY KEY AUTO_INCREMENT,
    category        VARCHAR(50) NOT NULL,    -- 代碼類別
    source_code     VARCHAR(20) NOT NULL,    -- 舊系統代碼
    target_code     VARCHAR(50) NOT NULL,    -- 新系統代碼
    description     VARCHAR(200),            -- 說明
    effective_date  DATE DEFAULT CURRENT_DATE,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    UNIQUE KEY uk_mapping (category, source_code)
);

-- 範例資料
INSERT INTO migration_code_mapping (category, source_code, target_code, description) VALUES
('CUST_TYPE', '1', 'INDIVIDUAL', '個人客戶'),
('CUST_TYPE', '2', 'CORPORATE', '法人客戶'),
('CUST_TYPE', '3', 'VIP', 'VIP客戶'),
('ACCT_STATUS', 'A', 'ACTIVE', '有效'),
('ACCT_STATUS', 'C', 'CLOSED', '已結清'),
('ACCT_STATUS', 'F', 'FROZEN', '凍結');

-- 建立對應查詢函數
CREATE FUNCTION fn_get_target_code(
    p_category VARCHAR(50),
    p_source_code VARCHAR(20)
) RETURNS VARCHAR(50)
BEGIN
    DECLARE v_target_code VARCHAR(50);
    
    SELECT target_code INTO v_target_code
    FROM migration_code_mapping
    WHERE category = p_category 
      AND source_code = p_source_code;
    
    IF v_target_code IS NULL THEN
        RETURN CONCAT('UNMAPPED_', p_source_code);
    END IF;
    
    RETURN v_target_code;
END;

-- 檢查是否有未對應的代碼
SELECT DISTINCT 
    'CUST_TYPE' AS category,
    c.cust_type AS source_code,
    COUNT(*) AS affected_rows
FROM old_customer c
LEFT JOIN migration_code_mapping m 
    ON m.category = 'CUST_TYPE' 
    AND m.source_code = c.cust_type
WHERE m.id IS NULL
GROUP BY c.cust_type;

-- 歷史資料分區表設計範例
CREATE TABLE transaction_history (
    txn_id          BIGINT NOT NULL,
    acct_no         VARCHAR(20) NOT NULL,
    txn_date        DATE NOT NULL,
    amount          DECIMAL(18,2),
    -- ... 其他欄位
    PRIMARY KEY (txn_id, txn_date)
) PARTITION BY RANGE (txn_date) (
    PARTITION p_2024 VALUES LESS THAN ('2025-01-01'),
    PARTITION p_2025 VALUES LESS THAN ('2026-01-01'),
    PARTITION p_2026 VALUES LESS THAN ('2027-01-01'),
    PARTITION p_future VALUES LESS THAN MAXVALUE
);

-- 分批轉置控制表
CREATE TABLE migration_batch_control (
    batch_id        VARCHAR(20) PRIMARY KEY,
    table_name      VARCHAR(100) NOT NULL,
    batch_seq       INT NOT NULL,
    start_key       VARCHAR(100),
    end_key         VARCHAR(100),
    total_rows      BIGINT DEFAULT 0,
    processed_rows  BIGINT DEFAULT 0,
    status          VARCHAR(20) DEFAULT 'PENDING',
    start_time      TIMESTAMP,
    end_time        TIMESTAMP,
    error_message   TEXT,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP
);

-- 建立分批（依 CUST_ID 範圍）
INSERT INTO migration_batch_control (batch_id, table_name, batch_seq, start_key, end_key)
SELECT 
    CONCAT('CUST_', LPAD(ROW_NUMBER() OVER(), 3, '0')),
    'CUSTOMER_MASTER',
    ROW_NUMBER() OVER(),
    MIN(cust_id),
    MAX(cust_id)
FROM (
    SELECT 
        cust_id,
        NTILE(10) OVER (ORDER BY cust_id) AS batch_group
    FROM old_customer
) t
GROUP BY batch_group;

// CDC 監聽器概念範例（使用 Debezium 概念）
public class CustomerCdcHandler {
    
    @KafkaListener(topics = "dbserver.schema.customer")
    public void handleCustomerChange(ChangeEvent event) {
        switch (event.getOperation()) {
            case CREATE:
                migrateNewCustomer(event.getAfter());
                break;
            case UPDATE:
                updateMigratedCustomer(event.getBefore(), event.getAfter());
                break;
            case DELETE:
                markCustomerDeleted(event.getBefore());
                break;
        }
    }
    
    private void migrateNewCustomer(CustomerRecord record) {
        // 轉換並寫入新系統
        NewCustomer newCustomer = transform(record);
        newCustomerRepository.save(newCustomer);
        
        // 記錄轉置日誌
        migrationLogRepository.save(new MigrationLog(
            record.getCustId(), 
            "CREATE", 
            LocalDateTime.now()
        ));
    }
}

@Service
@Transactional
public class DualWriteService {
    
    @Autowired
    private OldCustomerRepository oldRepo;
    
    @Autowired
    private NewCustomerRepository newRepo;
    
    @Autowired
    private ComparisonService comparisonService;
    
    public void createCustomer(CustomerRequest request) {
        // 1. 寫入舊系統
        OldCustomer oldCustomer = oldRepo.save(
            mapToOldCustomer(request)
        );
        
        // 2. 寫入新系統
        NewCustomer newCustomer = newRepo.save(
            mapToNewCustomer(request)
        );
        
        // 3. 記錄比對資訊
        comparisonService.recordForComparison(
            oldCustomer.getId(),
            newCustomer.getId(),
            "CREATE"
        );
    }
}

-- 轉置執行記錄表
CREATE TABLE migration_execution_log (
    execution_id    VARCHAR(36) PRIMARY KEY,
    batch_id        VARCHAR(20),
    table_name      VARCHAR(100),
    start_key       VARCHAR(100),
    end_key         VARCHAR(100),
    status          VARCHAR(20),  -- RUNNING, SUCCESS, FAILED, ROLLBACK
    processed_count BIGINT,
    error_count     BIGINT,
    start_time      TIMESTAMP,
    end_time        TIMESTAMP,
    error_detail    TEXT,
    can_rerun       BOOLEAN DEFAULT TRUE,
    rerun_count     INT DEFAULT 0
);

-- Re-run 查詢：找出需要重跑的批次
SELECT 
    batch_id,
    table_name,
    start_key,
    end_key,
    rerun_count
FROM migration_execution_log
WHERE status = 'FAILED'
  AND can_rerun = TRUE
  AND rerun_count < 3  -- 最多重跑 3 次
ORDER BY batch_id;

@Service
public class MigrationCheckpointService {
    
    @Autowired
    private CheckpointRepository checkpointRepo;
    
    /**
     * 建立檢查點
     */
    public String createCheckpoint(String batchId, String tableName) {
        Checkpoint checkpoint = new Checkpoint();
        checkpoint.setCheckpointId(UUID.randomUUID().toString());
        checkpoint.setBatchId(batchId);
        checkpoint.setTableName(tableName);
        checkpoint.setCreatedAt(LocalDateTime.now());
        checkpoint.setStatus("CREATED");
        
        // 記錄當前進度
        checkpoint.setLastProcessedKey(
            getLastProcessedKey(tableName)
        );
        
        return checkpointRepo.save(checkpoint).getCheckpointId();
    }
    
    /**
     * 從檢查點恢復
     */
    public void resumeFromCheckpoint(String checkpointId) {
        Checkpoint checkpoint = checkpointRepo.findById(checkpointId)
            .orElseThrow(() -> new RuntimeException("Checkpoint not found"));
        
        log.info("Resume from checkpoint: {}, lastKey: {}", 
            checkpointId, checkpoint.getLastProcessedKey());
        
        // 從上次位置繼續
        migrationService.migrateFrom(
            checkpoint.getTableName(),
            checkpoint.getLastProcessedKey()
        );
    }
}

@Service
@Slf4j
public class DataExtractService {
    
    @Autowired
    private JdbcTemplate sourceJdbc;
    
    @Autowired
    private StagingRepository stagingRepo;
    
    /**
     * 分頁抽取大量資料
     */
    public ExtractResult extractCustomers(String batchId, int pageSize) {
        ExtractResult result = new ExtractResult(batchId);
        result.setStartTime(LocalDateTime.now());
        
        int offset = 0;
        int totalExtracted = 0;
        
        try {
            while (true) {
                // 分頁查詢
                String sql = """
                    SELECT cust_id, cust_name, cust_type, 
                           create_date, modify_date
                    FROM customer_master
                    ORDER BY cust_id
                    OFFSET ? ROWS FETCH NEXT ? ROWS ONLY
                    """;
                
                List<Map<String, Object>> rows = sourceJdbc.queryForList(
                    sql, offset, pageSize
                );
                
                if (rows.isEmpty()) {
                    break;
                }
                
                // 寫入 Staging
                List<StagingCustomer> stagingData = rows.stream()
                    .map(this::mapToStaging)
                    .collect(Collectors.toList());
                
                stagingRepo.batchInsert(stagingData);
                
                totalExtracted += rows.size();
                offset += pageSize;
                
                log.info("Extracted {} records, total: {}", 
                    rows.size(), totalExtracted);
            }
            
            result.setStatus("SUCCESS");
            result.setExtractedCount(totalExtracted);
            
        } catch (Exception e) {
            log.error("Extract failed at offset {}", offset, e);
            result.setStatus("FAILED");
            result.setErrorMessage(e.getMessage());
        }
        
        result.setEndTime(LocalDateTime.now());
        return result;
    }
    
    private StagingCustomer mapToStaging(Map<String, Object> row) {
        StagingCustomer staging = new StagingCustomer();
        staging.setSourceCustId((String) row.get("CUST_ID"));
        staging.setSourceCustName((String) row.get("CUST_NAME"));
        staging.setSourceCustType((String) row.get("CUST_TYPE"));
        staging.setSourceCreateDate((Date) row.get("CREATE_DATE"));
        staging.setExtractTime(LocalDateTime.now());
        staging.setProcessStatus("PENDING");
        return staging;
    }
}

/**
 * 轉換規則介面
 */
public interface TransformRule<S, T> {
    T transform(S source);
    boolean validate(S source);
    String getRuleName();
}

/**
 * 日期轉換規則
 */
@Component
public class DateTransformRule implements TransformRule<String, LocalDate> {
    
    private static final DateTimeFormatter SOURCE_FORMAT = 
        DateTimeFormatter.ofPattern("yyyyMMdd");
    
    @Override
    public LocalDate transform(String source) {
        if (source == null || source.trim().isEmpty()) {
            return null;
        }
        return LocalDate.parse(source.trim(), SOURCE_FORMAT);
    }
    
    @Override
    public boolean validate(String source) {
        if (source == null || source.trim().isEmpty()) {
            return true; // 允許空值
        }
        try {
            transform(source);
            return true;
        } catch (DateTimeParseException e) {
            return false;
        }
    }
    
    @Override
    public String getRuleName() {
        return "DATE_YYYYMMDD_TO_LOCALDATE";
    }
}

/**
 * 代碼對應轉換規則
 */
@Component
public class CodeMappingRule implements TransformRule<String, String> {
    
    @Autowired
    private CodeMappingRepository codeRepo;
    
    private String category;
    
    public CodeMappingRule category(String category) {
        this.category = category;
        return this;
    }
    
    @Override
    public String transform(String source) {
        return codeRepo.findTargetCode(category, source)
            .orElse("UNMAPPED_" + source);
    }
    
    @Override
    public boolean validate(String source) {
        return codeRepo.existsMapping(category, source);
    }
    
    @Override
    public String getRuleName() {
        return "CODE_MAPPING_" + category;
    }
}

@Service
@Slf4j
public class DataTransformService {
    
    @Autowired
    private StagingRepository stagingRepo;
    
    @Autowired
    private TransformRuleEngine ruleEngine;
    
    @Autowired
    private ErrorRepository errorRepo;
    
    @Transactional
    public TransformResult transformCustomers(String batchId) {
        TransformResult result = new TransformResult(batchId);
        result.setStartTime(LocalDateTime.now());
        
        int successCount = 0;
        int errorCount = 0;
        
        // 取得待轉換資料
        List<StagingCustomer> stagingData = stagingRepo
            .findByBatchIdAndStatus(batchId, "PENDING");
        
        for (StagingCustomer staging : stagingData) {
            try {
                // 執行轉換
                NewCustomer newCustomer = transformSingleRecord(staging);
                
                // 更新 Staging 狀態
                staging.setProcessStatus("TRANSFORMED");
                staging.setTransformedData(toJson(newCustomer));
                stagingRepo.save(staging);
                
                successCount++;
                
            } catch (TransformException e) {
                // 記錄錯誤
                staging.setProcessStatus("ERROR");
                staging.setErrorMessage(e.getMessage());
                stagingRepo.save(staging);
                
                errorRepo.save(new TransformError(
                    batchId,
                    staging.getSourceCustId(),
                    e.getRuleName(),
                    e.getMessage()
                ));
                
                errorCount++;
            }
        }
        
        result.setSuccessCount(successCount);
        result.setErrorCount(errorCount);
        result.setEndTime(LocalDateTime.now());
        
        log.info("Transform completed: success={}, error={}", 
            successCount, errorCount);
        
        return result;
    }
    
    private NewCustomer transformSingleRecord(StagingCustomer staging) {
        NewCustomer target = new NewCustomer();
        
        // 1. 直接對應
        target.setCustomerNo(staging.getSourceCustId().trim());
        
        // 2. 清理轉換
        target.setName(cleanName(staging.getSourceCustName()));
        
        // 3. 代碼對應
        target.setCustomerType(
            ruleEngine.transform("CUST_TYPE", staging.getSourceCustType())
        );
        
        // 4. 日期轉換
        target.setCreatedAt(
            ruleEngine.transform("DATE", staging.getSourceCreateDate())
        );
        
        // 5. 系統欄位
        target.setMigrationBatchId(staging.getBatchId());
        target.setMigrationTime(LocalDateTime.now());
        
        return target;
    }
    
    private String cleanName(String name) {
        if (name == null) return null;
        // 全形轉半形、去除多餘空白
        return StringUtils.normalizeSpace(
            CharUtils.toHalfWidth(name)
        );
    }
}

@Service
@Slf4j
public class DataLoadService {
    
    @Autowired
    private StagingRepository stagingRepo;
    
    @Autowired
    private NewCustomerRepository targetRepo;
    
    @PersistenceContext
    private EntityManager entityManager;
    
    private static final int BATCH_SIZE = 1000;
    
    @Transactional
    public LoadResult loadCustomers(String batchId) {
        LoadResult result = new LoadResult(batchId);
        result.setStartTime(LocalDateTime.now());
        
        List<StagingCustomer> stagingData = stagingRepo
            .findByBatchIdAndStatus(batchId, "TRANSFORMED");
        
        int loadedCount = 0;
        int errorCount = 0;
        
        for (int i = 0; i < stagingData.size(); i++) {
            StagingCustomer staging = stagingData.get(i);
            
            try {
                // 解析轉換後資料
                NewCustomer customer = fromJson(
                    staging.getTransformedData(), 
                    NewCustomer.class
                );
                
                // 執行 Upsert
                upsertCustomer(customer);
                
                staging.setProcessStatus("LOADED");
                staging.setLoadTime(LocalDateTime.now());
                loadedCount++;
                
            } catch (Exception e) {
                staging.setProcessStatus("LOAD_ERROR");
                staging.setErrorMessage(e.getMessage());
                errorCount++;
                log.error("Load error for {}", staging.getSourceCustId(), e);
            }
            
            stagingRepo.save(staging);
            
            // 批次提交
            if ((i + 1) % BATCH_SIZE == 0) {
                entityManager.flush();
                entityManager.clear();
                log.info("Committed batch at index {}", i + 1);
            }
        }
        
        // 最後一批提交
        entityManager.flush();
        
        result.setLoadedCount(loadedCount);
        result.setErrorCount(errorCount);
        result.setEndTime(LocalDateTime.now());
        
        return result;
    }
    
    private void upsertCustomer(NewCustomer customer) {
        Optional<NewCustomer> existing = targetRepo
            .findByCustomerNo(customer.getCustomerNo());
        
        if (existing.isPresent()) {
            // Update
            NewCustomer target = existing.get();
            target.setName(customer.getName());
            target.setCustomerType(customer.getCustomerType());
            target.setUpdatedAt(LocalDateTime.now());
            targetRepo.save(target);
        } else {
            // Insert
            customer.setCreatedAt(LocalDateTime.now());
            targetRepo.save(customer);
        }
    }
}

-- Staging Table 設計範例
CREATE TABLE stg_customer (
    -- Staging 控制欄位
    stg_id              BIGINT AUTO_INCREMENT PRIMARY KEY,
    batch_id            VARCHAR(20) NOT NULL,
    extract_time        TIMESTAMP NOT NULL,
    process_status      VARCHAR(20) DEFAULT 'PENDING',
    
    -- 來源資料欄位（原始格式保留）
    src_cust_id         VARCHAR(20),
    src_cust_name       VARCHAR(100),
    src_cust_type       VARCHAR(10),
    src_create_date     VARCHAR(20),
    src_raw_data        TEXT,           -- 原始資料 JSON
    
    -- 轉換後資料欄位
    tgt_customer_no     VARCHAR(20),
    tgt_name            VARCHAR(100),
    tgt_customer_type   VARCHAR(50),
    tgt_created_at      TIMESTAMP,
    tgt_transformed_data TEXT,          -- 轉換後資料 JSON
    
    -- 處理追蹤欄位
    transform_time      TIMESTAMP,
    load_time           TIMESTAMP,
    error_code          VARCHAR(20),
    error_message       TEXT,
    retry_count         INT DEFAULT 0,
    
    -- 索引
    INDEX idx_batch_status (batch_id, process_status),
    INDEX idx_src_cust_id (src_cust_id)
);

@Service
@Slf4j
public class RetryableTransformService {
    
    private static final int MAX_RETRY = 3;
    private static final long RETRY_DELAY_MS = 1000;
    
    @Autowired
    private TransformService transformService;
    
    @Autowired
    private ErrorRepository errorRepo;
    
    public TransformResult transformWithRetry(StagingCustomer staging) {
        int retryCount = 0;
        Exception lastException = null;
        
        while (retryCount < MAX_RETRY) {
            try {
                return transformService.transform(staging);
                
            } catch (RetryableException e) {
                retryCount++;
                lastException = e;
                log.warn("Retryable error, attempt {}/{}: {}", 
                    retryCount, MAX_RETRY, e.getMessage());
                
                if (retryCount < MAX_RETRY) {
                    sleep(RETRY_DELAY_MS * retryCount); // 指數退避
                }
                
            } catch (DataException e) {
                // 資料錯誤，不重試
                log.error("Data error for {}: {}", 
                    staging.getSourceCustId(), e.getMessage());
                return handleDataError(staging, e);
                
            } catch (Exception e) {
                // 系統錯誤，不重試
                log.error("System error for {}", staging.getSourceCustId(), e);
                throw new MigrationException("System error", e);
            }
        }
        
        // 重試次數用盡
        return handleRetryExhausted(staging, lastException);
    }
    
    private TransformResult handleDataError(
            StagingCustomer staging, DataException e) {
        
        errorRepo.save(new TransformError(
            staging.getBatchId(),
            staging.getSourceCustId(),
            "DATA_ERROR",
            e.getMessage(),
            e.getFieldName(),
            e.getFieldValue()
        ));
        
        return TransformResult.error(staging.getSourceCustId(), e.getMessage());
    }
    
    private TransformResult handleRetryExhausted(
            StagingCustomer staging, Exception e) {
        
        errorRepo.save(new TransformError(
            staging.getBatchId(),
            staging.getSourceCustId(),
            "RETRY_EXHAUSTED",
            "Max retry reached: " + e.getMessage()
        ));
        
        return TransformResult.error(staging.getSourceCustId(), 
            "Retry exhausted");
    }
    
    private void sleep(long ms) {
        try {
            Thread.sleep(ms);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
        }
    }
}

-- 轉置錯誤記錄表
CREATE TABLE migration_error_log (
    error_id        BIGINT AUTO_INCREMENT PRIMARY KEY,
    batch_id        VARCHAR(20) NOT NULL,
    source_table    VARCHAR(100),
    source_key      VARCHAR(100),
    error_phase     VARCHAR(20),    -- EXTRACT, TRANSFORM, LOAD
    error_type      VARCHAR(50),    -- DATA_ERROR, RETRY_EXHAUSTED, etc.
    error_code      VARCHAR(20),
    error_message   TEXT,
    field_name      VARCHAR(100),
    field_value     TEXT,
    raw_data        TEXT,
    created_at      TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    resolved_at     TIMESTAMP,
    resolved_by     VARCHAR(50),
    resolution_note TEXT,
    
    INDEX idx_batch_phase (batch_id, error_phase),
    INDEX idx_error_type (error_type)
);

-- 查詢未解決的錯誤
SELECT 
    error_phase,
    error_type,
    COUNT(*) AS error_count
FROM migration_error_log
WHERE batch_id = 'BATCH_001'
  AND resolved_at IS NULL
GROUP BY error_phase, error_type
ORDER BY error_count DESC;

-- 1. 總筆數比對
SELECT 
    'SOURCE' AS system,
    COUNT(*) AS record_count
FROM old_customer
UNION ALL
SELECT 
    'TARGET' AS system,
    COUNT(*) AS record_count
FROM new_customer
WHERE migration_batch_id IS NOT NULL;

-- 2. 批次筆數比對
SELECT 
    s.batch_id,
    s.source_count,
    t.target_count,
    e.error_count,
    CASE 
        WHEN s.source_count = t.target_count + e.error_count 
        THEN 'PASS' 
        ELSE 'FAIL' 
    END AS validation_result
FROM (
    SELECT batch_id, COUNT(*) AS source_count
    FROM stg_customer
    GROUP BY batch_id
) s
LEFT JOIN (
    SELECT migration_batch_id, COUNT(*) AS target_count
    FROM new_customer
    GROUP BY migration_batch_id
) t ON s.batch_id = t.migration_batch_id
LEFT JOIN (
    SELECT batch_id, COUNT(*) AS error_count
    FROM migration_error_log
    WHERE error_phase = 'LOAD'
    GROUP BY batch_id
) e ON s.batch_id = e.batch_id;

## 筆數驗證報告

### 執行日期：2026-02-02
### 批次編號：BATCH_001

| 項目 | 來源筆數 | 目標筆數 | 錯誤筆數 | 差異 | 結果 |
|------|---------|---------|---------|------|------|
| CUSTOMER | 1,234,567 | 1,234,500 | 67 | 0 | ✅ PASS |
| ACCOUNT | 3,456,789 | 3,456,789 | 0 | 0 | ✅ PASS |
| TRANSACTION | 50,000,000 | 49,999,985 | 10 | 5 | ⚠️ WARN |

### 差異說明
- TRANSACTION 差異 5 筆：經確認為重複資料，已過濾

-- 金額驗證查詢
WITH source_sum AS (
    SELECT 
        'SOURCE' AS system,
        SUM(balance) AS total_balance,
        SUM(txn_amount) AS total_txn_amount,
        COUNT(DISTINCT acct_no) AS account_count
    FROM old_account
),
target_sum AS (
    SELECT 
        'TARGET' AS system,
        SUM(balance) AS total_balance,
        SUM(txn_amount) AS total_txn_amount,
        COUNT(DISTINCT account_no) AS account_count
    FROM new_account
    WHERE migration_batch_id IS NOT NULL
)
SELECT 
    s.total_balance AS source_balance,
    t.total_balance AS target_balance,
    s.total_balance - t.total_balance AS balance_diff,
    CASE 
        WHEN ABS(s.total_balance - t.total_balance) < 0.01 
        THEN 'PASS' 
        ELSE 'FAIL' 
    END AS balance_check,
    s.total_txn_amount AS source_txn,
    t.total_txn_amount AS target_txn,
    s.total_txn_amount - t.total_txn_amount AS txn_diff
FROM source_sum s
CROSS JOIN target_sum t;

-- 依客戶類型分群驗證
SELECT 
    COALESCE(s.cust_type, t.customer_type) AS customer_type,
    s.source_balance,
    t.target_balance,
    s.source_balance - COALESCE(t.target_balance, 0) AS diff,
    CASE 
        WHEN ABS(s.source_balance - COALESCE(t.target_balance, 0)) < 0.01 
        THEN '✅ PASS' 
        ELSE '❌ FAIL' 
    END AS result
FROM (
    SELECT 
        cust_type,
        SUM(balance) AS source_balance
    FROM old_account oa
    JOIN old_customer oc ON oa.cust_id = oc.cust_id
    GROUP BY cust_type
) s
FULL OUTER JOIN (
    SELECT 
        customer_type,
        SUM(balance) AS target_balance
    FROM new_account na
    JOIN new_customer nc ON na.customer_id = nc.id
    WHERE na.migration_batch_id IS NOT NULL
    GROUP BY customer_type
) t ON s.cust_type = (
    SELECT source_code 
    FROM migration_code_mapping 
    WHERE category = 'CUST_TYPE' 
      AND target_code = t.customer_type
);

-- 1. 找出來源有但目標沒有的資料（遺失）
SELECT s.cust_id AS missing_key
FROM old_customer s
LEFT JOIN new_customer t 
    ON s.cust_id = t.customer_no
WHERE t.id IS NULL
  AND s.status = 'A';  -- 只檢查有效資料

-- 2. 找出目標有但來源沒有的資料（多餘）
SELECT t.customer_no AS extra_key
FROM new_customer t
LEFT JOIN old_customer s 
    ON t.customer_no = s.cust_id
WHERE s.cust_id IS NULL
  AND t.migration_batch_id IS NOT NULL;

-- 3. 關鍵欄位比對（找出不一致）
SELECT 
    s.cust_id,
    s.cust_name AS source_name,
    t.name AS target_name,
    s.balance AS source_balance,
    t.balance AS target_balance
FROM old_customer s
JOIN new_customer t 
    ON s.cust_id = t.customer_no
WHERE s.cust_name <> t.name
   OR ABS(s.balance - t.balance) > 0.01;

-- 比對結果統計
WITH comparison AS (
    SELECT 
        s.cust_id,
        CASE 
            WHEN t.id IS NULL THEN 'MISSING'
            WHEN s.cust_name <> t.name THEN 'NAME_MISMATCH'
            WHEN ABS(s.balance - t.balance) > 0.01 THEN 'BALANCE_MISMATCH'
            ELSE 'MATCH'
        END AS compare_result
    FROM old_customer s
    LEFT JOIN new_customer t 
        ON s.cust_id = t.customer_no
)
SELECT 
    compare_result,
    COUNT(*) AS count,
    ROUND(COUNT(*) * 100.0 / SUM(COUNT(*)) OVER(), 2) AS percentage
FROM comparison
GROUP BY compare_result
ORDER BY count DESC;

-- 1. 隨機抽樣 100 筆
SELECT *
FROM new_customer
WHERE migration_batch_id = 'BATCH_001'
ORDER BY RANDOM()
LIMIT 100;

-- 2. 分層抽樣（每種類型各 20 筆）
WITH ranked AS (
    SELECT 
        *,
        ROW_NUMBER() OVER (
            PARTITION BY customer_type 
            ORDER BY RANDOM()
        ) AS rn
    FROM new_customer
    WHERE migration_batch_id = 'BATCH_001'
)
SELECT *
FROM ranked
WHERE rn <= 20;

-- 3. 風險抽樣（高金額）
SELECT *
FROM new_account
WHERE migration_batch_id = 'BATCH_001'
  AND balance > 1000000
ORDER BY balance DESC
LIMIT 50;

## 抽樣驗證表單

### 抽樣條件
- 批次：BATCH_001
- 抽樣數量：100 筆
- 抽樣方式：分層隨機抽樣

### 驗證結果

| 序號 | 客戶編號 | 來源姓名 | 目標姓名 | 來源餘額 | 目標餘額 | 驗證結果 | 備註 |
|------|---------|---------|---------|---------|---------|---------|------|
| 1 | C001 | 王小明 | 王小明 | 50,000 | 50,000 | ✅ | - |
| 2 | C002 | 李大華 | 李大華 | 120,000 | 120,000 | ✅ | - |
| 3 | C003 | 張三　 | 張三 | 30,000 | 30,000 | ⚠️ | 全形空白已處理 |

### 驗證人員：_______________
### 驗證日期：_______________
### 簽核：_______________

@Service
@Slf4j
public class ValidationReportService {
    
    @Autowired
    private ValidationRepository validationRepo;
    
    public ValidationReport generateReport(String batchId) {
        ValidationReport report = new ValidationReport();
        report.setBatchId(batchId);
        report.setGeneratedAt(LocalDateTime.now());
        
        // 1. 筆數驗證
        RecordCountResult countResult = validateRecordCount(batchId);
        report.setRecordCountResult(countResult);
        
        // 2. 金額驗證
        AmountCheckResult amountResult = validateAmount(batchId);
        report.setAmountCheckResult(amountResult);
        
        // 3. Key 比對
        KeyCompareResult keyResult = compareKeys(batchId);
        report.setKeyCompareResult(keyResult);
        
        // 4. 計算整體狀態
        report.setOverallStatus(calculateOverallStatus(
            countResult, amountResult, keyResult
        ));
        
        // 5. 儲存報表
        validationRepo.saveReport(report);
        
        // 6. 產生 Markdown 檔案
        generateMarkdownReport(report);
        
        return report;
    }
    
    private void generateMarkdownReport(ValidationReport report) {
        StringBuilder md = new StringBuilder();
        
        md.append("# 資料轉置驗證報告\n\n");
        md.append(String.format("- **批次編號**：%s\n", report.getBatchId()));
        md.append(String.format("- **產生時間**：%s\n", report.getGeneratedAt()));
        md.append(String.format("- **整體狀態**：%s\n\n", 
            report.getOverallStatus().getDisplayName()));
        
        // 筆數驗證區塊
        md.append("## 1. 筆數驗證\n\n");
        md.append("| 項目 | 來源 | 目標 | 差異 | 結果 |\n");
        md.append("|------|------|------|------|------|\n");
        for (RecordCountItem item : report.getRecordCountResult().getItems()) {
            md.append(String.format("| %s | %,d | %,d | %,d | %s |\n",
                item.getTableName(),
                item.getSourceCount(),
                item.getTargetCount(),
                item.getDifference(),
                item.isPassed() ? "✅" : "❌"
            ));
        }
        
        // 金額驗證區塊
        md.append("\n## 2. 金額驗證\n\n");
        // ... 類似格式
        
        // 輸出檔案
        String fileName = String.format("validation_report_%s.md", 
            report.getBatchId());
        writeToFile(fileName, md.toString());
    }
    
    private ValidationStatus calculateOverallStatus(
            RecordCountResult count, 
            AmountCheckResult amount,
            KeyCompareResult key) {
        
        if (!count.isPassed() || !amount.isPassed()) {
            return ValidationStatus.FAILED;
        }
        
        if (key.getMismatchCount() > 0) {
            return ValidationStatus.WARNING;
        }
        
        return ValidationStatus.PASSED;
    }
}

-- Dashboard 用彙總查詢
CREATE VIEW v_migration_dashboard AS
SELECT 
    batch_id,
    -- 筆數狀態
    SUM(CASE WHEN record_count_passed THEN 1 ELSE 0 END) AS count_pass,
    SUM(CASE WHEN NOT record_count_passed THEN 1 ELSE 0 END) AS count_fail,
    
    -- 金額狀態
    SUM(CASE WHEN amount_check_passed THEN 1 ELSE 0 END) AS amount_pass,
    SUM(CASE WHEN NOT amount_check_passed THEN 1 ELSE 0 END) AS amount_fail,
    
    -- 錯誤統計
    SUM(error_count) AS total_errors,
    
    -- 進度
    ROUND(
        SUM(CASE WHEN status = 'LOADED' THEN 1 ELSE 0 END) * 100.0 / 
        COUNT(*), 2
    ) AS completion_pct,
    
    -- 時間
    MIN(start_time) AS start_time,
    MAX(end_time) AS end_time,
    TIMESTAMPDIFF(MINUTE, MIN(start_time), MAX(end_time)) AS duration_min
    
FROM migration_batch_control
GROUP BY batch_id;

-- 客戶資料轉置 Stored Procedure
CREATE OR REPLACE PROCEDURE sp_migrate_customer(
    p_batch_id IN VARCHAR2,
    p_start_key IN VARCHAR2,
    p_end_key IN VARCHAR2,
    p_result OUT VARCHAR2
)
AS
    v_count NUMBER := 0;
    v_error_count NUMBER := 0;
    v_start_time TIMESTAMP := SYSTIMESTAMP;
BEGIN
    -- 記錄開始
    UPDATE migration_batch_control
    SET status = 'RUNNING', start_time = v_start_time
    WHERE batch_id = p_batch_id;
    COMMIT;
    
    -- 執行轉置（使用 MERGE）
    MERGE INTO new_customer t
    USING (
        SELECT 
            cust_id AS customer_no,
            TRIM(cust_name) AS name,
            CASE cust_type
                WHEN '1' THEN 'INDIVIDUAL'
                WHEN '2' THEN 'CORPORATE'
                ELSE 'UNKNOWN'
            END AS customer_type,
            TO_TIMESTAMP(create_dt, 'YYYYMMDD') AS created_at,
            p_batch_id AS migration_batch_id
        FROM old_customer
        WHERE cust_id BETWEEN p_start_key AND p_end_key
          AND status = 'A'
    ) s
    ON (t.customer_no = s.customer_no)
    WHEN MATCHED THEN
        UPDATE SET 
            t.name = s.name,
            t.customer_type = s.customer_type,
            t.updated_at = SYSTIMESTAMP
    WHEN NOT MATCHED THEN
        INSERT (customer_no, name, customer_type, created_at, migration_batch_id)
        VALUES (s.customer_no, s.name, s.customer_type, s.created_at, s.migration_batch_id);
    
    v_count := SQL%ROWCOUNT;
    COMMIT;
    
    -- 更新批次狀態
    UPDATE migration_batch_control
    SET status = 'SUCCESS',
        processed_rows = v_count,
        end_time = SYSTIMESTAMP
    WHERE batch_id = p_batch_id;
    COMMIT;
    
    p_result := 'SUCCESS: ' || v_count || ' rows processed';
    
EXCEPTION
    WHEN OTHERS THEN
        ROLLBACK;
        UPDATE migration_batch_control
        SET status = 'FAILED',
            error_message = SQLERRM,
            end_time = SYSTIMESTAMP
        WHERE batch_id = p_batch_id;
        COMMIT;
        
        p_result := 'FAILED: ' || SQLERRM;
END;
/

-- 1. 使用 Hint 優化大量資料處理
INSERT /*+ APPEND PARALLEL(4) */ INTO new_customer
SELECT /*+ PARALLEL(4) */
    cust_id, cust_name, cust_type, create_dt
FROM old_customer;

-- 2. 分批處理避免 Undo 爆滿
DECLARE
    v_batch_size NUMBER := 10000;
BEGIN
    LOOP
        INSERT INTO new_customer
        SELECT * FROM old_customer
        WHERE ROWNUM <= v_batch_size
          AND cust_id NOT IN (SELECT customer_no FROM new_customer);
        
        EXIT WHEN SQL%ROWCOUNT = 0;
        COMMIT;
    END LOOP;
END;

-- 3. 停用 Index 加速載入
ALTER INDEX idx_customer_name UNUSABLE;
-- 執行大量 INSERT
ALTER INDEX idx_customer_name REBUILD;

-- 4. 使用 NOLOGGING 加速（需謹慎）
ALTER TABLE new_customer NOLOGGING;
INSERT /*+ APPEND */ INTO new_customer ...;
ALTER TABLE new_customer LOGGING;

# migration_dag.py
from airflow import DAG
from airflow.operators.python import PythonOperator
from airflow.operators.sql import SQLCheckOperator
from airflow.utils.dates import days_ago
from datetime import timedelta

default_args = {
    'owner': 'data-team',
    'depends_on_past': False,
    'email_on_failure': True,
    'email': ['admin@company.com'],
    'retries': 3,
    'retry_delay': timedelta(minutes=5),
}

dag = DAG(
    'customer_migration',
    default_args=default_args,
    description='Customer data migration pipeline',
    schedule_interval=None,  # Manual trigger
    start_date=days_ago(1),
    tags=['migration'],
)

def extract_customers(**context):
    """Extract customers from source system"""
    from migration.extract import CustomerExtractor
    
    batch_id = context['params']['batch_id']
    extractor = CustomerExtractor()
    result = extractor.extract(batch_id)
    
    context['ti'].xcom_push(key='extract_count', value=result.count)
    return result

def transform_customers(**context):
    """Transform customer data"""
    from migration.transform import CustomerTransformer
    
    batch_id = context['params']['batch_id']
    transformer = CustomerTransformer()
    result = transformer.transform(batch_id)
    
    context['ti'].xcom_push(key='transform_count', value=result.count)
    return result

def load_customers(**context):
    """Load customers to target system"""
    from migration.load import CustomerLoader
    
    batch_id = context['params']['batch_id']
    loader = CustomerLoader()
    result = loader.load(batch_id)
    
    context['ti'].xcom_push(key='load_count', value=result.count)
    return result

# Task definitions
extract_task = PythonOperator(
    task_id='extract_customers',
    python_callable=extract_customers,
    dag=dag,
)

transform_task = PythonOperator(
    task_id='transform_customers',
    python_callable=transform_customers,
    dag=dag,
)

load_task = PythonOperator(
    task_id='load_customers',
    python_callable=load_customers,
    dag=dag,
)

validate_count = SQLCheckOperator(
    task_id='validate_count',
    conn_id='target_db',
    sql="""
        SELECT CASE 
            WHEN source_count = target_count THEN 1 
            ELSE 0 
        END
        FROM migration_summary
        WHERE batch_id = '{{ params.batch_id }}'
    """,
    dag=dag,
)

# Task dependencies
extract_task >> transform_task >> load_task >> validate_count

@Configuration
@EnableBatchProcessing
public class CustomerMigrationBatchConfig {
    
    @Autowired
    private JobBuilderFactory jobBuilderFactory;
    
    @Autowired
    private StepBuilderFactory stepBuilderFactory;
    
    @Autowired
    private DataSource sourceDataSource;
    
    @Autowired
    private DataSource targetDataSource;
    
    @Bean
    public Job customerMigrationJob() {
        return jobBuilderFactory.get("customerMigrationJob")
            .incrementer(new RunIdIncrementer())
            .listener(new JobCompletionListener())
            .flow(customerMigrationStep())
            .end()
            .build();
    }
    
    @Bean
    public Step customerMigrationStep() {
        return stepBuilderFactory.get("customerMigrationStep")
            .<OldCustomer, NewCustomer>chunk(1000)
            .reader(customerReader())
            .processor(customerProcessor())
            .writer(customerWriter())
            .faultTolerant()
            .skipLimit(100)
            .skip(DataIntegrityViolationException.class)
            .retryLimit(3)
            .retry(DeadlockLoserDataAccessException.class)
            .listener(new ChunkListener())
            .build();
    }
    
    @Bean
    @StepScope
    public JdbcPagingItemReader<OldCustomer> customerReader() {
        JdbcPagingItemReader<OldCustomer> reader = new JdbcPagingItemReader<>();
        reader.setDataSource(sourceDataSource);
        reader.setFetchSize(1000);
        reader.setPageSize(1000);
        
        Map<String, Order> sortKeys = new HashMap<>();
        sortKeys.put("cust_id", Order.ASCENDING);
        
        OraclePagingQueryProvider queryProvider = new OraclePagingQueryProvider();
        queryProvider.setSelectClause("cust_id, cust_name, cust_type, create_dt");
        queryProvider.setFromClause("old_customer");
        queryProvider.setWhereClause("status = 'A'");
        queryProvider.setSortKeys(sortKeys);
        
        reader.setQueryProvider(queryProvider);
        reader.setRowMapper(new OldCustomerRowMapper());
        
        return reader;
    }
    
    @Bean
    public ItemProcessor<OldCustomer, NewCustomer> customerProcessor() {
        return new CustomerTransformProcessor();
    }
    
    @Bean
    public JdbcBatchItemWriter<NewCustomer> customerWriter() {
        JdbcBatchItemWriter<NewCustomer> writer = new JdbcBatchItemWriter<>();
        writer.setDataSource(targetDataSource);
        writer.setSql("""
            INSERT INTO new_customer 
                (customer_no, name, customer_type, created_at, migration_batch_id)
            VALUES 
                (:customerNo, :name, :customerType, :createdAt, :migrationBatchId)
            """);
        writer.setItemSqlParameterSourceProvider(
            new BeanPropertyItemSqlParameterSourceProvider<>()
        );
        return writer;
    }
}

/**
 * 轉換處理器
 */
@Component
public class CustomerTransformProcessor 
        implements ItemProcessor<OldCustomer, NewCustomer> {
    
    @Autowired
    private CodeMappingService codeMappingService;
    
    @Value("#{jobParameters['batchId']}")
    private String batchId;
    
    @Override
    public NewCustomer process(OldCustomer source) throws Exception {
        NewCustomer target = new NewCustomer();
        
        // 轉換邏輯
        target.setCustomerNo(source.getCustId().trim());
        target.setName(cleanName(source.getCustName()));
        target.setCustomerType(
            codeMappingService.map("CUST_TYPE", source.getCustType())
        );
        target.setCreatedAt(parseDate(source.getCreateDt()));
        target.setMigrationBatchId(batchId);
        
        return target;
    }
    
    private String cleanName(String name) {
        if (name == null) return null;
        return StringUtils.normalizeSpace(name);
    }
    
    private LocalDateTime parseDate(String dateStr) {
        if (dateStr == null || dateStr.isEmpty()) return null;
        return LocalDate.parse(dateStr, 
            DateTimeFormatter.ofPattern("yyyyMMdd"))
            .atStartOfDay();
    }
}

/**
 * 固定長度檔案解析器
 */
@Component
public class FixedLengthFileParser {
    
    /**
     * Layout 定義
     */
    private static final FieldLayout[] CUSTOMER_LAYOUT = {
        new FieldLayout("custId", 0, 10),
        new FieldLayout("custName", 10, 40),
        new FieldLayout("custType", 50, 2),
        new FieldLayout("idNo", 52, 10),
        new FieldLayout("birthDate", 62, 8),
        new FieldLayout("createDate", 70, 8)
    };
    
    public List<Map<String, String>> parseFile(Path filePath, Charset charset) 
            throws IOException {
        
        List<Map<String, String>> records = new ArrayList<>();
        
        try (BufferedReader reader = Files.newBufferedReader(filePath, charset)) {
            String line;
            int lineNum = 0;
            
            while ((line = reader.readLine()) != null) {
                lineNum++;
                try {
                    Map<String, String> record = parseLine(line);
                    record.put("_lineNum", String.valueOf(lineNum));
                    records.add(record);
                } catch (Exception e) {
                    log.error("Parse error at line {}: {}", lineNum, e.getMessage());
                    // 記錄錯誤但繼續處理
                }
            }
        }
        
        return records;
    }
    
    private Map<String, String> parseLine(String line) {
        Map<String, String> record = new LinkedHashMap<>();
        
        for (FieldLayout field : CUSTOMER_LAYOUT) {
            int endPos = Math.min(field.start + field.length, line.length());
            String value = "";
            
            if (field.start < line.length()) {
                value = line.substring(field.start, endPos).trim();
            }
            
            record.put(field.name, value);
        }
        
        return record;
    }
    
    @Data
    @AllArgsConstructor
    private static class FieldLayout {
        private String name;
        private int start;
        private int length;
    }
}

/**
 * 大檔案串流處理（避免 OOM）
 */
@Service
public class LargeFileProcessor {
    
    private static final int BATCH_SIZE = 10000;
    
    public void processLargeFile(Path filePath, Consumer<List<String>> batchHandler) 
            throws IOException {
        
        try (Stream<String> lines = Files.lines(filePath, Charset.forName("Big5"))) {
            
            List<String> batch = new ArrayList<>(BATCH_SIZE);
            
            lines.forEach(line -> {
                batch.add(line);
                
                if (batch.size() >= BATCH_SIZE) {
                    batchHandler.accept(new ArrayList<>(batch));
                    batch.clear();
                }
            });
            
            // 處理最後一批
            if (!batch.isEmpty()) {
                batchHandler.accept(batch);
            }
        }
    }
    
    /**
     * 使用範例
     */
    public void example() throws IOException {
        processLargeFile(Path.of("/data/large_file.txt"), batch -> {
            log.info("Processing batch of {} records", batch.size());
            // 處理邏輯
        });
    }
}

/**
 * 資料比對工具
 */
@Component
public class DataComparisonTool {
    
    @Autowired
    private JdbcTemplate sourceJdbc;
    
    @Autowired
    private JdbcTemplate targetJdbc;
    
    /**
     * 比對兩個查詢結果
     */
    public ComparisonResult compare(
            String sourceQuery, 
            String targetQuery,
            List<String> keyColumns,
            List<String> compareColumns) {
        
        ComparisonResult result = new ComparisonResult();
        
        // 取得來源資料
        Map<String, Map<String, Object>> sourceData = executeQuery(
            sourceJdbc, sourceQuery, keyColumns
        );
        
        // 取得目標資料
        Map<String, Map<String, Object>> targetData = executeQuery(
            targetJdbc, targetQuery, keyColumns
        );
        
        // 找出遺失的 Key
        Set<String> missingKeys = new HashSet<>(sourceData.keySet());
        missingKeys.removeAll(targetData.keySet());
        result.setMissingKeys(missingKeys);
        
        // 找出多餘的 Key
        Set<String> extraKeys = new HashSet<>(targetData.keySet());
        extraKeys.removeAll(sourceData.keySet());
        result.setExtraKeys(extraKeys);
        
        // 比對欄位值
        List<FieldMismatch> mismatches = new ArrayList<>();
        for (String key : sourceData.keySet()) {
            if (targetData.containsKey(key)) {
                Map<String, Object> sourceRow = sourceData.get(key);
                Map<String, Object> targetRow = targetData.get(key);
                
                for (String col : compareColumns) {
                    Object sourceVal = sourceRow.get(col);
                    Object targetVal = targetRow.get(col);
                    
                    if (!Objects.equals(sourceVal, targetVal)) {
                        mismatches.add(new FieldMismatch(
                            key, col, sourceVal, targetVal
                        ));
                    }
                }
            }
        }
        result.setMismatches(mismatches);
        
        return result;
    }
    
    private Map<String, Map<String, Object>> executeQuery(
            JdbcTemplate jdbc, 
            String query, 
            List<String> keyColumns) {
        
        Map<String, Map<String, Object>> result = new LinkedHashMap<>();
        
        jdbc.query(query, rs -> {
            Map<String, Object> row = new HashMap<>();
            ResultSetMetaData meta = rs.getMetaData();
            
            for (int i = 1; i <= meta.getColumnCount(); i++) {
                row.put(meta.getColumnName(i), rs.getObject(i));
            }
            
            // 組合 Key
            String key = keyColumns.stream()
                .map(col -> String.valueOf(row.get(col)))
                .collect(Collectors.joining("|"));
            
            result.put(key, row);
        });
        
        return result;
    }
}

@ExtendWith(MockitoExtension.class)
class CustomerTransformProcessorTest {
    
    @Mock
    private CodeMappingService codeMappingService;
    
    @InjectMocks
    private CustomerTransformProcessor processor;
    
    @BeforeEach
    void setUp() {
        // 設定 Mock 行為
        when(codeMappingService.map("CUST_TYPE", "1"))
            .thenReturn("INDIVIDUAL");
        when(codeMappingService.map("CUST_TYPE", "2"))
            .thenReturn("CORPORATE");
        when(codeMappingService.map("CUST_TYPE", "X"))
            .thenReturn("UNMAPPED_X");
    }
    
    @Test
    @DisplayName("正常資料轉換")
    void testNormalTransform() throws Exception {
        // Given
        OldCustomer source = new OldCustomer();
        source.setCustId("C0001     ");  // 含空白
        source.setCustName("  王小明  ");
        source.setCustType("1");
        source.setCreateDt("20240101");
        
        // When
        NewCustomer result = processor.process(source);
        
        // Then
        assertThat(result.getCustomerNo()).isEqualTo("C0001");
        assertThat(result.getName()).isEqualTo("王小明");
        assertThat(result.getCustomerType()).isEqualTo("INDIVIDUAL");
        assertThat(result.getCreatedAt())
            .isEqualTo(LocalDateTime.of(2024, 1, 1, 0, 0));
    }
    
    @Test
    @DisplayName("日期格式異常處理")
    void testInvalidDateFormat() {
        // Given
        OldCustomer source = new OldCustomer();
        source.setCustId("C0002");
        source.setCustName("李大華");
        source.setCustType("1");
        source.setCreateDt("2024/01/01");  // 錯誤格式
        
        // When & Then
        assertThatThrownBy(() -> processor.process(source))
            .isInstanceOf(DateTimeParseException.class);
    }
    
    @Test
    @DisplayName("空值處理")
    void testNullValues() throws Exception {
        // Given
        OldCustomer source = new OldCustomer();
        source.setCustId("C0003");
        source.setCustName(null);
        source.setCustType("2");
        source.setCreateDt("");
        
        // When
        NewCustomer result = processor.process(source);
        
        // Then
        assertThat(result.getCustomerNo()).isEqualTo("C0003");
        assertThat(result.getName()).isNull();
        assertThat(result.getCustomerType()).isEqualTo("CORPORATE");
        assertThat(result.getCreatedAt()).isNull();
    }
    
    @Test
    @DisplayName("未定義代碼處理")
    void testUnmappedCode() throws Exception {
        // Given
        OldCustomer source = new OldCustomer();
        source.setCustId("C0004");
        source.setCustName("張三");
        source.setCustType("X");  // 未定義代碼
        source.setCreateDt("20240101");
        
        // When
        NewCustomer result = processor.process(source);
        
        // Then
        assertThat(result.getCustomerType()).isEqualTo("UNMAPPED_X");
    }
    
    @ParameterizedTest
    @DisplayName("各種日期格式測試")
    @CsvSource({
        "20240101, 2024-01-01",
        "20241231, 2024-12-31",
        "20200229, 2020-02-29"  // 閏年
    })
    void testDateFormats(String input, String expected) throws Exception {
        OldCustomer source = new OldCustomer();
        source.setCustId("C0005");
        source.setCustName("Test");
        source.setCustType("1");
        source.setCreateDt(input);
        
        NewCustomer result = processor.process(source);
        
        assertThat(result.getCreatedAt().toLocalDate())
            .isEqualTo(LocalDate.parse(expected));
    }
}

@SpringBootTest
@Testcontainers
@ActiveProfiles("test")
class CustomerMigrationIntegrationTest {
    
    @Container
    static OracleContainer sourceDb = new OracleContainer("oracle:19c")
        .withInitScript("sql/source-init.sql");
    
    @Container
    static PostgreSQLContainer targetDb = new PostgreSQLContainer("postgres:15")
        .withInitScript("sql/target-init.sql");
    
    @Autowired
    private JobLauncherTestUtils jobLauncherTestUtils;
    
    @Autowired
    private JdbcTemplate targetJdbc;
    
    @Test
    @DisplayName("完整轉置流程測試")
    void testFullMigrationFlow() throws Exception {
        // Given: 準備測試資料
        setupTestData();
        
        // When: 執行轉置 Job
        JobParameters params = new JobParametersBuilder()
            .addString("batchId", "TEST_BATCH_001")
            .addLong("time", System.currentTimeMillis())
            .toJobParameters();
        
        JobExecution execution = jobLauncherTestUtils.launchJob(params);
        
        // Then: 驗證結果
        assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        
        // 驗證筆數
        Integer targetCount = targetJdbc.queryForObject(
            "SELECT COUNT(*) FROM new_customer WHERE migration_batch_id = ?",
            Integer.class, "TEST_BATCH_001"
        );
        assertThat(targetCount).isEqualTo(100);
        
        // 驗證金額
        BigDecimal totalBalance = targetJdbc.queryForObject(
            "SELECT SUM(balance) FROM new_account WHERE migration_batch_id = ?",
            BigDecimal.class, "TEST_BATCH_001"
        );
        assertThat(totalBalance).isEqualByComparingTo(new BigDecimal("1234567.89"));
    }
    
    @Test
    @DisplayName("錯誤資料處理測試")
    void testErrorHandling() throws Exception {
        // Given: 插入異常資料
        insertInvalidData();
        
        // When
        JobParameters params = new JobParametersBuilder()
            .addString("batchId", "TEST_BATCH_002")
            .addLong("time", System.currentTimeMillis())
            .toJobParameters();
        
        JobExecution execution = jobLauncherTestUtils.launchJob(params);
        
        // Then: 確認 Job 完成（錯誤被跳過）
        assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        
        // 確認錯誤被記錄
        Integer errorCount = targetJdbc.queryForObject(
            "SELECT COUNT(*) FROM migration_error_log WHERE batch_id = ?",
            Integer.class, "TEST_BATCH_002"
        );
        assertThat(errorCount).isGreaterThan(0);
    }
    
    @Test
    @DisplayName("Checkpoint 恢復測試")
    void testCheckpointRecovery() throws Exception {
        // Given: 執行到一半中斷
        JobParameters params = new JobParametersBuilder()
            .addString("batchId", "TEST_BATCH_003")
            .addLong("time", System.currentTimeMillis())
            .toJobParameters();
        
        // 模擬中斷
        simulateInterruption();
        
        // When: 重新啟動
        JobExecution execution = jobLauncherTestUtils.launchJob(params);
        
        // Then: 確認從 Checkpoint 恢復
        assertThat(execution.getStatus()).isEqualTo(BatchStatus.COMPLETED);
        
        // 確認沒有重複資料
        Integer duplicateCount = targetJdbc.queryForObject(
            """
            SELECT COUNT(*) FROM (
                SELECT customer_no, COUNT(*) AS cnt
                FROM new_customer
                WHERE migration_batch_id = ?
                GROUP BY customer_no
                HAVING COUNT(*) > 1
            ) t
            """,
            Integer.class, "TEST_BATCH_003"
        );
        assertThat(duplicateCount).isEqualTo(0);
    }
}

## UAT 驗證檢核表

### 基本資訊
- 專案名稱：_______________
- 轉置批次：_______________
- 驗證日期：_______________
- 驗證人員：_______________

### 一、技術驗證 ☑

| 項目 | 預期值 | 實際值 | 結果 | 備註 |
|------|--------|--------|------|------|
| 來源筆數 | 1,000,000 | | □ Pass □ Fail | |
| 目標筆數 | 1,000,000 | | □ Pass □ Fail | |
| 錯誤筆數 | < 100 | | □ Pass □ Fail | |
| 金額總和 | 123,456,789.00 | | □ Pass □ Fail | |
| 執行時間 | < 2 小時 | | □ Pass □ Fail | |

### 二、業務驗證 ☑

| 驗證情境 | 測試案例 | 預期結果 | 實際結果 | 結果 |
|---------|---------|---------|---------|------|
| 客戶查詢 | 查詢客戶 C001 | 顯示完整資料 | | □ Pass □ Fail |
| 帳戶餘額 | 帳戶 A001 餘額 | 50,000.00 | | □ Pass □ Fail |
| 交易明細 | 近 3 個月交易 | 顯示 15 筆 | | □ Pass □ Fail |

### 三、抽樣驗證 ☑

| 序號 | 客戶編號 | 驗證項目 | 來源值 | 目標值 | 結果 |
|------|---------|---------|--------|--------|------|
| 1 | | 姓名 | | | □ Match |
| 2 | | 餘額 | | | □ Match |
| 3 | | 狀態 | | | □ Match |

### 四、簽核

| 角色 | 姓名 | 簽名 | 日期 |
|------|------|------|------|
| 技術負責人 | | | |
| 業務負責人 | | | |
| 專案經理 | | | |

## 資料轉置上線前檢核清單

### 📋 T-7（上線前 7 天）

#### 環境確認
- [ ] 正式環境資料庫連線測試通過
- [ ] 正式環境帳號權限確認
- [ ] 備份機制確認可運作
- [ ] 監控告警設定完成

#### 程式確認
- [ ] 轉置程式已部署至正式環境
- [ ] 程式版本與 UAT 版本一致
- [ ] 設定檔已更新為正式環境參數
- [ ] Log 輸出路徑確認

#### 文件確認
- [ ] Mapping 文件最終版已確認
- [ ] 操作手冊已完成
- [ ] Rollback 手冊已完成
- [ ] 緊急聯絡清單已更新

### 📋 T-3（上線前 3 天）

#### 資料確認
- [ ] 來源資料筆數確認
- [ ] 來源資料品質檢測通過
- [ ] 代碼對應表已載入
- [ ] 測試資料已清除

#### 演練確認
- [ ] 正式環境演練已完成
- [ ] 演練時間符合預期
- [ ] 演練驗證報告已產出
- [ ] 演練問題已修復

### 📋 T-1（上線前 1 天）

#### 最終確認
- [ ] 上線計畫已發送相關人員
- [ ] 維護公告已發布
- [ ] 相關系統已通知
- [ ] 值班人員已確認

#### 備份確認
- [ ] 來源資料備份完成
- [ ] 目標資料庫備份完成
- [ ] 備份可還原已驗證

### 📋 D-Day（上線當天）

#### 上線前
- [ ] 停止相關應用服務
- [ ] 確認無進行中交易
- [ ] 建立最新備份點
- [ ] 記錄開始時間

#### 上線中
- [ ] 執行資料抽取 → 完成時間：______
- [ ] 執行資料轉換 → 完成時間：______
- [ ] 執行資料載入 → 完成時間：______
- [ ] 執行筆數驗證 → 結果：______
- [ ] 執行金額驗證 → 結果：______

#### 上線後
- [ ] 啟動相關應用服務
- [ ] 執行功能測試
- [ ] 業務確認通過
- [ ] 監控觀察正常

### 📋 D+1（上線後 1 天）

#### 確認事項
- [ ] 無異常錯誤回報
- [ ] 監控指標正常
- [ ] 業務運作正常
- [ ] 上線報告已產出

### ✅ 最終簽核

| 檢核項目 | 簽核人 | 簽核時間 |
|---------|--------|---------|
| 技術檢核完成 | | |
| 業務確認完成 | | |
| 上線許可 | | |

// 建立自訂字元映射表
Map<Character, String> customMapping = new HashMap<>();
customMapping.put('\uE000', "堃");  // 私有碼區字元
customMapping.put('\uE001', "燊");

public String convertEncoding(String input) {
    StringBuilder result = new StringBuilder();
    for (char c : input.toCharArray()) {
        if (customMapping.containsKey(c)) {
            result.append(customMapping.get(c));
        } else {
            result.append(c);
        }
    }
    return result.toString();
}

正常格式：20240101
異常格式：2024/1/1、24-01-01、00000000、99999999

public LocalDate parseDateFlexible(String dateStr) {
    if (dateStr == null || dateStr.trim().isEmpty()) {
        return null;
    }
    
    // 處理特殊值
    if ("00000000".equals(dateStr) || "99999999".equals(dateStr)) {
        return null;
    }
    
    // 嘗試多種格式
    List<DateTimeFormatter> formatters = Arrays.asList(
        DateTimeFormatter.ofPattern("yyyyMMdd"),
        DateTimeFormatter.ofPattern("yyyy/M/d"),
        DateTimeFormatter.ofPattern("yy-MM-dd"),
        DateTimeFormatter.ofPattern("yyyy-MM-dd")
    );
    
    for (DateTimeFormatter formatter : formatters) {
        try {
            return LocalDate.parse(dateStr.trim(), formatter);
        } catch (DateTimeParseException e) {
            // 繼續嘗試下一個格式
        }
    }
    
    // 全部失敗，記錄錯誤
    throw new DataException("無法解析日期: " + dateStr);
}

// 優化前：單筆處理
for (Transaction txn : transactions) {
    if (!exists(txn.getId())) {
        insert(txn);
    }
}

// 優化後：批次處理 + 並行
@Async
public CompletableFuture<Integer> processBatch(List<Transaction> batch) {
    // 批次檢查存在性
    Set<String> existingIds = findExistingIds(
        batch.stream().map(Transaction::getId).collect(toList())
    );
    
    // 過濾出需要新增的
    List<Transaction> toInsert = batch.stream()
        .filter(t -> !existingIds.contains(t.getId()))
        .collect(toList());
    
    // 批次寫入
    batchInsert(toInsert);
    
    return CompletableFuture.completedFuture(toInsert.size());
}

/**
 * 業務驗證用查詢介面
 */
@RestController
@RequestMapping("/api/migration/verify")
public class VerificationController {
    
    @Autowired
    private VerificationService verifyService;
    
    /**
     * 單筆資料比對
     */
    @GetMapping("/customer/{custId}")
    public CustomerCompareResult compareCustomer(@PathVariable String custId) {
        return verifyService.compareCustomer(custId);
    }
    
    /**
     * 抽樣驗證報表
     */
    @GetMapping("/sample-report")
    public SampleReport getSampleReport(
            @RequestParam String batchId,
            @RequestParam(defaultValue = "100") int sampleSize) {
        return verifyService.generateSampleReport(batchId, sampleSize);
    }
    
    /**
     * 匯出驗證資料（Excel）
     */
    @GetMapping("/export")
    public ResponseEntity<byte[]> exportVerificationData(
            @RequestParam String batchId) {
        byte[] excelData = verifyService.exportToExcel(batchId);
        
        return ResponseEntity.ok()
            .header(HttpHeaders.CONTENT_DISPOSITION, 
                "attachment; filename=verification_" + batchId + ".xlsx")
            .contentType(MediaType.APPLICATION_OCTET_STREAM)
            .body(excelData);
    }
}

-- 轉置稽核表
CREATE TABLE migration_audit_log (
    audit_id        BIGINT AUTO_INCREMENT PRIMARY KEY,
    batch_id        VARCHAR(20) NOT NULL,
    action_type     VARCHAR(50) NOT NULL,  -- EXTRACT, TRANSFORM, LOAD, VERIFY
    action_detail   TEXT,
    source_table    VARCHAR(100),
    target_table    VARCHAR(100),
    affected_rows   BIGINT,
    executed_by     VARCHAR(50),
    executed_at     TIMESTAMP DEFAULT CURRENT_TIMESTAMP,
    client_ip       VARCHAR(45),
    checksum        VARCHAR(64),  -- 資料校驗碼
    
    INDEX idx_batch (batch_id),
    INDEX idx_time (executed_at)
);

-- 記錄稽核日誌
INSERT INTO migration_audit_log 
    (batch_id, action_type, action_detail, source_table, target_table, 
     affected_rows, executed_by, checksum)
VALUES 
    ('BATCH_001', 'LOAD', 'Customer migration completed', 
     'old_customer', 'new_customer', 1234567, 'system', 
     SHA2(CONCAT('BATCH_001', '1234567', NOW()), 256));

/**
 * 資料血緣記錄
 */
@Entity
@Table(name = "data_lineage")
public class DataLineage {
    
    @Id
    @GeneratedValue(strategy = GenerationType.IDENTITY)
    private Long id;
    
    private String batchId;
    
    // 來源資訊
    private String sourceSystem;
    private String sourceTable;
    private String sourceKey;
    private String sourceChecksum;
    
    // 目標資訊
    private String targetSystem;
    private String targetTable;
    private String targetKey;
    private String targetChecksum;
    
    // 轉換資訊
    private String transformRules;  // JSON 格式
    private LocalDateTime transformTime;
    
    // 追蹤資訊
    private String createdBy;
    private LocalDateTime createdAt;
}

/**
 * 血緣查詢服務
 */
@Service
public class DataLineageService {
    
    /**
     * 追蹤資料來源
     */
    public LineageTrace traceBack(String targetTable, String targetKey) {
        List<DataLineage> lineage = lineageRepo
            .findByTargetTableAndTargetKey(targetTable, targetKey);
        
        LineageTrace trace = new LineageTrace();
        trace.setTargetTable(targetTable);
        trace.setTargetKey(targetKey);
        
        for (DataLineage l : lineage) {
            trace.addSource(new SourceInfo(
                l.getSourceSystem(),
                l.getSourceTable(),
                l.getSourceKey(),
                l.getTransformRules()
            ));
        }
        
        return trace;
    }
}

/**
 * 敏感資料處理
 */
@Service
public class SensitiveDataHandler {
    
    /**
     * 身分證字號脫敏
     */
    public String maskIdNumber(String idNo) {
        if (idNo == null || idNo.length() < 10) {
            return idNo;
        }
        // A123456789 → A12****789
        return idNo.substring(0, 3) + "****" + idNo.substring(7);
    }
    
    /**
     * 加密儲存
     */
    public String encrypt(String plainText) {
        // 使用 AES-256 加密
        return aesEncryptor.encrypt(plainText);
    }
    
    /**
     * 轉置時的敏感資料處理
     */
    public NewCustomer processSensitiveData(OldCustomer source, boolean isMasked) {
        NewCustomer target = new NewCustomer();
        
        // 一般欄位直接轉
        target.setCustomerNo(source.getCustId());
        target.setName(source.getCustName());
        
        // 敏感欄位處理
        if (isMasked) {
            target.setIdNumber(maskIdNumber(source.getIdNo()));
        } else {
            target.setIdNumberEncrypted(encrypt(source.getIdNo()));
        }
        
        return target;
    }
}

## 資料轉置合規檢查清單

### 個人資料保護
- [ ] 敏感欄位已識別並標註
- [ ] 傳輸過程已加密（TLS 1.2+）
- [ ] 靜態資料已加密或脫敏
- [ ] 存取權限已設定最小權限原則
- [ ] 個資存取已記錄稽核日誌

### 資料完整性
- [ ] 所有資料都有驗證機制
- [ ] 金額類資料採用 Decimal 避免精度問題
- [ ] 關鍵欄位設定 NOT NULL 約束
- [ ] 參照完整性已驗證

### 稽核要求
- [ ] 轉置執行記錄完整
- [ ] 變更歷程可追溯
- [ ] 錯誤處理記錄保存
- [ ] 驗證報告已存檔並簽核

### 資料保存
- [ ] 資料保存期限已確認
- [ ] 歷史資料歸檔機制已建立
- [ ] 資料刪除機制已設計
- [ ] 備份還原機制已驗證

## 專案啟動檢查清單

### 範圍確認
- [ ] 轉置範圍已明確定義並簽核
- [ ] 舊系統資料清冊已完成
- [ ] 新系統資料模型已確認
- [ ] 轉置時程已規劃

### 團隊組成
- [ ] 專案經理已指派
- [ ] 技術負責人已指派
- [ ] 業務窗口已確認
- [ ] DBA 支援已協調

### 環境準備
- [ ] 開發環境已建置
- [ ] 測試環境已建置
- [ ] 來源資料存取權限已取得
- [ ] 目標資料庫已建立

## 分析設計檢查清單

### 舊系統分析
- [ ] 所有資料來源已盤點
- [ ] Table/File 結構已文件化
- [ ] Key 與關聯已分析
- [ ] 資料品質已檢測
- [ ] 資料量已統計

### 新系統設計
- [ ] 資料模型已設計
- [ ] Mapping 規則已定義
- [ ] 代碼對應表已建立
- [ ] 歷史資料策略已確認

### 轉置策略
- [ ] 轉置模式已決定（Big Bang / Parallel Run）
- [ ] 批次策略已規劃
- [ ] Rollback 機制已設計
- [ ] 驗證機制已設計

## 開發測試檢查清單

### 程式開發
- [ ] Extract 程式已完成
- [ ] Transform 程式已完成
- [ ] Load 程式已完成
- [ ] Error Handling 已實作
- [ ] Checkpoint 機制已實作

### 單元測試
- [ ] 轉換邏輯測試通過
- [ ] 邊界條件測試通過
- [ ] 錯誤處理測試通過
- [ ] 效能測試通過

### 整合測試
- [ ] End-to-End 測試通過
- [ ] 筆數驗證通過
- [ ] 金額驗證通過
- [ ] Rollback 測試通過

## UAT 檢查清單

### 測試準備
- [ ] UAT 環境已建置
- [ ] 測試資料已準備
- [ ] 測試案例已確認
- [ ] 業務人員已訓練

### 測試執行
- [ ] 技術驗證通過
- [ ] 業務驗證通過
- [ ] 抽樣驗證通過
- [ ] 效能驗證通過

### 問題處理
- [ ] 所有問題已記錄
- [ ] 關鍵問題已修復
- [ ] 修復後回歸測試通過
- [ ] UAT 簽核已完成

## 上線檢查清單

### 上線前（T-1）
- [ ] 上線計畫已發送
- [ ] 維護公告已發布
- [ ] 備份已完成
- [ ] Rollback 腳本已準備
- [ ] 值班人員已確認

### 上線中（D-Day）
- [ ] 系統已停機
- [ ] 資料抽取完成 → 時間：______
- [ ] 資料轉換完成 → 時間：______
- [ ] 資料載入完成 → 時間：______
- [ ] 驗證通過 → 結果：______
- [ ] 系統已啟動
- [ ] 功能測試通過

### 上線後（D+1）
- [ ] 監控正常
- [ ] 無異常回報
- [ ] 業務確認正常
- [ ] 上線報告已產出

-- B.1.1 完整的資料品質檢測腳本
-- =====================================================

-- 1. 空值檢測
SELECT 
    'NULL_CHECK' AS check_type,
    column_name,
    COUNT(*) AS null_count,
    ROUND(COUNT(*) * 100.0 / (SELECT COUNT(*) FROM target_table), 2) AS null_pct
FROM (
    SELECT 'CUST_NAME' AS column_name FROM target_table WHERE cust_name IS NULL
    UNION ALL
    SELECT 'CUST_TYPE' FROM target_table WHERE cust_type IS NULL
    UNION ALL
    SELECT 'CREATE_DATE' FROM target_table WHERE create_date IS NULL
) t
GROUP BY column_name;

-- 2. 重複值檢測
SELECT 
    'DUPLICATE_CHECK' AS check_type,
    cust_id,
    COUNT(*) AS dup_count
FROM target_table
GROUP BY cust_id
HAVING COUNT(*) > 1;

-- 3. 格式驗證（身分證字號）
SELECT 
    'FORMAT_CHECK' AS check_type,
    'ID_NO' AS column_name,
    COUNT(*) AS invalid_count
FROM target_table
WHERE id_no IS NOT NULL
  AND NOT REGEXP_LIKE(id_no, '^[A-Z][12][0-9]{8}$');

-- 4. 範圍驗證
SELECT 
    'RANGE_CHECK' AS check_type,
    'BALANCE' AS column_name,
    COUNT(*) AS invalid_count
FROM target_table
WHERE balance < 0;

-- 5. 參照完整性
SELECT 
    'FK_CHECK' AS check_type,
    'CUST_ID' AS column_name,
    COUNT(*) AS orphan_count
FROM account a
LEFT JOIN customer c ON a.cust_id = c.cust_id
WHERE c.cust_id IS NULL;

-- B.2.1 新舊資料比對腳本
-- =====================================================

-- 1. 筆數比對
WITH counts AS (
    SELECT 'SOURCE' AS system, COUNT(*) AS cnt FROM old_customer
    UNION ALL
    SELECT 'TARGET', COUNT(*) FROM new_customer WHERE migration_batch_id IS NOT NULL
)
SELECT 
    MAX(CASE WHEN system = 'SOURCE' THEN cnt END) AS source_count,
    MAX(CASE WHEN system = 'TARGET' THEN cnt END) AS target_count,
    MAX(CASE WHEN system = 'SOURCE' THEN cnt END) - 
    MAX(CASE WHEN system = 'TARGET' THEN cnt END) AS diff
FROM counts;

-- 2. 金額比對
SELECT 
    SUM(s.balance) AS source_balance,
    SUM(t.balance) AS target_balance,
    SUM(s.balance) - SUM(t.balance) AS diff,
    CASE 
        WHEN ABS(SUM(s.balance) - SUM(t.balance)) < 0.01 THEN 'PASS'
        ELSE 'FAIL'
    END AS result
FROM old_account s
FULL OUTER JOIN new_account t 
    ON s.acct_no = t.account_no;

-- 3. 欄位值比對
SELECT 
    s.cust_id,
    'NAME_MISMATCH' AS mismatch_type,
    s.cust_name AS source_value,
    t.name AS target_value
FROM old_customer s
JOIN new_customer t ON s.cust_id = t.customer_no
WHERE TRIM(s.cust_name) <> TRIM(t.name)
UNION ALL
SELECT 
    s.cust_id,
    'TYPE_MISMATCH',
    s.cust_type,
    t.customer_type
FROM old_customer s
JOIN new_customer t ON s.cust_id = t.customer_no
WHERE s.cust_type <> (
    SELECT source_code 
    FROM migration_code_mapping 
    WHERE category = 'CUST_TYPE' AND target_code = t.customer_type
);

-- B.3.1 轉置進度查詢
-- =====================================================

-- 1. 批次進度總覽
SELECT 
    batch_id,
    table_name,
    status,
    total_rows,
    processed_rows,
    ROUND(processed_rows * 100.0 / NULLIF(total_rows, 0), 2) AS progress_pct,
    start_time,
    end_time,
    TIMESTAMPDIFF(MINUTE, start_time, COALESCE(end_time, NOW())) AS duration_min
FROM migration_batch_control
ORDER BY batch_id;

-- 2. 錯誤統計
SELECT 
    batch_id,
    error_phase,
    error_type,
    COUNT(*) AS error_count,
    COUNT(CASE WHEN resolved_at IS NOT NULL THEN 1 END) AS resolved_count
FROM migration_error_log
GROUP BY batch_id, error_phase, error_type
ORDER BY batch_id, error_count DESC;

-- 3. 每小時處理量趨勢
SELECT 
    DATE_FORMAT(load_time, '%Y-%m-%d %H:00') AS hour,
    COUNT(*) AS records_loaded
FROM stg_customer
WHERE process_status = 'LOADED'
GROUP BY DATE_FORMAT(load_time, '%Y-%m-%d %H:00')
ORDER BY hour;