extract_keywords.py

def flatten(items, seqtypes=(list, tuple)):
    try:
        for i, x in enumerate(items):
            while isinstance(x, seqtypes):    
                items[i:i+1] = x
                x = items[i]
    except IndexError:
        pass
    return items

aliases = [
  #('canonical name', ['aliases', ...])
  ('почта россия', ['почта', 'почта рф', 'пр', 'gh']),
  ('почта россия трекинг', ['пр трекинг', 'почта трекинг', 'пр трэкинг', 'почта трэкинг']),
  ('реестр почта', ['реестр пр', 'реестр почта россии']),
  ('реестр пэк', []),
  ('реквизиты', []),
  ('пешкарики', []),
  ('импорт лидов директ', []),
  ('яндекс доставка экспресс', ['яндекс доставка express', 'яд экспресс', 'ядоставка экспресс']),
  ('яндекс доставка ndd', ['яд ндд', 'я доставка ндд', 'ядоставка ндд', 'модуль ндд']),
  ('яндекс метрика', ['яндекс метрика импорт']),
  ('альфабанк', ['альфа банк', 'alfabank', 'альфа']),
  ('импорт лидов facebook', ['импорт лидов fb', 'загрузка лидов fb', 'лиды фейсбук', 'импорт лидов фб', 'fb lead']),
  ('маркетинговые расходы', ['расходы', 'загрузка расходов']),
  ('cloudpayments', ['клауд', 'клаудпеймент', 'клаудпейментс']),
  ('robokassa', ['робокасса', 'робокаса']),
  ('sipuni', ['сипуни', 'сипьюни']),
  ('mailchimp', ['майлчимп', 'мейлчим', 'мейлчимп']),
  ('unisender', ['юнисендер']),
  ('яндекс аудитории', ['экспорт аудитории', 'экспорт яндекс аудитории']),
  ('экспорт facebook', ['экспорт сегментов facebook', 'экспорт fb', 'экспорт фейсбук', 'экспорт аудиторий фб', 'fb экспорт']),
  ('экспорт вк', ['экспорт сегментов vkontakte', 'экспорт vk', 'экспорт контакте'])  
]

vocab_raw = flatten([[k] + keywords for k, keywords in aliases])

import string
import pymorphy3

morph = None
def normalize_word(word):
    if word == 'лид':
        return word
    global morph
    if morph is None:
        morph = pymorphy3.MorphAnalyzer()
    return morph.parse(word)[0].normal_form

def tokenize_sentence(text):
    # remove punctuation
    text = text.translate(str.maketrans(string.punctuation, ' ' * len(string.punctuation)))
    # tokenize
    return [normalize_word(word) for word in text.split()]
  
def normalize_sentence(text):
    return " ".join(tokenize_sentence(text))

def canonical_keywords(keywords):
    """
    replace keyword aliases with canonical keyword names
    """
    result = []
    for k in keywords:
        k = normalize_sentence(k)
        for canonical_name, alias_names in aliases:
            canonical_name = normalize_sentence(canonical_name)
            for a in alias_names:
                a = normalize_sentence(a)
                #print('a', a)
                if a == k:
                    result.append(canonical_name)
                    break
            else:
                continue
            break
        else:
            result.append(k)
    return result

def merge_keywords(keywords):
    """
    remove subkeywords
    """
    result = []
    sorted_keywords = sorted(keywords, key=len, reverse=True)
    
    for k in sorted_keywords:
        for rk in result:
            if rk.lower().startswith(k):
                break
        else:
            result.append(k)
            continue

    return result


vectorizer = None
kw_model = None

def init_keyword_extractor():
    global vectorizer
    global kw_model

    from keybert import KeyBERT
    import spacy
    from sklearn.feature_extraction.text import CountVectorizer

    kw_model = KeyBERT(model=spacy.load("ru_core_news_sm", exclude=['tokenizer', 'tagger', 'parser', 'ner', 'attribute_ruler', 'lemmatizer']))
    vocab = [" ".join(tokenize_sentence(s)) for s in vocab_raw]
    vectorizer = CountVectorizer(ngram_range=(1, 4), vocabulary=vocab, tokenizer=tokenize_sentence)

def extract_keywords(text):
    global vectorizer
    global kw_model

    if vectorizer is None or kw_model is None:
        init_keyword_extractor()

    keywords = [k for k, score in kw_model.extract_keywords(text, vectorizer=vectorizer)]
    return merge_keywords(canonical_keywords(keywords))