adding meatdata and allowed lists

app.py

@@ -1,35 +1,33 @@
 import spacy
 import streamlit as st
-from flair.data import Sentence
-from flair.models import SequenceTagger
 import re 
 import logging
-from presidio_analyzer.nlp_engine import NlpEngineProvider
 from presidio_anonymizer import AnonymizerEngine
 from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
 from annotated_text import annotated_text
 from flair_recognizer import FlairRecognizer
 
-# Render Streamlit page
+
+###############################
+#### Render Streamlit page ####
+###############################
+
 st.title("Anonymise your text!")
 st.markdown(
     "This mini-app anonymises text using Flair. You can find the code on [GitHub(WIP)](#)"
 )
+
 # Configure logger
 logging.basicConfig(format="\n%(asctime)s\n%(message)s", level=logging.INFO, force=True)
 
-@st.cache(suppress_st_warning=True, allow_output_mutation=True, show_spinner=False)
-def load_tagger():
-    return SequenceTagger.load("flair/ner-english-large")
+
+##############################
+###### Define functions ######
+##############################
 
 @st.cache(allow_output_mutation=True,show_spinner=False)
 def analyzer_engine():
     """Return AnalyzerEngine."""
-    # registry = RecognizerRegistry()
-    # flair_recognizer = FlairRecognizer()
-    # registry.load_predefined_recognizers()
-    # registry.add_recognizer(flair_recognizer)
-    # analyzer = AnalyzerEngine(registry=registry, supported_languages=["en"])
     analyzer = AnalyzerEngine()
     flair_recognizer = FlairRecognizer()
     analyzer.registry.add_recognizer(flair_recognizer)
@@ -42,143 +40,215 @@ def analyze(**kwargs):
         kwargs["entities"] = None
     return analyzer_engine().analyze(**kwargs)
 
-def annotate(text, analyze_results,st_entities):
+def annotate():
+    text = st.session_state.text 
+    analyze_results = st.session_state.analyze_results
     tokens = []
+    starts=[]
     # sort by start index
     results = sorted(analyze_results, key=lambda x: x.start)
     for i, res in enumerate(results):
-        if i == 0:
-            tokens.append(text[:res.start])
-
-        # append entity text and entity type
-        tokens.append((text[res.start: res.end], res.entity_type))
-
-        # if another entity coming i.e. we're not at the last results element, add text up to next entity
-        if i != len(results) - 1:
-            tokens.append(text[res.end:results[i+1].start])
-        # if no more entities coming, add all remaining text
-        else:
-            tokens.append(text[res.end:])
+        # if we already have an entity for this token don't add another
+        if res.start not in starts:
+            if i == 0:
+                tokens.append(text[:res.start])
+
+            # append entity text and entity type
+            tokens.append((text[res.start: res.end], res.entity_type))
+
+            # if another entity coming i.e. we're not at the last results element, add text up to next entity
+            if i != len(results) - 1:
+                tokens.append(text[res.end:results[i+1].start])
+            # if no more entities coming, add all remaining text
+            else:
+                tokens.append(text[res.end:])
+
+            # append this token to the list so we don't repeat results per token
+            starts.append(res.start)
     return tokens
 
 def get_supported_entities():
     """Return supported entities from the Analyzer Engine."""
     return analyzer_engine().get_supported_entities()
 
-st_entities = st.sidebar.multiselect(
-    label="Which entities to look for?",
-    options=get_supported_entities(),
-    default=list(get_supported_entities()),
-)
-
-def analyze_text(text: str, st_entities: str):
-    if not text:
+def analyze_text():
+    if not st.session_state.text:
         st.session_state.text_error = "Please enter your text"
         return
-    
+
     with text_spinner_placeholder:
         with st.spinner("Please wait while your text is being analysed..."):
-            logging.info(f"This is the text being analysed: {text}")
+            logging.info(f"This is the text being analysed: {st.session_state.text}")
+            st.session_state.text_error = ""
+            st.session_state.n_requests += 1
             analyze_results = analyze(
-                text=text,
+                text=st.session_state.text,
                 entities=st_entities,
                 language="en",
                 return_decision_process=False,
             )
-            st.session_state.annotated_tokens = annotate(text, analyze_results,st_entities)
+
+            # if st.session_state.metadata:
+            #     analyze_results = include_manual_input(analyze_results)
+
+            if st.session_state.allowed_words:
+                analyze_results = exclude_manual_input(analyze_results)
+
+            st.session_state.analyze_results = analyze_results
            
-            # st.session_state.text_analys=annotated_text(*annotated_tokens)
             logging.info(
-                f"text: {text}{metadata}{white_listed_words}\n"
-                f"tokens: {st.session_state.annotated_tokens}\n"
+                f"analyse results: {st.session_state.analyze_results}\n"
+            )
+            
+
+# def include_manual_input(analyze_results):
+#     analyze_results_extended=[]
+#     logging.info(
+#                 f"analyse results before adding extra words: {analyze_results}\n"
+#             )
+#     for word in st.session_state.text.split():
+#         if word in st.session_state.metadata:
+#             [m.start() for m in re.finditer('test', 'test test test test')]
+#             analyze_results_extended.append("type: MANUAL, start: 0, end: 3, score: 1.0")
+#     logging.info(
+#                 f"analyse results after adding allowed words: {analyze_results_extended}\n"
+#             )
+#     return analyze_results
+
+def exclude_manual_input(analyze_results):
+    analyze_results_fltered=[]
+    logging.info(
+                f"analyse results before removing allowed words: {analyze_results}\n"
             )
