Recetario Práctico de Ingeniería de Datos: Manipulación de Archivos CSV, JSON y TXT con Python y Pandas

Ingeniería industrial de procesos, , , ,

Procesamiento de Datos Estructurados con Python y Pandas

1. Manipulación de DataFrames con Pandas

1.1. Carga y Filtrado de Datos JSON

Instrucción: Carga el fichero JSON en un DataFrame, modo lectura, con codificación 'utf-8'.

import json
import pandas as pd

ruta_archivo_json = 'T4/data/usuarios.json'

with open(ruta_archivo_json, 'r', encoding="utf-8") as archivo_json:
    datos_json = json.load(archivo_json)

df = pd.DataFrame(datos_json)

Instrucción: Crea un nuevo objeto DataFrame, mostrando solo los usuarios activos (activo = True) y con edad mayor a 40.

df_filtrado2 = df[(df["activo"] == True) & (df["edad"] > 40)]
print(df_filtrado2)

1.2. Creación y Exploración de DataFrames (CSV)

Instrucción: Creación de un DataFrame a partir de un diccionario de datos.

data = {datos que te den}
df = pd.DataFrame(data)

Instrucción: Carga el fichero CSV en un DataFrame usando Pandas.

df = pd.read_csv("T4/data/alumnos.csv")
Inspección del DataFrame
  • Inspecciona la estructura del DataFrame:
print(df.info())
Muestra las primeras 3 filas:
print(df.head(3))
Visualiza estadísticas básicas:
print(df.describe())
Selección de Columnas
  1. Muestra solo la columna CiudadDestino.
print(df["CiudadDestino"])
Muestra las columnas CiudadOrigen y Medio.
print(df[["CiudadOrigen", "Medio"]])

1.3. Filtrado y Transformación de Datos

Filtrado Condicional
  • Muestra solo los viajes cuyo medio sea Tren.
print(df[df["Medio"] == "Tren"])
Muestra solo los viajes cuya distancia sea mayor a 500 km.
print(df[df["Distancia_km"] > 500])
Filtra los viajes cuya ciudad de origen sea Madrid.
print(df[df["CiudadOrigen"] == "Madrid"])
Creación y Modificación de Columnas
  • Crea una columna llamada Precio-por_km mediante operaciones.
df["Precio-por_km"] = df["Precio"] / df["Distancia_km"]
Crea la columna Promedio de notas.
df["Promedio"] = (df["Matematicas"] + df["Lengua"] + df["Ciencias"]) / 3
Muestra por pantalla el alumno y su Promedio.
print(df.loc[:, ["alumno", "Promedio"]])
Crea una nueva columna llamada Aprobado y asigna 1 si el promedio es mayor a 5, 0 si no.
df["Aprobado"] = (df["Promedio"] > 5).astype(int)
print(df)
Reemplazar valores en una columna (ejemplo: Bus por Autobús).
df["Medio"] = df["Medio"].replace("Bus", "Autobús")
Eliminar columna (ejemplo: Precio_por_km).
df = df.drop(columns=["Precio_por_km"])
print(df.head())

1.4. Exportación de DataFrames

Instrucción: Exportar datos filtrados y el DataFrame completo.

# Guarda en un archivo nuevo (df_nuevo.csv) únicamente los viajes desde Madrid.
Df_nuevo = df[df["CiudadOrigen"] == "Madrid"]

# Guarda el DataFrame filtrado en un archivo CSV, sin incluir el índice.
Df_nuevo.to_csv("df_nuevo.csv", index=False)

1.5. Transformación de JSON a JSON usando Pandas

Instrucción: Cargar un archivo JSON, crear una nueva columna de precio de venta y exportar el resultado a un nuevo JSON.

import pandas as pd

# Cargamos el archivo
df = pd.read_json("T4/data/productos.json")
print(df.head())

# Crea la nueva columna basada en el precio de compra (incremento del 20%)
df["precio_venta"] = df["precio_compra"] + (df["precio_compra"] * 0.2)

# Imprime todo el DataFrame
print(df)

# Guarda en un nuevo archivo json llamado “precio_venta.json”
df.to_json("T4/data/precio_venta.json", orient="records", indent=4, index=False)

# Muestra en pantalla solo aquellos productos que tengan una cantidad menor a 10.
print(df.loc[df["cantidad"] < 10, ["id", "nombre", "cantidad"]])

# Utiliza el mismo filtro de antes, pero para almacenar ese resultado en una nueva variable (objeto DataFrame).
df_nuevo = df.loc[df["cantidad"] < 10, ["id", "nombre", "cantidad"]]

# Inspecciona el nuevo DataFrame
print(df_nuevo.info())

# Guarda este DataFrame como un nuevo archivo json
df_nuevo.to_json("T4/data/pendiente_compra.json", orient="records", indent=4, index=False)

2. Operaciones con Archivos CSV (Módulo Estándar)

2.1. Búsqueda Condicional en CSV

Instrucción: Encontrar un registro e imprimir otros datos usando el primero como condición (ejemplo: buscar empleados nacidos en 1980).

