presidio_streamlit.py

"""Streamlit app pro anonymizaci českých textů s využitím Presidio."""
import logging
import os
import traceback

import dotenv
import pandas as pd
import streamlit as st
import streamlit.components.v1 as components
from annotated_text import annotated_text
from streamlit_tags import st_tags

from openai_fake_data_generator import OpenAIParams
from presidio_helpers import (
    get_supported_entities,
    analyze,
    anonymize,
    annotate,
    create_fake_data,
    analyzer_engine,
)

st.set_page_config(
    page_title="Presidio demo pro české texty",
    layout="wide",
    initial_sidebar_state="expanded",
    menu_items={
        "About": "https://microsoft.github.io/presidio/",
    },
)

dotenv.load_dotenv()
logger = logging.getLogger("presidio-streamlit")

allow_other_models = os.getenv("ALLOW_OTHER_MODELS", False)

# Sidebar
st.sidebar.header(
    """
Anonymizace osobních údajů v českých textech s [Microsoft Presidio](https://microsoft.github.io/presidio/)
"""
)

model_help_text = """
    Vyberte model pro rozpoznávání pojmenovaných entit (NER) pro detekci osobních údajů v českých textech.
    Presidio podporuje různé NER balíčky, jako jsou spaCy, Huggingface, Stanza a Flair,
    stejně jako služby jako Azure Text Analytics PII.
    """
st_ta_key = st_ta_endpoint = ""

model_list = [
    "spacy/en_core_web_sm",
    "iiiorg/piiranha-v1-detect-personal-information",
    "FacebookAI/xlm-roberta-large-finetuned-conll03-english",
]
if not allow_other_models:
    model_list.pop()
# Výběr modelu
st_model = st.sidebar.selectbox(
    "NER model",
    model_list,
    index=1,
    help=model_help_text,
)

# Extrakce balíčku modelu
st_model_package = st_model.split("/")[0]

# Odstranění prefixu balíčku (pokud je potřeba)
st_model = (
    st_model
    if st_model_package.lower() not in ("spacy", "iiiorg")
    else "/".join(st_model.split("/")[1:])
)

if st_model == "Other":
    st_model_package = st.sidebar.selectbox(
        "NER model OSS balíček", options=["spacy", "iiiorg"]
    )
    st_model = st.sidebar.text_input(f"Název NER modelu", value="")

st.sidebar.warning("Poznámka: Stažení modelů může chvíli trvat.")

analyzer_params = (st_model_package, st_model, st_ta_key, st_ta_endpoint)
logger.debug(f"analyzer_params: {analyzer_params}")

st_operator = st.sidebar.selectbox(
    "Přístup k anonymizaci",
    ["redact", "replace", "synthesize", "highlight", "mask", "hash", "encrypt"],
    index=1,
    help="""
    Vyberte způsob úpravy textu po identifikaci osobních údajů.\n
    - Redact: Kompletně odstranit osobní údaj\n
    - Replace: Nahradit osobní údaj konstantou, např. <OSOBA>\n
    - Synthesize: Nahradit falešnými hodnotami (vyžaduje OpenAI klíč)\n
    - Highlight: Zobrazí původní text se zvýrazněnými osobními údaji\n
    - Mask: Nahradí požadovaný počet znaků hvězdičkou (nebo jiným znakem)\n
    - Hash: Nahradí hashem osobního údaje\n
    - Encrypt: Nahradí AES šifrováním osobního údaje, umožňující reverzní proces
         """,
)
st_mask_char = "*"
st_number_of_chars = 15
st_encrypt_key = "WmZq4t7w!z%C&F)J"

open_ai_params = None

logger.debug(f"st_operator: {st_operator}")

