app.py

import os
import gradio as gr
from transformers import pipeline
import spacy
import lib.read_pdf
import pandas as pd
import re
import matplotlib.pyplot as plt
import matplotlib.patches as patches
import io
# Initialize spaCy model
nlp = spacy.load('en_core_web_sm')
nlp.add_pipe('sentencizer')

def split_in_sentences(text):
    doc = nlp(text)
    return [str(sent).strip() for sent in doc.sents]

def make_spans(text, results):
    results_list = [res['label'] for res in results]
    facts_spans = list(zip(split_in_sentences(text), results_list))
    return facts_spans

# Initialize pipelines
summarizer = pipeline("summarization", model="human-centered-summarization/financial-summarization-pegasus")
fin_model = pipeline("sentiment-analysis", model='yiyanghkust/finbert-tone', tokenizer='yiyanghkust/finbert-tone')
fin_model_bis = pipeline("sentiment-analysis", model='ProsusAI/finbert', tokenizer='ProsusAI/finbert')
table_to_text = pipeline('text2text-generation', model='google/flan-t5-large')

def summarize_text(text):
    resp = summarizer(text)
    return resp[0]['summary_text']

def text_to_sentiment(text):
    sentiment = fin_model(text)[0]["label"]
    return sentiment

def fin_ext(text):
    results = fin_model(split_in_sentences(text))
    return make_spans(text, results)
def fin_ext_bis(text):
    results = fin_model_bis(split_in_sentences(text))
    return make_spans(text, results)

def extract_and_paragraph(pdf1, pdf2, paragraph):
    if not pdf1 or not pdf2:
        return [], []

    pdf1_path = os.path.join(PDF_FOLDER, pdf1)
    pdf2_path = os.path.join(PDF_FOLDER, pdf2)

    # Extract and format paragraphs
    paragraphs_1 = lib.read_pdf.extract_and_format_paragraphs(pdf1_path)
    paragraphs_2 = lib.read_pdf.extract_and_format_paragraphs(pdf2_path)

    start_keyword = "Main risks to"
    end_keywords = ["4. Appendix", "Annex:", "4. Annex", "Detailed tables", "ACKNOWLEDGEMENTS", "STATISTICAL ANNEX", "PROSPECTS BY MEMBER STATES"]

    start_index1, end_index1 = lib.read_pdf.find_text_range(paragraphs_1, start_keyword, end_keywords)
    start_index2, end_index2 = lib.read_pdf.find_text_range(paragraphs_2, start_keyword, end_keywords)
    paragraphs_1 = lib.read_pdf.extract_relevant_text(paragraphs_1, start_index1, end_index1)
    paragraphs_2 = lib.read_pdf.extract_relevant_text(paragraphs_2, start_index2, end_index2)
    if paragraph:
        paragraphs_1 = lib.read_pdf.split_text_into_paragraphs(paragraphs_1, 0)
        paragraphs_2 = lib.read_pdf.split_text_into_paragraphs(paragraphs_2, 0)
    
    return paragraphs_1, paragraphs_2

# Gradio interface setup
PDF_FOLDER = "data"

def get_pdf_files(folder):
    return [f for f in os.listdir(folder) if f.endswith('.pdf')]

def show(name):
    return f"{name}"

def get_excel_files(folder):
    return [f for f in os.listdir(folder) if f.endswith('.xlsx')]

def get_sheet_names(file):
    xls = pd.ExcelFile(os.path.join(PDF_FOLDER, file))
    return gr.update(choices=xls.sheet_names)

    
