app.py

import os
import subprocess
import gradio as gr
import wget
from ftlangdetect import detect
from cleantext import clean
from keybert import KeyBERT
from keyphrase_vectorizers import KeyphraseCountVectorizer
from sklearn.feature_extraction.text import CountVectorizer
from functools import partial
from sentence_transformers import SentenceTransformer


## models sentence-bert multilingual

# fonte SBERT: https://www.sbert.net/docs/pretrained_models.html#multi-lingual-models
# models na Hugging Face model hub (https://huggingface.co/sentence-transformers/...)
# old: paraphrase-multilingual-MiniLM-L12-v2
model_id = ["paraphrase-multilingual-mpnet-base-v2", "sentence-transformers/LaBSE", "distiluse-base-multilingual-cased-v1"]
model_name = ["SBERT multilingual", "LaBSE", "DistilBERT mltilingual (v1)"]


## get KeyBERT model

kw_model_0 = KeyBERT(model=model_id[0])
#kw_model_1 = KeyBERT(model=model_id[1])
#kw_model_2 = KeyBERT(model=model_id[2])
kw_model = {
    0: kw_model_0,
    #1: kw_model_1,
    #2: kw_model_2
    }


## max_seq_length

# get max_seq_length of the KeyBERT model
#if isinstance(kw_model_0.model.embedding_model, SentenceTransformer):
#  max_seq_length_0 = kw_model_0.model.embedding_model.max_seq_length
# change max_seq_length
#kw_model_0.model.embedding_model.max_seq_length = 512
#num_tokens = kw_model_0.model.embedding_model.tokenize([doc_original])['input_ids'].shape[1]


## spacy (pipeline)

import spacy
# Portuguese pipeline optimized for CPU. Components: tok2vec, morphologizer, parser, lemmatizer (trainable_lemmatizer), senter, ner, attribute_ruler.
spacy_pipeline = "pt_core_news_lg"
# download spacy pipeline (https://spacy.io/models/pt)
os.system(f"python -m spacy download {spacy_pipeline}")

# Load tokenizer, tagger, parser, NER and word vectors
#os.system("python -m spacy download pt_core_news_lg")
nlp = spacy.load(spacy_pipeline)

# Add the component to the pipeline
# "nlp" Object is used to create documents with linguistic annotations.
nlp.add_pipe('sentencizer')


## download stop words in Portuguese
output = subprocess.run(["python", "stopwords.py"], capture_output=True, text=True)
stop_words = list(eval(output.stdout))


## Part-of-Speech Tagging for Portuguese 
# (https://melaniewalsh.github.io/Intro-Cultural-Analytics/05-Text-Analysis/Multilingual/Portuguese/03-POS-Keywords-Portuguese.html)
#pos_pattern = '<NUM.*>*<NOUN.*>*<ADJ.*>*<ADP.*>*<NOUN.*>*<NUM.*>*<NOUN.*>*<ADJ.*>*'
pos_pattern = '<PROPN.*>*<N.*>*<ADJ.*>*'


# vectorizer options
vectorizer_options = ["keyword", "3gramword", "nounfrase"]


