Implement a simple CLI

.envrc

@@ -1 +1,5 @@
+if ! has nix_direnv_version || ! nix_direnv_version 3.0.6; then
+    source_url "https://raw.githubusercontent.com/nix-community/nix-direnv/3.0.6/direnvrc" "sha256-RYcUJaRMf8oF5LznDrlCXbkOQrywm0HDv1VjYGaJGdM="
+fi
+
 use flake

src/cli.py

@@ -0,0 +1,71 @@
+import argparse
+from presidio_analyzer.analyzer_engine import AnalyzerEngine
+from presidio_anonymizer.anonymizer_engine import AnonymizerEngine
+from analyzer_engine.csv_analyzer_engine import CSVAnalyzerEngine
+from presidio_anonymizer import BatchAnonymizerEngine
+from config.nlp_engine_config import FlairNLPEngine
+
+NLP_ENGINE = "flair/ner-english-large"
+
+
+def analyze(args):
+    nlp_engine = FlairNLPEngine(NLP_ENGINE)
+    analyzer_results = None
+
+    if args.text:
+        nlp_engine, registry = nlp_engine.create_nlp_engine()
+        engine = AnalyzerEngine(registry=registry, nlp_engine=nlp_engine)
+
+        analyzer_results = engine.analyze(
+            text=args.text,
+            language=args.language
+        )
+    else:
+        engine = CSVAnalyzerEngine(nlp_engine)
+
+        analyzer_results = engine.analyze_csv(
+            csv_full_path=args.filepath,
+            language=args.language
+        )
+
+    print(analyzer_results)
+    return analyzer_results
+
+
+def anonymize(args):
+    analyzer_results = analyze(args)
+    anonymized_results = None
+
+    if args.text:
+        anonymizer = AnonymizerEngine()
+        anonymized_results = anonymizer.anonymize(args.text, analyzer_results)
+    else:
+        anonymizer = BatchAnonymizerEngine()
+        anonymized_results = anonymizer.anonymize_dict(analyzer_results)
+
+    print(anonymized_results)
+    return anonymized_results
+
+
+def main():
+    parser = argparse.ArgumentParser(description="A CLI for detecting and Anonymizing PII.")
+    subparsers = parser.add_subparsers(required=True)
+
+    analyzer_parser = subparsers.add_parser("analyze", description="Analyze inputs and return PII detection results")
+    analyzer_parser.add_argument("--filepath", required=False, type=str, metavar="FILE")
+    analyzer_parser.add_argument("--text", required=False, type=str)
+    analyzer_parser.add_argument("--language", required=False, type=str, default="en")
+    analyzer_parser.set_defaults(func=analyze)
+
+    anonymizer_parser = subparsers.add_parser("anonymize", description="Anonymize inputs")
+    anonymizer_parser.add_argument("--filepath", required=False, type=str, metavar="FILE")
+    anonymizer_parser.add_argument("--text", required=False, type=str)
+    anonymizer_parser.add_argument("--language", required=False, type=str, default="en")
+    anonymizer_parser.set_defaults(func=anonymize)
+
+    args = parser.parse_args()
+    args.func(args)
+
+
+if __name__ == '__main__':
+    main()

src/recognizer/flair_recognizer.py

@@ -4,6 +4,7 @@
 2. https://huggingface.co/spaces/presidio/presidio_demo/blob/main/flair_recognizer.py
 
 '''
+import logging
 from typing import Optional, List, Tuple, Set
 
 from presidio_analyzer import (
@@ -62,10 +63,10 @@ def __init__(
         elif model and not model_path:
             self.model = model
         elif not model and model_path:
-            print(f"Loading model from {model_path}")
+            logging.info(f"Loading model from {model_path}")
             self.model = SequenceTagger.load(model_path)
         else:
-            print(f"Loading model for language {supported_language}")
+            logging.info(f"Loading model for language {supported_language}")
             self.model = SequenceTagger.load(
                 self.MODEL_LANGUAGES.get(supported_language)
             )