def process_and_compare(file1, sheet1, file2, sheet2):
    def process_file(file_path, sheet_name):
        # Extract year from file name
        year = int(re.search(r'(\d{4})', file_path).group(1))
        
        # Load the Excel file
        df = pd.read_excel(os.path.join(PDF_FOLDER, file_path), sheet_name=sheet_name, index_col=0)
        
        # Define expected columns based on extracted year
        historical_col = f'Historical {year - 1}'
        baseline_cols = [f'Baseline {year}', f'Baseline {year + 1}', f'Baseline {year + 2}']
        adverse_cols = [f'Adverse {year}', f'Adverse {year + 1}', f'Adverse {year + 2}']
        level_deviation_col = f'Level Deviation {year + 2}'

        # Drop rows and reset index
        df = df.iloc[4:].reset_index(drop=True)

        # Define the new column names
        new_columns = ['Country', 'Code', historical_col] + baseline_cols + adverse_cols + ['Adverse Cumulative', 'Adverse Minimum', level_deviation_col]
        
        # Ensure the number of columns matches
        if len(df.columns) == len(new_columns):
            df.columns = new_columns
        else:
            raise ValueError(f"Expected {len(new_columns)} columns, but found {len(df.columns)} columns in the data.")
        columns = ['Country', f'Adverse {year}', f'Adverse {year+1}', f'Adverse {year+2}', 'Adverse Cumulative']
        return df, df[columns]

    # Process both files
    global stored_df1, stored_df2
    df1, stored_df1 = process_file(file1, sheet1)
    df2, stored_df2 = process_file(file2, sheet2)
    year1 = int(re.search(r'(\d{4})', file1).group(1))
    year2 = int(re.search(r'(\d{4})', file2).group(1))

    # Merge dataframes on 'Country'
    merged_df = pd.merge(df2, df1, on='Country', suffixes=(f'_{year1}', f'_{year2}'))
    merged_df['Difference adverse cumulative growth'] = merged_df[f'Adverse Cumulative_{year2}'] - merged_df[f'Adverse Cumulative_{year1}']
    # Ensure data types are correct
    merged_df['Country'] = merged_df['Country'].astype(str)
    merged_df['Difference adverse cumulative growth'] = pd.to_numeric(merged_df['Difference adverse cumulative growth'], errors='coerce')

    # Create histogram plot with color coding
    fig, ax = plt.subplots(figsize=(12, 8))
    colors = plt.get_cmap('tab20').colors  # Use a colormap with multiple colors
    num_countries = len(merged_df['Country'])
    
    bars = ax.bar(merged_df['Country'], merged_df['Difference adverse cumulative growth'], color=colors[:num_countries])
    
    # Add a legend
    handles = [patches.Patch(color=color, label=country) for color, country in zip(colors[:num_countries], merged_df['Country'])]
    ax.legend(handles=handles, title='Countries', bbox_to_anchor=(1.05, 1), loc='upper left')
    
    ax.set_title(f'Histogram of Difference between Adverse cumulative growth of {year2} and {year1} for {sheet1}')
    ax.set_xlabel('Country')
    ax.set_ylabel('Difference')
    plt.xticks(rotation=90)

    # Save plot to a file
    file_path = 'output/plot.png'
    plt.savefig(file_path, format='png', bbox_inches='tight')
    plt.close()

    return file_path, gr.update(choices=stored_df1.Country.values.tolist()), gr.update(choices=stored_df2.Country.values.tolist())

def find_sentences_with_keywords(text, keywords):
    # Split text into sentences using regular expression to match sentence-ending punctuation
    sentences = re.split(r'(?<=[.!?])\s+', text)
    
    matched_sentences = set()  # Use a set to store unique sentences
    
    # For each keyword, find sentences that contain the keyword as a whole word
    for keyword in keywords:
        keyword_pattern = re.compile(rf'\b{re.escape(keyword)}\b', re.IGNORECASE)  # Using word boundaries

        for sentence in sentences:
            if keyword_pattern.search(sentence):
                matched_sentences.add(sentence)  # Add to set to ensure uniqueness

    return list(matched_sentences)  # Convert set back to list for consistent output


# Main function to process both PDFs based on the Excel file names and the sheet name
def process_pdfs_and_analyze_sentiment(file1, file2, sheet):
    # Extract text from both PDFs based on the file name
    pdf_file1 = file1.replace(".xlsx", ".pdf")
    pdf_file2 = file2.replace(".xlsx", ".pdf")
    text1, text2 =extract_and_paragraph(pdf_file1, pdf_file2, False)
    # Use sheet name as the keyword to find relevant sentences
    keywords = {
        'GDP': ['GDP'], 
        'HICP': ['HICP'], 
        'RRE prices': ['RRE', 'residential'], 
        'CRE prices': ['CRE', 'commercial'], 
        'Unemployment': ['unemployment']
    }
    selected_keywords = keywords.get(sheet, [])

    # Find sentences containing the keywords
    sentences1 = find_sentences_with_keywords(text1, selected_keywords)
    sentences2 = find_sentences_with_keywords(text2, selected_keywords)

    # Concatenate all sentences for each PDF
    text_pdf1 = "\n".join(sentences1)
    text_pdf2 = "\n".join(sentences2)
    
    # Perform sentiment analysis on the extracted sentences for each PDF
    result_pdf1 = fin_ext_bis(text_pdf1)
    result_pdf2 = fin_ext_bis(text_pdf2)
    
    return result_pdf1, result_pdf2
#def change_choices(df):
#    return gr.update(choices=df.Country.values.tolist())
    