import csv

ruta_fichero_empleado = "T3/data/datos_empleados.csv"
anio_a_buscar = '1980'

with open(ruta_fichero_empleado, "r", encoding="utf-8") as f:
    # Apertura del fichero modo csv, con delimitador ','
    reader = csv.reader(f, delimiter=',')
    
    # Recorrer el fichero modo lista.
    for empleado in reader:
        # Condición si el año a buscar está en la fecha de nacimiento (índice 3).
        if anio_a_buscar in empleado[3]:
            # Muestra por pantalla si se da la condición.
            print(empleado[2], empleado[0], empleado[1])

2.2. Filtrado y Exportación de CSV

Instrucción: Leer un CSV, encontrar filas con un dato específico y guardar el resultado en otro CSV.

import csv

ruta_fichero_empleado = "T3/data/datos_empleados.csv"
ruta_fichero_empleado_filtrado = "T3/data/datos_empleados_filtrado.csv"
lista_profesiones = ['Especialista en IA', 'Ingeniero de Datos']
lista_perfiles = []

with open(ruta_fichero_empleado, "r", encoding="utf-8") as f:
    # Apertura del fichero modo csv, con delimitador ','
    reader = csv.reader(f, delimiter=',')
    
    # Recorrer el fichero modo lista.
    for empleado in reader:
        # Obtiene la profesión (índice 4)
        profesion = empleado[4]
        
        # Comprueba que la profesión está en la lista lista_profesiones:
        if profesion in lista_profesiones:
            # Añade elemento a la lista_perfiles.
            lista_perfiles.append(empleado)

with open(ruta_fichero_empleado_filtrado, "w", encoding="utf-8", newline="\n") as f:
    # Crea el escritor CSV
    escritor_csv = csv.writer(f)
    
    # Escribe el contenido de la lista
    escritor_csv.writerows(lista_perfiles)

2.3. Recuento de Profesiones y Exportación a JSON

Instrucción: Realizar el recuento de profesiones en un CSV.

import csv

ruta_fichero_empleado = "T3/data/datos_empleados.csv"
conteo = {'Consultor': 0, 'Analista': 0, 'Ingeniero de Datos': 0}

with open(ruta_fichero_empleado, "r", encoding="utf-8") as f:
    # Apertura del fichero modo csv, con delimitador ','
    reader = csv.reader(f, delimiter=',')
    
    # Recorrer el fichero modo lista.
    for empleado in reader:
        # Obtiene la profesión
        profesion = empleado[4]
        
        # Comprueba que la profesión está en el diccionario conteo:
        if profesion in conteo:
            # Incrementa el valor del diccionario.
            conteo[profesion] = conteo[profesion] + 1

print(conteo)

Instrucción: Guardar el recuento de profesiones en un archivo JSON, añadiendo la ruta del fichero original.

import json
# Nota: Se requiere definir 'ruta_fichero_salida_json'

with open(ruta_fichero_salida_json, 'w', encoding="utf-8") as f:
    conteo["Fichero original"] = ruta_fichero_empleado
    json.dump(conteo, f, indent=4)

3. Manipulación de Archivos JSON (Módulo Estándar)

3.1. Lectura, Modificación y Guardado de Usuarios

Instrucción: Lee usuarios de un JSON, incrementa la edad, recalcula la categoría y guarda el resultado en un nuevo JSON.

import json

ruta_fichero_usuarios = "T3/data/usuarios.json"
ruta_fichero_usuarios_modificado = "T3/data/usuarios_modificado.json"

# 1. Leer el archivo JSON completo
with open(ruta_fichero_usuarios, "r", encoding="utf-8") as f:
    # 2. Leemos el contenido en la variable usuarios
    usuarios = json.load(f)
    
    # 3. Iteramos cada elemento del JSON en la variable usuario
    for usuario in usuarios:
        # 4. Comprobamos si el usuario está activo.
        if usuario['activo'] is True:    
            # 5. Actualizamos la edad del usuario.
            usuario['edad'] = usuario['edad'] + 1
            
            # 6. Creamos la categoría con el valor correcto.
            if usuario['edad'] < 30:
                usuario['categoria'] = 'Joven'
            elif usuario['edad'] >= 60:
                usuario['categoria'] = 'Senior'
            else:
                usuario['categoria'] = 'Adulto'

# 7. Escribir en un fichero JSON completo
with open(ruta_fichero_usuarios_modificado, "w", encoding="utf-8") as f:
    json.dump(usuarios, f, indent=4, ensure_ascii=False)
    print("Fichero creado con éxito!")

3.2. Adición de un Nuevo Usuario a JSON

Instrucción: Añadir un usuario nuevo incrementando el valor del ID del último registro.

import json

ruta_fichero_usuarios = "T3/data/usuarios.json"
nuevo_usuario = {
    "id": 0,
    "nombre": "Manuel",
    "edad": 22,
    "pais": "España",
    "activo": False
}

# 1. Leer el archivo JSON completo
with open(ruta_fichero_usuarios, "r", encoding="utf-8") as f:
    usuarios = json.load(f)