# function principal (keywords)
def get_kw_html(doc, top_n, diversity, vectorizer_option, model_id, pos_pattern):

  # lowercase
  lowercase = False

  ## define o vectorizer
  def get_vectorizer(vectorizer_option):
      
    # one word 
    if vectorizer_option == "keyword":    
      vectorizer = CountVectorizer(
          ngram_range=(1, 1),
          stop_words=stop_words, 
          lowercase=lowercase
          )
      
    # upt to 3-gram 
    elif vectorizer_option == "3gramword":    
      vectorizer = CountVectorizer(
          ngram_range=(1, 3),
          #stop_words=stop_words, 
          lowercase=lowercase
          )
      
    # proper noun / noun (adjective) phrase
    elif vectorizer_option == "nounfrase":
      vectorizer = KeyphraseCountVectorizer(
          spacy_pipeline=spacy_pipeline, 
          #stop_words=stop_words, 
          pos_pattern=pos_pattern, 
          lowercase=lowercase
          )   
      
    return vectorizer

  # function to clean text of document
  def get_lang(doc):
    doc = clean(doc,
                fix_unicode=True,               # fix various unicode errors
                to_ascii=False,                  # transliterate to closest ASCII representation
                lower=True,                    # lowercase text
                no_line_breaks=True,           # fully strip line breaks as opposed to only normalizing them
                no_urls=True,                  # replace all URLs with a special token
                no_emails=False,                # replace all email addresses with a special token
                no_phone_numbers=False,         # replace all phone numbers with a special token
                no_numbers=False,               # replace all numbers with a special token
                no_digits=False,                # replace all digits with a special token
                no_currency_symbols=False,      # replace all currency symbols with a special token
                no_punct=False,                 # remove punctuations
                replace_with_punct="",          # instead of removing punctuations you may replace them
                replace_with_url="<URL>",
                replace_with_email="<EMAIL>",
                replace_with_phone_number="<PHONE>",
                replace_with_number="<NUMBER>",
                replace_with_digit="0",
                replace_with_currency_symbol="<CUR>",
                lang="pt"                       # set to 'de' for German special handling
                )
    res = detect(text=str(doc), low_memory=False)
    lang = res["lang"]
    score = res["score"]
  
    return lang, score

  def get_passages(doc):
    # method: https://github.com/UKPLab/sentence-transformers/blob/b86eec31cf0a102ad786ba1ff31bfeb4998d3ca5/examples/applications/retrieve_rerank/in_document_search_crossencoder.py#L19
    doc = doc.replace("\r\n", "\n").replace("\n", " ")
    doc = nlp(doc)

    paragraphs = []
    for sent in doc.sents:
      if len(sent.text.strip()) > 0:
        paragraphs.append(sent.text.strip())

    window_size = 2
    passages = []
    paragraphs = [paragraphs]
    for paragraph in paragraphs:
        for start_idx in range(0, len(paragraph), window_size):
            end_idx = min(start_idx+window_size, len(paragraph))
            passages.append(" ".join(paragraph[start_idx:end_idx]))

    return passages

  # keywords
  def get_kw(doc, kw_model=kw_model[model_id], top_n=top_n, diversity=diversity, vectorizer=get_vectorizer(vectorizer_option)):
    
    keywords = kw_model.extract_keywords(
        doc, 
        vectorizer = vectorizer,
        use_mmr = True,
        diversity = diversity,
        top_n = top_n,
        )
    
    return keywords

  def get_embeddings(doc, candidates, kw_model=kw_model[model_id]):

    doc_embeddings = kw_model.model.embed([doc])
    word_embeddings = kw_model.model.embed(candidates)
  
    # doc_embeddings, word_embeddings = kw_model.extract_embeddings(docs=doc, candidates = candidates,
    #                                                               keyphrase_ngram_range = (1, 100),
    #                                                               stop_words = None,
    #                                                               min_df = 1,
    #                                                               )

    return doc_embeddings, word_embeddings

  # highlight
  def get_html(keywords, doc=doc):

    # ordering of lists (from longest keywords to shortest ones)
    list3 = [keyword[0] for keyword in keywords]
    list2 = [len(item.split()) for item in list3]
    list1 = list(range(len(list2)))
    list2, list1 = (list(t) for t in zip(*sorted(zip(list2, list1))))
    list1 = list1[::-1]
    keywords_list = [list3[idx] for idx in list1]

    # converting doc to html format
    html_doc = doc
    for idx,keyword in enumerate(keywords_list):
      if sum([True if keyword in item else False for item in keywords_list[:idx]]) == 0:
        if keyword not in '<span style="color: black; background-color: yellow; padding:2px">' and keyword not in '</span>':
          html_doc = html_doc.replace(keyword, '<span style="color: black; background-color: yellow; padding:2px">' + keyword + '</span>')
    html_doc = '<p style="font-size:120%; line-height:120%">' + html_doc + '</p>'

    return html_doc

  # if isinstance(kw_model_0.model.embedding_model, SentenceTransformer):
  #   num_tokens = kw_model_0.model.embedding_model.tokenize([doc])['input_ids'].shape[1]


  ## main

  # empty doc
  if len(doc) == 0:

    # get keywords and highlighted text
    keywords, keywords_list_json = [("",0.)], {"":0.}
    html_doc = '<p style="font-size:150%; line-height:120%"></p>'
    label = "O texto do documento não pode estar vazio. Recomece, por favor."

  else:
    
    # detect lang
    lang, score = get_lang(doc)

    # error in lang detect
    if lang!="pt" or score<0.9:

      # get keywords and highlighted text
      keywords, keywords_list_json = [("",0.)], {"":0.}
      html_doc = '<p style="font-size:150%; line-height:120%"></p>'
      label = "O APP não tem certeza de que o texto do documento está em português. Recomece com um texto em português, por favor."
    
    # text not empty and in the correct language
    else:

      # get passages
      passages= get_passages(doc)
      num_passages = len(passages)

      # parameters
      candidates_list = list()
      passages_embeddings = dict()
      candidates_embeddings_list = list()

      # get keywords, candidates and their embeddings
      for i,passage in enumerate(passages):
        keywords = get_kw(passage)

        candidates = [keyword for keyword,prob in keywords]
        candidates_list.extend(candidates)

        passages_embeddings[i], candidates_embeddings = get_embeddings(passage, candidates)
        candidates_embeddings_list.extend(candidates_embeddings)

      if len(candidates_list) > 0:

        # get unique candidates 
        candidates_unique_list = list(set(candidates_list))
        candidates_embeddings_unique_list = [candidates_embeddings_list[candidates_list.index(candidate)] for candidate in candidates_unique_list]
        num_candidates_unique = len(candidates_unique_list)
        
        # get distances between the candidates and respectively all the passages
        # Maximal Marginal Relevance (MMR)
        from keybert._mmr import mmr
        from keybert._maxsum import max_sum_distance
        keywords_list = list()
        for i in range(num_passages):
          keywords_list.append(mmr(passages_embeddings[i],
                                  candidates_embeddings_unique_list,
                                  candidates_unique_list,
                                  num_candidates_unique,
                                  diversity = 0)
          )

        # get the average distances between the candidates and the passages (1 distance by candidate)
        keywords_with_distance_list = dict()
        for i in range(num_passages):
          for keyword, prob in keywords_list[i]:
            if i == 0: keywords_with_distance_list[keyword] = prob
            else: keywords_with_distance_list[keyword] += prob

        # get top_n keywords with prob
        keywords_list_sorted = {k: v for k, v in sorted(keywords_with_distance_list.items(), key=lambda item: item[1], reverse=True)}
        keywords_with_distance_list_sorted = [(keyword, round(keywords_with_distance_list[keyword]/num_passages, 4)) for keyword in keywords_list_sorted]
        keywords_with_distance_list_sorted = keywords_with_distance_list_sorted[:top_n]
      
        # main keyword
        label = f"A palavra/frase chave com a maior similaridade é <span style='color: black; background-color: yellow; padding:2px'>{keywords_with_distance_list_sorted[0][0]}</span>."

        # json for printing
        keywords_list_json = {keyword:prob for keyword, prob in keywords_with_distance_list_sorted}

        # get html doc
        html_doc = get_html(keywords_with_distance_list_sorted)

      else:

        label, keywords_list_json, html_doc = "O APP não encontrou de palavras/frases chave no texto.", {"":0.}, ""

  return label, keywords_list_json, html_doc

