app.py

from os import makedirs, remove
from os.path import exists, dirname
from functools import cache
import json
import streamlit as st
from googleapiclient.discovery import build
from slugify import slugify
from transformers import pipeline
import uuid
import spacy
from spacy.matcher import PhraseMatcher

from beautiful_soup.beautiful_soup import get_url_content


@cache
def google_search_api_request( query ):
    """
    Request Google Search API with query and return results.
    """

    api_key = st.secrets["google_search_api_key"]
    cx = st.secrets["google_search_engine_id"]
    service = build(
        "customsearch",
        "v1",
        developerKey=api_key,
        cache_discovery=False
    )

    # Exclude PDFs from search results.
    query = query + ' -filetype:pdf'

    return service.cse().list(
        q=query,
        cx=cx,
        num=5,
        lr='lang_en', # lang_de
        fields='items(title,link),searchInformation(totalResults)'
        ).execute()


def search_results( query ):
    """
    Request Google Search API with query and return results. Results are cached in files.
    """
    file_path = 'search-results/' + slugify( query ) + '.json'

    results = []
    makedirs(dirname(file_path), exist_ok=True)
    if exists( file_path ):
        with open( file_path, 'r' ) as results_file:
            results = json.load( results_file )
    else:
        search_result = google_search_api_request( query )
        if int( search_result['searchInformation']['totalResults'] ) > 0:
            results = search_result['items']
            with open( file_path, 'w' ) as results_file:
                json.dump( results, results_file )

    if len( results ) == 0:
        raise Exception('No results found.')
    
    return results

def get_summary( url_id, content ):
    file_path = 'summaries/' + url_id + '.json'
    makedirs(dirname(file_path), exist_ok=True)
    if exists( file_path ):
        with open( file_path, 'r' ) as file:
            summary = json.load( file )
    else:
        summary = generate_summary( content )

        with open( file_path, 'w' ) as file:
            json.dump( summary, file )
    
    return summary

def generate_summary( content, max_length = 200 ):
    """
    Generate summary for content.
    """
    try:
        summarizer = pipeline("summarization", model="sshleifer/distilbart-cnn-12-6")
        # https://huggingface.co/docs/transformers/v4.18.0/en/main_classes/pipelines#transformers.SummarizationPipeline
        summary = summarizer(content, max_length, min_length=30, do_sample=False, truncation=True)
    except Exception as exception:
        raise exception
    
    return summary

def exception_notice( exception ):
    """
    Helper function for exception notices.
    """
    query_params = st.experimental_get_query_params()
    if 'debug' in query_params.keys() and query_params['debug'][0] == 'true':
        st.exception(exception)
    else:
        st.warning(str(exception))

def is_keyword_in_string( keywords, string ):
    """
    Checks if string contains keyword.
    """
    for keyword in keywords:
        if keyword in string:
            return True
    return False

def filter_sentences_by_keywords( strings, keywords ):
    nlp = spacy.load("en_core_web_sm")
    matcher = PhraseMatcher(nlp.vocab)
    phrases = keywords
    patterns = [nlp(phrase) for phrase in phrases]
    matcher.add("QueryList", patterns)

    sentences = []
    for string in strings:
        # Exclude short sentences
        string_length = len( string.split(' ') )
        if string_length < 5:
            continue
        doc = nlp(string)
        for sentence in doc.sents:
            matches = matcher(nlp(sentence.text))
            for match_id, start, end in matches:
                if nlp.vocab.strings[match_id] in ["QueryList"]:
                    sentences.append(sentence.text) 

    return sentences

def split_content_into_chunks( sentences ):
    """
    Split content into chunks.
    """
    chunk  = ''
    word_count = 0
    chunks = []
    for sentence in sentences:
        current_word_count = len(sentence.split(' '))
        if word_count + current_word_count > 512:
            st.write("Number of words(tokens): {}".format(word_count))
            chunks.append(chunk)
            chunk = ''
            word_count = 0

        word_count += current_word_count
        chunk += sentence + ' '

    st.write("Number of words(tokens): {}".format(word_count))
    chunks.append(chunk)

    return chunks

def main():
    st.title('Racoon Search')
    query = st.text_input('Search query')
    query_params = st.experimental_get_query_params()

    if query :
        with st.spinner('Loading search results...'):
            try:
                results = search_results( query )
            except Exception as exception:
                exception_notice(exception)
                return

        number_of_results = len( results )
        st.success( 'Found {} results for "{}".'.format( number_of_results, query ) )

        if 'debug' in query_params.keys() and query_params['debug'][0] == 'true':
            with st.expander("Search results JSON"):
                if st.button('Delete search result cache', key=query + 'cache'):
                    remove( 'search-results/' + slugify( query ) + '.json' )
                st.json( results )

        progress_bar = st.progress(0)

        st.header('Search results')
        st.markdown('---')

        # for result in results:
        for index, result in enumerate(results):
            with st.container():
                st.markdown('### ' + result['title'])
                url_id = uuid.uuid5( uuid.NAMESPACE_URL, result['link'] ).hex
                try:
                    strings   = get_url_content( result['link'] )
                    keywords  = query.split(' ')
                    sentences = filter_sentences_by_keywords( strings, keywords )
                    chunks    = split_content_into_chunks( sentences )

                    number_of_chunks = len( chunks )
                    if number_of_chunks > 1:
                        max_length = int( 512 / len( chunks ) )
                        st.write("Max length: {}".format(max_length))

                        content = ''
                        for chunk in chunks:
                            chunk_length = len( chunk.split(' ') )
                            chunk_max_length = 200
                            if chunk_length < max_length:
                                chunk_max_length = int( chunk_length / 2 )
                            chunk_summary = generate_summary( chunk, min( max_length, chunk_max_length )  )
                            for summary in chunk_summary:
                                content += summary['summary_text'] + ' '
                    else:
                        content = chunks[0]

                    summary = get_summary( url_id, content )

                except Exception as exception:
                    exception_notice(exception)

                progress_bar.progress( ( index + 1 ) / number_of_results )

                col1, col2, col3 = st.columns(3)
                with col1:
                    st.markdown('[Website Link]({})'.format(result['link']))

                with col2:
                    if st.button('Delete content from cache', key=url_id + 'content'):
                        remove( 'page-content/' + url_id + '.txt' )

                with col3:
                    if st.button('Delete summary from cache', key=url_id + 'summary'):
                        remove( 'summaries/' + url_id + '.json' )

                st.markdown('---')
 

if __name__ == '__main__':
    main()