Migracion de Datos

-- Exportar tabla completa a CSV via SQL
-- (usando CPYTOIMPF desde CL o query directo)

-- Metodo 1: Query directo via JDBC/ODBC
SELECT
  CUSID,
  TRIM(CUSNAM) AS customer_name,
  TRIM(CUSADR) AS address,
  CUSBAL AS balance,
  CHAR(OPNDAT, ISO) AS open_date,
  CASE CUSTS
    WHEN 'A' THEN 'Active'
    WHEN 'I' THEN 'Inactive'
    WHEN 'C' THEN 'Closed'
  END AS status
FROM PRODLIB.CUSTMAST
WHERE CUSTS <> 'C'
ORDER BY CUSID;

-- Metodo 2: Unload a archivo delimitado
CREATE TABLE QTEMP.EXPORT_CUSTOMERS AS (
  SELECT CUSID, CUSNAM, CUSADR, CUSBAL
  FROM PRODLIB.CUSTMAST
) WITH DATA;

-- Exportar con CPYTOIMPF (metodo CL nativo)
CPYTOIMPF FROMFILE(PRODLIB/CUSTMAST)
          TOSTMF('/home/exports/custmast.csv')
          MBROPT(*REPLACE)
          STMFCCSID(1208)        -- UTF-8
          RCDDLM(*CRLF)
          DTAFMT(*DLM)
          STRDLM(*DBLQ)
          FLDDLM(',')

-- Exportar con SQL
RUNSQLSTM SRCSTMF('/home/scripts/export_cust.sql')
          COMMIT(*NONE)
          NAMING(*SQL)

import pyodbc
import pandas as pd

# Conectar y extraer
conn = pyodbc.connect('DSN=IBMi_PROD')
df = pd.read_sql("""
    SELECT CUSID, CUSNAM, CUSADR, CUSBAL,
           OPNDAT, CUSTS
    FROM PRODLIB.CUSTMAST
""", conn)

# Transformaciones
df['customer_name'] = df['CUSNAM'].str.strip()
df['address'] = df['CUSADR'].str.strip()
df['balance'] = df['CUSBAL'].astype(float)
df['open_date'] = pd.to_datetime(
    df['OPNDAT'], format='%Y%m%d'
)
df['status'] = df['CUSTS'].map({
    'A': 'Active', 'I': 'Inactive', 'C': 'Closed'
})

# Validar
assert df['CUSID'].is_unique, "IDs duplicados!"
assert df['balance'].notna().all(), "Balances nulos!"

# Cargar en PostgreSQL
from sqlalchemy import create_engine
pg_engine = create_engine(
    'postgresql://user:pass@host/db'
)
df.to_sql('customers', pg_engine,
          if_exists='replace', index=False)

-- Leer cambios de las ultimas 24 horas
SELECT
  jrn.JOURNAL_ENTRY_TYPE AS operation,
  jrn.ENTRY_TIMESTAMP AS change_time,
  jrn.CUSID,
  jrn.CUSNAM,
  jrn.CUSBAL
FROM TABLE(
  QSYS2.DISPLAY_JOURNAL(
    JOURNAL_LIBRARY => 'MYLIB',
    JOURNAL_NAME => 'MYJRN',
    STARTING_TIMESTAMP =>
      CURRENT_TIMESTAMP - 24 HOURS,
    JOURNAL_ENTRY_TYPES => 'PT,UP,DL',
    JOURNAL_CODES => 'R',
    FILE_NAME => 'CUSTMAST'
  )
) AS jrn
ORDER BY jrn.ENTRY_TIMESTAMP;

-- Tipos de operacion:
-- PT = Put (INSERT)
-- UP = Update
-- DL = Delete
-- Los campos contienen el AFTER image
-- (con IMAGES(*BOTH) tambien se obtiene el BEFORE)

-- Ver el CCSID de una tabla
SELECT TABLE_SCHEMA, TABLE_NAME, COLUMN_NAME,
       CCSID, DATA_TYPE, CHARACTER_MAXIMUM_LENGTH
FROM QSYS2.SYSCOLUMNS
WHERE TABLE_SCHEMA = 'MYLIB'
  AND TABLE_NAME = 'CUSTMAST';

-- Convertir datos al extraer
SELECT
  CAST(CUSNAM AS VARCHAR(40) CCSID 1208) AS name_utf8,
  CAST(CUSADR AS VARCHAR(60) CCSID 1208) AS addr_utf8
FROM MYLIB.CUSTMAST;

-- Cambiar CCSID de un archivo IFS
CHGATR OBJ('/home/exports/data.csv')
       ATR(*CCSID)
       VALUE(1208)

def decode_packed_decimal(data: bytes, decimals: int) -> float:
    """Decodifica packed decimal de IBM i."""
    result = 0
    for i, byte in enumerate(data):
        if i < len(data) - 1:
            # Cada byte tiene 2 digitos
            high = (byte >> 4) & 0x0F
            low = byte & 0x0F
            result = result * 100 + high * 10 + low
        else:
            # Ultimo byte: 1 digito + signo
            high = (byte >> 4) & 0x0F
            sign = byte & 0x0F
            result = result * 10 + high
            if sign in (0x0D, 0x0B):  # D o B = negativo
                result = -result

    return result / (10 ** decimals)

# Ejemplo: b'\x01\x23\x45\x67\x0F' = 12345.67
value = decode_packed_decimal(
    b'\x01\x23\x45\x67\x0F', decimals=2
)
print(value)  # 12345.67

-- 1. Validar conteos
-- En IBM i (origen):
SELECT COUNT(*) AS total_origin FROM PRODLIB.CUSTMAST;
-- En PostgreSQL (destino):
SELECT COUNT(*) AS total_dest FROM public.customers;
-- Deben coincidir!

-- 2. Validar sumas de control
-- En IBM i:
SELECT
  COUNT(*) AS rows,
  SUM(CUSBAL) AS sum_balance,
  MIN(CUSBAL) AS min_balance,
  MAX(CUSBAL) AS max_balance,
  COUNT(DISTINCT CUSTS) AS distinct_status
FROM PRODLIB.CUSTMAST;

-- En destino: misma query, comparar resultados

-- 3. Validar datos especificos (sampling)
-- Comparar 100 registros aleatorios
SELECT CUSID, CUSNAM, CUSBAL
FROM PRODLIB.CUSTMAST
ORDER BY RAND()
FETCH FIRST 100 ROWS ONLY;

-- 4. Detectar truncamientos
SELECT CUSID, CUSNAM
FROM PRODLIB.CUSTMAST
WHERE LENGTH(TRIM(CUSNAM)) > 40;  -- limite del destino

-- 5. Detectar caracteres problematicos
SELECT CUSID, CUSNAM, HEX(CUSNAM) AS hex_value
FROM PRODLIB.CUSTMAST
WHERE CUSNAM LIKE '%?%'  -- caracteres no convertidos