app.py

import gradio as gr
from io import BytesIO
import torch
import os
import pdfplumber
import re
from transformers import pipeline, AutoModelForQuestionAnswering, AutoTokenizer
from transformers import BertTokenizer, EncoderDecoderModel

def process_pdf(path):
    results_dict = {}
    results_dict["1. Kurzbeschreibung"] = \
        read_section(path, "1. Kurzbeschreibung", "2. Einordnung des Moduls")
    # results_dict["2. Einordnung des Moduls"] = \
    #     read_section(path, "Einordnung des Moduls",
    #                 "Entwicklungsmaßnahmen im konkreten Interventionsbereich des Moduls")
    # results_dict["2.2 Andere Entwicklungsmaßnahmen im konkreten Interventionsbereich des Moduls"] = \
    #     read_section(path, "Entwicklungsmaßnahmen im konkreten Interventionsbereich des Moduls",
    #                 "3. Entwicklungen im Interventionsbereich")
    # results_dict["3. Entwicklungen im Interventionsbereich"] = \
    #     read_section(path, "3. Entwicklungen im Interventionsbereich",
    #                 "4.1 Bewertungen von Zielen, Zielgruppen, Wirkungshypothesen und Indikatoren")
    results_dict["4.1 Bewertungen von Zielen, Zielgruppen, Wirkungshypothesen und Indikatoren"] = \
        read_section(path, "4.1 Bewertungen von Zielen, Zielgruppen, Wirkungshypothesen und Indikatoren",
                    "4.2 ")
    results_dict["4.2 Umgesetzte Maßnahmen / Aktivitäten während des Berichtszeitraums"] = \
        read_section(path, "4.2 ", "4.3 ")
    # results_dict["4.3 Umsetzung von Maßnahmen zur Sicherstellung der nachhaltigen Wirksamkeit des Vorhabens"] = \
    #     read_section(path, "4.3",
    #                 "4.4 Laufzeit und Zeitplan")
    # results_dict["4.4 Laufzeit und Zeitplan"] = \
    #     read_section(path, "4.4 Laufzeit und Zeitplan", "4.5")
    # results_dict["4.5 Entstandene Kosten und Kostenverschiebungen"] = \
    #     read_section(path, "4.5", "4.6")
    results_dict["4.6 Bewertung der Wirkungen und Risiken"] = \
        read_section(path, "4.6 ", "5. Übergeordnete Empfehlungen")
    results_dict["5. Übergeordnete Empfehlungen"] = \
        read_section(path, "5. Übergeordnete Empfehlungen",
                    "5.2 Lernerfahrungen, die für die Länderstrategie und zukünftige")
    # results_dict["5.2 Lernerfahrungen, die für die Länderstrategie und zukünftige EZ-Programme interessant sein könnten"] = \
    #     read_section(path, "5.2 Lernerfahrungen", "6. Testat")
    # results_dict["6. Testat (TZ)"] = \
    #     read_section(path, "6. Testat", "Anlage 1: Wirkungsmatrix des Moduls")
    return results_dict

def read_section(path, wanted_section, next_section):
    
    doc = pdfplumber.open(BytesIO(path)) 
    start_page = []
    end_page = []

    for page in range(len(doc.pages)):
        if len(doc.pages[page].search(wanted_section, return_chars = False, case = False)) > 0:
            start_page.append(page)
        if len(doc.pages[page].search(next_section, return_chars = False, case = False)) > 0:
            end_page.append(page)
    print(wanted_section)
    print(max(start_page))
    print(max(end_page)+1)

    text = []
    for page_num in range(max(start_page), max(end_page)+1):
        page = doc.pages[page_num]
        text.append(page.extract_text())
    text = " ".join(text)
    text = text.replace("\n", " ")
    # print(wanted_section + str(extract_between(text, wanted_section, next_section)))
    return wanted_section + str(extract_between(text, wanted_section, next_section))

def extract_between(text, start_string, end_string):
    pattern = re.escape(start_string) + '(.*?)' + re.escape(end_string)
    match = re.search(pattern, text, re.DOTALL)
    if match:
        return match.group(1)
    else:
        return None

