Fix circular import and NameError issues

api.py

@@ -1,26 +1,33 @@
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel, Field
-from typing import List, Dict, Tuple, Any, Optional
-import os  # To potentially load environment variables if needed
+from typing import List, Dict, Tuple, Any, Optional, Union
+import sys  # Add sys import
+import os  # Add os import
 
-# Make sure these imports work and don't cause errors themselves
+# --- Define/Import Pipeline Type FIRST ---
 try:
-    from utils import mask_pii
-    from models import load_model_pipeline, predict_category, Pipeline  # Ensure Pipeline is imported if type hint used
+    from sklearn.pipeline import Pipeline
+except ImportError:
+    Pipeline = object  # type: ignore
+
+# Add the project root to the Python path to allow imports
+# sys.path.append(os.path.dirname(os.path.abspath(__file__))) # This might not be needed depending on execution context
+
+# Import the processing function from utils
+try:
+    # Assuming utils.py is in the same directory or PYTHONPATH is set correctly
+    from utils import process_email_request, load_spacy_model, load_model_pipeline  # Import necessary functions
 except ImportError as e:
-    print(f"Error importing from utils or models in api.py: {e}")
-    # Optionally raise the error to make it obvious during startup
-    # raise e
-except Exception as e:
-    print(f"Unexpected error during imports in api.py: {e}")
-    # raise e
+    print(f"ERROR in api.py: Could not import from utils. Details: {e}")
+    # Define dummy function if import fails to allow FastAPI to start
+    def process_email_request(email_body: str):
+        return {"error": f"Failed to import processing function: {e}"}
+    # Define dummy loaders
+    def load_spacy_model(): print("Dummy spacy loader called"); return None
+    def load_model_pipeline(): print("Dummy pipeline loader called"); return None
 
-# --- FastAPI App ---
-# >>>>> THIS LINE IS CRUCIAL <<<<<
-app = FastAPI()
-# >>>>> MUST BE NAMED 'app' <<<<<
+app = FastAPI(title="Email PII Classifier API", version="1.0.0")
 
-# --- Pydantic Models ---
 class EmailInput(BaseModel):
     email_body: str = Field(..., example="Hello, my name is Jane Doe and my email is jane.doe@example.com. I have a billing question.")
 
@@ -29,64 +36,56 @@ class MaskedEntity(BaseModel):
     classification: str = Field(..., example="full_name")
     entity: str = Field(..., example="Jane Doe")
 
-class ClassificationOutput(BaseModel):
+class EmailResponse(BaseModel):
     input_email_body: str
     list_of_masked_entities: List[MaskedEntity]
     masked_email: str
     category_of_the_email: str
 
-# --- Load Model at Startup ---
-# Ensure load_model_pipeline() exists and works
-model_pipeline: Optional[Pipeline] = load_model_pipeline()
-
-# --- API Endpoints ---
-@app.get("/")
-async def read_root():
-    return {"message": "Email Classification API is running. Use the /classify/ endpoint."}
+# --- Load models on startup ---
+# It's often better to load models once when the app starts
+@app.on_event("startup")
+async def startup_event():
+    print("FastAPI startup: Loading models...")
+    load_spacy_model()       # Load spaCy model
+    load_model_pipeline()    # Load classification pipeline
+    print("FastAPI startup: Model loading complete.")
 
-@app.post("/classify/", response_model=ClassificationOutput)
+# --- API Endpoint ---
+@app.post("/classify_email/", response_model=Union[EmailResponse, Dict[str, str]])  # Allow dict for error response
 async def classify_email(email_input: EmailInput):
     """
-    Accepts an email body, masks PII, classifies the email,
-    and returns the results in the specified JSON format.
+    Receives email body, performs PII masking and classification.
     """
-    if not model_pipeline:
-        raise HTTPException(status_code=503, detail="Model pipeline not available. Please train/load it first.")
-
-    input_email = email_input.email_body
-
     try:
-        # 1. Mask PII
-        masked_email_body, entities = mask_pii(input_email)
-
-        # Ensure entities match the Pydantic model format
-        validated_entities = [MaskedEntity(**entity) for entity in entities]
+        print("Received request for /classify_email/")  # Log request
+        result = process_email_request(email_input.email_body)
 
