Funcionando con il calculo de afinidad en funciones base

This commit is contained in:
Miguel 2024-11-18 12:11:36 +01:00
parent 751825a659
commit 2d21e09f19
4 changed files with 121 additions and 159 deletions

View File

@ -38,6 +38,10 @@ def obtener_nombre_idioma(codigo_corto):
def exportar_para_traduccion(config: TranslationConfig): def exportar_para_traduccion(config: TranslationConfig):
"""
Exporta los textos del archivo maestro para su traducción, realizando validaciones
y cálculos de afinidad en lotes. Los textos idénticos reciben afinidad 1 automáticamente.
"""
master_path = config.get_master_path() master_path = config.get_master_path()
if not os.path.exists(master_path): if not os.path.exists(master_path):
print("El archivo maestro no existe.") print("El archivo maestro no existe.")
@ -46,6 +50,7 @@ def exportar_para_traduccion(config: TranslationConfig):
configurar_detector_idiomas() configurar_detector_idiomas()
df_maestro = fc.read_dataframe_with_cleanup_retries(master_path) df_maestro = fc.read_dataframe_with_cleanup_retries(master_path)
# Preparar DataFrame de exportación
df_export = pd.DataFrame() df_export = pd.DataFrame()
primera_columna = df_maestro.columns[0] primera_columna = df_maestro.columns[0]
df_export[primera_columna] = df_maestro[primera_columna] df_export[primera_columna] = df_maestro[primera_columna]
@ -53,7 +58,7 @@ def exportar_para_traduccion(config: TranslationConfig):
columna_propuesta = f"{config.codigo_idioma_seleccionado}_Propuesto" columna_propuesta = f"{config.codigo_idioma_seleccionado}_Propuesto"
df_export[columna_propuesta] = df_maestro[config.codigo_idioma_seleccionado] df_export[columna_propuesta] = df_maestro[config.codigo_idioma_seleccionado]
# Add validation columns if source and target languages are different # Agregar columnas de validación si los idiomas son diferentes
if config.codigo_columna_maestra != config.codigo_idioma_seleccionado: if config.codigo_columna_maestra != config.codigo_idioma_seleccionado:
df_export["Validation_Error"] = "" df_export["Validation_Error"] = ""
df_export["Affinity_Score"] = None df_export["Affinity_Score"] = None
@ -62,9 +67,7 @@ def exportar_para_traduccion(config: TranslationConfig):
# Agregar columna del idioma secundario # Agregar columna del idioma secundario
if config.codigo_idioma_secundario in df_maestro.columns: if config.codigo_idioma_secundario in df_maestro.columns:
df_export[config.codigo_idioma_secundario] = df_maestro[ df_export[config.codigo_idioma_secundario] = df_maestro[config.codigo_idioma_secundario]
config.codigo_idioma_secundario
]
ruta_export = config.get_translate_path() ruta_export = config.get_translate_path()
@ -74,9 +77,15 @@ def exportar_para_traduccion(config: TranslationConfig):
worksheet = writer.sheets["Sheet1"] worksheet = writer.sheets["Sheet1"]
# Inmovilizar paneles en A2 # Inmovilizar paneles en A2
worksheet.freeze_panes = 'A2' worksheet.freeze_panes = "A2"
# Configurar estilos
wrap_alignment = Alignment(wrap_text=True, vertical="top") wrap_alignment = Alignment(wrap_text=True, vertical="top")
red_fill = PatternFill(start_color="FF0000", end_color="FF0000", fill_type="solid")
yellow_fill = PatternFill(start_color="FFFF00", end_color="FFFF00", fill_type="solid")
blue_fill = PatternFill(start_color="ADD8E6", end_color="ADD8E6", fill_type="solid")
# Ajustar anchos de columna
for col in worksheet.columns: for col in worksheet.columns:
max_length = 0 max_length = 0
column = col[0].column_letter column = col[0].column_letter
@ -87,101 +96,107 @@ def exportar_para_traduccion(config: TranslationConfig):
text_length = len(str(cell.value)) text_length = len(str(cell.value))
if text_length > 50: if text_length > 50:
cell.alignment = wrap_alignment cell.alignment = wrap_alignment
text_length = min( text_length = min(50, max(len(word) for word in str(cell.value).split()))
50, max(len(word) for word in str(cell.value).split())
)
max_length = max(max_length, text_length) max_length = max(max_length, text_length)
except: except:
pass pass
adjusted_width = min(50, max_length + 2) adjusted_width = min(50, max_length + 2)
worksheet.column_dimensions[column].width = ( worksheet.column_dimensions[column].width = adjusted_width if adjusted_width > 8 else 8
adjusted_width if adjusted_width > 8 else 8
)
red_fill = PatternFill(
start_color="FF0000", end_color="FF0000", fill_type="solid"
)
yellow_fill = PatternFill(
start_color="FFFF00", end_color="FFFF00", fill_type="solid"
)
blue_fill = PatternFill(
start_color="ADD8E6", end_color="ADD8E6", fill_type="solid"
)
total_rows = worksheet.max_row - 1
progress_bar = fc.ProgressBar(
total_rows, prefix="Procesando filas:", suffix="Completado"
)
# Primera fase: Procesar detección de idioma y recopilar textos para afinidad
texts_to_check = {} # Para textos que necesitan cálculo de afinidad
identical_texts = {} # Para textos idénticos (afinidad 1)
texto_a_filas = defaultdict(list) texto_a_filas = defaultdict(list)
inconsistencias = 0 inconsistencias = 0
afinidad_baja = 0 afinidad_baja = 0
for row in range(2, worksheet.max_row + 1): progress_bar = fc.ProgressBar(
texto = worksheet.cell(row=row, column=2).value worksheet.max_row - 1, prefix="Procesando textos:", suffix="Completado"
if texto: )
# Language detection
texto_limpio = fc.limpiar_texto(config.codigo_tipo_PLC, texto)
if texto == texto_limpio:
texto_a_filas[texto].append(row)
idioma_detectado = detectar_idioma(texto, config.codigo_tipo_PLC) for row in range(2, worksheet.max_row + 1):
idioma_esperado = fc.idiomas_shortcodefromcode( texto_original = worksheet.cell(row=row, column=1).value
config.codigo_idioma_seleccionado texto_propuesto = worksheet.cell(row=row, column=2).value
)
if ( if texto_original and texto_propuesto:
idioma_detectado != "unknown" # Detección de idioma
and idioma_detectado != idioma_esperado texto_limpio = fc.limpiar_texto(config.codigo_tipo_PLC, texto_propuesto)
): if texto_propuesto == texto_limpio:
texto_a_filas[texto_propuesto].append(row)
idioma_detectado = detectar_idioma(texto_propuesto, config.codigo_tipo_PLC)
idioma_esperado = fc.idiomas_shortcodefromcode(config.codigo_idioma_seleccionado)
if idioma_detectado != "unknown" and idioma_detectado != idioma_esperado:
worksheet.cell(row=row, column=2).fill = blue_fill worksheet.cell(row=row, column=2).fill = blue_fill
nombre_idioma = obtener_nombre_idioma(idioma_detectado) nombre_idioma = obtener_nombre_idioma(idioma_detectado)
worksheet.cell( worksheet.cell(
row=row, row=row,
column=df_export.columns.get_loc("Idioma_Detectado") + 1, column=df_export.columns.get_loc("Idioma_Detectado") + 1
).value = nombre_idioma ).value = nombre_idioma
# Validation checks for different languages # Recopilar textos para afinidad si los idiomas son diferentes
if config.codigo_columna_maestra != config.codigo_idioma_seleccionado: if config.codigo_columna_maestra != config.codigo_idioma_seleccionado:
texts_to_check = {} if pd.notnull(texto_propuesto) and texto_propuesto.strip() != "":
batch_size = 20 # Compactar los textos para comparación
texto_original_comp = fc.compactar_celda_traducida(config.codigo_tipo_PLC, str(texto_original))
texto_propuesto_comp = fc.compactar_celda_traducida(config.codigo_tipo_PLC, str(texto_propuesto))
for row in range(2, worksheet.max_row + 1): # Si los textos son idénticos después de compactar, afinidad automática de 1
clave = worksheet.cell(row=row, column=1).value if texto_original_comp == texto_propuesto_comp:
texto = worksheet.cell(row=row, column=2).value identical_texts[texto_original] = row
if pd.notnull(texto) and texto.strip() != "": else:
texts_to_check[clave] = texto texts_to_check[texto_original] = texto_propuesto
if len(texts_to_check) >= batch_size:
try:
affinities = fc.affinity_batch_openai(
config.codigo_tipo_PLC,
texts_to_check,
openai_client,
logger,
)
for check_row, (key, score) in enumerate(
affinities.items(), start=2
):
col_idx = (
df_export.columns.get_loc("Affinity_Score")
+ 1
)
worksheet.cell(
row=check_row, column=col_idx
).value = score
if score < 1:
worksheet.cell(
row=check_row, column=2
).fill = yellow_fill
afinidad_baja += 1
except Exception as e:
logger.error(f"Error en lote de afinidad: {str(e)}")
texts_to_check.clear()
progress_bar.increment() progress_bar.increment()
# Mark duplicate cells in bold progress_bar.finish()
# Segunda fase: Procesar textos idénticos y calcular afinidades en lote
if config.codigo_columna_maestra != config.codigo_idioma_seleccionado:
# Asignar afinidad 1 a textos idénticos
logger.info(f"Asignando afinidad 1 a {len(identical_texts)} textos idénticos")
for _, row in identical_texts.items():
col_idx = df_export.columns.get_loc("Affinity_Score") + 1
worksheet.cell(row=row, column=col_idx).value = 1.0
# Calcular afinidades para textos diferentes
if texts_to_check:
logger.info(f"Calculando afinidad para {len(texts_to_check)} textos diferentes")
try:
affinities = fc.calculate_batch_affinities(
texts_to_check,
config.codigo_tipo_PLC,
openai_client,
logger
)
# Aplicar resultados de afinidad
progress_bar = fc.ProgressBar(
len(affinities), prefix="Aplicando afinidades:", suffix="Completado"
)
for texto_original, afinidad in affinities.items():
row = next(row for row in range(2, worksheet.max_row + 1)
if worksheet.cell(row=row, column=1).value == texto_original)
col_idx = df_export.columns.get_loc("Affinity_Score") + 1
worksheet.cell(row=row, column=col_idx).value = afinidad
if afinidad < 1:
worksheet.cell(row=row, column=2).fill = yellow_fill
afinidad_baja += 1
progress_bar.increment()
progress_bar.finish()
except Exception as e:
logger.error(f"Error en el cálculo de afinidad por lotes: {str(e)}")
print(f"Error en el cálculo de afinidad por lotes: {str(e)}")
# Marcar celdas duplicadas
bold_font = Font(bold=True) bold_font = Font(bold=True)
celdas_duplicadas = 0 celdas_duplicadas = 0
for filas in texto_a_filas.values(): for filas in texto_a_filas.values():
@ -191,27 +206,18 @@ def exportar_para_traduccion(config: TranslationConfig):
cell.font = bold_font cell.font = bold_font
celdas_duplicadas += len(filas) celdas_duplicadas += len(filas)
progress_bar.finish() # Imprimir resumen
print(f"\nArchivo exportado para traducción: {ruta_export}") print(f"\nArchivo exportado para traducción: {ruta_export}")
print("Las celdas con idioma incorrecto han sido marcadas en azul.") print("Las celdas con idioma incorrecto han sido marcadas en azul.")
print( print("Se ha añadido el nombre del idioma detectado cuando es diferente del esperado.")
"Se ha añadido el nombre del idioma detectado cuando es diferente del esperado." print(f"Se ha agregado la columna del idioma secundario ({config.codigo_idioma_secundario}) al final de la planilla.")
)
print(
f"Se ha agregado la columna del idioma secundario ({config.codigo_idioma_secundario}) al final de la planilla."
)
if config.codigo_columna_maestra != config.codigo_idioma_seleccionado:
print(
f"Se encontraron {inconsistencias} celdas con errores de validación (marcadas en rojo)"
)
print(
f"Se encontraron {afinidad_baja} celdas con afinidad menor a 1 (marcadas en amarillo)"
)
print(
f"Se han marcado {celdas_duplicadas} celdas en negrita por tener texto duplicado."
)
if config.codigo_columna_maestra != config.codigo_idioma_seleccionado:
print(f"Se encontraron {len(identical_texts)} textos idénticos (afinidad 1)")
print(f"Se encontraron {inconsistencias} celdas con errores de validación (marcadas en rojo)")
print(f"Se encontraron {afinidad_baja} celdas con afinidad menor a 1 (marcadas en amarillo)")
print(f"Se han marcado {celdas_duplicadas} celdas en negrita por tener texto duplicado.")
def run(config: TranslationConfig): def run(config: TranslationConfig):
global logger global logger