def get_kw_html_0(doc, top_n, diversity, vectorizer_option, model_id=0, pos_pattern=pos_pattern):
  return get_kw_html(doc, top_n, diversity, vectorizer_option, model_id, pos_pattern)


title = "Extração das palavras/frases chave em português"

description = '<p>(17/12/2022) Forneça seu próprio texto em português e o APP vai fazer a extração das palavras/frases chave com as maiores similaridades ao texto.</p>\
<p>Este aplicativo usa os modelos seguintes:\
<br />- <a href="https://huggingface.co/sentence-transformers/paraphrase-multilingual-mpnet-base-v2">SBERT multilingual</a>,\
<br />- <a href="https://maartengr.github.io/KeyBERT/index.html">KeyBERT</a> para calcular as similaridades entre as palavras/frases chave e o texto do documento.</p>'

# examples
doc_original_0 = """
As contas de pelo menos seis jornalistas norte-americanos que cobrem tecnologia foram suspensas pelo Twitter na noite desta quinta-feira (15). Os profissionais escrevem sobre o tema para diversos veículos de comunicação dos Estados Unidos, como os jornais 'The New York Times' e 'Washington Post'.

A rede social afirmou apenas que suspende contas que violam as regras, mas não deu mais detalhes sobre os bloqueios.

Assim que comprou o Twitter, Elon Musk disse defender a liberdade de expressão, e reativou, inclusive, a conta do ex-presidente Donald Trump, suspensa desde o ataque ao Capitólio, em 2021.

Os jornalistas que tiveram as contas bloqueadas questionaram o compromisso de Musk com a liberdade de expressão.

Eles encararam o bloqueio como uma retaliação de Musk às críticas que o bilionário vem recebendo pela forma como está conduzindo a rede social: com demissões em massa e o desmonte de áreas, como o conselho de confiança e segurança da empresa.

Metade dos funcionários do Twitter foram demitidos desde que ele assumiu o comando da empresa e outros mil pediram demissão.
      """

