Upload 6 files

preprocess/utils/common/brand_matching.py

@@ -0,0 +1,137 @@
+from tqdm import tqdm
+import re
+from ahocorasick import Automaton
+from rapidfuzz import fuzz, process
+
+
+def contains_full_word(word, text, case_sensitive=True):
+    """
+    Проверяет, содержится ли слово word в строке text как отдельное слово.
+    Параметр case_sensitive задаёт, учитывать ли регистр.
+    """
+    flags = 0 if case_sensitive else re.IGNORECASE
+    pattern = r'\b' + re.escape(word) + r'\b'
+    return re.search(pattern, text, flags) is not None
+
+
+def unwrap_brands(products):
+    res={}
+    #brands=items['brand'].unique()
+    new_brands=sorted([x for x in products['brand'].unique() if isinstance(x, str)], key=len)
+    #items['new_brand'].unique() if isinstance(x, str)], key=len)
+
+    for i in tqdm(new_brands):
+        for j in new_brands:
+            if contains_full_word(i, j, case_sensitive=False):
+                if i != j:
+                    #if len(i)>1:#i != 'А' and i  != "Я":
+                        res[j]=i
+    return res
+
+
+
+def split_n_match(products, items, th_len=3):
+    result={}
+    conditionally_spited=[]
+    for i in tqdm(items['brand'].unique()):
+        if '/' in i:
+            conditionally_spited.append(i)
+    for i in tqdm(products['brand'].unique()):
+        for j in conditionally_spited:
+            if len(i)>th_len and contains_full_word(i,j):
+                result[j]=i
+    return result
+
+
+
+def fill_brands_in_dataframe(brands, df, col_name='new_brand', is_brand=True):
+    """
+    Заполняет колонку 'brand' в DataFrame найденными брендами.
+
+    :param brands: Список брендов.
+    :param df: DataFrame с колонками ['id', 'brand', 'name', ...].
+    :return: DataFrame с обновлённой колонкой 'brand'.
+    """
+    # Инициализируем автомат для быстрого поиска брендов
+    automaton = Automaton()
+
+    # Добавляем бренды в автомат
+    for idx, brand in enumerate(brands):
+        if isinstance(brand, str) and brand:
+            automaton.add_word(brand.lower(), (idx, brand))
+
+    automaton.make_automaton()
+
+    def find_brand(name):
+        """
+        Находит лучший бренд для данного имени.
+        """
+        matched_brands = set()
+        for _, (_, brand) in automaton.iter(name.lower()):
+            # Проверяем, что бренд встречается как отдельное слово
+            if re.search(rf'\b{re.escape(brand.lower())}\b', name.lower()):
+                matched_brands.add(brand)
+
+        # Возвращаем бренд с максимальной длиной (более точное совпадение)
+        return max(matched_brands, key=len) if matched_brands else None
+
+    # Обновляем колонку brand только для пустых значений
+    # df['new_brand'] = df.apply(
+    #     lambda row: find_brand(row['name']), #if pd.isna(row['brand']) else row['brand'],
+    #     axis=1
+    # )
+    if is_brand==True:
+        df[col_name] = df.apply(lambda row: find_brand(row['name']) or row['brand'], axis=1)
+    else:
+        df[col_name] = df.apply(lambda row: find_brand(row['name']) or None, axis=1)
+
+
+
+def get_same_brands(products, items):
+    comp_list=[]
+    #not_comp_prods=[]
+    #not_comp_items=[]
+    prod_brand_list=list(products['brand'].unique())
+    items_brand_list=list(items['new_brand'].unique())
+    for i in tqdm(prod_brand_list):
+        if i in items_brand_list:
+            comp_list.append(i)
+
+    return comp_list, prod_brand_list, items_brand_list
+
+
+def match_brands_improved(items_brands, prods_brands, threshold=85):
+    """
+    Улучшенный алгоритм сопоставления брендов с учётом нечёткого поиска и фильтрации ошибок.
+
+    :param items_brands: Список брендов из датафрейма items.
+    :param prods_brands: Список брендов из датафрейма prods.
+    :param threshold: Порог сходства для нечёткого поиска.
+    :return: Словарь соответствий {бренд из items: ближайший бренд из prods}.
+    """
+    brand_mapping = {}
+
+    for item_brand in tqdm(items_brands):
+        if isinstance(item_brand, str):
+            # Разделяем бренд на части
+            parts = [part.strip() for part in re.split(r"[\/\(\)]", item_brand) if part.strip()]
+            best_match = None
+            best_score = 0
+
+            for part in parts:
+                match, score, _ = process.extractOne(part, prods_brands, scorer=fuzz.ratio)
+                # Фильтрация по длине строк и порогу
+                if score >= threshold and abs(len(part) - len(match)) / len(part) <= 0.3:
+                    if score > best_score:
+                        best_match = match
+                        best_score = score
+
+            # Сохранение результата
+            if best_match:
+                brand_mapping[item_brand] = best_match#, best_score)
+
+    return brand_mapping
+
+
+
+

