Cómo procesé 40GB de datos de comercio exterior sin pagar un centavo (gracias a DuckDB). Caso Real: España.

#Consolidar todos los archivos nuevos de forma incremental
import duckdb
import time
import glob
import os
import pandas as pd

# --- CONFIGURATION ---
# Point to the PARENT folder containing year folders (2015, 2016, etc.)
ROOT_PATH = r"G:\Bases\España\Datos" 
CATALOG_PATH = r"G:\Bases\España\Datos\Catalogos"
DB_FILE = "trade_database.duckdb"
FINAL_OUTPUT = "consolidated_trade_history.parquet"

def fix_path(path):
    return path.replace('\\', '/')

def run_production_pipeline():
    con = duckdb.connect(DB_FILE)
    print("🚀 Starting Production Pipeline...")

    # ---------------------------------------------------------
    # 1. SETUP & HISTORY TRACKING
    # ---------------------------------------------------------
    # Create the Master Table if it doesn't exist
    con.execute("""
        CREATE TABLE IF NOT EXISTS clean_trade (
            flujo VARCHAR,
            año INT,
            mes INT,
            fecha DATE,
            cod_pais VARCHAR,
            nombre_pais VARCHAR,
            cod_provincia VARCHAR,
            nombre_provincia VARCHAR,
            cod_taric VARCHAR,
            descripcion_taric VARCHAR,
            euros DOUBLE,
            dolares DOUBLE,
            kilogramos DOUBLE,
            source_file VARCHAR
        )
    """)

    # Create a Log Table to remember processed files
    con.execute("CREATE TABLE IF NOT EXISTS processed_log (filename VARCHAR, process_date TIMESTAMP)")
    
    # Get list of already processed files
    processed_files = set(row[0] for row in con.execute("SELECT filename FROM processed_log").fetchall())
    print(f"📜 System remembers {len(processed_files)} previously processed files.")

    # ---------------------------------------------------------
    # 2. SCAN FOR NEW DATA (2015-2025+)
    # ---------------------------------------------------------
    # Look for year folders 20xx
    year_folders = glob.glob(os.path.join(ROOT_PATH, "20*"))
    
    candidate_files = []
    for year_folder in year_folders:
        # Get CSVs strictly inside the year folder (ignore subfolders like 'Trimestre')
        files = glob.glob(os.path.join(year_folder, "*.csv"))
        candidate_files.extend(files)

    # Filter: Keep ONLY files we haven't seen yet
    new_files = [f for f in candidate_files if fix_path(f) not in processed_files]
    
    if not new_files:
        print("✅ No new files found. Data is up to date.")
    else:
        print(f"⚡ Found {len(new_files)} NEW files. Starting incremental processing...")
        
        # ---------------------------------------------------------
        # 3. PROCESS NEW FILES (Incremental Batch)
        # ---------------------------------------------------------
        # Load Catalogs (Only needed if we have new files to process)
        p_paises = fix_path(os.path.join(CATALOG_PATH, "PAISES.csv"))
        p_sectores = fix_path(os.path.join(CATALOG_PATH, "SECTORES.csv"))
        p_taric = fix_path(os.path.join(CATALOG_PATH, "TARIC.csv"))
        p_provincias = fix_path(os.path.join(CATALOG_PATH, "PROVINCIAS.csv"))

        con.execute(f"""
            CREATE OR REPLACE TABLE paises AS SELECT * FROM read_csv('{p_paises}', delim='\\t', encoding='UTF-16', header=True, ignore_errors=True);
            CREATE OR REPLACE TABLE sectores AS SELECT * FROM read_csv('{p_sectores}', delim='\\t', encoding='UTF-16', header=True, types={{'cod_sec': 'VARCHAR'}});
            CREATE OR REPLACE TABLE taric_cat AS SELECT * FROM read_csv('{p_taric}', delim='\\t', encoding='UTF-16', header=True, types={{'cod_taric': 'VARCHAR'}});
            CREATE OR REPLACE TABLE provincias AS SELECT * FROM read_csv('{p_provincias}', delim='\\t', encoding='UTF-16', header=True, types={{'cod_provincia': 'VARCHAR'}});
        """)

        # Sanitize paths for SQL
        clean_paths = [fix_path(f) for f in new_files]
        files_sql = "[" + ", ".join([f"'{f}'" for f in clean_paths]) + "]"

        # ---------------------------------------------------------
        # A. Robust Load (Smart Encoding Detection)
        # ---------------------------------------------------------
        print("⏳ Reading CSVs with auto-recovery (Mixed Encodings)...")
        
        # 1. Create the Staging Table structure first (Empty)
        # We perform a dummy read of the first file just to get the columns/types
        first_file = clean_paths[0]
        con.execute(f"""
            CREATE OR REPLACE TABLE staging_raw AS 
            SELECT * FROM read_csv('{first_file}', 
                delim='\\t', 
                encoding='UTF-16', 
                header=True, 
                filename=True,
                types={{'cod_taric': 'VARCHAR', 'pais': 'VARCHAR', 'provincia': 'VARCHAR'}}
            ) LIMIT 0;
        """)

        # 2. Loop through every file and try to append it safely
        success_count = 0
        error_count = 0
        
        for f in clean_paths:
            try:
                # Attempt 1: Try UTF-16 (Most common for your data)
                con.execute(f"""
                    INSERT INTO staging_raw 
                    SELECT * FROM read_csv('{f}', 
                        delim='\\t', 
                        encoding='UTF-16', 
                        header=True, 
                        filename=True,
                        types={{'cod_taric': 'VARCHAR', 'pais': 'VARCHAR', 'provincia': 'VARCHAR'}}
                    )
                """)
                success_count += 1
                
            except Exception as e:
                # If UTF-16 fails, the file is likely UTF-8 or Latin-1
                print(f"⚠️ UTF-16 failed for {os.path.basename(f)}. Retrying with Latin-1/Auto...")
                try:
                    # Attempt 2: Let DuckDB Auto-Detect (Handles UTF-8/Latin-1)
                    con.execute(f"""
                        INSERT INTO staging_raw 
                        SELECT * FROM read_csv('{f}', 
                            delim='\\t', 
                            auto_detect=True, -- Let DuckDB figure it out
                            header=True, 
                            filename=True,
                            types={{'cod_taric': 'VARCHAR', 'pais': 'VARCHAR', 'provincia': 'VARCHAR'}}
                        )
                    """)
                    print(f"   ✅ Recovered {os.path.basename(f)} using Auto-Detect.")
                    success_count += 1
                except Exception as e2:
                    print(f"❌ FAILED to read {os.path.basename(f)}. Skipping.")
                    print(f"   Error: {e2}")
                    error_count += 1

        print(f"📦 Staging complete. Success: {success_count} | Failed: {error_count}")

        # C. Update Log
        print("📝 Updating history log...")
        for f in clean_paths:
            con.execute(f"INSERT INTO processed_log VALUES ('{f}', current_timestamp)")

 # ---------------------------------------------------------
    # 4. QUALITY CHECK (Corregido)
    # ---------------------------------------------------------
    print("\n🔍 --- DATA QUALITY REPORT ---")
    
    # Check 1: Row count by Year
    print("1. Rows per Year:")
    # CORRECCIÓN: Usamos .df() para convertir a Pandas y luego imprimir
    df_yearly = con.execute("SELECT año, count(*) as records, round(sum(euros)/1e9, 2) as billion_euros FROM clean_trade GROUP BY año ORDER BY año").df()
    print(df_yearly)

    # Check 2: Missing Countries?
    print("\n2. Rows with Unknown Country (Should be 0):")
    unknowns = con.execute("SELECT count(*) FROM clean_trade WHERE nombre_pais IS NULL").fetchone()[0]
    print(f"   -> {unknowns} rows")

    # Check 3: Missing Products?
    print("3. Rows with Unknown TARIC codes:")
    unknown_taric = con.execute("SELECT count(*) FROM clean_trade WHERE descripcion_taric IS NULL").fetchone()[0]
    print(f"   -> {unknown_taric} rows")

    # ---------------------------------------------------------
    # 5. EXPORT CONSOLIDATED FILE
    # ---------------------------------------------------------
    # Only export if we added new files OR if the file doesn't exist
    if new_files or not os.path.exists(FINAL_OUTPUT):
        print(f"\n💾 Writing FULL consolidated history to {FINAL_OUTPUT}...")
        # Overwrite the big file with the latest master data
        con.execute(f"COPY clean_trade TO '{FINAL_OUTPUT}' (FORMAT PARQUET)")
        print("✅ Consolidated File Updated.")
    else:
        print("\n⏩ No new data, skipping export to save time.")

    con.close()

if __name__ == "__main__":
    run_production_pipeline()