adding presidio

app.py

@@ -1,21 +1,98 @@
+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
 st.title("Anonymise your text!")
 st.markdown(
-    "This mini-app anonymises text using Bert. You can find the code on [GitHub(WIP)](#)"
+    "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)
+@st.cache(suppress_st_warning=True, allow_output_mutation=True, show_spinner=False)
 def load_tagger():
     return SequenceTagger.load("flair/ner-english-large")
 
+@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)
+
+    return analyzer
+
+def analyze(**kwargs):
+    """Analyze input using Analyzer engine and input arguments (kwargs)."""
+    if "entities" not in kwargs or "All" in kwargs["entities"]:
+        kwargs["entities"] = None
+    return analyzer_engine().analyze(**kwargs)
+
+def annotate(text, analyze_results,st_entities):
+    tokens = []
+    # 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:])
+    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:
+        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}")
+            analyze_results = analyze(
+                text=text,
+                entities=st_entities,
+                language="en",
+                return_decision_process=False,
+            )
+            st.session_state.annotated_tokens = annotate(text, analyze_results,st_entities)
+           
+            # 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"
+            )
+
 def anonymise_text(text: str, metadata: str = "", white_listed_words: str = ""):
     """anonymise text"""
     if st.session_state.n_requests >= 50:
@@ -42,7 +119,8 @@ def anonymise_text(text: str, metadata: str = "", white_listed_words: str = ""):
             
             # else:
             # load tagger
-            tagger = load_tagger() 
+            tagger = load_tagger()
+            # tagger = load_tagger() 
             sentence = Sentence(text)
             # predict NER tags
             tagger.predict(sentence)
@@ -56,15 +134,16 @@ def anonymise_text(text: str, metadata: str = "", white_listed_words: str = ""):
             st.session_state.text_anon = text_anon
             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}"
             )
-# def anonymise_text(text: str, metadata: str = "", white_listed_words: str = ""):
-#     st.session_state.text_anon = "this is anonymised"
 
 if "text" not in st.session_state:
     st.session_state.text = ""
 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 "n_requests" not in st.session_state:
@@ -79,6 +158,14 @@ white_listed_words = st.text_input(
     label="Data to be ignored (optional)",
     placeholder="inspirational",
 )