preprocess/utils/common/extracters.py

@@ -0,0 +1,66 @@
+import re
+
+
+def extract_years(text):
+    """
+    Извлекает сочетание числа и слова, указывающего возраст (например: '50 лет', '21 years').
+    """
+    # Регулярное выражение ищет числа и слова 'лет' или 'years' с учетом регистра
+    match = re.search(r'\b(?<!\d)(\d{1,2})\s*(лет|years)\b', text, re.IGNORECASE)
+    if match:
+        # Приводим слово 'лет' или 'years' к исходному регистру
+        return f"{match.group(1)} {match.group(2)}"
+    return None
+
+def extract_production_year(text):
+    """
+    Извлекает год производства (четырехзначное число в диапазоне 1900–2099) из строки.
+    Например: '2019'.
+    """
+    match = re.search(r'\b(19\d{2}|20\d{2})\b', text)
+    if match:
+        return match.group(1)
+    return None
+
+def extract_alcohol_content(text):
+    """
+    Извлекает содержание алкоголя из строки.
+    Например: '40%'.
+    """
+    match = re.search(r'(\d{1,2}(?:[.,]\d+)?\s*%)', text)
+    if match:
+        # Заменяем запятую на точку для единообразия (если нужно)
+        return match.group(1).replace(' ', '').replace(',', '.')
+    return None
+
+
+def is_volume(value):
+    """
+    Проверяет, является ли значение валидным объемом (<= 10 литров).
+    """
+    try:
+        volume = float(value)
+        return volume if volume <= 10 else None
+    except ValueError:
+        return None
+
+def extract_volume_or_number(text):
+    """
+    Извлекает объем в литрах или число с плавающей точкой из строки.
+    Например: '0,75л', '0.5', или '1,5 л'.
+    """
+    # Попытка найти объем с буквой 'л' или без пробела перед ней
+    match_with_l = re.search(r'(\d+(?:[\.,]\d+)?\s*[лЛ]|(?:\d+(?:[\.,]\d+)?[лЛ]))', text)
+    if match_with_l:
+        return is_volume(match_with_l.group(1).replace(',', '.').replace('л', '').replace('Л', '').strip())
+
+    # Если не найдено, ищем просто число с плавающей точкой
+    match_number = re.search(r'(?<!№)\b(\d{1,2}(?:[\.,]\d+))\b(?!\s*(№|-er|er|\d{3,}))', text)
+    if match_number:
+        return is_volume(match_number.group(1).replace(',', '.'))
+
+    return None
+
+
+
+

preprocess/utils/common/parallel_brand_mutching.py

