Miguel
/
HMI_Translate_Helper_wMaster_in_Excel
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

Algunos cambios para adaptar a TranslationConfig

This commit is contained in:
Miguel 2024-10-12 14:51:41 +02:00
parent 97de43cba7
commit 8fbffef047
1 changed files with 30 additions and 59 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

89
x3_llm_auto_translate.py
View File

@ -106,13 +106,13 @@ def translate_batch_openai(texts_dict, source_lang, target_lang):
return dict(zip(texts_dict.keys(), translations)) return dict(zip(texts_dict.keys(), translations))
def affinity_batch_openai(tipo_PLC, texts_dict): def affinity_batch_openai(codigo_tipo_PLC, texts_dict):
system_prompt = ( system_prompt = (
"Evaluate the semantic similarity between the following table of pairs of texts in json format on a scale from 0 to 1. " "Evaluate the semantic similarity between the following table of pairs of texts in json format on a scale from 0 to 1. "
"Return the similarity scores for every row in JSON format as a list of numbers, without any additional text or formatting." "Return the similarity scores for every row in JSON format as a list of numbers, without any additional text or formatting."
) )
original_list = [ original_list = [
fc.compactar_celda_traducida(tipo_PLC, key) for key in texts_dict.keys() fc.compactar_celda_traducida(codigo_tipo_PLC, key) for key in texts_dict.keys()
] ]
re_translated_list = list(texts_dict.values()) re_translated_list = list(texts_dict.values())
@ -204,6 +204,7 @@ def calcular_afinidad(tipo_PLC, texto1, texto2):
def main(config: TranslationConfig): def main(config: TranslationConfig):
global logger
df = fc.read_dataframe_with_cleanup_retries(config.get_translate_path()) df = fc.read_dataframe_with_cleanup_retries(config.get_translate_path())
source_col = config.codigo_columna_maestra source_col = config.codigo_columna_maestra
@ -254,8 +255,7 @@ def main(config: TranslationConfig):
logger.info(f"Número total de textos a traducir: {num_texts}") logger.info(f"Número total de textos a traducir: {num_texts}")
print(f"Número total de textos a traducir: {num_texts}") print(f"Número total de textos a traducir: {num_texts}")
# Traducciones # Traducciones
# Hacer las traducciones via LLM en batch
translations = {} translations = {}
for start_idx in range(0, num_texts, batch_size): for start_idx in range(0, num_texts, batch_size):
end_idx = min(start_idx + batch_size, num_texts) end_idx = min(start_idx + batch_size, num_texts)
@ -263,101 +263,73 @@ def main(config: TranslationConfig):
logger.info(f"Traduciendo: celdas desde {start_idx} a {end_idx}.") logger.info(f"Traduciendo: celdas desde {start_idx} a {end_idx}.")
print(f"Traduciendo : celdas desde: {start_idx} a :{end_idx}.") print(f"Traduciendo : celdas desde: {start_idx} a :{end_idx}.")
retries = 4 # Número de intentos totales (1 inicial + 1 reintento) retries = 4
for attempt in range(retries): for attempt in range(retries):
try: try:
batch_translations = translate_batch_openai( batch_translations = translate_batch_openai(
batch_texts, batch_texts,
fc.idiomas_idiomafromcode(codigo_columna_maestra), fc.idiomas_idiomafromcode(config.codigo_columna_maestra),
target_lang, fc.idiomas_idiomafromcode(config.codigo_idioma_seleccionado)
) )
translations.update(batch_translations) translations.update(batch_translations)
break # Si la traducción es exitosa, salimos del bucle de reintentos break
except Exception as e: except Exception as e:
if attempt < retries - 1: # Si no es el último intento if attempt < retries - 1:
logger.warning( logger.warning(f"Error en el intento {attempt + 1} de traducción de celdas desde {start_idx} a {end_idx}: {e}. Reintentando...")
f"Error en el intento {attempt + 1} de traducción de celdas desde {start_idx} a {end_idx}: {e}. Reintentando..." print(f"Error en el intento {attempt + 1} de traducción de celdas desde {start_idx} a {end_idx}: {e}. Reintentando...")
)
print(
f"Error en el intento {attempt + 1} de traducción de celdas desde {start_idx} a {end_idx}: {e}. Reintentando..."
)
time.sleep(3) time.sleep(3)
else: # Si es el último intento else:
logger.error( logger.error(f"Error en todos los intentos de traducción de celdas desde {start_idx} a {end_idx}: {e}")
f"Error en todos los intentos de traducción de celdas desde {start_idx} a {end_idx}: {e}" print(f"Error en todos los intentos de traducción de celdas desde {start_idx} a {end_idx}: {e}")
)
print(
f"Error en todos los intentos de traducción de celdas desde {start_idx} a {end_idx}: {e}"
)
logger.info(f"Número total de traducciones recibidas: {len(translations)}") logger.info(f"Número total de traducciones recibidas: {len(translations)}")
# Traduccion inversa
# Actualizar el DataFrame con las traducciones y hacemos la Traduccion inversa # Actualizar el DataFrame con las traducciones y hacemos la Traduccion inversa
for index, row in tqdm( for index, row in tqdm(df.iterrows(), total=df.shape[0], desc="Procesando traducciones"):
df.iterrows(), total=df.shape[0], desc="Procesando traducciones"
):
celda_clave = str(row[source_col]) celda_clave = str(row[source_col])
if celda_clave in translations: if celda_clave in translations:
df.at[index, target_col] = translations[celda_clave] df.at[index, target_col] = translations[celda_clave]
# Realizar la traducción de verificación con Google Translate
try: try:
google_translation = google_translate( google_translation = google_translate(
translations[celda_clave], translations[celda_clave],
fc.idiomas_shortcodefromcode(codigo_columna_maestra), fc.idiomas_shortcodefromcode(config.codigo_columna_maestra)
) )
df.at[index, check_translate_col] = google_translation df.at[index, check_translate_col] = google_translation
except Exception as e: except Exception as e:
logger.error( logger.error(f"Error en la traducción de Google para el texto '{celda_clave}': {e}")
f"Error en la traducción de Google para el texto '{celda_clave}': {e}"
)
df.at[index, check_translate_col] = "Error en la traducción" df.at[index, check_translate_col] = "Error en la traducción"
df.at[index, affinity_col] = 0.0 df.at[index, affinity_col] = 0.0
# Afinidades # Afinidades
# Se calculan las Afinidades
affinities = {} affinities = {}
for start_idx in range(0, num_texts, batch_size): # num_text for start_idx in range(0, num_texts, batch_size):
end_idx = min(start_idx + batch_size, num_texts) end_idx = min(start_idx + batch_size, num_texts)
batch_texts = dict(list(texts_to_translate.items())[start_idx:end_idx]) batch_texts = dict(list(texts_to_translate.items())[start_idx:end_idx])
logger.info(f"Afinidad: celdas desde {start_idx} a {end_idx}.") logger.info(f"Afinidad: celdas desde {start_idx} a {end_idx}.")
print(f"Afinidad: celdas desde: {start_idx} a :{end_idx}.") print(f"Afinidad: celdas desde: {start_idx} a :{end_idx}.")
retries = 2 # Número de intentos totales (1 inicial + 1 reintento) retries = 2
for attempt in range(retries): for attempt in range(retries):
try: try:
batch_affinities = affinity_batch_openai(tipo_PLC, batch_texts) batch_affinities = affinity_batch_openai(config.codigo_tipo_PLC, batch_texts)
affinities.update(batch_affinities) affinities.update(batch_affinities)
break # Si la llamada es exitosa, salimos del bucle de reintentos break
except Exception as e: except Exception as e:
if attempt < retries - 1: # Si no es el último intento if attempt < retries - 1:
logger.warning( logger.warning(f"Error en el intento {attempt + 1} de Afinidad de celdas desde {start_idx} a {end_idx}: {e}. Reintentando...")
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...")
)
print(
f"Error en el intento {attempt + 1} de Afinidad de celdas desde {start_idx} a {end_idx}: {e}. Reintentando..."
)
time.sleep(3) time.sleep(3)
else: # Si es el último intento else:
logger.error( logger.error(f"Error en todos los intentos de Afinidad de celdas desde {start_idx} a {end_idx}: {e}")
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}")
)
print(
f"Error en todos los intentos de Afinidad de celdas desde {start_idx} a {end_idx}: {e}"
)
# Intentar individualmente si falla en batch
for key, value in batch_texts.items(): for key, value in batch_texts.items():
try: try:
score = calcular_afinidad(tipo_PLC, key, value) score = calcular_afinidad(config.codigo_tipo_PLC, key, value)
affinities[key] = score affinities[key] = score
except Exception as ind_e: except Exception as ind_e:
affinities[key] = "0" affinities[key] = "0"
logger.error( logger.error(f"Error en el cálculo individual de Afinidad para el texto '{key}': {ind_e}")
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}")
)
print(
f"Error en el cálculo individual de Afinidad para el texto '{key}': {ind_e}"
)
# Actualizar el DataFrame con las Afinidades # Actualizar el DataFrame con las Afinidades
for index, row in df.iterrows(): for index, row in df.iterrows():
@ -370,7 +342,6 @@ def main(config: TranslationConfig):
logger.info(f"Archivo traducido guardado en: {output_path}") logger.info(f"Archivo traducido guardado en: {output_path}")
print(f"Archivo traducido guardado en: {output_path}") print(f"Archivo traducido guardado en: {output_path}")
def run(config: TranslationConfig): def run(config: TranslationConfig):
global logger global logger
logger = fc.configurar_logger(config.work_dir) logger = fc.configurar_logger(config.work_dir)