View File

@ -232,70 +232,26 @@ def main(config: TranslationConfig):
update_progress.finish() update_progress.finish()
# Inicializar ProgressBar para la fase de cálculo de afinidad # Afinidades
affinity_progress = fc.ProgressBar( # Los textos ya vienen del proceso de traducción
num_texts, prefix="Calculando afinidad:", suffix="Completado" texts_to_check = {}
for key, translated_text in translations.items():
if pd.notna(translated_text) and str(translated_text).strip() != "":
texts_to_check[key] = translated_text
# Calcular afinidades usando LLM
affinities_dict = fc.calculate_batch_affinities(
texts_to_check,
config.codigo_tipo_PLC,
openai_client,
logger
) )
# Afinidades # Asignar resultados al DataFrame
affinities = {}
for start_idx in range(0, num_texts, batch_size):
end_idx = min(start_idx + batch_size, num_texts)
batch_texts = dict(list(texts_to_translate.items())[start_idx:end_idx])
logger.info(f"Afinidad: celdas desde {start_idx} a {end_idx}.")
retries = 2
for attempt in range(retries):
try:
batch_affinities = fc.affinity_batch_openai(
config.codigo_tipo_PLC, batch_texts, openai_client, logger
)
affinities.update(batch_affinities)
break
except Exception as e:
if attempt < retries - 1:
logger.warning(
f"Error en el intento {attempt + 1} de Afinidad de celdas desde {start_idx} a {end_idx}: {e}. Reintentando..."
)
print(
f"Error en el intento {attempt + 1} de Afinidad de celdas desde {start_idx} a {end_idx}: {e}. Reintentando..."
)
time.sleep(3)
else:
logger.error(
f"Error en todos los intentos de Afinidad de celdas desde {start_idx} a {end_idx}: {e}"
)
print(
f"Error en todos los intentos de Afinidad de celdas desde {start_idx} a {end_idx}: {e}"
)
for key, value in batch_texts.items():
try:
score = fc.calcular_afinidad(
config.codigo_tipo_PLC,
key,
value,
openai_client,
logger,
)
affinities[key] = score
except Exception as ind_e:
affinities[key] = "0"
logger.error(
f"Error en el cálculo individual de Afinidad para el texto '{key}': {ind_e}"
)
print(
f"Error en el cálculo individual de Afinidad para el texto '{key}': {ind_e}"
)
affinity_progress.increment()
affinity_progress.finish()
# Actualizar el DataFrame con las Afinidades
for index, row in df.iterrows(): for index, row in df.iterrows():
celda_clave = str(row[source_col]) key = str(row[source_col])
if celda_clave in affinities: if key in affinities_dict:
df.at[index, affinity_col] = affinities[celda_clave] df.at[index, affinity_col] = affinities_dict[key]
output_path = config.get_auto_translate_path() output_path = config.get_auto_translate_path()