208 lines
8.2 KiB
Python
208 lines
8.2 KiB
Python
import pandas as pd
|
|
from openai import OpenAI
|
|
import os
|
|
import re
|
|
import logging
|
|
from openai_api_key import api_key
|
|
from x2_master_export2translate import transformar_texto
|
|
import ollama
|
|
import json
|
|
|
|
client = OpenAI(api_key=api_key())
|
|
|
|
# Diccionario de idiomas
|
|
IDIOMAS = {
|
|
1: ("English", "en-GB"),
|
|
2: ("Portuguese", "pt-PT"),
|
|
3: ("Spanish", "es-ES"),
|
|
4: ("Russian", "ru-RU"),
|
|
5: ("French", "fr-FR"),
|
|
6: ("German", "de-DE"),
|
|
}
|
|
|
|
|
|
def configurar_logger():
|
|
logger = logging.getLogger("translate_logger")
|
|
logger.setLevel(logging.DEBUG) # Cambiado a DEBUG para más información
|
|
os.makedirs(".\\data", exist_ok=True)
|
|
fh = logging.FileHandler(".\\data\\translate_log.log", encoding="utf-8")
|
|
fh.setLevel(logging.DEBUG)
|
|
formatter = logging.Formatter("%(asctime)s - %(levelname)s - %(message)s")
|
|
fh.setFormatter(formatter)
|
|
logger.addHandler(fh)
|
|
return logger
|
|
|
|
|
|
logger = configurar_logger()
|
|
|
|
|
|
def mostrar_idiomas():
|
|
print("Selecciona el idioma de destino:")
|
|
for numero, (nombre, _) in IDIOMAS.items():
|
|
print(f"{numero}: {nombre}")
|
|
|
|
|
|
def translate_text(text, source_lang, target_lang):
|
|
logger.info(
|
|
f"Solicitando traducción de {source_lang} a {target_lang} para el texto: {text}"
|
|
)
|
|
response = client.chat.completions.create(
|
|
model="gpt-3.5-turbo",
|
|
messages=[
|
|
{"role": "system", "content": f"You are a translator."},
|
|
{
|
|
"role": "user",
|
|
"content": f"Translate the following text from {source_lang} to {target_lang} while preserving special fields like <> and <#>. This texts are for an HMI industrial machine: {text}",
|
|
},
|
|
],
|
|
max_tokens=150,
|
|
temperature=0.3,
|
|
)
|
|
translated_text = response.choices[0].message.content.strip()
|
|
logger.info(f"Respuesta recibida: {translated_text}")
|
|
return 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_openai(batch_texts, source_lang, target_lang):
|
|
# Aquí se asume que esta función maneja una lista de textos y devuelve una lista de traducciones
|
|
translations = []
|
|
for text in batch_texts:
|
|
translation = translate_text(text, source_lang, target_lang)
|
|
translations.append(translation)
|
|
return translations
|
|
|
|
def translate_batch_openai(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}"
|
|
)
|
|
request_payload = json.dumps({"texts": texts, "source_lang": source_lang, "target_lang": target_lang})
|
|
response = client.chat.completions.create(
|
|
model= "gpt-4o-mini", # "gpt-3.5-turbo",
|
|
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):
|
|
raise ValueError("La cantidad de traducciones recibidas no coincide con la cantidad de textos enviados.")
|
|
return translations
|
|
|
|
def translate_batch(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 texto_requiere_traduccion(texto):
|
|
palabras = re.findall(r"\b\w{4,}\b", texto)
|
|
campos_especiales = re.findall(r"<.*?>", texto)
|
|
requiere_traduccion = len(palabras) > 0 or len(campos_especiales) != len(
|
|
re.findall(r"<#>", texto)
|
|
)
|
|
logger.debug(
|
|
f"Decisión de traducción para texto '{texto}': {'Sí' if requiere_traduccion else 'No'} (palabras > 3 letras: {len(palabras) > 0}, solo campos especiales: {len(campos_especiales) == len(re.findall(r'<#>', texto))})"
|
|
)
|
|
return requiere_traduccion
|
|
|
|
|
|
def main(file_path, target_lang_code,target_lang, traducir_todo, batch_size=10):
|
|
df = pd.read_excel(file_path)
|
|
source_col = "it-IT"
|
|
source_translated_col = target_lang_code
|
|
target_col = f"{target_lang_code} Translated"
|
|
|
|
# Asegurarse de que la columna de destino existe
|
|
if target_col not in df.columns:
|
|
df[target_col] = None
|
|
|
|
texts_to_translate = []
|
|
indices_to_translate = []
|
|
|
|
for index, row in df.iterrows():
|
|
source_text = str(row[source_col])
|
|
source_translated_text = str(row[source_translated_col]) if source_translated_col in df.columns else ""
|
|
processed_text = transformar_texto(source_text)
|
|
|
|
if traducir_todo:
|
|
# Traducir todas las celdas del idioma de destino
|
|
if texto_requiere_traduccion(processed_text):
|
|
texts_to_translate.append(processed_text)
|
|
indices_to_translate.append(index)
|
|
else:
|
|
# Traducir solo las celdas vacías en el idioma de destino original
|
|
if pd.isna(row[source_translated_col]) or source_translated_text.strip() == "":
|
|
if texto_requiere_traduccion(processed_text):
|
|
texts_to_translate.append(processed_text)
|
|
indices_to_translate.append(index)
|
|
|
|
num_texts = len(texts_to_translate)
|
|
logger.info(f"Número total de textos a traducir: {num_texts}")
|
|
print(f"Número total de textos a traducir: {num_texts}")
|
|
|
|
translations = []
|
|
for start_idx in range(0, num_texts, batch_size):
|
|
end_idx = min(start_idx + batch_size, num_texts)
|
|
batch_texts = texts_to_translate[start_idx:end_idx]
|
|
logger.info(f"Traduciendo: celdas desde {start_idx} a {end_idx}.")
|
|
print(f"Traduciendo : celdas desde: {start_idx} a :{end_idx}.")
|
|
try:
|
|
batch_translations = translate_batch_openai(batch_texts, 'Italian', target_lang_code)
|
|
translations.extend(batch_translations)
|
|
except Exception as e:
|
|
logger.error(f"Error en la traducción de celdas desde {start_idx} a {end_idx}: {e}")
|
|
print(f"Error en la traducción de celdas desde {start_idx} a {end_idx}: {e}")
|
|
continue
|
|
|
|
logger.info(f"Número total de traducciones recibidas: {len(translations)}")
|
|
|
|
if len(translations) != len(indices_to_translate):
|
|
logger.warning(f"Desajuste entre el número de traducciones ({len(translations)}) y el número de índices ({len(indices_to_translate)})")
|
|
|
|
for i, index in enumerate(indices_to_translate):
|
|
if i < len(translations):
|
|
df.at[index, target_col] = translations[i]
|
|
else:
|
|
logger.error(f"No hay traducción disponible para el índice {index}")
|
|
|
|
output_path = os.path.join(os.path.dirname(file_path), '3_master_export2translate_translated.xlsx')
|
|
df.to_excel(output_path, index=False)
|
|
logger.info(f"Archivo traducido guardado en: {output_path}")
|
|
print(f"Archivo traducido guardado en: {output_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
batch_size = 20
|
|
translate_file = ".\\data\\2_master_export2translate.xlsx"
|
|
|
|
mostrar_idiomas()
|
|
seleccion_idioma = int(input("Introduce el número del idioma de destino: "))
|
|
if seleccion_idioma not in IDIOMAS:
|
|
print("Selección inválida.")
|
|
else:
|
|
target_lang, target_lang_code = IDIOMAS[seleccion_idioma]
|
|
traducir_todo = (
|
|
input("¿Desea traducir todas las celdas (s/n)? ").strip().lower() == "s"
|
|
)
|
|
main(translate_file, target_lang_code,target_lang, traducir_todo, batch_size)
|