preprocess/utils/common/utils.py

import re

from tqdm import tqdm

'''def get_delimiter(file_path):
    with open(file_path, 'r') as f:
        sample = f.read(1024)  # читаем часть файла для анализа
        dialect = csv.Sniffer().sniff(sample)
    return dialect.delimiter'''

def get_delimiter(file_path):
    with open(file_path, 'r', encoding="utf-8") as f:
        ln = f.readline()
        if ',' in ln:
            return ','
        if ';' in ln:
            return ';'
        if '\t' in ln:
            return '\t'
        if '|' in ln:
            return '|'

    raise ValueError(None, "Error parsing CSV file. Cannot detect delimiter")

def remove_quotes(text):
    return re.sub(r'["\']', '', text)


def remove_l(text):
    result = re.sub(r'\bл\b', '', text, flags=re.IGNORECASE)

  # Убираем возможные лишние пробелы, возникающие после удаления
    result = re.sub(r'\s{2,}', ' ', result).strip()
    return result


def clean_wine_name(name):
    """
    Удаляет в конце строки отдельно стоящие буквы (однобуквенные слова), не входящие в состав других слов.
    Например, "токай   л" превратится в "токай".
    """
    # Регулярное выражение ищет:
    # \s+        – один или несколько пробельных символов;
    # \b         – граница слова;
    # [A-Za-zА-ЯЁа-яё] – ровно одна буква (латинская или кириллическая);
    # \b         – граница слова;
    # \s*$       – любые пробелы до конца строки.
    return re.sub(r'\s+\b[A-Za-zА-ЯЁа-яё]\b\s*$', '', name)


def find_full_word(text, word_list):
    """
    Ищет первое полное вхождение слова из word_list в строке text.
    Возвращает найденное слово или None, если совпадение не найдено.
    """
    for word in word_list:
        pattern = r'\b' + re.escape(word) + r'\b'
        if re.search(pattern, text, re.IGNORECASE):
            return word
    return None


def merge_wine_type(items, colors=None, color_merge_dict=None):    
    result=[]
    for row in tqdm(items.iterrows()):
        try:
            #print("merge_wine_type:" + str(row))
            if row[1]['type_wine'] is not None:
                color=find_full_word(row[1]['type_wine'], colors)
                if color is not None:
                    result.append(color)
                else:
                    color=find_full_word(row[1]['name'], colors)
                    if color is not None:
                        result.append(color)
                    else:
                        result.append(None)
            else:
                color=find_full_word(row[1]['name'], colors)
                if color is not None:
                    result.append(color)
                else:
                    result.append(None)
        except Exception as ex:
            print(ex)
            result.append(None)

    items['new_type_wine']=result
    items['new_type_wine']=items['new_type_wine'].replace(color_merge_dict)


def merge_types(items, products, type_merge_dict={}, sub_alco_types=["Бренди", "Шампань", "Шампанское"]):
    alco_types=[i.strip().lower() for i in products['type'].unique()]
    alco_types.append('ликёр')
    result=[]

    for row in tqdm(items.iterrows()):
        try:
            # Parameter 'sub_alco_types' specifies specific alcohol types that usually specified
            # in product / item name along with "parent" type and in this case this subtype should have priority
            # For example, "Вино Шампано Ле Брён де Нёвиль", or "Бренди де Херес"
            if sub_alco_types:
                type_in_name=find_full_word(row[1]['name'], sub_alco_types)
                if type_in_name is not None:
                    result.append(type_in_name)
                    continue

            type_in_name=find_full_word(row[1]['name'], alco_types)
            if type_in_name is not None:
                result.append(type_in_name)
                continue
            if row[1]['type'] is not None:
                type_in_type=find_full_word(row[1]['type'], alco_types)
                if type_in_type is not None:
                    result.append(type_in_type)
                else:
                    result.append(row[1]['type'])
            else:
                result.append(None)
        except Exception as ex:
            print(ex)
            result.append(None)

    items['new_type']=result
    #items['new_type']=items['new_type'].replace({'ликёр': 'ликер', None: 'unmatched'})
    items['new_type'] = items['new_type'].replace(type_merge_dict)


def trim_name(text, words_to_remove):
    """
    Удаляет из текста только те слова, которые полностью совпадают с элементами списка words_to_remove.

    :param text: Исходная строка.
    :param words_to_remove: Список слов, которые необходимо удалить.
    :return: Обновлённая строка с удалёнными словами.
    """
    # Создаём регулярное выражение, которое ищет любое из указанных слов как отдельное слово.
    # Используем re.escape, чтобы экранировать спецсимволы в словах.
    pattern = r'\b(?:' + '|'.join(re.escape(word) for word in words_to_remove) + r')\b'
    #print("Pattern: " + pattern)

    # Заменяем найденные полные слова на пустую строку.
    new_text = re.sub(pattern, '', text, flags=re.IGNORECASE)

    # Убираем лишние пробелы, возникающие после удаления слов.
    new_text = re.sub(r'\s+', ' ', new_text).strip()

    return new_text


def name_trimmer(df, prcess_text, types_and_others):
    result={}
    gbs=[]
    sours=[]
    for idx, row in tqdm(df.iterrows()):
        #print("Name1: " + str(row['name']))
        text, alcohol, volume_or_number, years, production_year, gb, color, sour=prcess_text(str(row['name']))
        #print("Name2: " + text)
        text=trim_name(text, types_and_others).replace(',','').replace('.','')
        #print("Name3: " + text)
        result[row['id']]=text.lower().strip() #remove_l(text).lower().strip()

        gbs.append(gb)
        sours.append(sour)
    return result, gbs, sours