# 2.1 Obtenemos el ID del último usuario
ultimo = usuarios[-1] 

# 2.2 Incrementamos el valor del ID para el nuevo usuario.
nuevo_usuario["id"] = ultimo["id"] + 1

# 2.3 Añadimos el nuevo usuario a la lista
usuarios.append(nuevo_usuario)

# 3. Sobreescribimos el fichero.
with open(ruta_fichero_usuarios, "w", encoding="utf-8") as f:
    json.dump(usuarios, f, indent=4, ensure_ascii=False)

4. Gestión de Archivos de Texto Plano (TXT)

4.1. Conteo de Palabras en un Fichero TXT

Instrucción: Función para contar el número total de palabras en el contenido de un archivo.

def cuenta_palabras(contenido):
    lista_palabras = contenido.split()
    total = len(lista_palabras)
    return total

# Asumiendo que 'ruta_absoluta' está definida
with open(ruta_absoluta, "r", encoding="utf-8") as f:
    contenido = f.read()
    total = cuenta_palabras(contenido)
    print(f"El fichero '{f.name}' tiene {total} palabras.")

4.2. Búsqueda Básica de Palabras

Instrucción: Función para buscar la frecuencia de una palabra específica.

def busca_palabra(palabra: str, contenido: str) -> int:
    lista_palabras = contenido.split()
    total = 0
    if palabra in lista_palabras:
        total = lista_palabras.count(palabra)
    return total

# Asumiendo que 'ruta_absoluta' está definida
with open(ruta_absoluta, "r", encoding="utf-8") as f:
    contenido = f.read()
    palabra_a_buscar = str(input("¿Qué palabra busco? \n"))
    total = busca_palabra(palabra_a_buscar, contenido)
    print(f"El fichero '{f.name}' tiene {total} ocurrencias de '{palabra_a_buscar}'.")

4.3. Búsqueda de Palabras Optimizada (Ignorando Puntuación y Caso)

Instrucción: Función mejorada para buscar la frecuencia de una palabra, ignorando mayúsculas/minúsculas y signos de puntuación.

def busca_palabra_mejorado(palabra: str, contenido: str) -> int:
# Reemplazo de puntuación por espacios
    contenido = contenido.replace(",", " ").replace(".", " ").replace("*", " ")
    
# Conversión de contenido y palabra a minúsculas
    contenido = contenido.lower()
    palabra = palabra.lower()
    
# Crea la lista separada por espacios
    lista_palabras = contenido.split()
    total = 0
    
# Comprueba que la palabra está en la lista y cuenta las ocurrencias
    if palabra in lista_palabras:
        total = lista_palabras.count(palabra)
    return total

# Asumiendo que 'ruta_absoluta' está definida
with open(ruta_absoluta, "r", encoding="utf-8") as f:
    contenido = f.read()
    palabra_a_buscar = str(input("¿Qué palabra busco? \n"))
    total = busca_palabra_mejorado(palabra_a_buscar, contenido)
    print(f"El fichero '{f.name}' tiene {total} ocurrencias de '{palabra_a_buscar}'.")

4.4. Extracción de Líneas y Creación de Ficheros

Instrucción: Crear un fichero de salida guardando las primeras N líneas de otro fichero.

fichero_salida = "output.txt"

def extrae_lineas(n_lineas: int):
# Asumiendo que 'ruta_absoluta' está definida
    with open(ruta_absoluta, "r", encoding='utf-8') as f:
# Leer el fichero línea a línea
        contenido = f.readlines()
        
# Seleccionamos las primeras N líneas
    lista_contenido = contenido[:n_lineas]
    
# Escribimos el fichero de salida
    with open(fichero_salida, "w", encoding='utf-8') as w:
        w.writelines(lista_contenido)

n_lineas = int(input("¿Cuántas líneas extraigo del fichero? \n"))
extrae_lineas(n_lineas)

4.5. Creación de Ficheros de Log de Usuario

Instrucción: Implementación de una clase Usuario y una función para registrar eventos de login en un archivo de log.

from datetime import datetime

class Usuario:
    def __init__(self, nombre: str):
        self.nombre = nombre

def abre_fichero(usuario):
    # Obtener la fecha y hora actuales en el formato solicitado
    fecha_hora = datetime.now().strftime("%Y-%m-%d %H:%M:%S")
    
    # Construir la línea de log
    linea_log = f"{fecha_hora} LOGIN {usuario.nombre}\n"
    
    # Abrir el fichero en modo 'append' para no sobrescribir los registros anteriores
    with open("log.txt", "a", encoding="utf-8") as f:
        f.write(linea_log)
    
    print(f"Registro añadido al log: {linea_log.strip()}")

usuario1 = Usuario("usuario1")
usuario2 = Usuario("usuario2")

abre_fichero(usuario1) 
abre_fichero(usuario2)

Deja una respuesta

Tu dirección de correo electrónico no será publicada. Los campos obligatorios están marcados con *

Este sitio usa Akismet para reducir el spam. Aprende cómo se procesan los datos de tus comentarios.