358 lines
14 KiB
Python
358 lines
14 KiB
Python
import pandas as pd
|
|
from openai import OpenAI
|
|
import os
|
|
from openai_api_key import openai_api_key
|
|
from google_api_key import google_api_key
|
|
import ollama
|
|
import json
|
|
from google.cloud import translate_v2 as translate
|
|
from google.oauth2 import service_account
|
|
import html
|
|
from tqdm import tqdm
|
|
import PyLibrary.funciones_comunes as fc
|
|
import time
|
|
import PyLibrary.funciones_comunes as fc
|
|
from translation_config import TranslationConfig
|
|
from openai import OpenAI
|
|
from tqdm import tqdm
|
|
|
|
openai_client = OpenAI(api_key=openai_api_key())
|
|
GOOGLE_APPLICATION_CREDENTIALS = "translate-431108-020c17463fbb.json"
|
|
batch_size = 20
|
|
|
|
# Definir el logger a nivel de módulo
|
|
logger = None
|
|
|
|
|
|
def init_google_translate_client():
|
|
if os.path.exists(GOOGLE_APPLICATION_CREDENTIALS):
|
|
# Usar credenciales de cuenta de servicio
|
|
credentials = service_account.Credentials.from_service_account_file(
|
|
GOOGLE_APPLICATION_CREDENTIALS
|
|
)
|
|
return translate.Client(credentials=credentials)
|
|
else:
|
|
raise ValueError(
|
|
"No se han proporcionado credenciales válidas para Google Translate"
|
|
)
|
|
|
|
|
|
google_translate_client = init_google_translate_client()
|
|
|
|
|
|
def google_translate(text, target_language):
|
|
result = google_translate_client.translate(text, target_language=target_language)
|
|
translated_text = result["translatedText"]
|
|
return html.unescape(translated_text)
|
|
|
|
|
|
def read_system_prompt():
|
|
try:
|
|
with open(".\\data\\system_prompt.txt", "r", encoding="utf-8") as file:
|
|
return file.read().strip()
|
|
except FileNotFoundError:
|
|
logger.warning(
|
|
"Archivo system_prompt.txt no encontrado. Usando prompt por defecto."
|
|
)
|
|
return "You are a translator."
|
|
|
|
|
|
def translate_batch_ollama(texts, source_lang, target_lang):
|
|
joined_text = "\n".join(texts)
|
|
system_prompt = read_system_prompt()
|
|
logger.info(
|
|
f"Solicitando traducción de {source_lang} a {target_lang} para el lote de textos:\n{joined_text}"
|
|
)
|
|
response = ollama.generate(
|
|
model="llama3.1",
|
|
prompt=f"Translate the following texts from {source_lang} to {target_lang} while preserving special fields like <> and <#>. {system_prompt}: \n\n{joined_text}",
|
|
)
|
|
|
|
translations = response["response"].strip().split("\n")
|
|
logger.info(f"Respuestas recibidas:\n{translations}")
|
|
return translations
|
|
|
|
|
|
def translate_batch_openai(texts_dict, source_lang, target_lang):
|
|
system_prompt = read_system_prompt()
|
|
texts_list = list(texts_dict.values())
|
|
|
|
request_payload = json.dumps(
|
|
{"texts": texts_list, "source_lang": source_lang, "target_lang": target_lang}
|
|
)
|
|
logger.info(
|
|
f"Solicitando traducción de {source_lang} a {target_lang} para el lote de textos:\n{request_payload}"
|
|
)
|
|
|
|
response = openai_client.chat.completions.create(
|
|
model="gpt-4o-mini",
|
|
messages=[
|
|
{"role": "system", "content": f"You are a translator.{system_prompt}."},
|
|
{"role": "user", "content": request_payload},
|
|
],
|
|
max_tokens=1500,
|
|
temperature=0.3,
|
|
)
|
|
response_payload = json.loads(response.choices[0].message.content.strip())
|
|
translations = response_payload.get("texts", [])
|
|
logger.info(f"Respuestas recibidas:\n{translations}")
|
|
|
|
if len(translations) != len(texts_list):
|
|
raise ValueError(
|
|
"La cantidad de traducciones recibidas no coincide con la cantidad de textos enviados."
|
|
)
|
|
|
|
return dict(zip(texts_dict.keys(), translations))
|
|
|
|
|
|
def affinity_batch_openai(codigo_tipo_PLC, texts_dict):
|
|
system_prompt = (
|
|
"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."
|
|
)
|
|
original_list = [
|
|
fc.compactar_celda_traducida(codigo_tipo_PLC, key) for key in texts_dict.keys()
|
|
]
|
|
re_translated_list = list(texts_dict.values())
|
|
|
|
request_payload = json.dumps(
|
|
{"original": original_list, "compared": re_translated_list}
|
|
)
|
|
logger.info(f"Solicitando Afinidad para el lote de textos:\n{request_payload}")
|
|
|
|
response = openai_client.chat.completions.create(
|
|
model="gpt-4o-mini",
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": system_prompt,
|
|
},
|
|
{"role": "user", "content": request_payload},
|
|
],
|
|
max_tokens=1500,
|
|
temperature=0.3,
|
|
)
|
|
response_content = response.choices[0].message.content
|
|
|
|
# Limpiar y convertir el contenido de la respuesta
|
|
cleaned_response_content = response_content.strip().strip("'```json").strip("```")
|
|
|
|
# Intentar convertir el contenido a JSON
|
|
try:
|
|
response_payload = json.loads(cleaned_response_content)
|
|
except json.JSONDecodeError:
|
|
raise ValueError("La respuesta no se pudo decodificar como JSON.")
|
|
|
|
# Manejar diferentes formatos de respuesta
|
|
if isinstance(response_payload, dict) and "similarity_scores" in response_payload:
|
|
scores = response_payload["similarity_scores"]
|
|
elif isinstance(response_payload, list):
|
|
scores = response_payload
|
|
else:
|
|
raise ValueError("Formato de respuesta inesperado.")
|
|
|
|
logger.info(f"Respuestas recibidas:\n{scores}")
|
|
|
|
if len(scores) != len(original_list):
|
|
raise ValueError(
|
|
"La cantidad de afinidades recibidas no coincide con la cantidad de textos enviados."
|
|
)
|
|
|
|
return dict(zip(texts_dict.keys(), scores))
|
|
|
|
|
|
# Función que calcula la afinidad entre dos textos
|
|
def calcular_afinidad(tipo_PLC, texto1, texto2):
|
|
system_prompt = (
|
|
"Evaluate the semantic similarity between the following pair of texts on a scale from 0 to 1. "
|
|
"Return the similarity score as a single number."
|
|
)
|
|
|
|
original_text = fc.compactar_celda_traducida(tipo_PLC, texto1)
|
|
compared_text = texto2
|
|
|
|
request_payload = json.dumps({"original": original_text, "compared": compared_text})
|
|
logger.info(f"Solicitando afinidad para el par de textos:\n{request_payload}")
|
|
|
|
response = openai_client.chat.completions.create(
|
|
model="gpt-4o-mini",
|
|
messages=[
|
|
{
|
|
"role": "system",
|
|
"content": system_prompt,
|
|
},
|
|
{"role": "user", "content": request_payload},
|
|
],
|
|
max_tokens=1500,
|
|
temperature=0.3,
|
|
)
|
|
response_content = response.choices[0].message.content
|
|
|
|
# Limpiar y convertir el contenido de la respuesta
|
|
cleaned_response_content = response_content.strip().strip("'```json").strip("```")
|
|
|
|
# Intentar convertir el contenido a JSON
|
|
try:
|
|
score = float(cleaned_response_content)
|
|
except ValueError:
|
|
raise ValueError(
|
|
f"La respuesta no se pudo decodificar como un número: {cleaned_response_content}"
|
|
)
|
|
|
|
return score
|
|
|
|
|
|
def main(config: TranslationConfig):
|
|
global logger
|
|
df = fc.read_dataframe_with_cleanup_retries(config.get_translate_path())
|
|
|
|
source_col = config.codigo_columna_maestra
|
|
source_translated_col = config.codigo_idioma_seleccionado
|
|
target_col = f"{config.codigo_idioma_seleccionado} Translated"
|
|
check_translate_col = f"{config.codigo_idioma_seleccionado} CheckTranslate"
|
|
affinity_col = f"{config.codigo_idioma_seleccionado} Affinity"
|
|
|
|
# Asegurarse de que la columna de destino existe
|
|
for col in [target_col, check_translate_col, affinity_col]:
|
|
if col not in df.columns:
|
|
df[col] = None
|
|
|
|
texts_to_translate = {}
|
|
|
|
# Inicializar ProgressBar para la fase de preparación
|
|
prep_progress = fc.ProgressBar(len(df), prefix='Preparando textos:', suffix='Completado')
|
|
|
|
for index, row in df.iterrows():
|
|
celda_clave = str(row[source_col])
|
|
source_translated_text = str(row[source_translated_col]) if source_translated_col in df.columns else ""
|
|
celda_clave_compactada = fc.compactar_celda_traducida(config.codigo_tipo_PLC, celda_clave)
|
|
|
|
if config.traducir_todo:
|
|
if fc.texto_requiere_traduccion(config.codigo_tipo_PLC, celda_clave_compactada, logger):
|
|
df.at[index, source_translated_col] = ""
|
|
texts_to_translate[celda_clave] = celda_clave_compactada
|
|
else:
|
|
if pd.isna(row[source_translated_col]) or source_translated_text.strip() == "":
|
|
if fc.texto_requiere_traduccion(config.codigo_tipo_PLC, celda_clave_compactada, logger) or fc.texto_con_campos_especiales(config.codigo_tipo_PLC, celda_clave_compactada):
|
|
texts_to_translate[celda_clave] = celda_clave_compactada
|
|
|
|
prep_progress.update(index + 1)
|
|
|
|
prep_progress.finish()
|
|
|
|
num_texts = len(texts_to_translate)
|
|
logger.info(f"Número total de textos a traducir: {num_texts}")
|
|
print(f"\nNúmero total de textos a traducir: {num_texts}")
|
|
|
|
# Inicializar ProgressBar para la fase de traducción
|
|
trans_progress = fc.ProgressBar(num_texts, prefix='Traduciendo:', suffix='Completado')
|
|
|
|
# Traducciones
|
|
translations = {}
|
|
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"Traduciendo: celdas desde {start_idx} a {end_idx}.")
|
|
|
|
retries = 4
|
|
for attempt in range(retries):
|
|
try:
|
|
batch_translations = translate_batch_openai(
|
|
batch_texts,
|
|
fc.idiomas_idiomafromcode(config.codigo_columna_maestra),
|
|
fc.idiomas_idiomafromcode(config.codigo_idioma_seleccionado)
|
|
)
|
|
translations.update(batch_translations)
|
|
break
|
|
except Exception as e:
|
|
if attempt < retries - 1:
|
|
logger.warning(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)
|
|
else:
|
|
logger.error(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}")
|
|
|
|
trans_progress.update(end_idx)
|
|
|
|
trans_progress.finish()
|
|
logger.info(f"Número total de traducciones recibidas: {len(translations)}")
|
|
|
|
# Inicializar ProgressBar para la fase de actualización del DataFrame
|
|
update_progress = fc.ProgressBar(len(df), prefix='Actualizando DataFrame:', suffix='Completado')
|
|
|
|
# Actualizar el DataFrame con las traducciones y hacemos la Traduccion inversa
|
|
for index, row in df.iterrows():
|
|
celda_clave = str(row[source_col])
|
|
if celda_clave in translations:
|
|
df.at[index, target_col] = translations[celda_clave]
|
|
try:
|
|
google_translation = google_translate(
|
|
translations[celda_clave],
|
|
fc.idiomas_shortcodefromcode(config.codigo_columna_maestra)
|
|
)
|
|
df.at[index, check_translate_col] = google_translation
|
|
except Exception as e:
|
|
logger.error(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, affinity_col] = 0.0
|
|
update_progress.increment()
|
|
|
|
update_progress.finish()
|
|
# Inicializar ProgressBar para la fase de cálculo de afinidad
|
|
affinity_progress = fc.ProgressBar(num_texts, prefix='Calculando afinidad:', suffix='Completado')
|
|
|
|
# Afinidades
|
|
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 = affinity_batch_openai(config.codigo_tipo_PLC, batch_texts)
|
|
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 = calcular_afinidad(config.codigo_tipo_PLC, key, value)
|
|
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():
|
|
celda_clave = str(row[source_col])
|
|
if celda_clave in affinities:
|
|
df.at[index, affinity_col] = affinities[celda_clave]
|
|
|
|
output_path = config.get_auto_translate_path()
|
|
fc.save_dataframe_with_retries(df, output_path=output_path)
|
|
logger.info(f"Archivo traducido guardado en: {output_path}")
|
|
print(f"\nArchivo traducido guardado en: {output_path}")
|
|
|
|
def run(config: TranslationConfig):
|
|
global logger
|
|
logger = fc.configurar_logger(config.work_dir)
|
|
script_name = os.path.basename(__file__)
|
|
print(f"\rIniciando: {script_name}\r")
|
|
main(config)
|
|
|
|
if __name__ == "__main__":
|
|
import menu_pasos_traduccion
|
|
menu_pasos_traduccion.main() |