partie_prenante_carte.py

import streamlit as st
import pandas as pd
import numpy as np
import re

import streamlit as st
from dotenv import load_dotenv
from PyPDF2 import PdfReader
from langchain.text_splitter import CharacterTextSplitter,RecursiveCharacterTextSplitter
from langchain_experimental.text_splitter import SemanticChunker
from langchain_community.embeddings import OpenAIEmbeddings
from langchain_community.vectorstores import FAISS
from langchain_community.chat_models import ChatOpenAI
from langchain.llms import HuggingFaceHub
from langchain import hub
from langchain_core.output_parsers import StrOutputParser
from langchain_core.runnables import RunnablePassthrough
from langchain_community.document_loaders import WebBaseLoader
from langchain_core.prompts.prompt import PromptTemplate
import altair as alt
from session import set_partie_prenante
import os
from streamlit_vertical_slider import vertical_slider
from pp_viz import display_viz
from high_chart import test_chart

load_dotenv()

def get_docs_from_website(urls):
    loader = WebBaseLoader(urls, header_template={
      'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/102.0.0.0 Safari/537.36',
    })
    try:
        docs = loader.load()
        return docs
    except Exception as e:
        return None
    

def get_doc_chunks(docs):
    # Split the loaded data
    # text_splitter = RecursiveCharacterTextSplitter(
    #                             chunk_size=500, 
    #                             chunk_overlap=100)

    text_splitter = SemanticChunker(OpenAIEmbeddings())
    
    docs = text_splitter.split_documents(docs)
    return docs
    

def get_vectorstore_from_docs(doc_chunks):
    embedding = OpenAIEmbeddings(model="text-embedding-3-large")
    vectorstore = FAISS.from_documents(documents=doc_chunks, embedding=embedding)
    return vectorstore

def get_conversation_chain(vectorstore):
    llm = ChatOpenAI(model="gpt-4o",temperature=0.5, max_tokens=2048)
    retriever=vectorstore.as_retriever()

    prompt = hub.pull("rlm/rag-prompt")
    # Chain
    rag_chain = (
        {"context": retriever , "question": RunnablePassthrough()}
        | prompt
        | llm
    )
    return rag_chain

# FILL THE PROMPT FOR THE QUESTION VARIABLE THAT WILL BE USED IN THE RAG PROMPT, ATTENTION NOT CONFUSE WITH THE RAG PROMPT
def fill_promptQ_template(input_variables, template):
    prompt = PromptTemplate(input_variables=["BRAND_NAME","BRAND_DESCRIPTION"], template=template)
    return prompt.format(BRAND_NAME=input_variables["BRAND_NAME"], BRAND_DESCRIPTION=input_variables["BRAND_DESCRIPTION"])

def text_to_list(text):
    lines = text.replace("- ","").split('\n')
    
    lines = [line.split() for line in lines]
    items = [[' '.join(line[:-1]),line[-1]] for line in lines]

    # Assuming `items` is the list of items
    for item in items:
        item[1] = re.sub(r'\D', '', item[1])
    return items

def delete_pp(pps):
    for pp in pps:
        for i in range(len(st.session_state['pp_grouped'])):
            if st.session_state['pp_grouped'][i]['name'] == pp:
                del st.session_state['pp_grouped'][i]
                break

def display_list_urls():
    for index, item in enumerate(st.session_state["urls"]):
        emp = st.empty()  # Create an empty placeholder
        col1, col2 = emp.columns([7, 3])  # Divide the space into two columns

        # Button to delete the entry, placed in the second column
        if col2.button("❌", key=f"but{index}"):

            st.session_state["save"] = True
            temp  = st.session_state['parties_prenantes'][index]
            delete_pp(temp)
            del st.session_state.urls[index]
            del st.session_state["parties_prenantes"][index]
            st.experimental_rerun()  # Rerun the app to update the display

        if len(st.session_state.urls) > index:
            # Instead of using markdown, use an expander in the first column
            with col1.expander(f"URL {index}: {item}"):
                pp = st.session_state["parties_prenantes"][index]
                st.write(pd.DataFrame(pp, columns=["Partie prenante"]))
        else:
            emp.empty()  # Clear the placeholder if the index exceeds the list
    