def format_section1(section1_text):
    result_section1_dict = {}
    result_section1_dict['TOPIC'] = extract_between(section1_text, "Sektor", "EZ-Programm")
    result_section1_dict['PROGRAM'] = extract_between(section1_text, "Sektor", "EZ-Programm")
    result_section1_dict['PROJECT DESCRIPTION'] = extract_between(section1_text, "EZ-Programmziel", "Datum der letzten BE")
    result_section1_dict['PROJECT NAME'] = extract_between(section1_text, "Modul", "Modulziel")
    result_section1_dict['OBJECTIVE'] = extract_between(section1_text, "Modulziel", "Berichtszeitraum")
    result_section1_dict['PROGRESS'] = extract_between(section1_text, "Zielerreichung des Moduls", "Massnahme im Zeitplan")
    result_section1_dict['STATUS'] = extract_between(section1_text, "Massnahme im Zeitplan", "Risikoeinschätzung")
    result_section1_dict['RECOMMENDATIONS'] = extract_between(section1_text, "Vorschläge zur Modulanpas-", "Voraussichtliche")
    return result_section1_dict

def initialize_question_answering():
    model_name = "deepset/gelectra-large-germanquad"
    model = AutoModelForQuestionAnswering.from_pretrained(model_name)
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    qa_pipeline = pipeline("question-answering", model=model, tokenizer=tokenizer)
    return qa_pipeline

def answer_questions_section_1(text, language="de"):
    qa_pipeline = initialize_question_answering()
    questions = [
        "Welche ist die Projektnummer des Moduls?",
        "Welches ist das Titel des Moduls?",
        "Welches ist das Sektor oder das Kernthema?",
        "Welches ist das Land?",
        "Zu welchem Program oder Programm gehort das Projekt?",
        # "Welche Durchführungsorganisation aus den 4 Varianten 'giz', 'kfw', 'ptb' und 'bgr' implementiert das Projekt?"
        "Wurde das Ziel des Moduls erreicht?", # "In dem Dokument was steht bei Zielerreichung des Moduls?",
        "Welche ist die Risikoeinschätzung des Moduls?",
        "Ist die Maßnahme im Zeitplan?"
        # "In dem Dokument was steht bei Sektor?",
        # "In dem Dokument was steht von 'EZ-Programm' bis 'EZ-Programmziel'?",
        # "In dem Dokument was steht bei EZ-Programmziel?",
        # "In dem Dokument in dem Abschnitt '1. Kurzbeschreibung' was steht bei Modul?",
        # "In dem Dokument in dem Abschnitt '1. Kurzbeschreibung' was steht bei Maßnahme im Zeitplan?",
        # "In dem Dokument was steht bei Vorschläge zur Modulanpassung?",
        # "In dem Dokument in dem Abschnitt 'Anlage 1: Wirkungsmatrix des Moduls' was steht unter Laufzeit als erstes Datum?",
        # "In dem Dokument in dem Abschnitt 'Anlage 1: Wirkungsmatrix des Moduls' was steht unter Laufzeit als zweites Datum?"
    ]
    answers_dict = {}
    for question in questions:
        result = qa_pipeline(question=question, context=text)
        print(f"Question: {question}")
        print(f"Answer: {result['answer']}\n")
        answers_dict[question] = result['answer']
    return answers_dict

def summarize_german_text(text):
    model_name = "mrm8488/bert2bert_shared-german-finetuned-summarization"
    tokenizer = BertTokenizer.from_pretrained(model_name)
    model = EncoderDecoderModel.from_pretrained(model_name)
    inputs = tokenizer(text, padding="max_length", truncation=True, max_length=512, return_tensors="pt")
    summary_ids = model.generate(inputs['input_ids'], num_beams=4, max_length=200, early_stopping=True)
    summary = tokenizer.decode(summary_ids[0], skip_special_tokens=True)
    return summary

def extract_details(path):
    sections_dict = process_pdf(path)
    results = answer_questions_section_1(sections_dict["1. Kurzbeschreibung"])
    # results["Section 4.1 summary"] = summarize_german_text(sections_dict["4.1 Bewertungen von Zielen, Zielgruppen, Wirkungshypothesen und Indikatoren"])
    # results["Section 4.2 summary"] = summarize_german_text(sections_dict["4.2 Umgesetzte Maßnahmen / Aktivitäten während des Berichtszeitraums"])
    # results["Section 4.6 summary"] = summarize_german_text(sections_dict["4.6 Bewertung der Wirkungen und Risiken"])
    # results["Section 5.1 summary"] = summarize_german_text(sections_dict["5. Übergeordnete Empfehlungen"])
    return results
    # for key, answer in results.items():
    #     print(f"{key}: {answer}")

if __name__ == "__main__":
    demo = gr.Interface(fn=extract_details, 
                     inputs=gr.File(type="binary", label="Upload PDF"),
                     outputs=gr.Textbox(label="Extracted Text"),
                     title="PDF Text Extractor",
                     description="Upload a PDF file to extract.")
    demo.launch()