-        # 2. Classify the masked email
-        predicted_class = predict_category(masked_email_body, model_pipeline)
+        if "error" in result:
+            # Return a 500 error if processing failed internally
+            raise HTTPException(status_code=500, detail=result["error"])
 
-        # 3. Construct the response
-        response = ClassificationOutput(
-            input_email_body=input_email,
-            list_of_masked_entities=validated_entities,
-            masked_email=masked_email_body,
-            category_of_the_email=predicted_class
-        )
-        return response
+        # Validate response structure before returning (optional but good practice)
+        # This assumes process_email_request returns the correct structure on success
+        return EmailResponse(**result)
 
+    except HTTPException as http_exc:
+        # Re-raise HTTP exceptions (like the 500 error above)
+        raise http_exc
     except Exception as e:
-        # Log the exception for debugging
-        print(f"Error processing request: {e}")
-        # Optionally include more detail in the error response during development
-        raise HTTPException(status_code=500, detail=f"Internal Server Error: {str(e)}")
+        print(f"Unexpected error in /classify_email endpoint: {e}")
+        # import traceback # Uncomment for detailed debugging
+        # print(traceback.format_exc()) # Uncomment for detailed debugging
+        # Return a generic 500 error for other unexpected issues
+        raise HTTPException(status_code=500, detail=f"An internal server error occurred: {str(e)}")
+
+# --- Root Endpoint ---
+@app.get("/")
+async def read_root():
+    return {"message": "Welcome to the Email PII Classifier API. Use the /docs endpoint for details."}
 
-# --- Running the API (for local development) ---
-# You'll typically run this using uvicorn from the terminal:
-# uvicorn api:app --reload
-# The block below is useful if you want to run `python api.py` directly sometimes,
-# but `uvicorn` command is standard for development.
+# --- Optional: Add uvicorn runner for local testing ---
 if __name__ == "__main__":
     import uvicorn
     print("Starting Uvicorn server directly (for debugging)...")
-    # Use host="0.0.0.0" to make it accessible within Codespace network
     uvicorn.run(app, host="0.0.0.0", port=8000)

models.py

@@ -9,7 +9,6 @@ from pathlib import Path
 import re
 from fastapi import FastAPI, HTTPException
 from pydantic import BaseModel
-from utils import clean_text_for_classification
 import spacy
 import pickle
 
@@ -55,6 +54,15 @@ def load_model_pipeline() -> Optional[Pipeline]:
         print("Please train and save the model pipeline first.")
     return model_pipeline
 
+# --- Text Cleaning Function ---
+def clean_text_for_classification(text: str) -> str:
+    """Basic text cleaning."""
+    text = text.lower()
+    text = re.sub(r'<.*?>', '', text) # Remove HTML tags
+    text = re.sub(r'[^a-z\s]', '', text) # Remove non-alpha and non-whitespace
+    text = re.sub(r'\s+', ' ', text).strip() # Normalize whitespace
+    return text
+
 # --- Mask PII Function ---
 def mask_pii(text: str, nlp_model: spacy.language.Language) -> Tuple[str, List[Dict[str, Any]]]:
     """
@@ -90,22 +98,21 @@ def mask_pii(text: str, nlp_model: spacy.language.Language) -> Tuple[str, List[D
 def predict_category(text: str, pipeline: Pipeline) -> str:
     """
     Predicts the category of the text using the loaded classification pipeline.