+    for token in analyze_results:
+        if st.session_state.text[token.start:token.end] not in st.session_state.allowed_words:
+            analyze_results_fltered.append(token)
+    logging.info(
+                f"analyse results after removing allowed words: {analyze_results_fltered}\n"
+            )
+    return analyze_results_fltered
+
+
+@st.cache(allow_output_mutation=True)
+def anonymizer_engine():
+    """Return AnonymizerEngine."""
+    return AnonymizerEngine()
 
-def anonymise_text(text: str, metadata: str = "", white_listed_words: str = ""):
-    """anonymise text"""
+def anonymise_text():
     if st.session_state.n_requests >= 50:
         st.session_state.text_error = "Too many requests. Please wait a few seconds before anonymising more text."
         logging.info(f"Session request limit reached: {st.session_state.n_requests}")
         st.session_state.n_requests = 1
-        return
 
-    st.session_state.text = ""
     st.session_state.text_error = ""
 
-    if not text:
+    if not st.session_state.text:
         st.session_state.text_error = "Please enter your text"
         return
-    
+
+    if not st.session_state.analyze_results:
+        analyze_text()
+
     with text_spinner_placeholder:
         with st.spinner("Please wait while your text is being anonymised..."):
-            
-            # flagged = openai.moderate(prompt)
-            # if flagged:
-            #     st.session_state.text_error = "Input flagged as inappropriate."
-            #     logging.info(f"Topic: {topic}{mood_output}{style_output}\n")
-            #     return
-            
-            # else:
-            # load tagger
-            tagger = load_tagger()
-            # tagger = load_tagger() 
-            sentence = Sentence(text)
-            # predict NER tags
-            tagger.predict(sentence)
-            # iterate over entities and redact
-            enitities=[e.text for e in sentence.get_spans('ner')]
-            regex = re.compile('|'.join(map(re.escape, enitities)))
-            text_anon = regex.sub("<PID>", text)
-            
+            anon_results = anonymizer_engine().anonymize(st.session_state.text, st.session_state.analyze_results)
             st.session_state.text_error = ""
             st.session_state.n_requests += 1
-            st.session_state.text_anon = text_anon
+            st.session_state.anon_results = anon_results
             logging.info(
-                f"text: {text}{metadata}{white_listed_words}\n"
-                f"entities: {sentence.get_spans('ner')}\n"
-                f"text anonymised: {st.session_state.text_anon}"
+                f"text anonymised: {st.session_state.anon_results}"
             )
 
-if "text" not in st.session_state:
-    st.session_state.text = ""
+def clear_results():
+    st.session_state.anon_results=""
+    st.session_state.analyze_results=""
+
+##############################
+#### Initialize variables ####
+##############################
+
 if "text_error" not in st.session_state:
     st.session_state.text_error = ""
-if "annotated_tokens" not in st.session_state:
-    st.session_state.annotated_tokens = ""
-if "text_anon" not in st.session_state:
-    st.session_state.text_anon = ""
+if "analyze_results" not in st.session_state:
+    st.session_state.analyze_results = ""
+if "anon_results" not in st.session_state:
+    st.session_state.anon_results = ""
 if "n_requests" not in st.session_state:
     st.session_state.n_requests = 0
 
-text = st.text_input(label="Text to be anonymised", placeholder="Write your text here")
-metadata = st.text_input(
+##############################
+####### Page arguments #######
+##############################
+
+# Every widget with a key is automatically added to Session State
+
+# In Streamlit, interacting with a widget triggers a rerun and variables defined
+# in the code get reinitialized after each rerun.
+
+# If a callback function is associated with a widget then a change in the widget 
+# triggers the following sequence: First the callback function is executed and then
+# the app executes from top to bottom.
+
+st.text_input(
+    label="Text", 
+    placeholder="Write your text here", 
+    key='text',
+    on_change=clear_results
+)
+st.text_input(
     label="Data to be redacted (optional)",
-    placeholder="inspirational",
+    placeholder="John, Mary, London",
+    key='metadata',
+    on_change=clear_results
 )
-white_listed_words = st.text_input(
+st.text_input(
     label="Data to be ignored (optional)",
-    placeholder="inspirational",
+    placeholder="NHS, GEL, Lab",
+    key='allowed_words',
+    on_change=clear_results
 )
 
-# button return true when clicked
-analyze_now = st.button(
-    label="Analyse text",
-    type="primary",
-    on_click=analyze_text,
-    args=(text,st_entities,),
+st_entities = st.sidebar.multiselect(
+    label="Which entities to look for?",
+    options=get_supported_entities(),
+    default=list(get_supported_entities()),
 )
+
+##############################
+######## Page buttons ########
+##############################
+
 # button return true when clicked
-anonymise_now = st.button(
-    label="Anonymise text",
-    type="primary",
-    on_click=anonymise_text,
-    args=(text, metadata, white_listed_words),
-)
+
+col1, col2 = st.columns(2)
+
+with col1:
+    analyze_now = st.button(
+        label="Analyse text",
+        type="primary",
+        on_click=analyze_text,
+    )
+
+with col2:
+    anonymise_now = st.button(
+        label="Anonymise text",
+        type="primary",
+        on_click=anonymise_text,
+    )
+
+##############################
+######## Page actions ########
+##############################
+
 text_spinner_placeholder = st.empty()
 if st.session_state.text_error:
     st.error(st.session_state.text_error)
-if analyze_now:
-     # annotated_tokens
-    annotated_text(*st.session_state.annotated_tokens)
-if st.session_state.text_anon:
-    st.markdown("""---""")
-    st.text_area(label="Text anonymised", value=st.session_state.text_anon, height=100)
+
+with col1:
+    if st.session_state.analyze_results:
+        annotated_tokens=annotate()
+        annotated_text(*annotated_tokens)
+        st.write(st.session_state.analyze_results)
+with col2:
+    if st.session_state.anon_results:
+        st.write(st.session_state.anon_results.text)