app.py

import fasttext
import gradio as gr
import joblib
import json as js
import omikuji
import os
import re
from collections import defaultdict
from huggingface_hub import snapshot_download
from typing import List, Tuple, Dict

from install_packages import download_model

download_model('https://dl.fbaipublicfiles.com/fasttext/supervised-models/lid.176.bin', 'lid.176.bin')

# Download the model files from Hugging Face
for repo_id in ['kapllan/omikuji-bonsai-parliament-de-spacy', 'kapllan/omikuji-bonsai-parliament-fr-spacy',
                'kapllan/omikuji-bonsai-parliament-it-spacy']:
    if not os.path.exists(repo_id):
        os.makedirs(repo_id)
    model_dir = snapshot_download(repo_id=repo_id, local_dir=repo_id)

lang_model = fasttext.load_model('lid.176.bin')

with open('./id2label.json', 'r') as f:
    id2label = js.load(f)

with open('topics_hierarchy.json', 'r') as f:
    topics_hierarchy = js.load(f)


def map_language(language: str) -> str:
    language_mapping = {'de': 'German',
                        'it': 'Italian',
                        'fr': 'French'}
    if language in language_mapping.keys():
        return language_mapping[language]
    else:
        return language


def find_model(language: str):
    vectorizer, model = None, None
    if language in ['de', 'fr', 'it']:
        path_to_vectorizer = f'./kapllan/omikuji-bonsai-parliament-{language}-spacy/vectorizer'
        path_to_model = f'./kapllan/omikuji-bonsai-parliament-{language}-spacy/omikuji-model'
        vectorizer = joblib.load(path_to_vectorizer)
        model = omikuji.Model.load(path_to_model)
    return vectorizer, model


def predict_lang(text: str) -> str:
    text = re.sub(r'\n', '', text)  # Remove linebreaks because fasttext cannot process that otherwise
    predictions = lang_model.predict(text, k=1)  # returns top 2 matching languages
    language = predictions[0][0]  # returns top 2 matching languages
    language = re.sub(r'__label__', '', language)  # returns top 2 matching languages
    return language


def predict_topic(text: str) -> [List[str], str]:
    results = []
    language = predict_lang(text)
    vectorizer, model = find_model(language)
    language = map_language(language)
    if vectorizer is not None:
        texts = [text]
        vector = vectorizer.transform(texts)
        for row in vector:
            if row.nnz == 0:  # All zero vector, empty result
                continue
            feature_values = [(col, row[0, col]) for col in row.nonzero()[1]]
            for subj_id, score in model.predict(feature_values, top_k=1000):
                results.append((id2label[str(subj_id)], round(score,2)))
    return results, language


def get_row_color(type: str):
    if 'main' in type.lower():
        return 'background-color: darkgrey;'
    if 'sub' in type.lower():
        return 'background-color: lightgrey;'


def generate_html_table(topics: List[Tuple[str, str, float]]):
    html = '<table style="width:100%; border: 1px solid black; border-collapse: collapse;">'

    html += '<tr><th>Type</th><th>Topic</th><th>Score</th></tr>'
    for type, topic, score in topics:
        color = get_row_color(type)
        topic = f"<strong>{topic}</strong>" if 'main' in type.lower() else topic
        type = f"<strong>{type}</strong>" if 'main' in type.lower() else type
        score = f"<strong>{score}</strong>" if 'main' in type.lower() else score
        html += f'<tr style="{color}"><td>{type}</td><td>{topic}</td><td>{score}</td></tr>'
    html += '</table>'
    return html


def restructure_topics(topics: List[Tuple[str, float]]) -> List[Tuple[str, str, float]]:
    topics = [(str(x[0]).lower(), x[1]) for x in topics]
    topics_as_dict = {}
    for predicted_topic, score in topics:
        if str(predicted_topic).lower() in topics_hierarchy.keys():
            topics_as_dict[str(predicted_topic).lower()] = []

    for predicted_topic, score in topics:
        for main_topic, sub_topics in topics_hierarchy.items():
            if main_topic in topics_as_dict.keys() and predicted_topic != main_topic and predicted_topic in sub_topics:
                topics_as_dict[main_topic].append(predicted_topic)

    topics_restructured = []
    for predicted_main_topic, predicted_sub_topics in topics_as_dict.items():
        if len(predicted_sub_topics) > 0:
            score = [t for t in topics if t[0] == predicted_main_topic][0][1]
            topics_restructured.append(
                ('Main Topic', predicted_main_topic, score))
            predicted_sub_topics_with_scores = []
            for pst in predicted_sub_topics:
                score = [t for t in topics if t[0] == pst][0][1]
                entry = ('Sub Topic', pst, score)
                if entry not in predicted_sub_topics_with_scores:
                    predicted_sub_topics_with_scores.append(entry)
            for x in predicted_sub_topics_with_scores:
                topics_restructured.append(x)
    return topics_restructured


def topic_modeling(text: str, threshold: float) -> [List[str], str]:
    # Prepare labels and scores for the plot
    sorted_topics, language = predict_topic(text)
    if len(sorted_topics) > 0 and language in ['German', 'French', 'Italian']:
        sorted_topics = [t for t in sorted_topics if t[1] >= threshold]
    else:
        sorted_topics = []
    sorted_topics = restructure_topics(sorted_topics)
    sorted_topics = generate_html_table(sorted_topics)
    return sorted_topics, language


with gr.Blocks() as iface:
    gr.Markdown("# Topic Modeling")
    gr.Markdown("Enter a document and get each topic along with its score.")

    with gr.Row():
        with gr.Column():
            input_text = gr.Textbox(lines=10, placeholder="Enter a document")
            submit_button = gr.Button("Submit")
            threshold_slider = gr.Slider(minimum=0.0, maximum=1.0, step=0.01, label="Score Threshold",
                                         value=0.0)
            language_text = gr.Textbox(lines=1, placeholder="Detected language will be shown here...",
                                       interactive=False, label="Detected Language")
        with gr.Column():
            output_data = gr.HTML()

    submit_button.click(topic_modeling, inputs=[input_text, threshold_slider],
                        outputs=[output_data, language_text])

    # Launch the app
iface.launch(share=True)