def set_up_openai_synthesis():
    """Nastavení OpenAI API klíče a modelu pro syntézu textu."""

    if os.getenv("OPENAI_TYPE", default="openai") == "Azure":
        openai_api_type = "azure"
        st_openai_api_base = st.sidebar.text_input(
            "Azure OpenAI base URL",
            value=os.getenv("AZURE_OPENAI_ENDPOINT", default=""),
        )
        openai_key = os.getenv("AZURE_OPENAI_KEY", default="")
        st_deployment_id = st.sidebar.text_input(
            "Název nasazení", value=os.getenv("AZURE_OPENAI_DEPLOYMENT", default="")
        )
        st_openai_version = st.sidebar.text_input(
            "OpenAI verze",
            value=os.getenv("OPENAI_API_VERSION", default="2023-05-15"),
        )
    else:
        openai_api_type = "openai"
        st_openai_version = st_openai_api_base = None
        st_deployment_id = ""
        openai_key = os.getenv("OPENAI_KEY", default="")
    st_openai_key = st.sidebar.text_input(
        "OPENAI_KEY",
        value=openai_key,
        help="Více informací na https://help.openai.com/en/articles/4936850-where-do-i-find-my-secret-api-key",
        type="password",
    )
    st_openai_model = st.sidebar.text_input(
        "OpenAI model pro syntézu textu",
        value=os.getenv("OPENAI_MODEL", default="gpt-3.5-turbo-instruct"),
        help="Více informací zde: https://platform.openai.com/docs/models/",
    )
    return (
        openai_api_type,
        st_openai_api_base,
        st_deployment_id,
        st_openai_version,
        st_openai_key,
        st_openai_model,
    )

if st_operator == "mask":
    st_number_of_chars = st.sidebar.number_input(
        "počet znaků", value=st_number_of_chars, min_value=0, max_value=100
    )
    st_mask_char = st.sidebar.text_input(
        "Znak pro maskování", value=st_mask_char, max_chars=1
    )
elif st_operator == "encrypt":
    st_encrypt_key = st.sidebar.text_input("AES klíč", value=st_encrypt_key)
elif st_operator == "synthesize":
    (
        openai_api_type,
        st_openai_api_base,
        st_deployment_id,
        st_openai_version,
        st_openai_key,
        st_openai_model,
    ) = set_up_openai_synthesis()

    open_ai_params = OpenAIParams(
        openai_key=st_openai_key,
        model=st_openai_model,
        api_base=st_openai_api_base,
        deployment_id=st_deployment_id,
        api_version=st_openai_version,
        api_type=openai_api_type,
    )

st_threshold = st.sidebar.slider(
    label="Práh přijetí",
    min_value=0.0,
    max_value=1.0,
    value=0.35,
    help="Definujte práh pro přijetí detekce jako osobní údaj.",
)

st_return_decision_process = st.sidebar.checkbox(
    "Přidat vysvětlení analýzy k nálezům",
    value=False,
    help="Přidá rozhodovací proces do výstupní tabulky. "
    "Více informací najdete zde: https://microsoft.github.io/presidio/analyzer/decision_process/",
)

# Povolené a zakázané seznamy
st_deny_allow_expander = st.sidebar.expander(
    "Povolené a zakázané seznamy",
    expanded=False,
)

with st_deny_allow_expander:
    st_allow_list = st_tags(
        label="Přidat slova do povoleného seznamu", text="Zadejte slovo a stiskněte enter."
    )
    st.caption(
        "Povolené seznamy obsahují slova, která nejsou považována za osobní údaje, ale jsou jako takové detekována."
    )

    st_deny_list = st_tags(
        label="Přidat slova do zakázaného seznamu", text="Zadejte slovo a stiskněte enter."
    )
    st.caption(
        "Zakázané seznamy obsahují slova, která jsou považována za osobní údaje, ale nejsou jako takové detekována."
    )

# Hlavní panel
with st.expander("O této ukázce", expanded=False):
    st.info(
        """Presidio je open source přizpůsobitelný framework pro detekci a anonymizaci osobních údajů.
        \n\n[Kód](https://aka.ms/presidio) | 
        [Tutoriál](https://microsoft.github.io/presidio/tutorial/) | 
        [Instalace](https://microsoft.github.io/presidio/installation/) | 
        [FAQ](https://microsoft.github.io/presidio/faq/) |
        [Zpětná vazba](https://forms.office.com/r/9ufyYjfDaY) |"""
    )

    st.info(
        """
    Použijte tuto ukázku k:
    - Experimentování s různými hotovými modely a NLP balíčky.
    - Prozkoumání různých možností anonymizace, včetně redakce, maskování, šifrování a dalších.
    - Generování syntetického textu s Microsoft Presidio a OpenAI.
    - Konfiguraci povolených a zakázaných seznamů.
    
    Tato ukázková webová stránka ukazuje některé z možností Presidio.
    [Navštivte naši webovou stránku](https://microsoft.github.io/presidio) pro více informací,
    ukázek a možností nasazení.    
    """
    )

    st.markdown(
        "[![Pypi Downloads](https://img.shields.io/pypi/dm/presidio-analyzer.svg)](https://img.shields.io/pypi/dm/presidio-analyzer.svg)"  # noqa
        "[![MIT license](https://img.shields.io/badge/license-MIT-brightgreen.svg)](https://opensource.org/licenses/MIT)"
        "![GitHub Repo stars](https://img.shields.io/github/stars/microsoft/presidio?style=social)"
    )