@@ -0,0 +1,97 @@
+import re
+from ahocorasick import Automaton
+from rapidfuzz import fuzz, process
+from unidecode import unidecode
+from pqdm.threads import pqdm
+
+
+def normalize(text):
+    """
+    Приводит текст к нижнему регистру и транслитерирует его в латиницу.
+    """
+    return unidecode(text.lower())
+
+def build_regex_for_brands(brands):
+    """
+    Нормализует бренды и создаёт одно регулярное выражение для точного поиска.
+    Возвращает скомпилированный паттерн и словарь: нормализованное название -> оригинальное название.
+    """
+    norm_to_brand = {}
+    for brand in brands:
+        norm_brand = normalize(brand)
+        if norm_brand not in norm_to_brand:
+            norm_to_brand[norm_brand] = brand
+    pattern = re.compile(r'\b(?:' + '|'.join(re.escape(nb) for nb in norm_to_brand.keys()) + r')\b')
+    return pattern, norm_to_brand
+
+def process_string(s, regex_pattern, norm_to_brand, norm_brand_list, index_to_brand, threshold):
+    """
+    Обрабатывает одну строку:
+      1. Пытается найти бренд через регулярное выражение.
+      2. Если точного совпадения нет – разбивает строку и выполняет нечёткий поиск.
+    Возвращает кортеж: (исходная строка, найденный бренд или None).
+    """
+    norm_s = normalize(s)
+    # Пытаемся найти бренд через регулярное выражение
+    match = regex_pattern.search(norm_s)
+    if match:
+        return s, norm_to_brand[match.group(0)]
+
+    # Если точного совпадения нет, разбиваем строку по разделителям и анализируем части
+    parts = [part.strip() for part in re.split(r"[\/\(\)]", s) if part.strip()]
+    parts.append(s)  # анализ всей строки
+    best_match = None
+    best_score = 0
+    for part in parts:
+        norm_part = normalize(part)
+        res = process.extractOne(norm_part, norm_brand_list, scorer=fuzz.ratio, score_cutoff=threshold)
+        if res is not None:
+            match_norm, score, idx = res
+            if score > best_score:
+                best_match = index_to_brand[idx]
+                best_score = score
+                if best_score == 100:
+                    break
+    if best_match:
+        return s, best_match
+    return s, None
+
+def check_brands_in_strings_pqdm(strings, brands, threshold=85, n_jobs=8):
+    """
+    Поиск брендов в строках с учетом вариантов написания и транслитерации.
+    Использует предварительный поиск через регулярное выражение и, при необходимости,
+    нечёткий поиск. Обработка выполняется параллельно с отображением прогресса с помощью pqdm.
+
+    :param strings: Список строк для поиска брендов.
+    :param brands: Список брендов для поиска.
+    :param threshold: Порог сходства для нечёткого поиска.
+    :param n_jobs: Число рабочих потоков (или процессов, если использовать pqdm.processes).
+    :return: Словарь вида {строка: найденный бренд}.
+    """
+    # Подготавливаем список нормализованных брендов и сопоставление индексов с оригинальными брендами.
+    norm_brand_list = []
+    index_to_brand = []
+    for brand in brands:
+        norm_brand = normalize(brand)
+        norm_brand_list.append(norm_brand)
+        index_to_brand.append(brand)
+
+    # Создаем комбинированный паттерн для точного поиска.
+    regex_pattern, norm_to_brand = build_regex_for_brands(brands)
+
+    # Определяем вспомогательную функцию, закрывающую необходимые параметры.
+    def process_string_wrapper(s):
+        return process_string(s, regex_pattern, norm_to_brand, norm_brand_list, index_to_brand, threshold)
+
+    # Обрабатываем строки параллельно с отображением прогресса.
+    results = pqdm(strings, process_string_wrapper, n_jobs=n_jobs)
+
+    brand_mapping = {}
+    for s, matched_brand in results:
+        if matched_brand:
+            brand_mapping[s] = matched_brand
+    return brand_mapping
+
+
+
+

preprocess/utils/common/top_inserts.py

