app.py

import gradio as gr
import requests
import json
from base64 import b64encode
import fitz
import os
import pickle
import pytesseract
import numpy as np
from langchain import OpenAI
from langchain.text_splitter import RecursiveCharacterTextSplitter
from langchain.prompts import PromptTemplate
from langchain.chains.summarize import load_summarize_chain
from langchain.callbacks import get_openai_callback
import logging

logging.basicConfig(
    format='%(asctime)s %(levelname)-8s %(message)s',
    level=logging.INFO,
    datefmt='%Y-%m-%d %H:%M:%S')
chkey = os.environ["API_TOKEN"]
token = b64encode(f"{chkey}".encode('utf-8')).decode("ascii")

with gr.Blocks(gr.themes.Soft()) as demo:
    logging.info("*** App Starting ***")
    # Introduction
    intro = gr.Textbox(label="Introduction", interactive=False, value="An application to search for any Limited company in the UK, check when the latest Accounts are filed and summarize the account filing using OpenAI.\nUses UK Companies House API to search and get the company information. And Langchain's Summarization chain to create a summary. (Needs an OpenAI API key)")
    # Create input Text search box
    input_box = gr.Textbox(label="Input search string for a UK Company Name")
    # Button to initiate the Company Search
    search_btn = gr.Button("Search")
    # State variable to store the Document ID
    doc_id = gr.State()
    # Column to display the Company search result - List of companies
    with gr.Column(visible=False) as output_col:
        company_list_box = gr.Radio(choices=["Test1","Test2"],label="Company search result")
        #company_list = gr.DataFrame(headers=['Company Number','Company Name','Date Created','Registered Address'], interactive=False)
    
    # Text box to display the latest filing information
    display_filing = gr.Textbox(label="",interactive=False, visible=False)

    # Button to get the latest filing document
    submit_btn = gr.Button("Get latest filing", visible=False)

    # Text box Display the Document information
    display_filing_doc_info = gr.Textbox(label="",interactive=False, visible=False)

    # OpenAPI Key Input box
    openapi_key_input = gr.Textbox(label="OpenAI API Key", type='password', interactive=True, visible=False)

    # Button to initiate the processing of the Document - OPENAI Call initiated here too
    process_filing_btn = gr.Button("Summarize the Account filing", visible=False)

    # Text box showing the progress of the processing and the message
    processed_info = gr.Textbox(label="",interactive=False, visible=False)

    # Final Text box to display the summary of the Annual Accounts
    summary_text = gr.Textbox(label="Summary using OPENAI",interactive=False, visible=False)

    # Clear all fields to start again
    clear = gr.Button("Clear")

    # Function that does the Company search based on a search string. Gets the top 10 results 
    def company_search(text):
        logging.info("*** New Search Starting ***")
        logging.info(f'Search term : {text}')
        url = "https://api.company-information.service.gov.uk/advanced-search/companies?company_name_includes=" + text + "&company_status=active&size=10"
        logging.info(f'Calling Companies House API Advanced search : {url}')
        auth = f'Basic {token}'
        payload={}
        headers = {
            'Authorization': auth
        }
        response = requests.request("GET", url, headers=headers, data=payload)
        logging.info(f'API Response Code : {response.status_code}')
        select_resp = []
        if response.status_code == 200:
            resp = json.loads(response.text)
            for comp in resp["items"]:
                addr = []
                for key, value in comp["registered_office_address"].items():
                    addr.append(value) 
                select_resp.append(comp["company_number"] + " : " + comp["company_name"] + " : " + ', '.join(addr))
            resp_joined = (','.join(select_resp))
            logging.info(f'Response list : {resp_joined}')
            return {output_col: gr.update(visible=True), company_list_box: gr.update(choices=select_resp,interactive=True)}
        else:
            select_resp.append("No matching companies found")
            return {output_col: gr.update(visible=True), company_list_box: gr.update(choices=select_resp,interactive=False)}            
    
    # Function to get the Filing information of a selected company
    def company_selected(selected_company, docid):
        logging.info("* Company selected. Getting Filing History *")
        logging.info(f'User Selection : {selected_company}')
        regid = selected_company.split(' : ')[0]
        filings_url = "https://api.company-information.service.gov.uk/company/" + regid + "/filing-history?category=accounts&items_per_page=1"
        logging.info(f'Calling Companies House API Filings Endpoint : {filings_url}')
        auth = f'Basic {token}'
        payload={}
        headers = {
            'Authorization': auth
        }
        response = requests.request("GET", filings_url, headers=headers, data=payload)
        resp = json.loads(response.text)
        logging.info(f'API Response Code : {response.status_code}')
        if response.status_code == 200:
            if len(resp["items"])>0:
                resp_value = f'Latest filing done on {resp["items"][0]["date"]}.'
                if "links" in resp["items"][0]:
                    if "document_metadata" in resp["items"][0]["links"]:
                        docid = resp["items"][0]["links"]["document_metadata"].rsplit('/',1)[-1]
                        return {display_filing: gr.Textbox.update(visible=True, value=resp_value), submit_btn: gr.update(visible=True), doc_id : docid}
                    else:
                        resp_value += "But Document Metadata is not available."
                        return {display_filing: gr.Textbox.update(visible=True, value=resp_value), submit_btn: gr.update(visible=False), doc_id : "None"}
                else:
                    resp_value += "But Links to the filing not available."
                    return {display_filing: gr.Textbox.update(visible=True, value=resp_value), submit_btn: gr.update(visible=False), doc_id : "None"}
            else:
                return {display_filing: gr.Textbox.update(visible=True, value="No record of accounts filed for the company"), submit_btn: gr.update(visible=False), doc_id : "None"}
        else:
            return {display_filing: gr.Textbox.update(visible=True, value="No record of accounts filed for the company"), submit_btn: gr.update(visible=False), doc_id : "None"}

    # Function to get the Filing document related to the latest Annual Account filing
    def get_filing(docid):
        logging.info("* Getting Filing Document for latest filing *")
        doc_url = "https://document-api.company-information.service.gov.uk/document/" + docid + "/content"
        logging.info(f'Calling Companies House Documents API : {doc_url}')
        auth = f'Basic {token}'
        payload={}
        headers = {
            'Authorization': auth,
            'Accept': 'application/pdf'
        }
        response = requests.request("GET", doc_url, headers=headers, data=payload)
        logging.info(f'API Response Code : {response.status_code}')
        content_type = response.headers['Content-Type']
        resp_value = f'Filing document is of type {content_type}. '
        if content_type == 'application/pdf':
            filename = f'doc_{docid}.pdf'
            filepath = './data/'+filename
            with open(filepath, 'wb') as f:
                f.write(response.content)
            pdf_document = fitz.open(filepath)
            resp_value += f'PDF saved as: {filename}. There are a total of {pdf_document.page_count} pages'
            logging.info(resp_value)
            return {display_filing_doc_info: gr.Textbox.update(visible=True, value=resp_value), process_filing_btn: gr.update(visible=True), openapi_key_input: gr.update(visible=True), processed_info: gr.update(visible=True), doc_id : docid}
        else:
            resp_value += 'Work in progress to process these type of filings'
            logging.info(resp_value)
            return {display_filing_doc_info: gr.Textbox.update(visible=True, value=resp_value), process_filing_btn: gr.update(visible=False), openapi_key_input: gr.update(visible=False), processed_info: gr.update(visible=False), doc_id : "None"}

    # Function to initial the Langchain chain with call to OPENAI to Summarize the Annual report
    def langchain_summarize(contents,openai_api_key):
        logging.info("* Calling Langchain / OPENAI to get the summary *")
        concatenated_content = '`n`n'.join(contents)
        text_splitter = RecursiveCharacterTextSplitter(chunk_size=1200,chunk_overlap=20,length_function=len)
        docs = text_splitter.create_documents([concatenated_content])
        prompt_template = """You are a financial analyst, analyzing the Annual report submitted by the limited company at UK Companies House. Write a concise summary of the following report:


        {text}

        
        CONCISE SUMMARY:"""
        PROMPT = PromptTemplate(template=prompt_template, input_variables=["text"])
        llm = OpenAI(temperature=0, openai_api_key=openai_api_key)
        #docs = [Document(page_content=t) for t in contents]
        chain = load_summarize_chain(llm, chain_type="map_reduce", map_prompt=PROMPT, combine_prompt=PROMPT)
        with get_openai_callback() as cb:
            resp = chain.run(docs)
            tkn_text = f'*** Spent a total of {cb.total_tokens} tokens ***'
        return resp, tkn_text

    # Function to extract text from the document and call the LangChain processing function with the text array of pages
    def process_filing(docid, openai_api_key, progress=gr.Progress()):
        logging.info("* Processing the filing document *")
        progress(0,desc="Starting...")
        filepath = f'./data/doc_{docid}.pdf'
        pdf_document = fitz.open(filepath)
        text_path = f'./text/doc_{docid}.pkl'
        if os.path.exists(text_path):
            # retrieve list from file
            with open(text_path, 'rb') as f:
                contents = pickle.load(f)
        else:
            zoom_x = 2.0  # horizontal zoom
            zoom_y = 2.0  # vertical zoom
            mat = fitz.Matrix(zoom_x, zoom_y)  # zoom factor 2 in each dimension
            contents = []
            for page in progress.tqdm(pdf_document, desc="Processing pages from PDF..."):
                # Convert the page to a PNG image using PyMuPDF
                pix = page.get_pixmap(matrix=mat)
                img = np.frombuffer(pix.samples, dtype=np.uint8).reshape(pix.height,pix.width, pix.n)
                texts = pytesseract.image_to_string(img)
                contents.append(texts)
            # save list to file
            with open(text_path, 'wb') as f:
                pickle.dump(contents, f)

        resp_value = f'Total of {pdf_document.page_count} pages processed. '
        summary_path = f'./summary/doc_{docid}.txt'
        if os.path.exists(summary_path):
            with open(summary_path, 'r') as f:
                summary = f.read()
            logging.info(resp_value)
            return {processed_info: gr.Textbox.update(visible=True, value=resp_value), summary_text: gr.Textbox.update(visible=True, value=summary)}
        else:
            try:
                summary, tkn_text = langchain_summarize(contents, openai_api_key)
                resp_value += tkn_text
                logging.info(resp_value)
                with open(summary_path, 'wb') as f:
                    f.write(summary.encode())
                return {processed_info: gr.Textbox.update(visible=True, value=resp_value), summary_text: gr.Textbox.update(visible=True, value=summary)}
            except Exception as e:
                logging.info(e)
                resp_value += 'LLM Call failed. Please check the OpenAI key again'
                logging.info(resp_value)
                return {processed_info: gr.Textbox.update(visible=True, value=resp_value), summary_text: gr.Textbox.update(visible=False)}
            finally:
                logging.info(resp_value)
    
    def clear_screen():
        return {output_col: gr.update(visible=False),display_filing: gr.Textbox.update(visible=False),submit_btn: gr.update(visible=False), display_filing_doc_info:gr.update(visible=False), process_filing_btn:gr.update(visible=False),openapi_key_input:gr.update(visible=False),processed_info:gr.update(visible=False),summary_text:gr.update(visible=False)}

    search_btn.click(company_search,input_box,[company_list_box,output_col])
    company_list_box.change(company_selected,[company_list_box, doc_id],[display_filing, submit_btn, doc_id])
    submit_btn.click(get_filing,doc_id,[display_filing_doc_info, process_filing_btn, openapi_key_input, processed_info, doc_id])
    process_filing_btn.click(process_filing,[doc_id,openapi_key_input],[processed_info,summary_text])
    clear.click(clear_screen, None, [output_col,display_filing,submit_btn,display_filing_doc_info,process_filing_btn,openapi_key_input,processed_info,summary_text])

demo.queue(concurrency_count=3)
demo.launch()