+
+# button return true when clicked
+analyze_now = st.button(
+    label="Analyse text",
+    type="primary",
+    on_click=analyze_text,
+    args=(text,st_entities,),
+)
 # button return true when clicked
 anonymise_now = st.button(
     label="Anonymise text",
@@ -89,7 +176,9 @@ anonymise_now = st.button(
 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)

flair_recognizer.py

@@ -0,0 +1,219 @@
+import logging
+from typing import Optional, List, Tuple, Set
+
+from presidio_analyzer import (
+    RecognizerResult,
+    EntityRecognizer,
+    AnalysisExplanation,
+)
+from presidio_analyzer.nlp_engine import NlpArtifacts
+
+try:
+    from flair.data import Sentence
+    from flair.models import SequenceTagger
+except ImportError:
+    print("Flair is not installed")
+
+
+logger = logging.getLogger("presidio-analyzer")
+
+
+class FlairRecognizer(EntityRecognizer):
+    """
+    Wrapper for a flair model, if needed to be used within Presidio Analyzer.
+
+    :example:
+    >from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
+
+    >flair_recognizer = FlairRecognizer()
+
+    >registry = RecognizerRegistry()
+    >registry.add_recognizer(flair_recognizer)
+
+    >analyzer = AnalyzerEngine(registry=registry)
+
+    >results = analyzer.analyze(
+    >    "My name is Christopher and I live in Irbid.",
+    >    language="en",
+    >    return_decision_process=True,
+    >)
+    >for result in results:
+    >    print(result)
+    >    print(result.analysis_explanation)
+
+
+    """
+
+    ENTITIES = [
+        "LOCATION",
+        "PERSON",
+        "ORGANIZATION",
+        # "MISCELLANEOUS"   # - There are no direct correlation with Presidio entities.
+    ]
+
+    DEFAULT_EXPLANATION = "Identified as {} by Flair's Named Entity Recognition"
+
+    CHECK_LABEL_GROUPS = [
+        ({"LOCATION"}, {"LOC", "LOCATION"}),
+        ({"PERSON"}, {"PER", "PERSON"}),
+        ({"ORGANIZATION"}, {"ORG"}),
+        # ({"MISCELLANEOUS"}, {"MISC"}), # Probably not PII
+    ]
+
+    MODEL_LANGUAGES = {
+        "en": "flair/ner-english-large",
+        "es": "flair/ner-spanish-large",
+        "de": "flair/ner-german-large",
+        "nl": "flair/ner-dutch-large",
+    }
+
+    PRESIDIO_EQUIVALENCES = {
+        "PER": "PERSON",
+        "LOC": "LOCATION",
+        "ORG": "ORGANIZATION",
+        # 'MISC': 'MISCELLANEOUS'   # - Probably not PII
+    }
+
+    def __init__(
+        self,
+        supported_language: str = "en",
+        supported_entities: Optional[List[str]] = None,
+        check_label_groups: Optional[Tuple[Set, Set]] = None,
+        model: SequenceTagger = None,
+    ):
+        self.check_label_groups = (
+            check_label_groups if check_label_groups else self.CHECK_LABEL_GROUPS
+        )
+
+        supported_entities = supported_entities if supported_entities else self.ENTITIES
+        self.model = (
+            model
+            if model
+            else SequenceTagger.load(self.MODEL_LANGUAGES.get(supported_language))
+        )
+
+        super().__init__(
+            supported_entities=supported_entities,
+            supported_language=supported_language,
+            name="Flair Analytics",
+        )
+
+    def load(self) -> None:
+        """Load the model, not used. Model is loaded during initialization."""
+        pass
+
+    def get_supported_entities(self) -> List[str]:
+        """
+        Return supported entities by this model.
+
+        :return: List of the supported entities.
+        """
+        return self.supported_entities
+
+    # Class to use Flair with Presidio as an external recognizer.
+    def analyze(
+        self, text: str, entities: List[str], nlp_artifacts: NlpArtifacts = None
+    ) -> List[RecognizerResult]:
+        """
+        Analyze text using Text Analytics.
+
+        :param text: The text for analysis.
+        :param entities: Not working properly for this recognizer.
+        :param nlp_artifacts: Not used by this recognizer.
+        :param language: Text language. Supported languages in MODEL_LANGUAGES
+        :return: The list of Presidio RecognizerResult constructed from the recognized
+            Flair detections.
+        """
+
+        results = []
+
+        sentences = Sentence(text)
+        self.model.predict(sentences)
+
+        # If there are no specific list of entities, we will look for all of it.
+        if not entities:
+            entities = self.supported_entities
+
+        for entity in entities:
+            if entity not in self.supported_entities:
+                continue
+
+            for ent in sentences.get_spans("ner"):
+                if not self.__check_label(
+                    entity, ent.labels[0].value, self.check_label_groups
+                ):
+                    continue
+                textual_explanation = self.DEFAULT_EXPLANATION.format(
+                    ent.labels[0].value
+                )
+                explanation = self.build_flair_explanation(
+                    round(ent.score, 2), textual_explanation
+                )
+                flair_result = self._convert_to_recognizer_result(ent, explanation)
+
+                results.append(flair_result)
+
+        return results
+
+    def _convert_to_recognizer_result(self, entity, explanation) -> RecognizerResult:
+
+        entity_type = self.PRESIDIO_EQUIVALENCES.get(entity.tag, entity.tag)
+        flair_score = round(entity.score, 2)
+
+        flair_results = RecognizerResult(
+            entity_type=entity_type,
+            start=entity.start_position,
+            end=entity.end_position,
+            score=flair_score,
+            analysis_explanation=explanation,
+        )
+
+        return flair_results
+
+    def build_flair_explanation(
+        self, original_score: float, explanation: str
+    ) -> AnalysisExplanation:
+        """
+        Create explanation for why this result was detected.
+
+        :param original_score: Score given by this recognizer
+        :param explanation: Explanation string
+        :return:
+        """
+        explanation = AnalysisExplanation(
+            recognizer=self.__class__.__name__,
+            original_score=original_score,
+            textual_explanation=explanation,
+        )
+        return explanation
+
+    @staticmethod
+    def __check_label(
+        entity: str, label: str, check_label_groups: Tuple[Set, Set]
+    ) -> bool:
+        return any(
+            [entity in egrp and label in lgrp for egrp, lgrp in check_label_groups]
+        )
+
+
+if __name__ == "__main__":
+
+    from presidio_analyzer import AnalyzerEngine, RecognizerRegistry
+
+    flair_recognizer = (
+        FlairRecognizer()
+    )  # This would download a very large (+2GB) model on the first run
+
+    registry = RecognizerRegistry()
+    registry.add_recognizer(flair_recognizer)
+
+    analyzer = AnalyzerEngine(registry=registry)
+
+    results = analyzer.analyze(
+        "My name is Christopher and I live in Irbid.",
+        language="en",
+        return_decision_process=True,
+    )
+    for result in results:
+        print(result)
+        print(result.analysis_explanation)

requirements.txt

@@ -1 +1,6 @@
-flair==0.11
+flair==0.11
+presidio-anonymizer
+presidio-analyzer
+st-annotated-text
+spacy>=3.0.0,<4.0.0 
+https://github.com/explosion/spacy-models/releases/download/en_core_web_lg-3.0.0/en_core_web_lg-3.0.0.tar.gz#egg=en_core_web_lg