@@ -0,0 +1,66 @@
+from preprocess.utils.common.extracters import *
+from preprocess.utils.common.utils import *
+from preprocess.utils.common.parallel_brand_matching import *
+from tqdm import tqdm
+import re
+import math
+import numpy as np
+
+
+
+def top_inserts_matching(other_brands, p_brands, items, th=65):
+    replaced={}
+    for i in other_brands:
+        l=i.split('/')
+        if len(l)>2:
+            replaced[l[0].replace('Шато','')]=i
+        else:
+            if 'Шато' in i:
+                replaced[i.replace('Шато','')]=i
+
+    ob=[i.split('/')[0].replace('Шато','') for i in other_brands]
+    ob_in_pb=check_brands_in_strings_pqdm(ob, p_brands, threshold=th)
+
+    result={}
+    for k in ob_in_pb.keys():
+        if k in replaced.keys():
+            result[replaced[k]]=ob_in_pb[k]
+        else:
+            result[k]=ob_in_pb[k]
+
+    items.loc[items['new_name'].isin(result.keys()), 'new_brand'] = items['new_name'].map(result)
+
+
+def process_unbrended_names(items, p_brands, process_text, types, grape_varieties, onther_words):
+    result={}
+    for n in tqdm(items[items['new_brand'].isna()]['name'].values):
+
+        name, alcohol, volume_or_number, years, production_year, gb, color, sour=process_text(n)
+        #name, alcohol, volume_or_number, years, production_year, gb, color, sour=prcess_text('Вино Токай Фурминт п/сл. бел.0.75л')
+        name=trim_name(name, types)
+        name=trim_name(name, grape_varieties)
+        name=trim_name(name, onther_words)
+        name=name.replace('.','').replace(',','').replace('(','').replace(')','')
+        #result.append(clean_wine_name(name).strip())
+        result[n]=clean_wine_name(name).strip()
+
+
+
+    items['new_name']=None
+    items.loc[items['name'].isin(result.keys()), 'new_name'] = items['name'].map(result)
+
+    u_nn=list(items[~items['new_name'].isna()]['new_name'].unique())
+    res={}
+    for i in tqdm(u_nn):
+        lenta=len(items[items['new_name']==i])
+        if lenta>1:
+            res[i]=lenta
+
+    th=math.sqrt(((np.array(list(res.values())).mean()+np.array(list(res.values())).std())**2)//2)
+    other_brands=[i for i,j in res.items() if j>th]
+
+    reess=check_brands_in_strings_pqdm(other_brands, p_brands)
+
+    items.loc[items['new_name'].isin(reess.keys()), 'new_brand'] = items['new_name'].map(reess)
+
+    top_inserts_matching(other_brands, p_brands, items)

preprocess/utils/common/utils.py

@@ -0,0 +1,130 @@
+import re
+from tqdm import tqdm
+
+
+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:
+            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):
+    alco_types=[i.strip().lower() for i in products['type'].unique()]
+    alco_types.append('ликёр')
+    result=[]
+    for row in tqdm(items.iterrows()):
+        try:
+            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'})
+
+
+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)
+
+    # Заменяем найденные полные слова на пустую строку.
+    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()):
+        text, alcohol, volume_or_number, years, production_year, gb, color, sour=prcess_text(str(row['name']))
+        text=trim_name(text, types_and_others).replace(',','').replace('.','')
+        result[row['id']]=text.lower().strip() #remove_l(text).lower().strip()
+
+        gbs.append(gb)
+        sours.append(sour)
+    return result, gbs, sours

preprocess/utils/items/attrs.py

@@ -0,0 +1,40 @@
+def check_spark(row, col_name='name', types=['Игристое', 'игр']):
+    if col_name in row.keys():
+        for t in types:
+            if t.lower() in row[col_name].lower() and 'Пилигрим' not in row[col_name].lower():
+                return 'Игристое'
+        return None
+
+
+def check_color_and_sour(row, col_name='type_wine', types=['Белое', 'Розовое', 'Красное']):
+    if col_name in row.keys():
+        for t in types:
+            if t.lower() in row[col_name].lower():
+                return 'Вино'
+        return None
+
+
+def is_type_exist(row, types):
+    for t in types:
+        if t.lower() in row['type'].lower():  # Сравнение без учета регистра
+            return t
+    return None
+
+def check_type(row, types):
+    #checker=False
+    for t in types:
+        if t.lower() in row['name'].lower():  # Сравнение без учета регистра
+            return t
+    return None
+
+def get_type(row, types):
+    if 'type' not in row.keys():
+      return check_type(row, types)
+    elif 'type' in row.keys():
+      semi_res=is_type_exist(row, types)
+      if semi_res!=None:
+        return semi_res
+      else:
+        return check_type(row, types)
+    return None
+