Notes

import pyodbc
import pandas as pd
import time
from tqdm import tqdm

inicio = time.time() # iniciar cronómetro

# Configuración de conexión
servidor = 'MI_SERVIDOR\MI_INSTANCIA' # <--- Cambiar por su servidor
base_datos = 'MiBaseDeDatos' # <--- Cambiar por su base de datos
conn = pyodbc.connect(
f"DRIVER={{ODBC Driver 18 for SQL Server}};"
f"SERVER={servidor};DATABASE={base_datos};"
f"Trusted_Connection=yes;"
f"Encrypt=no;" # o usar TrustServerCertificate=yes si usas cifrado
)
cursor = conn.cursor()

# Obtener lista de tablas (esquema y nombre)
cursor.execute("""
SELECT TABLE_SCHEMA, TABLE_NAME
FROM INFORMATION_SCHEMA.TABLES
WHERE TABLE_TYPE = 'BASE TABLE';
""")
tablas = cursor.fetchall()

resultados = []
tamaño_lote = 20
total_tablas = len(tablas)
procesadas = 0

for i in range(0, total_tablas, tamaño_lote):
lote = tablas[i:i+tamaño_lote]
print(f"n📦 Procesando lote {i//tamaño_lote + 1} de {((total_tablas - 1) // tamaño_lote) + 1}...")

for esquema, nombre in tqdm(lote, desc="Analizando tablas", unit="tabla"):
tabla_full = f"{esquema}.{nombre}"
procesadas += 1
print(f"→ {procesadas}/{total_tablas} analizando: {tabla_full}")

# 1. Número de filas
cursor.execute(f"""
SELECT SUM(row_count)
FROM sys.dm_db_partition_stats
WHERE object_id = OBJECT_ID('{tabla_full}') AND index_id < 2;
""")
filas = cursor.fetchone()[0] or 0

# 2. Número de columnas
cursor.execute(f"""
SELECT COUNT(*)
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA='{esquema}' AND TABLE_NAME='{nombre}';
""")
cols = cursor.fetchone()[0] or 0

# 3. Particionamiento
cursor.execute(f"""
SELECT COUNT(DISTINCT partition_number)
FROM sys.partitions
WHERE object_id = OBJECT_ID('{tabla_full}') AND index_id < 2;
""")
num_part = cursor.fetchone()[0] or 0
particionado = "Sí" if num_part and num_part > 1 else "No"

# 4. Tipo de índice principal
cursor.execute(f"""
SELECT type, type_desc
FROM sys.indexes
WHERE object_id = OBJECT_ID('{tabla_full}') AND index_id IN (0, 1);
""")
idx_info = cursor.fetchall()
tipo_idx = "Heap"
for idx_type, idx_desc in idx_info:
if idx_type == 5:
tipo_idx = "Clustered Columnstore"
elif idx_type == 1:
tipo_idx = "Clustered Index"
elif idx_type == 0:
tipo_idx = "Heap"
cursor.execute(f"""
SELECT COUNT(*)
FROM sys.indexes
WHERE object_id = OBJECT_ID('{tabla_full}') AND type = 6;
""")
has_colstore = cursor.fetchone()[0] or 0
if has_colstore:
if tipo_idx == "Clustered Index":
tipo_idx = "Clustered Index + Nonclustered Columnstore"
elif tipo_idx == "Heap":
tipo_idx = "Heap + Nonclustered Columnstore"

# 5. Tamaño
cursor.execute(f"""
SELECT SUM(reserved_page_count)*8.0/1024
FROM sys.dm_db_partition_stats
WHERE object_id = OBJECT_ID('{tabla_full}');
""")
tam_mb = cursor.fetchone()[0] or 0.0
tam_mb = float(tam_mb) # conversión segura
if tam_mb >= 1024:
tam_gb = tam_mb / 1024.0
tam_str = f"{tam_gb:.2f} GB"
else:
tam_str = f"{tam_mb:.2f} MB"

# 6. Identificación de columna de fecha
cursor.execute(f"""
SELECT COLUMN_NAME, DATA_TYPE
FROM INFORMATION_SCHEMA.COLUMNS
WHERE TABLE_SCHEMA='{esquema}' AND TABLE_NAME='{nombre}';
""")
columnas_info = cursor.fetchall()
col_fecha = None
keywords_fecha = ["fecha", "date", "period"]
for col_name, data_type in columnas_info:
if data_type.lower() in ("int", "bigint") and any(kw in col_name.lower() for kw in keywords_fecha):
col_fecha = col_name
break
col_fecha = col_fecha or "N/A"

# 7. Sugerencia de optimización
sugerencia = []
if filas > 1000000:
if "Columnstore" in tipo_idx:
sugerencia.append("Utiliza columnstore, verificar mantenimiento (eliminar fragmentación, etc).")
else:
sugerencia.append("Considerar índice columnstore para compresión y aceleración de consultas.")
if tipo_idx.startswith("Heap"):
sugerencia.append("Agregar índice clustered para evitar heap y mejorar rendimiento.")
if particionado == "No" and col_fecha != "N/A" and filas > 1000000:
sugerencia.append(f"Evaluar particionado por {col_fecha} para manejar datos históricos.")
if cols > 50:
sugerencia.append("Tabla muy amplia; revisar si conviene normalizar o eliminar columnas no usadas.")
if filas < 1000 and tam_mb < 50:
sugerencia.append("Tabla pequeña; mantenimiento estándar (estadísticas, índices) suficiente.")
sugerencia_txt = " ".join(sugerencia) if sugerencia else "Sin acciones importantes sugeridas."

# Guardar resultado
resultados.append({
"Tabla": tabla_full,
"Filas": int(filas),
"Columnas": int(cols),
"Particiones": particionado,
"Índice": tipo_idx,
"Tamaño": tam_str,
"Columna_Fecha": col_fecha,
"Sugerencia": sugerencia_txt
})

# Pausa entre lotes
time.sleep(15)

# Exportar a Excel
df = pd.DataFrame(resultados)
writer = pd.ExcelWriter('AnalisisTablas.xlsx', engine='xlsxwriter')
df.to_excel(writer, sheet_name='Resumen', index=False)
worksheet = writer.sheets['Resumen']
workbook = writer.book

header_fmt = workbook.add_format({'bold': True})
worksheet.set_row(0, None, header_fmt)
worksheet.freeze_panes(1, 0)
worksheet.autofilter(0, 0, len(df), len(df.columns)-1)
for idx, col in enumerate(df.columns):
max_len = max(len(col), df[col].astype(str).map(len).max()) + 2
worksheet.set_column(idx, idx, max_len)
writer.save()

# Tiempo total
fin = time.time()
print(f"n✅ Análisis completado en {round((fin - inicio)/60, 2)} minutos.")