-
-    Args:
-        text: The input text (potentially masked).
-        pipeline: The loaded scikit-learn compatible pipeline object.
-
-    Returns:
-        The predicted category name as a string.
+    Applies cleaning before prediction.
     """
-    print(f"Executing predict_category for text: '{text[:50]}...'") # Add log
+    print(f"Executing predict_category for text: '{text[:50]}...'")
     try:
-        prediction = pipeline.predict([text])
+        # Clean the text first using the function now in this file
+        cleaned_text = clean_text_for_classification(text)
+        print(f"Cleaned text for prediction: '{cleaned_text[:50]}...'")
+
+        # Assuming the pipeline has a .predict() method
+        prediction = pipeline.predict([cleaned_text]) # Predict on cleaned text
         category = str(prediction[0]) if prediction else "Prediction failed"
     except Exception as e:
         print(f"Error during prediction: {e}")
         category = "Prediction Error"
-    print(f"predict_category result: {category}") # Add log
+    print(f"predict_category result: {category}")
     return category
 
 # --- Training Function ---

utils.py

@@ -1,27 +1,31 @@
 import re
 import spacy
-from typing import List, Dict, Tuple
+from typing import List, Dict, Tuple, Optional, Union
 import pickle
 from pathlib import Path
-import os  # Import os to handle potential path issues
+import os
 
-# Assuming mask_pii and predict_category are defined in models.py
+# --- Define/Import Pipeline Type FIRST ---
 try:
-    from models import mask_pii, predict_category
-    print("Successfully imported functions from models.py")  # Add this for confirmation
+    from sklearn.pipeline import Pipeline
+except ImportError:
+    Pipeline = object  # type: ignore
+
+# --- Import from models.py ---
+try:
+    from models import predict_category  # Keep this import
+    print("Successfully imported predict_category from models.py")
 except ImportError as e:
-    print(f"ERROR in utils.py: Could not import functions/classes from models.py. Details: {e}")
-    # Define dummy functions if import fails
-    def mask_pii(text, nlp_model): return "Masking failed", []
+    print(f"ERROR in utils.py: Could not import predict_category from models.py. Details: {e}")
     def predict_category(text, pipeline): return "Classification failed"
 
 # --- Model Loading ---
 MODEL_DIR = Path("saved_models")
 MODEL_PATH = MODEL_DIR / "email_classifier_pipeline.pkl"
-NLP_MODEL = None
-MODEL_PIPELINE = None
+NLP_MODEL: Optional[spacy.language.Language] = None
+MODEL_PIPELINE: Optional[Pipeline] = None  # Now Pipeline is defined
 
-def load_spacy_model():
+def load_spacy_model() -> Optional[spacy.language.Language]:
     """Loads the spaCy model."""
     global NLP_MODEL
     if NLP_MODEL is None:
@@ -41,7 +45,7 @@ def load_spacy_model():
                 NLP_MODEL = None  # Ensure it remains None if loading fails
     return NLP_MODEL
 
-def load_model_pipeline() -> Pipeline | None:
+def load_model_pipeline() -> Optional[Pipeline]:  # Now Pipeline is defined
     """Loads the classification pipeline from the .pkl file."""
     global MODEL_PIPELINE
     if MODEL_PIPELINE is None:
@@ -71,14 +75,14 @@ REGEX_PATTERNS = {
     "dob": r'\b(\d{1,2}[-/]\d{1,2}[-/]\d{2,4}|\d{4}[-/]\d{1,2}[-/]\d{1,2})\b'  # Basic DOB patterns
 }
 
-# --- PII Masking Function ---
-
-def mask_pii(text: str) -> Tuple[str, List[Dict]]:
+# --- PII Masking Function (Defined within utils.py) ---
+def mask_pii(text: str, nlp: spacy.language.Language) -> Tuple[str, List[Dict]]:
     """
     Detects and masks PII in the input text using spaCy NER and Regex.
 
     Args:
         text: The input email body string.
+        nlp: The spaCy language model.
 
     Returns:
         A tuple containing:
@@ -139,8 +143,7 @@ def mask_pii(text: str) -> Tuple[str, List[Dict]]:
 
     return masked_text, list_of_masked_entities
 
-# --- Main Processing Function for Gradio ---
-
+# --- Main Processing Function (Defined within utils.py) ---
 def process_email_request(email_body: str) -> dict:
     """
     Processes the input email body for PII masking and classification.
@@ -188,14 +191,3 @@ def process_email_request(email_body: str) -> dict:
             "error": f"An error occurred during processing: {str(e)}",
             "input_email_body": email_body
         }
-
-# --- Other Utility Functions (Add as needed) ---
-# E.g., text cleaning for classification model input
-def clean_text_for_classification(text: str) -> str:
-    """Basic text cleaning."""
-    text = text.lower()
-    text = re.sub(r'<.*?>', '', text)  # Remove HTML tags
-    text = re.sub(r'[^a-z\s]', '', text)  # Remove punctuation/numbers (adjust if needed)
-    text = re.sub(r'\s+', ' ', text).strip()  # Normalize whitespace
-    # Add stopword removal if necessary (import nltk stopwords)
-    return text