doc_original_1 = """
O bilionário Elon Musk restabeleceu neste sábado (17) as contas suspensas de jornalistas no Twitter. A súbita suspensão, um dia antes, provocou reações de entidades da sociedade e setores políticos, além de ameaças de sanção por parte da União Europeia.

O empresário, que comprou a rede social em outubro, acusou os repórteres de compartilhar informações privadas sobre seu paradeiro, sem apresentar provas.

Ainda na sexta (16), o empresário publicou uma enquete na rede social perguntando se as contas deveriam ser reativadas "agora" ou "em sete dias": 58,7% votaram pela retomada imediata; 41,3%, em sete dias.

"O povo falou. Contas envolvidas em doxing (revelação intencional e pública de informações pessoais sem autorização) com minha localização terão sua suspensão suspensa agora", tuitou o empresário neste sábado.

O g1 verificou a conta de alguns dos jornalistas suspensos, que pertencem a funcionários de veículos como a CNN, o The New York Times, o The Washington Post, e as páginas estavam ativas.

As exceções, até a última atualização desta reportagem, eram a conta @ElonJet, que rastreava o paradeiro do jato do próprio Elon Musk, e o perfil do criador, Jack Sweeney.

Ao acessar ambas as páginas, é possível visualizar a seguinte mensagem: "O Twitter suspende as contas que violam as Regras do Twitter".

A plataforma também suspendeu a conta da rede social Mastodon, concorrente do Twitter.

O Twitter Spaces também foi tirado do ar na sexta, após Musk ter sido questionado ao vivo sobre essas últimas decisões, informou a agência de notícias Bloomberg – o bilionário disse que o recurso voltou ao ar na tarde do mesmo dia.
      """

# parameters
num_results = 5
diversity = 0.3

examples = [
    [doc_original_0.strip(), num_results, diversity, vectorizer_options[0]],
    [doc_original_1.strip(), num_results, diversity, vectorizer_options[0]],
    #[doc_original_2.strip(), num_results, diversity, vectorizer_options[0]],
]

# parameters
num_results = 5
diversity = 0.3

# interfaces
interface_0 = gr.Interface(
    fn=get_kw_html_0,
    inputs=[
        gr.Textbox(lines=15, label="Texto do documento"), 
        gr.Slider(1, 20, value=num_results, step=1., label=f"Número das palavras/frases chave a procurar (0: mínimo - 20: máximo - padrão: {num_results})"),
        gr.Slider(0, 1, value=diversity, step=0.1, label=f"Diversidade entre as palavras/frases chave encontradas (0: mínimo - 1: máximo - padrão: {diversity})"),
        gr.Radio(choices=vectorizer_options, value=vectorizer_options[0], label=f"Tipo de resultados (keyword: lista de palavras únicas - 3gramword: lista de 1 a 3 palavras - nounfrase: lista de frases nominais)"),
        ],
    outputs=[
        gr.HTML(label=f"{model_name[0]}"),
        gr.Label(show_label=False),
        gr.HTML(),
        ]
)

# app
demo = gr.Parallel(
    interface_0,
    title=title,
    description=description,
    examples=examples,
    allow_flagging="never"
    )

if __name__ == "__main__":
    demo.launch()