import pandas as pd import csv import os def merge_bee_treatment_databases(db_oficial_path, nuevos_path, output_path): """ Integra los nuevos tratamientos en la base de datos oficial manteniendo la estructura original y añadiendo las nuevas columnas al final. Args: db_oficial_path: Ruta al archivo CSV de la base de datos oficial nuevos_path: Ruta al archivo CSV con los nuevos tratamientos output_path: Ruta donde se guardará el nuevo archivo CSV integrado """ print("Leyendo archivos CSV...") # Leer los archivos CSV directamente db_oficial = pd.read_csv(db_oficial_path, encoding='utf-8') # Leer el archivo de nuevos tratamientos con manejo de errores para columnas duplicadas try: nuevos = pd.read_csv(nuevos_path, encoding='utf-8') except pd.errors.ParserError as e: if "Duplicate names" in str(e): print("Detectando columnas duplicadas en el archivo de nuevos tratamientos...") # Leer el archivo sin pandas primero para identificar los encabezados with open(nuevos_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) headers = next(reader) # Verificar si hay encabezados duplicados header_counts = {} for header in headers: if header in header_counts: header_counts[header] += 1 else: header_counts[header] = 1 # Crear nuevos encabezados únicos unique_headers = [] header_counts_temp = {} for header in headers: if header in header_counts_temp: header_counts_temp[header] += 1 unique_headers.append(f"{header}_{header_counts_temp[header]}") else: header_counts_temp[header] = 0 unique_headers.append(header) # Leer el archivo con los encabezados únicos nuevos = pd.read_csv(nuevos_path, encoding='utf-8', names=unique_headers, header=0) print(f"Columnas duplicadas renombradas: {[h for h, c in header_counts.items() if c > 1]}") else: raise e print(f"Base de datos oficial: {len(db_oficial)} filas, {len(db_oficial.columns)} columnas") print(f"Encabezados en DB oficial: {db_oficial.columns.tolist()[:5]}...") print(f"Nuevos tratamientos: {len(nuevos)} filas, {len(nuevos.columns)} columnas") # Crear un DataFrame vacío con la estructura de la base oficial nuevos_formateados = pd.DataFrame(columns=db_oficial.columns) # Mapeo de columnas entre los dos archivos column_mapping = { 'Medicamento (listado extenso AEM)': 'Name', 'Active substances': 'Active Ingredient', 'Product status': None, # Nueva columna a añadir al final 'Target species': None, # Nueva columna a añadir al final 'Pharmaceutical form': None, # Nueva columna a añadir al final 'Routes of administration': 'Application Method', 'ATCvet code': None, # Nueva columna a añadir al final 'Legal status of supply': None, # Nueva columna a añadir al final 'Authorised in': None, # Nueva columna a añadir al final 'Last updated': None, # Nueva columna a añadir al final 'Package description': None, # Nueva columna a añadir al final 'Correspondencia EDU (tabla original)': None # Columna de mapeo, no se utiliza directamente } # Definir valores predeterminados para columnas obligatorias en la DB oficial default_values = { 'Type': 'Chemical', 'Efficacy': 80, 'Duration': 30, 'Implementation': 'Easy', 'Temperature Min': '10°C', 'Temperature Max': '30°C', 'Humidity Requirements': 'None', 'Equipment Needed': 'Basic', 'Colony Conditions': 'Any', 'Safety Equipment': 'Gloves, Mask', 'Environmental Precautions': 'Standard', 'Storage Requirements': 'Cool, dry place', 'Withdrawal Period': 0, 'Residue Risks': 'Low', 'Incompatibilities': 'None known', 'Number of Applications': 1, 'Days Between Applications': 0, 'Total Duration': 30, 'Best Time': 'Spring, Fall', 'Weather Conditions': 'Dry', 'Manufacturer': 'Various', 'Registration Number': 'Unknown', 'Cost Category': 'Medium', 'Resistance Risk': 'Medium', 'Side Effects': 'Minimal', 'Organic Compatible': 'No', 'Availability': 'Prescription', 'Jan': 'Yes', 'Feb': 'Yes', 'Mar': 'Yes', 'Apr': 'Yes', 'May': 'Yes', 'Jun': 'Yes', 'Jul': 'Yes', 'Aug': 'Yes', 'Sep': 'Yes', 'Oct': 'Yes', 'Nov': 'Yes', 'Dec': 'Yes', 'Real Time required': 30, 'Time required': 30, 'Organic': 'No', 'Conventional': 'Yes', 'Dadant': 'Yes', 'Warré': 'Yes', 'WBC': 'Yes', 'Langstroth': 'Yes', 'Kenyan': 'Yes', 'Strong (>6 frames covered with bees)': 'Yes', 'Weak (<5 frames covered with bees)': 'Yes', '6 frames (<6 frames covered with bees)': 'Yes', 'Nuclei (<6 frames covered with bees)': 'Yes', 'Varroa tolerant/VSH': 'No', 'January': 'Yes', 'February': 'Yes', 'March': 'Yes', 'April': 'Yes', 'May.1': 'Yes', 'June': 'Yes', 'July': 'Yes', 'August': 'Yes', 'September': 'Yes', 'October': 'Yes', 'November': 'Yes', 'Decembre': 'Yes', 'Authorised/Available': 'Yes' } # Mapeo de sustancias activas a configuraciones específicas active_substance_mapping = { 'amitraz': { 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'Medium', 'Efficacy': 90 }, 'tau-fluvalinate': { 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'Medium', 'Efficacy': 85 }, 'fluvalinate': { 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'Medium', 'Efficacy': 85 }, 'flumethrin': { 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'Medium', 'Efficacy': 85 }, 'coumaphos': { 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'High', 'Efficacy': 85 }, 'coumafos': { # Variante ortográfica 'Type': 'Chemical', 'Organic Compatible': 'No', 'Organic': 'No', 'Resistance Risk': 'High', 'Residue Risks': 'High', 'Efficacy': 85 }, 'thymol': { 'Type': 'Essential Oil', 'Organic Compatible': 'Yes', 'Organic': 'Yes', 'Resistance Risk': 'Low', 'Residue Risks': 'Low', 'Efficacy': 70 }, 'oxalic acid': { 'Type': 'Organic Acid', 'Organic Compatible': 'Yes', 'Organic': 'Yes', 'Resistance Risk': 'Low', 'Residue Risks': 'Low', 'Efficacy': 75 }, 'formic acid': { 'Type': 'Organic Acid', 'Organic Compatible': 'Yes', 'Organic': 'Yes', 'Resistance Risk': 'Low', 'Residue Risks': 'Low', 'Efficacy': 80 }, 'lactic acid': { 'Type': 'Organic Acid', 'Organic Compatible': 'Yes', 'Organic': 'Yes', 'Resistance Risk': 'Low', 'Residue Risks': 'Low', 'Efficacy': 65 } } # Mapeo adicional para formatos farmacéuticos pharmaceutical_form_to_implementation = { 'strip': 'Easy', 'gel': 'Easy', 'solution': 'Moderate', 'powder': 'Moderate', 'tablet': 'Easy', 'bee-hive strip': 'Easy', 'bee-hive solution': 'Moderate', 'bee-hive powder': 'Moderate', 'bee-hive dispersion': 'Moderate' } print("Procesando nuevos tratamientos...") # Procesar cada nuevo tratamiento y mapear sus datos a la estructura de la DB oficial for idx, treatment in nuevos.iterrows(): new_row = {} # Aplicar valores predeterminados for col in db_oficial.columns: new_row[col] = default_values.get(col, '') # Mapear las columnas que tienen correspondencia for nuevos_col, oficial_col in column_mapping.items(): if oficial_col and nuevos_col in nuevos.columns: new_row[oficial_col] = treatment[nuevos_col] # Asignar el nombre del medicamento (obligatorio) if 'Medicamento (listado extenso AEM)' in nuevos.columns: new_row['Name'] = treatment['Medicamento (listado extenso AEM)'] # Usar la correspondencia de tabla original si está disponible if 'Correspondencia EDU (tabla original)' in nuevos.columns: correspondence = treatment['Correspondencia EDU (tabla original)'] if correspondence and pd.notna(correspondence) and correspondence != 'No identificado en tabla original': # Buscar en la DB oficial los valores para este medicamento match = db_oficial[db_oficial['Name'] == correspondence] if not match.empty: for col in db_oficial.columns: if col not in ['Name']: # Mantener el nombre del nuevo medicamento new_row[col] = match.iloc[0][col] # Aplicar configuraciones basadas en la sustancia activa if 'Active substances' in nuevos.columns and pd.notna(treatment['Active substances']): active_substance = str(treatment['Active substances']).lower() # Verificar qué sustancia activa coincide y aplicar su configuración for substance, config in active_substance_mapping.items(): if substance in active_substance: for config_key, config_value in config.items(): new_row[config_key] = config_value break # Ajustar para combinaciones específicas if 'thymol' in active_substance and any(oil in active_substance for oil in ['oil', 'menthol', 'eucalyptus']): new_row['Type'] = 'Essential Oil Mix' if 'oxalic' in active_substance and 'formic' in active_substance: new_row['Type'] = 'Organic Acid Mix' new_row['Efficacy'] = 85 # Ajustar la implementación basado en la forma farmacéutica if 'Pharmaceutical form' in nuevos.columns and pd.notna(treatment['Pharmaceutical form']): form = str(treatment['Pharmaceutical form']).lower() for form_key, impl_value in pharmaceutical_form_to_implementation.items(): if form_key in form: new_row['Implementation'] = impl_value break # Mapear el método de aplicación desde la ruta de administración if 'Routes of administration' in nuevos.columns and pd.notna(treatment['Routes of administration']): route = str(treatment['Routes of administration']) if 'in-hive' in route.lower(): new_row['Application Method'] = 'In-hive application' elif 'oral' in route.lower(): new_row['Application Method'] = 'Oral application' elif 'cutaneous' in route.lower(): new_row['Application Method'] = 'Contact application' elif 'nebulisation' in route.lower(): new_row['Application Method'] = 'Evaporation' # Mapear el estado de autorización if 'Product status' in nuevos.columns and pd.notna(treatment['Product status']): status = str(treatment['Product status']).lower() if 'authorised' in status: new_row['Authorised/Available'] = 'Yes' else: new_row['Authorised/Available'] = 'No' # Ajustar la disponibilidad basado en el estado legal if 'Legal status of supply' in nuevos.columns and pd.notna(treatment['Legal status of supply']): legal_status = str(treatment['Legal status of supply']).lower() if 'prescription' in legal_status: new_row['Availability'] = 'Prescription' elif 'without' in legal_status and 'prescription' in legal_status: new_row['Availability'] = 'Over the counter' # Añadir la fila al DataFrame nuevos_formateados = pd.concat([nuevos_formateados, pd.DataFrame([new_row])], ignore_index=True) print(f"Se procesaron {len(nuevos_formateados)} nuevos tratamientos") # Identificar las columnas que solo existen en el archivo de nuevos tratamientos new_columns = [] for col in nuevos.columns: if col not in column_mapping.values() and column_mapping.get(col) is None and col != 'Correspondencia EDU (tabla original)': new_columns.append(col) # Añadir los nuevos tratamientos a la DB oficial combined_db = pd.concat([db_oficial, nuevos_formateados], ignore_index=True) # Añadir las nuevas columnas al final del DataFrame combinado for col in new_columns: combined_db[col] = '' # Copiar los valores de las nuevas columnas solo para los nuevos tratamientos for idx, treatment in nuevos.iterrows(): if col in nuevos.columns and idx + len(db_oficial) < len(combined_db): combined_db.loc[idx + len(db_oficial), col] = treatment[col] print(f"Base de datos final: {len(combined_db)} filas, {len(combined_db.columns)} columnas") # Guardar el resultado directamente en formato CSV combined_db.to_csv(output_path, index=False) print(f"Archivo guardado como {output_path}") # Verificar que el archivo se guardó correctamente try: # Leer el archivo recién creado para verificarlo df_check = pd.read_csv(output_path) # Verificar si la primera fila contiene encabezados duplicados first_row = df_check.iloc[0] headers = df_check.columns # Comprobar si la primera fila es similar a los encabezados headers_in_first_row = 0 for col, val in zip(headers[:10], first_row[:10]): # Verificar solo las primeras 10 columnas if pd.notna(val) and pd.notna(col): if str(val).lower() == str(col).lower() or ( isinstance(val, str) and isinstance(col, str) and (val.lower() in col.lower() or col.lower() in val.lower()) ): headers_in_first_row += 1 # Si más de la mitad de las primeras 10 columnas coinciden, eliminar la primera fila if headers_in_first_row >= 5: print("Detectada fila con encabezados duplicados, eliminando...") df_check = df_check.iloc[1:].reset_index(drop=True) df_check.to_csv(output_path, index=False) print("Fila duplicada eliminada correctamente.") # Mostrar información de verificación print(f"Verificación - Primeros 5 encabezados: {df_check.columns.tolist()[:5]}") if not df_check.empty: print(f"Verificación - Primeros 5 valores de la primera fila: {df_check.iloc[0].tolist()[:5]}") except Exception as e: print(f"Error al verificar el archivo guardado: {str(e)}") return combined_db # Función para verificar y corregir encabezados en caso necesario def verify_csv_format(file_path): """ Verifica que el formato del CSV sea correcto y que no tenga los encabezados automáticos Column1, Column2, etc. Args: file_path: Ruta del archivo CSV a verificar Returns: bool: True si el formato es correcto, False si necesita corrección """ try: with open(file_path, 'r', encoding='utf-8') as f: reader = csv.reader(f) headers = next(reader) # Verificar si la primera fila contiene "Column1", "Column2", etc. if headers[0] == "Column1" or headers[0].startswith("Column"): print(f"Advertencia: El archivo {file_path} tiene encabezados automáticos.") return False # Verificar si hay encabezados duplicados header_counts = {} for header in headers: if header in header_counts: header_counts[header] += 1 else: header_counts[header] = 1 duplicates = [h for h, c in header_counts.items() if c > 1] if duplicates: print(f"Advertencia: El archivo {file_path} tiene encabezados duplicados: {duplicates}") return False print(f"El archivo {file_path} tiene el formato correcto.") return True except Exception as e: print(f"Error al verificar el formato del archivo {file_path}: {str(e)}") return False # Función para corregir encabezados si es necesario def fix_csv_headers(input_file, output_file=None): """ Corrige los encabezados automáticos en un archivo CSV. Args: input_file: Ruta del archivo CSV con encabezados automáticos output_file: Ruta donde guardar el archivo corregido. Si es None, sobrescribe el archivo original. """ if output_file is None: output_file = input_file + '.temp' replace_original = True else: replace_original = False try: # Leer los encabezados para verificar si hay duplicados o son automáticos with open(input_file, 'r', encoding='utf-8') as f: reader = csv.reader(f) headers = next(reader) # Verificar si son encabezados automáticos is_auto_headers = headers[0] == "Column1" or headers[0].startswith("Column") # Verificar si hay encabezados duplicados header_counts = {} for header in headers: if header in header_counts: header_counts[header] += 1 else: header_counts[header] = 1 has_duplicates = any(count > 1 for count in header_counts.values()) if is_auto_headers: # Si son encabezados automáticos, usar la segunda fila como encabezados reales real_headers = next(reader) skip_first_row = True else: # Si no son automáticos pero tienen duplicados, crear encabezados únicos if has_duplicates: unique_headers = [] header_counts_temp = {} for header in headers: if header in header_counts_temp: header_counts_temp[header] += 1 unique_headers.append(f"{header}_{header_counts_temp[header]}") else: header_counts_temp[header] = 0 unique_headers.append(header) real_headers = unique_headers skip_first_row = False else: # No hay problemas con los encabezados print(f"El archivo {input_file} no necesita corrección.") return True # Leer todo el archivo de nuevo para reescribirlo with open(input_file, 'r', encoding='utf-8') as f_in, open(output_file, 'w', newline='', encoding='utf-8') as f_out: reader = csv.reader(f_in) writer = csv.writer(f_out) # Saltar la primera fila si es necesario if skip_first_row: next(reader) # Escribir los encabezados reales como primera fila writer.writerow(real_headers) # Copiar el resto del archivo for row in reader: writer.writerow(row) # Si se debe reemplazar el archivo original if replace_original: os.replace(output_file, input_file) print(f"Archivo {input_file} corregido y sobrescrito.") else: print(f"Archivo corregido guardado como {output_file}.") return True except Exception as e: print(f"Error al corregir los encabezados: {str(e)}") return False # Rutas de los archivos db_oficial_path = 'DB OFICIAL.csv' nuevos_path = 'NUEVOS.csv' output_path = 'DB_OFICIAL_INTEGRADA.csv' # Ejecutar la función principal if __name__ == "__main__": try: # Paso 0: Verificar y corregir los archivos de entrada si es necesario print("Verificando el formato de los archivos de entrada...") # Verificar y corregir DB OFICIAL.csv if not verify_csv_format(db_oficial_path): print("Corrigiendo encabezados en DB OFICIAL.csv...") if not fix_csv_headers(db_oficial_path): raise Exception("No se pudieron corregir los encabezados de DB OFICIAL.csv") # Verificar y corregir NUEVOS.csv if not verify_csv_format(nuevos_path): print("Corrigiendo encabezados en NUEVOS.csv...") if not fix_csv_headers(nuevos_path): raise Exception("No se pudieron corregir los encabezados de NUEVOS.csv") # Paso 1: Integrar los nuevos tratamientos print("\nIniciando la integración de nuevos tratamientos...") merge_bee_treatment_databases(db_oficial_path, nuevos_path, output_path) print("\n¡Integración completada con éxito!") # Paso 2: Verificar el formato del archivo resultante print("\nVerificando el formato del archivo resultante...") if not verify_csv_format(output_path): print("Se detectaron encabezados automáticos. Corrigiendo...") if fix_csv_headers(output_path): print("Encabezados corregidos con éxito.") else: print("No se pudieron corregir los encabezados automáticamente.") # Paso 3: Eliminar filas duplicadas de encabezados print("\nVerificando y eliminando filas duplicadas de encabezados...") try: # Leer el archivo resultante df = pd.read_csv(output_path) # Verificar si la primera fila contiene encabezados first_row = df.iloc[0] headers = df.columns # Comprobar si la primera fila es similar a los encabezados headers_in_first_row = sum(1 for col, val in zip(headers, first_row) if str(val).lower() == str(col).lower() or (isinstance(val, str) and isinstance(col, str) and (val.lower() in col.lower() or col.lower() in val.lower()))) if headers_in_first_row > len(headers) / 3: # Si más de un tercio de las columnas coinciden print("Eliminando fila duplicada de encabezados...") df = df.iloc[1:].reset_index(drop=True) df.to_csv(output_path, index=False) print("Fila duplicada eliminada correctamente.") else: print("No se detectaron filas duplicadas de encabezados.") except Exception as e: print(f"Error al verificar filas duplicadas: {str(e)}") print("\nProceso completado. El archivo final está listo para su uso.") except Exception as e: print(f"Error durante el proceso: {str(e)}")