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

from beautiful_soup.beautiful_soup import get_url_content

@cache
def google_search_api_request( query ):
    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,
        ).execute()

def search_results( query ):
    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 content_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:
        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=130, min_length=30, do_sample=False, truncation=True)
        except Exception as exception:
            raise exception

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

def exception_notice( exception ):
    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 ):
    for keyword in keywords:
        if keyword in string:
            return True
    return False

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(' ')
                    content = ''
                    for string in strings:
                        if is_keyword_in_string( keywords, string ):
                            content += string + '\n'
                    summary = content_summary( url_id, content )
                    for sentence in summary:
                        st.write(sentence['summary_text'])
                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()