analyzer_load_state = st.info("Spouštění Presidio analyzátoru...")

analyzer_load_state.empty()

# Načtení výchozího textu
with open("demo_text.txt", "r", encoding="utf-8") as f:
    demo_text = f.read()

# Vytvoření dvou sloupců pro před a po
col1, col2 = st.columns(2)

# Před:
col1.subheader("Vstup")
st_text = col1.text_area(
    label="Zadejte text", value=demo_text, height=400, key="text_input"
)

try:
    # Výběr entit
    st_entities_expander = st.sidebar.expander("Vyberte entity k vyhledání")
    st_entities = st_entities_expander.multiselect(
        label="Které entity hledat?",
        options=get_supported_entities(*analyzer_params),
        default=list(get_supported_entities(*analyzer_params)),
        help="Omezte seznam detekovaných osobních údajů. "
        "Tento seznam je dynamický a závisí na NER modelu a registrovaných rozpoznávačích. "
        "Více informací najdete zde: https://microsoft.github.io/presidio/analyzer/adding_recognizers/",
    )

    # Před
    analyzer_load_state = st.info("Spouštění Presidio analyzátoru...")
    analyzer = analyzer_engine(*analyzer_params)
    analyzer_load_state.empty()

    st_analyze_results = analyze(
        *analyzer_params,
        text=st_text,
        entities=st_entities,
        language="cs",
        score_threshold=st_threshold,
        return_decision_process=st_return_decision_process,
        allow_list=st_allow_list,
        deny_list=st_deny_list,
    )

    # Po
    if st_operator not in ("highlight", "synthesize"):
        with col2:
            st.subheader(f"Výstup")
            st_anonymize_results = anonymize(
                text=st_text,
                operator=st_operator,
                mask_char=st_mask_char,
                number_of_chars=st_number_of_chars,
                encrypt_key=st_encrypt_key,
                analyze_results=st_analyze_results,
            )
            st.text_area(
                label="Anonymizováno", value=st_anonymize_results.text, height=400
            )
    elif st_operator == "synthesize":
        with col2:
            st.subheader(f"OpenAI Generovaný výstup")
            fake_data = create_fake_data(
                st_text,
                st_analyze_results,
                open_ai_params,
            )
            st.text_area(label="Syntetická data", value=fake_data, height=400)
    else:
        st.subheader("Zvýrazněno")
        annotated_tokens = annotate(text=st_text, analyze_results=st_analyze_results)
        annotated_text(*annotated_tokens)

    # tabulka výsledků
    st.subheader(
        "Nálezy"
        if not st_return_decision_process
        else "Nálezy s rozhodovacími faktory"
    )
    if st_analyze_results:
        df = pd.DataFrame.from_records([r.to_dict() for r in st_analyze_results])
        df["text"] = [st_text[res.start : res.end] for res in st_analyze_results]

        df_subset = df[["entity_type", "text", "start", "end", "score"]].rename(
            {
                "entity_type": "Typ entity",
                "text": "Text",
                "start": "Začátek",
                "end": "Konec",
                "score": "Důvěryhodnost",
            },
            axis=1,
        )
       df_subset = pd.concat([df_subset, analysis_explanation_df], axis=1)
        st.dataframe(df_subset.reset_index(drop=True), use_container_width=True)
    else:
        st.text("Žádné nálezy")

except Exception as e:
    print(e)
    traceback.print_exc()
    st.error(e)

components.html(
    """
    <script type="text/javascript">
    (function(c,l,a,r,i,t,y){
        c[a]=c[a]||function(){(c[a].q=c[a].q||[]).push(arguments)};
        t=l.createElement(r);t.async=1;t.src="https://www.clarity.ms/tag/"+i;
        y=l.getElementsByTagName(r)[0];y.parentNode.insertBefore(t,y);
    })(window, document, "clarity", "script", "h7f8bp42n8");
    </script>
    """
)