Spaces:
Sleeping
Sleeping
| import logging | |
| import os | |
| import base64 | |
| import datetime | |
| import dotenv | |
| import pandas as pd | |
| import streamlit as st | |
| from streamlit_tags import st_tags | |
| from PyPDF2 import PdfReader, PdfWriter | |
| from presidio_helpers import ( | |
| analyzer_engine, | |
| get_supported_entities, | |
| analyze, | |
| anonymize, | |
| ) | |
| st.set_page_config( | |
| page_title="Presidio PHI De-identification", | |
| layout="wide", | |
| initial_sidebar_state="expanded", | |
| menu_items={"About": "https://microsoft.github.io/presidio/"}, | |
| ) | |
| dotenv.load_dotenv() | |
| logger = logging.getLogger("presidio-streamlit") | |
| # Sidebar | |
| st.sidebar.header("PHI De-identification with Presidio") | |
| model_help_text = "Select Named Entity Recognition (NER) model for PHI detection." | |
| model_list = [ | |
| ("flair/ner-english-large", "https://huggingface.co/flair/ner-english-large"), | |
| ("HuggingFace/obi/deid_roberta_i2b2", "https://huggingface.co/obi/deid_roberta_i2b2"), | |
| ("HuggingFace/StanfordAIMI/stanford-deidentifier-base", "https://huggingface.co/StanfordAIMI/stanford-deidentifier-base"), | |
| ] | |
| st_model = st.sidebar.selectbox( | |
| "NER model package", | |
| [model[0] for model in model_list], | |
| index=0, | |
| help=model_help_text, | |
| ) | |
| # Display HuggingFace link for selected model | |
| selected_model_url = next(url for model, url in model_list if model == st_model) | |
| st.sidebar.markdown(f"[View model on HuggingFace]({selected_model_url})") | |
| # Extract model package | |
| st_model_package = st_model.split("/")[0] | |
| st_model = st_model if st_model_package.lower() not in ("huggingface") else "/".join(st_model.split("/")[1:]) | |
| analyzer_params = (st_model_package, st_model) | |
| st.sidebar.warning("Note: Models might take some time to download on first run.") | |
| st_operator = st.sidebar.selectbox( | |
| "De-identification approach", | |
| ["replace", "redact", "mask"], | |
| index=0, | |
| help="Select PHI manipulation method.", | |
| ) | |
| st_threshold = st.sidebar.slider( | |
| label="Acceptance threshold", | |
| min_value=0.0, | |
| max_value=1.0, | |
| value=0.35, | |
| ) | |
| st_return_decision_process = st.sidebar.checkbox( | |
| "Add analysis explanations", | |
| value=False, | |
| ) | |
| # Allow and deny lists | |
| with st.sidebar.expander("Allowlists and denylists", expanded=False): | |
| st_allow_list = st_tags(label="Add words to allowlist", text="Enter word and press enter.") | |
| st_deny_list = st_tags(label="Add words to denylist", text="Enter word and press enter.") | |
| # PDF processing functions | |
| def get_timestamp_prefix(): | |
| central = pytz.timezone("US/Central") | |
| now = datetime.now(central) | |
| return now.strftime("%I%M%p_%d-%m-%y").upper() | |
| def save_pdf(pdf_input): | |
| """Save uploaded PDF to disk.""" | |
| try: | |
| original_name = pdf_input.name | |
| with open(original_name, "wb") as f: | |
| f.write(pdf_input.read()) | |
| return original_name | |
| except Exception as e: | |
| st.error(f"Failed to save PDF: {str(e)}") | |
| return None | |
| def read_pdf(pdf_path): | |
| """Read text from a PDF using PyPDF2.""" | |
| try: | |
| reader = PdfReader(pdf_path) | |
| text = "" | |
| for page in reader.pages: | |
| page_text = page.extract_text() or "" | |
| text += page_text + "\n" | |
| return text | |
| except Exception as e: | |
| st.error(f"Failed to read PDF: {str(e)}") | |
| return None | |
| def create_pdf(text, input_path, output_filename): | |
| """Create a PDF with anonymized text using PyPDF2.""" | |
| try: | |
| reader = PdfReader(input_path) | |
| writer = PdfWriter() | |
| for page in reader.pages: | |
| writer.add_page(page) | |
| with open(output_filename, "wb") as f: | |
| writer.write(f) | |
| return output_filename | |
| except Exception as e: | |
| st.error(f"Failed to create PDF: {str(e)}") | |
| return None | |
| # Main panel | |
| col1, col2 = st.columns(2) | |
| with col1: | |
| st.subheader("Input") | |
| uploaded_file = st.file_uploader("Upload PDF", type=["pdf"]) | |
| if uploaded_file: | |
| try: | |
| # Save PDF to disk | |
| pdf_path = save_pdf(uploaded_file) | |
| if not pdf_path: | |
| raise ValueError("Failed to save PDF") | |
| # Read PDF | |
| text = read_pdf(pdf_path) | |
| if not text: | |
| raise ValueError("No text extracted from PDF") | |
| # Initialize analyzer | |
| try: | |
| analyzer = analyzer_engine(*analyzer_params) | |
| except Exception as e: | |
| st.error(f"Failed to load model: {str(e)}") | |
| st.info("Ensure models are downloaded and check network/permissions.") | |
| raise | |
| # Analyze | |
| st_analyze_results = analyze( | |
| analyzer=analyzer, | |
| text=text, | |
| entities=get_supported_entities(*analyzer_params), | |
| language="en", | |
| score_threshold=st_threshold, | |
| return_decision_process=st_return_decision_process, | |
| allow_list=st_allow_list, | |
| deny_list=st_deny_list, | |
| ) | |
| # Process results | |
| phi_types = set(res.entity_type for res in st_analyze_results) | |
| if phi_types: | |
| st.success(f"Removed PHI types: {', '.join(phi_types)}") | |
| else: | |
| st.info("No PHI detected") | |
| # Anonymize | |
| anonymized_result = anonymize( | |
| text=text, | |
| operator=st_operator, | |
| analyze_results=st_analyze_results, | |
| ) | |
| # Generate output filename with timestamp | |
| timestamp = get_timestamp_prefix() | |
| output_filename = f"{timestamp}_{uploaded_file.name}" | |
| # Create new PDF | |
| pdf_output = create_pdf(anonymized_result.text, pdf_path, output_filename) | |
| if not pdf_output: | |
| raise ValueError("Failed to generate PDF") | |
| # Generate base64 download link | |
| try: | |
| with open(output_filename, "rb") as f: | |
| pdf_bytes = f.read() | |
| b64 = base64.b64encode(pdf_bytes).decode() | |
| href = f'<a href="data:application/pdf;base64,{b64}" download="{output_filename}">Download de-identified PDF</a>' | |
| st.markdown(href, unsafe_allow_html=True) | |
| except Exception as e: | |
| st.error(f"Error generating download link: {str(e)}") | |
| raise | |
| # Display findings | |
| with col2: | |
| st.subheader("Findings") | |
| if st_analyze_results: | |
| df = pd.DataFrame.from_records([r.to_dict() for r in st_analyze_results]) | |
| df["text"] = [text[res.start:res.end] for res in st_analyze_results] | |
| df_subset = df[["entity_type", "text", "start", "end", "score"]].rename( | |
| { | |
| "entity_type": "Entity type", | |
| "text": "Text", | |
| "start": "Start", | |
| "end": "End", | |
| "score": "Confidence", | |
| }, | |
| 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("No findings") | |
| # Clean up temporary file | |
| if os.path.exists(pdf_path): | |
| os.remove(pdf_path) | |
| except Exception as e: | |
| st.error(f"An error occurred: {str(e)}") | |
| logger.error(f"Processing error: {str(e)}") |