presidio_streamlit.py

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,
)

# Načtení proměnných prostředí
dotenv.load_dotenv()

# Konfigurace loggeru
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger("presidio-streamlit")

# Nastavení stránky Streamlit
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/",
    },
)

# Načtení konfiguračních proměnných
allow_other_models = os.getenv("ALLOW_OTHER_MODELS", "False").lower() == "true"

# Sidebar
st.sidebar.header("Anonymizace osobních údajů v českých textech s Microsoft Presidio")

# Výběr modelu
model_help_text = "Vyberte model pro rozpoznávání pojmenovaných entit (NER) pro detekci osobních údajů."
model_list = [
    "spacy/cs_core_news_sm",
    "iiiorg/piiranha-v1-detect-personal-information",
    "FacebookAI/xlm-roberta-large-finetuned-conll03-english",
]
if not allow_other_models:
    model_list.pop()

st_model = st.sidebar.selectbox("NER model", model_list, index=1, help=model_help_text)

# Zpracování výběru modelu
st_model_package = st_model.split("/")[0]
st_model = "/".join(st_model.split("/")[1:]) if st_model_package.lower() in ("spacy", "iiiorg") else st_model

st_ta_key = st_ta_endpoint = ""  # Placeholder pro případné použití Text Analytics
analyzer_params = (st_model_package, st_model, st_ta_key, st_ta_endpoint)
logger.debug(f"analyzer_params: {analyzer_params}")

# Výběr metody anonymizace
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ů."
)

# Nastavení pro jednotlivé metody anonymizace
st_mask_char = "*"
st_number_of_chars = 15
st_encrypt_key = "WmZq4t7w!z%C&F)J"
open_ai_params = None

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":
    # Implementace nastavení pro OpenAI syntézu
    pass

# Nastavení prahu důvěry
st_threshold = st.sidebar.slider("Práh přijetí", min_value=0.0, max_value=1.0, value=0.35)

# Volba pro zobrazení rozhodovacího procesu
st_return_decision_process = st.sidebar.checkbox("Přidat vysvětlení analýzy k nálezům", value=False)

# Povolené a zakázané seznamy
with st.sidebar.expander("Povolené a zakázané seznamy", expanded=False):
    st_allow_list = st_tags(label="Přidat slova do povoleného seznamu", text="Zadejte slovo a stiskněte enter.")
    st_deny_list = st_tags(label="Přidat slova do zakázaného seznamu", text="Zadejte slovo a stiskněte enter.")

# Hlavní část aplikace
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ů.")
    # Další informace o aplikaci...

# 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 vstup a výstup
col1, col2 = st.columns(2)

# Vstupní text
with col1:
    st.subheader("Vstup")
    st_text = st.text_area(label="Zadejte text", value=demo_text, height=400, key="text_input")

try:
    # Výběr entit k detekci
    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ů."
    )

    # Analýza textu
    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,
    )

    # Zobrazení výsledků
    with col2:
        st.subheader("Výstup")
        if st_operator not in ("highlight", "synthesize"):
            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":
            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)

    # Zobrazení tabulky s výsledky
    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,
        )
        if st_return_decision_process:
            analysis_explanation_df = pd.DataFrame.from_records(
                [r.analysis_explanation.to_dict() for r in st_analyze_results]
            )
            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:
    logger.error(f"Došlo k chybě: {str(e)}", exc_info=True)
    st.error(f"Došlo k chybě při zpracování: {str(e)}")

# Přidání skriptu pro sledování využití (volitelné)
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>
    """
)