Resum per importar fitxers CSV de comptabilitat a la base de dades autonomo_contabilidad amb Python

Resum per importar fitxers CSV de comptabilitat a la base de dades `autonomo_contabilidad` amb Python

Aquest script en Python importa dades d'un fitxer CSV (com `compartit 3tr-2025 - VENDA.csv`) a la base de dades MySQL `autonomo_contabilidad`, utilitzant l'esquema de taules `wp_contabilidad_clientes`, `wp_contabilidad_empleados`, `wp_contabilidad_productos`, `wp_contabilidad_ventas` i `wp_contabilidad_detalles_venta`.

**Passos clau**:
1. **Configuració**:
   - Instal·lar llibreries: `pip install pandas mysql-connector-python`.
   - Configurar credencials a `db_config` (usuari, contrasenya, host, port).
   - Especificar el camí del CSV (ex.: `C:\\Users\\joanp\\Documents\\compartit 3tr-2025 - VENDA.csv`).

2. **Processament del CSV**:
   - Llegir el CSV amb `pandas`.
   - Netejar columnes: convertir `import total` a float (eliminant ' €' i canviant ',' per '.'), i `data` a format `YYYY-MM-DD`.
   - Assumir IVA del 21% per calcular `subtotal` i `iva_monto`.

3. **Inserció a la base de dades**:
   - **Clients**: Inserir clients únics (`client`) a `wp_contabilidad_clientes`.
   - **Empleats**: Inserir empleats únics (`empleat`) a `wp_contabilidad_empleados` amb NIF provisional (UUID).
   - **Productes**: Inserir productes únics (`producte`) a `wp_contabilidad_productos` amb `protocol=''` (ajustar segons esquema).
   - **Vendes**: Inserir vendes a `wp_contabilidad_ventas` amb `fecha`, `cliente_id`, `subtotal`, `iva_porcentaje`, `iva_monto`, `total`, i `metode pagament` a `notas`.
   - **Detalls de vendes**: Inserir a `wp_contabilidad_detalles_venta` amb `venta_id`, `producto_id`, `cantidad=1`, `precio_unitario`, i `subtotal`.

4. **Gestió d'errors**:
   - Utilitzar `try-except` per capturar errors de connexió, CSV o SQL.
   - Afegir `cursor.fetchall()` i `buffered=True` per evitar errors com `Unread result found`.
   - Comprovar l'existència de taules abans d'inserir.

import pandas as pd
import mysql.connector
from datetime import datetime
import uuid
import logging

# Configurar logging per rastrejar errors i progrés
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

# Configuració de la connexió a la base de dades
db_config = {
    'host': 'localhost',  # Canvia si el teu host és diferent
    'user': 'joan',  # Substitueix amb el teu usuari de MySQL
    'password': 'queMm88/g62123',  # Substitueix amb la teva contrasenya de MySQL
    'database': 'autonomo_contabilidad',
    'port': 3306  # Canvia si utilitzes un port diferent
}

# Llegir el fitxer CSV
csv_file = 'C:\\Users\\joanp\\Documents\\compartit 3tr-2025 - VENDA.csv'
try:
    df = pd.read_csv(csv_file)
    logging.info(f"Fitxer CSV '{csv_file}' llegit correctament.")
except Exception as e:
    logging.error(f"Error en llegir el fitxer CSV: {e}")
    exit(1)

# Netejar i preprocessar les dades
def clean_currency(value):
    try:
        if isinstance(value, str):
            return float(value.replace(' €', '').replace(',', '.'))
        return float(value)
    except Exception as e:
        logging.error(f"Error en netejar valor monetari '{value}': {e}")
        return 0.0

def clean_date(date_str):
    try:
        return datetime.strptime(date_str, '%d/%m/%Y').strftime('%Y-%m-%d')
    except Exception as e:
        logging.error(f"Error en netejar data '{date_str}': {e}")
        return None

df['import total'] = df['import total'].apply(clean_currency)
df['data'] = df['data'].apply(clean_date)

# Eliminar files amb dates invàlides
df = df.dropna(subset=['data'])
logging.info(f"Total de files després de netejar dates: {len(df)}")

# Connectar a la base de dades
try:
    conn = mysql.connector.connect(**db_config)
    cursor = conn.cursor(buffered=True)  # Afegim buffered=True per gestionar resultats
    logging.info("Connexió a la base de dades establerta correctament.")
except mysql.connector.Error as e:
    logging.error(f"Error en connectar a la base de dades: {e}")
    exit(1)

# Comprovar si les taules existeixen
try:
    cursor.execute("SHOW TABLES LIKE 'wp_contabilidad_productos'")
    if not cursor.fetchone():
        logging.error("La taula wp_contabilidad_productos no existeix. Assegura't que l'esquema està creat.")
        cursor.close()
        conn.close()
        exit(1)
    cursor.fetchall()  # Consumir resultats pendents
except mysql.connector.Error as e:
    logging.error(f"Error en comprovar taules: {e}")
    cursor.close()
    conn.close()
    exit(1)