def generate_text(df, country, theme):
    # Filter the dataframe based on the country
    #for column in df.columns:
    #    if column != 'Country':
    #        df[column] = df[column].apply(lambda x: f"{x:.6f}%")
    #row = df[df['Country'] == country].iloc[0]
    def format_row_for_prompt(row):
        # Create a formatted string with colons and percentages
        formatted_row = []
        for col, value in row.items():
            if col != 'Country':  # Exclude 'Country' or format differently if needed
                if isinstance(value, (int, float)):  # Add percentage sign for numeric values
                    value_str = f"{value:.6f}%"
                else:
                    value_str = str(value)
                formatted_row.append(f"{col}: {value_str}")
            else:
                formatted_row.append(f"{col}: {value}")
        return "\n".join(formatted_row)
    # Convert the row to a string format for prompt
    row = df[df['Country'] == country].iloc[0]
    row_str = format_row_for_prompt(row)
    #row_str = row.to_string(index=True)
    print(row_str)
    simple_prompt = f"""
    Here is the data for {theme} in {country}:
    {row_str}
    
    Summarize the adverse growth for {theme} in {country}. Highlight any increase or decrease compared to previous years and include the cumulative result.
    """

    prompt = f"""
    Here is an example of how to describe adverse growth data for a given country:
    Country: Australia
    Adverse 1990: -0.43%
    Adverse 1991: -1.99%
    Adverse 1192: -1.20%
    Adverse Cumulative: -3.57%
    Topic: GDP

    Description:
    In the adverse scenario, the GDP growth in Australia was -0.43% in 1990. It decreased further to -1.99% in 1991, showing worsening conditions. There was a slight improvement to -1.20% in 1992. The total cumulative adverse growth is -3.57%.
    
    Now, using the following data for {theme} in {country}, describe the adverse growth:
    {row_str}
    Topic: {theme}
    
    Describe, using the similar pattern from the example, the changes for the provided country and years. Highlight how the values change year by year and whether they increased or decreased. Do not mention any other countries or years, and describe exactly what is in the table. Keep the description simple and direct.
    """
    prompt = f"""
    Here is an example of how to describe adverse growth data for a given country:
    
    Country: Australia
    Adverse 1990: -0.43%
    Adverse 1991: -1.99%
    Adverse 1992: -1.20%
    Adverse Cumulative: -3.57%
    Topic: GDP
    
    Description:
    In the adverse scenario, the GDP growth in Australia was -0.43% in 1990. It worsened to -1.99% in 1991 and slightly improved to -1.20% in 1992. The total cumulative adverse growth was -3.57%.
    
    Now, using the following data for {theme} in {country}, describe the adverse growth:
    {row_str}
    Topic: {theme}
    
    Describe the adverse growth in the provided data year by year. Ensure the description follows the pattern in the example, highlighting whether the values increased or decreased each year.
    """
    prompt1 = f"""
    Here is an example of how to describe adverse growth data for a given country:
    
    Country: Australia
    Adverse 2020: -0.43%
    Adverse 2021: -1.99%
    Adverse 2022: -1.20%
    Adverse Cumulative: -3.57%
    Topic: GDP
    
    Description:
    In the adverse scenario, the GDP growth in Australia was -0.43% in 2020. It decreased further to -1.99% in 2021, showing worsening conditions. However, there was a slight improvement to -1.20% in 2022. The total cumulative adverse growth is -3.57%.
    
    Now, using the following data for {theme} in {country}, perform the following:
    1. Highlight how the values change from year to year.
    2. Describe whether the values increased or decreased compared to the previous year.
    3. Indicate if the changes represent a worsening or improvement, and if this is strong or slight.
    4. Include the cumulative result.
    
    Data:
    {row_str}
    Topic: {theme}
    Make sure your description follows the example format and accurately reflects the data.
    """

    # Generate the descriptive text using the model
    result = table_to_text(prompt, max_length=240, temperature = 0.7, top_p = 0.3, do_sample = False)[0]['generated_text']
    
    return result
# Global variable
stored_paragraphs_1 = []
stored_paragraphs_2 = []

stored_df1 = []
stored_df2 = []

