script/ter_document.py - dev - v.bolshakov/AIEcosystem-Testing

script/ter_document.py
109 строк3 KB
Developer
fix(ter): use sentence_score instead of deprecated scorer method
05 авг 2026, 14:02
05 авг 2026, 14:02b216435
import pymongo
from pymongo import MongoClient
import re
import sacrebleu
import lxml.etree as ET
import pandas as pd
from typing import List, Dict, Tuple
import sys
import json

# Настройки подключения к MongoDB
MONGODB_URI = "mongodb://localhost:27017/"
DATABASE_NAME = "mydatabase"
COLLECTION_NAME = "xliff_segment"

def connect_to_mongodb():
    try:
        client = MongoClient(MONGODB_URI)
        client.admin.command('ping')
        print("Успешное подключение к MongoDB")
        return client
    except Exception as e:
        print(f"Ошибка подключения к MongoDB: {e}")
        return None

def calc_ter(ref, hyp):
    # Обработка пустых значений
    if not ref and not hyp:
        return 0.0
    if not ref or not hyp:
        return 100.0

    # Очистка от технических символов
    r = re.sub(r'<[^>]+>|\{[^}]*\}|\xa0', '', str(ref)).strip()
    h = re.sub(r'<[^>]+>|\{[^}]*\}|\xa0', '', str(hyp)).strip()

    # Если reference пустой после очистки
    if not r:
        return 100.0 if h else 0.0

    # Вычисление TER с использованием библиотеки SacreBLEU (с учетом shifts)
    ter = sacrebleu.metrics.TER()
    ter_score = ter.sentence_score(h, [r]).score

    # Вычисление TER в процентах
    return round(min(ter_score, 100.0), 2)

def process_documents(document_id):
    client = connect_to_mongodb()
    if not client:
        return

    try:
        db = client[DATABASE_NAME]
        collection = db[COLLECTION_NAME]

        filter_query = {"document_id": document_id}
        projection = {
            "_id": True,
            "pre_translation": True,
            "human_final": True
        }

        documents = list(collection.find(filter_query, projection))
        total_docs = len(documents)
        print(f"Найдено записей: {total_docs}")

        if total_docs == 0:
            print("Нет записей для обработки")
            return

        for doc in documents:
            doc_id = doc['_id']
            pre_translation = doc.get('pre_translation', '')
            human_final = doc.get('human_final', '')

            # Пропускаем, если оба поля пустые
            if not pre_translation and not human_final:
                continue

            # Вычисляем TER
            try:
                ter_score = calc_ter(human_final, pre_translation)

                # 3. Обновляем запись в базе
                result = collection.update_one(
                    {"_id": doc_id},
                    {"$set": {"ter": ter_score}}
                )

            except Exception as e:
                print(f"Ошибка при обработке документа {doc_id}: {e}")


    except Exception as e:
        print(f"Критическая ошибка: {e}")
    finally:
        client.close()
        print("Соединение с MongoDB закрыто")

def main():
    args = sys.argv;
    document_id = args[1]

    # Выполняем основную обработку
    process_documents(int(document_id))

if __name__ == "__main__":
    main()