# Inserir clients únics
clients = df['client'].unique()
client_ids = {}
try:
    for client in clients:
        cursor.execute("SELECT id FROM wp_contabilidad_clientes WHERE nombre = %s", (client,))
        result = cursor.fetchone()
        cursor.fetchall()  # Consumir qualsevol resultat pendent
        if not result:
            cursor.execute(
                "INSERT INTO wp_contabilidad_clientes (nombre, fecha_creacion) VALUES (%s, NOW())",
                (client,)
            )
            client_ids[client] = cursor.lastrowid
            logging.info(f"Client '{client}' inserit amb ID {client_ids[client]}.")
        else:
            client_ids[client] = result[0]
except mysql.connector.Error as e:
    logging.error(f"Error en inserir clients: {e}")
    conn.rollback()
    cursor.close()
    conn.close()
    exit(1)

# Inserir empleats únics
employees = df['empleat'].unique()
employee_ids = {}
try:
    for employee in employees:
        cursor.execute("SELECT id FROM wp_contabilidad_empleados WHERE nombre = %s", (employee,))
        result = cursor.fetchone()
        cursor.fetchall()  # Consumir qualsevol resultat pendent
        if not result:
            cursor.execute(
                "INSERT INTO wp_contabilidad_empleados (nombre, nif, fecha_creacion) VALUES (%s, %s, NOW())",
                (employee, str(uuid.uuid4())[:20])  # Generar un NIF provisional
            )
            employee_ids[employee] = cursor.lastrowid
            logging.info(f"Empleat '{employee}' inserit amb ID {employee_ids[employee]}.")
        else:
            employee_ids[employee] = result[0]
except mysql.connector.Error as e:
    logging.error(f"Error en inserir empleats: {e}")
    conn.rollback()
    cursor.close()
    conn.close()
    exit(1)

# Inserir productes únics
products = df['producte'].unique()
product_ids = {}
default_category_id = 7  # ID de categoria per defecte segons l'esquema
try:
    for product in products:
        cursor.execute("SELECT id FROM wp_contabilidad_productos WHERE nombre = %s", (product,))
        result = cursor.fetchone()
        cursor.fetchall()  # Consumir qualsevol resultat pendent
        if not result:
            cursor.execute(
                """
                INSERT INTO wp_contabilidad_productos 
                (nombre, id_categoria_producto, precio, stock, protocol, created_at, updated_at) 
                VALUES (%s, %s, %s, %s, %s, NOW(), NOW())
                """,
                (product, default_category_id, 0.00, 0, '')  # Afegim protocol com cadena buida
            )
            product_ids[product] = cursor.lastrowid
            logging.info(f"Producte '{product}' inserit amb ID {product_ids[product]}.")
        else:
            product_ids[product] = result[0]
except mysql.connector.Error as e:
    logging.error(f"Error en inserir productes: {e}")
    conn.rollback()
    cursor.close()
    conn.close()
    exit(1)

# Inserir vendes i detalls de vendes
try:
    for index, row in df.iterrows():
        # Calcular subtotal i IVA (assumim IVA del 21%)
        total = row['import total']
        iva_porcentaje = 21.00
        subtotal = round(total / 1.21, 2)
        iva_monto = round(total - subtotal, 2)

        # Inserir a wp_contabilidad_ventas
        cursor.execute(
            """
            INSERT INTO wp_contabilidad_ventas 
            (fecha, cliente_id, subtotal, iva_porcentaje, iva_monto, total, notas, fecha_creacion, empleado_id)
            VALUES (%s, %s, %s, %s, %s, %s, %s, NOW(), %s)
            """,
            (
                row['data'],
                client_ids[row['client']],
                subtotal,
                iva_porcentaje,
                iva_monto,
                total,
                f"Metode de pagament: {row['metode pagament']}",
                employee_ids[row['empleat']]
            )
        )
        venta_id = cursor.lastrowid
        logging.info(f"Venda inserida amb ID {venta_id} per al client {row['client']}.")

        # Inserir a wp_contabilidad_detalles_venta
        cursor.execute(
            """
            INSERT INTO wp_contabilidad_detalles_venta 
            (venta_id, producto_id, cantidad, precio_unitario, subtotal)
            VALUES (%s, %s, %s, %s, %s)
            """,
            (
                venta_id,
                product_ids[row['producte']],
                1,  # Assumim quantitat 1
                subtotal,
                subtotal
            )
        )
        logging.info(f"Detall de venda inserit per al producte {row['producte']}.")
except mysql.connector.Error as e:
    logging.error(f"Error en inserir vendes o detalls: {e}")
    conn.rollback()
    cursor.close()
    conn.close()
    exit(1)

# Confirmar canvis i tancar connexió
try:
    conn.commit()
    logging.info("Totes les dades s'han importat correctament a la base de dades.")
except mysql.connector.Error as e:
    logging.error(f"Error en confirmar canvis: {e}")
    conn.rollback()
finally:
    cursor.close()
    conn.close()

print("Importació completada amb èxit.")

**Notes**:
- Assegura't que l'esquema (`ESTRUCTURA.sql`) està aplicat.
- Ajusta `protocol` a `wp_contabilidad_productos` segons el tipus de dades requerit.
- Per a futurs fitxers, actualitza `csv_file` i verifica l'IVA o quantitats si varien.

Aquest resum és ideal per enganxar a un formulari amb camp Trix per documentar el procés. Si necessites més detalls o ajustos, pots ampliar el codi o consultar l'esquema complet!