Initial commit: FastAPI app, model, utils, Dockerfile

Dockerfile

@@ -0,0 +1,19 @@
+FROM python:3.10-slim
+
+# 1) Set working directory
+WORKDIR /app
+
+# 2) Copy everything into the container
+COPY . /app
+
+# 3) Install requirements
+RUN pip install --no-cache-dir -r requirements.txt
+
+# 4) Download spaCy model
+RUN python -m spacy download en_core_web_sm
+
+# 5) Expose port 7860 (FastAPI default)
+EXPOSE 7860
+
+# 6) Start Uvicorn with your FastAPI app
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

main.py

@@ -0,0 +1,29 @@
+from fastapi import FastAPI
+from pydantic import BaseModel
+from utils import mask_pii, demask_pii
+from models import load_model, classify_email
+
+app = FastAPI()
+model = load_model()
+
+class EmailRequest(BaseModel):
+    input_email_body: str
+
+@app.post("/classify")
+def classify(request: EmailRequest):
+    # Mask PII
+    masked_email, entities = mask_pii(request.input_email_body)
+    
+    # Classify email
+    category = classify_email(model, masked_email)
+    
+    # Demask back to original
+    demasked_email = demask_pii(masked_email, entities)
+    
+    return {
+        "input_email_body": request.input_email_body,
+        "list_of_masked_entities": entities,
+        "masked_email": masked_email,
+        "category_of_the_email": category,
+        "demasked_email": demasked_email
+    }

models.py

@@ -0,0 +1,15 @@
+from huggingface_hub import hf_hub_download
+from joblib import load
+from sklearn.pipeline import Pipeline
+
+import os
+os.environ["HF_HUB_ENABLE_HF_TRANSFER"] = "1"  # Add at top
+MODEL_REPO = "adrian7305/email-classifier"
+MODEL_FILE = "model.joblib"
+
+def load_model() -> Pipeline:
+    model_path = hf_hub_download(repo_id=MODEL_REPO, filename=MODEL_FILE)
+    return load(model_path)
+
+def classify_email(model: Pipeline, email: str) -> str:
+    return str(model.predict([email])[0])

requirements.txt

@@ -0,0 +1,11 @@
+fastapi==0.109.0
+uvicorn==0.27.0
+python-multipart==0.0.9
+spacy==3.7.4
+scikit-learn==1.6.1
+joblib==1.3.2
+regex==2023.12.25
+pandas==2.1.4
+huggingface-hub==0.20.3
+python-dotenv==1.0.0
+https://github.com/explosion/spacy-models/releases/download/en_core_web_sm-3.7.0/en_core_web_sm-3.7.0-py3-none-any.whl

utils.py

@@ -0,0 +1,74 @@
+import re
+import spacy
+from typing import Tuple, List, Dict
+
+nlp = spacy.load("en_core_web_sm")
+
+PATTERN_ORDER = [
+    ("credit_debit_no", r"\b(?:\d[ -]*?){13,19}\b"),
+    ("aadhar_num",      r"\b\d{4}[\s-]?\d{4}[\s-]?\d{4}\b"),
+    ("phone_number",    r"(?:(?:\+91|0)[-\s]?)?[6-9]\d{4}[-\s]?\d{5}"),
+    ("email",           r"\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Za-z]{2,}\b"),
+    ("dob",             r"\b(?:0?[1-9]|1[0-2])[\/-](?:0?[1-9]|[12][0-9]|3[01])[\/-](?:\d{4}|\d{2})\b"),
+    ("expiry_no",       r"\b(?:0[1-9]|1[0-2])[\/-]?(?:\d{2}|\d{4})\b"),
+    ("cvv_no",          r"\b\d{3,4}\b"),
+]
+
+def mask_pii(text: str) -> Tuple[str, List[Dict]]:
+    entities: List[Dict] = []
+    occupied_spans: List[Tuple[int, int]] = []
+    masked_text = text
+
+    def overlaps_existing(start: int, end: int) -> bool:
+        for os_, oe_ in occupied_spans:
+            if not (end <= os_ or start >= oe_):
+                return True
+        return False
+
+    for pii_type, pattern in PATTERN_ORDER:
+        for match in re.finditer(pattern, text):
+            start, end = match.span()
+            if not overlaps_existing(start, end):
+                entities.append({
+                    "position": [start, end],
+                    "classification": pii_type,
+                    "entity": text[start:end]
+                })
+                occupied_spans.append((start, end))
+
+    doc = nlp(text)
+    for ent in doc.ents:
+        if ent.label_ == "PERSON":
+            start, end = ent.start_char, ent.end_char
+            if not overlaps_existing(start, end):
+                entities.append({
+                    "position": [start, end],
+                    "classification": "full_name",
+                    "entity": ent.text
+                })
+                occupied_spans.append((start, end))
+
+    entities.sort(key=lambda x: x["position"][0], reverse=True)
+
+    for entity in entities:
+        start, end = entity["position"]
+        placeholder = f"[{entity['classification']}]"
+        masked_text = masked_text[:start] + placeholder + masked_text[end:]
+
+    return masked_text, entities
+def demask_pii(masked_text: str, entities: List[Dict]) -> str:
+    """
+    Given `masked_text` (with placeholders like “[email]”) and the
+    `entities` list (each entry has position, classification, entity),
+    restore the original substrings at their exact positions.
+    """
+    result = masked_text
+    # Sort in ascending order of start‐index, so that earlier replacements
+    # don’t break the indices of later ones.
+    for ent in sorted(entities, key=lambda x: x["position"][0]):
+        start, end = ent["position"]
+        placeholder = f"[{ent['classification']}]"
+        original = ent["entity"]
+        # Replace the placeholder at the exact location with the original text.
+        result = result[:start] + original + result[start + len(placeholder):]
+    return result