def display_list_pps():
    for index, item in enumerate(st.session_state["pp_grouped"]):
        emp = st.empty()
        col1, col2 = emp.columns([7, 3])

        if col2.button("❌", key=f"butp{index}"):

            del st.session_state["pp_grouped"][index]
            st.experimental_rerun()

        if len(st.session_state["pp_grouped"]) > index:
            name = st.session_state["pp_grouped"][index]["name"]
            col1.markdown(f"{name}")
        else:
            emp.empty()


                
def extract_pp(urls,input_variables):
    template_extraction_PP = '''
    Objectif : identifiez tout les noms de marques qui sont des parties prenantes de la marque suivante pour développer un marketing de coopération (co-op marketing)

    Le nom de la marque de référence est le suivant : {BRAND_NAME}
    Son activité est la suivante : {BRAND_DESCRIPTION}

    TA REPONSE DOIT ETRE SOUS FORME DE LISTE DE NOMS DE MARQUES SANS NUMEROTATION ET SEPARES PAR DES SAUTS DE LIGNE

    SI TU NE TROUVES PAS DE NOM DE MARQUE, REPONDS "444"
    '''
    #don't forget to add the input variables from the maim function

    docs = get_docs_from_website(urls)

    if docs == None:
        return "445"

    #get text chunks
    text_chunks = get_doc_chunks(docs)

    #create vectorstore
    vectorstore = get_vectorstore_from_docs(text_chunks)

    chain = get_conversation_chain(vectorstore)

    question = fill_promptQ_template(input_variables, template_extraction_PP)

    response = chain.invoke(question)

    # version plus poussée a considérer
    # each item in the list is a list with the name of the brand and the similarity percentage
    # partie_prenante = text_to_list(response.content)

    #version simple
    partie_prenante = response.content.replace("- ","").split('\n')

    return partie_prenante

def format_pp_add_viz(pp):
    y = 50
    x = 50
    for i in range(len(st.session_state['pp_grouped'])):
        if st.session_state['pp_grouped'][i]['y'] == y and st.session_state['pp_grouped'][i]['x'] == x:
            y += 5
        if y > 95:
            y = 50
            x += 5
        if st.session_state['pp_grouped'][i]['name'] == pp:
            return None
    else:
        st.session_state['pp_grouped'].append({'name':pp, 'x':x,'y':y})

def add_pp(new_pp, default_value=50):
    new_pp = sorted(new_pp)
    new_pp = [item.lower().capitalize() for item in new_pp]
    st.session_state['parties_prenantes'].append(new_pp)
    for pp in new_pp:
        format_pp_add_viz(pp)

def add_pp_input_text():
    new_pp = st.text_input("Ajouter une partie prenante")
    if st.button("Ajouter"):
        st.session_state["save"] = True
        format_pp_add_viz(new_pp)

import re

def complete_and_verify_url(partial_url):
    # Regex pattern for validating a URL
    regex = re.compile(
        r'^(?:http|ftp)s?://'  # http:// or https://
        r'(?:(?:[A-Z0-9](?:[A-Z0-9-]{0,61}[A-Z0-9])?\.)+[A-Z]{2,6}\.?|'  # domain
        r'localhost|'  # localhost...
        r'\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3})'  # ...or ip
        r'(?::\d+)?'  # optional port
        r'(?:/?|[/?]\S+)$', re.IGNORECASE)

    # Complete the URL if it doesn't have http:// or https://
    if not partial_url.startswith(('http://', 'https://', 'www.')):
        if not partial_url.startswith('www.'):
            complete_url = 'https://www.' + partial_url
        else:
            complete_url = 'https://' + partial_url

    elif partial_url.startswith('www.'):
        complete_url = 'https://' + partial_url

    else:
        complete_url = partial_url

    # Check if the URL is valid
    if re.match(regex, complete_url):
        return (True, complete_url)
    else:
        return (False, complete_url)