with gr.Blocks() as demo:
    with gr.Tab("Methodology"):
        gr.Markdown("""
        ## Macro-economy Adverse Scenario Comparison from EBA Reports
        
        This application allows the user to compare two reports from text contents or from tables. It's divided into two tabs.
        
        **First Tab: Text Comparisons**
        
        - Select two PDFs. Each PDF's text content will be extracted into paragraphs.
        - Select a paragraph from one PDF, and find the most similar paragraph from the other PDF using a specific method.
        - For a selected paragraph, compute summarization using the **FinPEGASUS model**.
        - For a selected paragraph, compute sentiment analysis of the paragraph, and for each sentence, classify into three classes (Positive, Negative, Neutral) using two different fine-tuned **FinBERT models**:
          - [ProsusAI/finbert](https://huggingface.co/ProsusAI/finbert)
          - [yiyanghkust/finbert-tone](https://huggingface.co/yiyanghkust/finbert-tone)
        
        **Second Tab: Table Comparisons**
        
        - Select two Excel files and a sheet name.
        - For the two selected tables, compute the difference of the cumulative adverse growth rate over their respective three years for the selected sheet name (topic).
        - For the selected topic (sheet name), find related sentences in the associated PDF text that mention the topic, and classify them by sentiment.
        - For a selected country and topic, describe the adverse growth rate trend over three years using the [**google/flan-t5-base**](https://huggingface.co/google/flan-t5-base).
        """)
    with gr.Tab("Financial Report Text Analysis"):
        gr.Markdown("## Paragraph Extraction and Analysis on Adverse Macro-Economy Scenarios")

        with gr.Row():
            with gr.Column():
                gr.Markdown("### Step 1: Upload and Extract Paragraphs")
                pdf1 = gr.Dropdown(choices=get_pdf_files(PDF_FOLDER), label="Select PDF 1")
                pdf2 = gr.Dropdown(choices=get_pdf_files(PDF_FOLDER), label="Select PDF 2")
                extract_button = gr.Button("Extract Paragraphs")
                
            with gr.Column():
                gr.Markdown("### Step 2: Select Paragraphs for Analysis")
                paragraph_1_dropdown = gr.Dropdown(label="Select Paragraph from PDF 1")
                paragraph_2_dropdown = gr.Dropdown(label="Select Paragraph from PDF 2")

                def update_paragraphs(pdf1, pdf2):
                    stored_paragraphs_1, stored_paragraphs_2 = extract_and_paragraph(pdf1, pdf2, True)
                    return [f"Paragraph {i+1}: {p[:100]}..." for i, p in enumerate(stored_paragraphs_1)], \
                           [f"Paragraph {i+1}: {p[:100]}..." for i, p in enumerate(stored_paragraphs_2)]

                extract_button.click(update_paragraphs, inputs=[pdf1, pdf2], outputs=[paragraph_1_dropdown, paragraph_2_dropdown])

        with gr.Row():
            with gr.Column():
                gr.Markdown("### PDF 1 Analysis")
                summarize_btn1 = gr.Button("Summarize Text from PDF 1")
                summary_textbox_1 = gr.Textbox(label="Summary for PDF 1", lines=2)
                sentiment_btn1 = gr.Button("Classify Financial Tone from PDF 1")
                sentiment_textbox_1 = gr.Textbox(label="Tone Classification for PDF 1", lines=1)

                summarize_btn1.click(process_paragraph_1_sum, inputs=paragraph_1_dropdown, outputs=summary_textbox_1)
                sentiment_btn1.click(process_paragraph_1_sent, inputs=paragraph_1_dropdown, outputs=sentiment_textbox_1)

            with gr.Column():
                gr.Markdown("### PDF 2 Analysis")
                summarize_btn2 = gr.Button("Summarize Text from PDF 2")
                summary_textbox_2 = gr.Textbox(label="Summary for PDF 2", lines=2)
                sentiment_btn2 = gr.Button("Classify Financial Tone from PDF 2")
                sentiment_textbox_2 = gr.Textbox(label="Tone Classification for PDF 2", lines=1)

                summarize_btn2.click(process_paragraph_2_sum, inputs=paragraph_2_dropdown, outputs=summary_textbox_2)
                sentiment_btn2.click(process_paragraph_2_sent, inputs=paragraph_2_dropdown, outputs=sentiment_textbox_2)

    with gr.Tab("Financial Report Table Analysis"):
        gr.Markdown("## Excel Data Comparison and Topic Analysis")

        with gr.Row():
            with gr.Column():
                gr.Markdown("### Step 1: Upload Excel Files")
                file1 = gr.Dropdown(choices=get_excel_files(PDF_FOLDER), label="Select Excel File 1")
                file2 = gr.Dropdown(choices=get_excel_files(PDF_FOLDER), label="Select Excel File 2")
                sheet = gr.Dropdown(choices=["GDP", "HICP", "RRE prices", "Unemployment", "CRE prices"], label="Select Sheet")

            with gr.Column():
                gr.Markdown("### Step 2: Select a Country for Adverse Growth Analysis")
                country_1_dropdown = gr.Dropdown(label="Select Country from Excel File 1")
                country_2_dropdown = gr.Dropdown(label="Select Country from Excel File 2")

        with gr.Row():
            gr.Markdown("### Step 3: Compare Data and Generate Reports")
            comparison_button = gr.Button("Compare Data")
            text_result_df1 = gr.Textbox(label="Adverse Growth Report for Excel File 1", lines=4)
            text_result_df2 = gr.Textbox(label="Adverse Growth Report for Excel File 2", lines=4)

            comparison_button.click(fn=process_and_compare, inputs=[file1, sheet, file2], outputs=[text_result_df1, text_result_df2])

demo.launch()