def display_pp():

    load_dotenv()
    
    #check if brand name and description are already set
    if "Nom de la marque" not in st.session_state:
        st.session_state["Nom de la marque"] = ""

    #check if urls and partie prenante are already set
    if "urls" not in st.session_state:
        st.session_state["urls"] = []
    if "parties_prenantes" not in st.session_state:
        st.session_state['parties_prenantes'] = []
    if "pp_grouped" not in st.session_state: #servira pour le plot et la cartographie des parties prenantes, regroupe sans doublons
        st.session_state['pp_grouped'] = []
    
    if "save" not in st.session_state:
        st.session_state["save"] = False

    st.header("Parties prenantes de la marque")
    #set brand name and description
    brand_name = st.text_input("Nom de la marque", st.session_state["Nom de la marque"])
    st.session_state["Nom de la marque"] = brand_name

    option = st.radio("Source", ("A partir de votre site web", "A partir de vos documents entreprise"))

    #if the user chooses to extract from website
    if option == "A partir de votre site web":

        url = st.text_input("Ajouter une URL")

        #if the user clicks on the button
        if st.button("ajouter"):
            st.session_state["save"] = True
            #complete and verify the url
            is_valid,url = complete_and_verify_url(url)
            if not is_valid:
                st.error("URL invalide")
            elif url in st.session_state["urls"] :
                st.error("URL déjà ajoutée")
            
            else:
                docs = get_docs_from_website(url)
                if docs is None:
                    st.error("Aucune url trouvée ou erreur lors de la récupération du contenu")
                else:
                # Création de l'expander
                    with st.expander("Cliquez ici pour éditer et voir le document"):
                        cleaned_text = re.sub(r'\n\n+', '\n\n', docs[0].page_content.strip())
                        text_value = st.text_area("Modifier le texte ci-dessous:", value=cleaned_text, height=300)
                        if st.button('Sauvegarder'):
                            st.success("Texte sauvegardé avec succès!")
                    
                    with st.spinner("Processing..."):

                        #handle the extraction
                        input_variables = {"BRAND_NAME": brand_name, "BRAND_DESCRIPTION": ""}
                        partie_prenante = extract_pp([url], input_variables)

                        if "444" in partie_prenante: #444 is the code for no brand found , chosen
                            st.error("Aucune partie prenante trouvée")
                        elif "445" in partie_prenante: #445 is the code for no website found with the given url
                            st.error("Aucun site web trouvé avec l'url donnée")
                        else:
                            partie_prenante = sorted(partie_prenante)
                            st.session_state["urls"].append(url)
                            add_pp(partie_prenante)
                    
                    # alphabet = [ pp[0] for pp in partie_prenante]
                    # pouvoir = [ 50 for _ in range(len(partie_prenante))]
                    # df = pd.DataFrame({'partie_prenante': partie_prenante, 'pouvoir': pouvoir, 'code couleur': partie_prenante})
                    # st.write(df)

                    # c = (
                    # alt.Chart(df)
                    # .mark_circle(size=300)
                    # .encode(x="partie_prenante", y=alt.Y("pouvoir",scale=alt.Scale(domain=[0,100])), color="code couleur")
                    # )
                    # st.subheader("Vertical Slider")
                    # age = st.slider("How old are you?", 0, 130, 25)
                    # st.write("I'm ", age, "years old")

                    # disp_vertical_slider(partie_prenante)
                    # st.altair_chart(c, use_container_width=True)
    display_list_urls()
    with st.expander("Liste des parties prenantes"):
        add_pp_input_text()
        display_list_pps()
    test_chart()