Delete app.py

app.py

@@ -1,177 +0,0 @@
-import os
-os.environ.setdefault("HOME", "/data")
-os.environ.setdefault("XDG_CACHE_HOME", "/data/.cache")
-os.environ.setdefault("HF_HOME", "/data/.cache")
-os.environ.setdefault("TRANSFORMERS_CACHE", "/data/.cache")
-os.environ.setdefault("TORCH_HOME", "/data/.cache")
-
-from fastapi import FastAPI
-from fastapi.responses import JSONResponse
-from pydantic import BaseModel
-from transformers import AutoTokenizer, AutoModelForSequenceClassification
-from huggingface_hub import hf_hub_download
-import joblib
-import torch
-import re
-import numpy as np
-import pandas as pd
-try:
-    import xgboost as xgb  # type: ignore
-except Exception:
-    xgb = None  # optional; required if bundle uses xgboost
-
-
-MODEL_ID = os.environ.get("MODEL_ID", "Perth0603/phishing-email-mobilebert")
-URL_REPO = os.environ.get("URL_REPO", "Perth0603/Random-Forest-Model-for-PhishingDetection")
-URL_REPO_TYPE = os.environ.get("URL_REPO_TYPE", "model")  # model|space|dataset
-# NOTE: set to your artifact filename, e.g. rf_url_phishing_xgboost_bst.joblib
-URL_FILENAME = os.environ.get("URL_FILENAME", "rf_url_phishing_xgboost_bst.joblib")
-
-# Ensure writable cache directory for HF/torch inside Spaces Docker
-CACHE_DIR = os.environ.get("HF_CACHE_DIR", "/data/.cache")
-os.makedirs(CACHE_DIR, exist_ok=True)
-
-app = FastAPI(title="Phishing Text Classifier", version="1.0.0")
-
-
-class PredictPayload(BaseModel):
-    inputs: str
-
-
-# Lazy singletons for model/tokenizer
-_tokenizer = None
-_model = None
-_url_bundle = None  # holds dict: {model, feature_cols, url_col, label_col, model_type}
-
-
-def _load_url_model():
-    global _url_bundle
-    if _url_bundle is None:
-        # Prefer local artifact if present (e.g., committed into the Space repo)
-        local_path = os.path.join(os.getcwd(), URL_FILENAME)
-        if os.path.exists(local_path):
-            _url_bundle = joblib.load(local_path)
-            return
-        # Download model artifact from HF Hub
-        model_path = hf_hub_download(
-            repo_id=URL_REPO,
-            filename=URL_FILENAME,
-            repo_type=URL_REPO_TYPE,
-            cache_dir=CACHE_DIR,
-        )
-        _url_bundle = joblib.load(model_path)
-
-
-# URL feature engineering (must match training)
-_SUSPICIOUS_TOKENS = ["login", "verify", "secure", "update", "bank", "pay", "account", "webscr"]
-_ipv4_pattern = re.compile(r'(?:\d{1,3}\.){3}\d{1,3}')
-
-def _engineer_features(df: pd.DataFrame, url_col: str, feature_cols: list[str] | None = None) -> pd.DataFrame:
-    s = df[url_col].astype(str)
-    out = pd.DataFrame(index=df.index)
-    out['url_len'] = s.str.len().fillna(0)
-    out['count_dot'] = s.str.count(r'\.')
-    out['count_hyphen'] = s.str.count('-')
-    out['count_digit'] = s.str.count(r'\d')
-    out['count_at'] = s.str.count('@')
-    out['count_qmark'] = s.str.count('\?')
-    out['count_eq'] = s.str.count('=')
-    out['count_slash'] = s.str.count('/')
-    out['digit_ratio'] = (out['count_digit'] / out['url_len'].replace(0, np.nan)).fillna(0)
-    out['has_ip'] = s.str.contains(_ipv4_pattern).astype(int)
-    for tok in _SUSPICIOUS_TOKENS:
-        out[f'has_{tok}'] = s.str.contains(tok, case=False, regex=False).astype(int)
-    out['starts_https'] = s.str.startswith('https').astype(int)
-    out['ends_with_exe'] = s.str.endswith('.exe').astype(int)
-    out['ends_with_zip'] = s.str.endswith('.zip').astype(int)
-    return out if feature_cols is None else out[feature_cols]
-
-
-def _load_model():
-    global _tokenizer, _model
-    if _tokenizer is None or _model is None:
-        _tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
-        _model = AutoModelForSequenceClassification.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
-        # Warm-up
-        with torch.no_grad():
-            _ = _model(**_tokenizer(["warm up"], return_tensors="pt")).logits
-
-
-@app.get("/")
-def root():
-    return {"status": "ok", "model": MODEL_ID}
-
-
-@app.post("/predict")
-def predict(payload: PredictPayload):
-    try:
-        _load_model()
-        with torch.no_grad():
-            inputs = _tokenizer([payload.inputs], return_tensors="pt", truncation=True, max_length=512)
-            logits = _model(**inputs).logits
-            probs = torch.softmax(logits, dim=-1)[0]
-            score, idx = torch.max(probs, dim=0)
-    except Exception as e:
-        return JSONResponse(status_code=500, content={"error": str(e)})
-
-    # Map common ids to labels (kept generic; your config also has these)
-    id2label = {0: "LEGIT", 1: "PHISH"}
-    label = id2label.get(int(idx), str(int(idx)))
-    return {"label": label, "score": float(score)}
-
-
-class PredictUrlPayload(BaseModel):
-    url: str
-
-
-@app.post("/predict-url")
-def predict_url(payload: PredictUrlPayload):
-    try:
-        _load_url_model()
-        bundle = _url_bundle
-        if not isinstance(bundle, dict) or 'model' not in bundle:
-            raise RuntimeError("Loaded URL artifact is not a bundle dict with 'model'.")
-        model = bundle['model']
-        feature_cols = bundle.get('feature_cols') or []
-        url_col = bundle.get('url_col') or 'url'
-        model_type = bundle.get('model_type') or ''
-
-        row = pd.DataFrame({url_col: [payload.url]})
-        feats = _engineer_features(row, url_col, feature_cols)
-
-        score = None
-        label = None
-
-        if isinstance(model_type, str) and model_type == 'xgboost_bst':
-            if xgb is None:
-                raise RuntimeError("xgboost is not installed but required for this model bundle.")
-            dmat = xgb.DMatrix(feats)
-            proba = float(model.predict(dmat)[0])
-            score = proba
-            label = "PHISH" if score >= 0.5 else "LEGIT"
-        elif hasattr(model, "predict_proba"):
-            proba = model.predict_proba(feats)[0]
-            if len(proba) == 2:
-                score = float(proba[1])
-                label = "PHISH" if score >= 0.5 else "LEGIT"
-            else:
-                max_idx = int(np.argmax(proba))
-                score = float(proba[max_idx])
-                label = "PHISH" if max_idx == 1 else "LEGIT"
-        else:
-            pred = model.predict(feats)[0]
-            if isinstance(pred, (int, float, np.integer, np.floating)):
-                label = "PHISH" if int(pred) == 1 else "LEGIT"
-                score = 1.0 if label == "PHISH" else 0.0
-            else:
-                up = str(pred).strip().upper()
-                if up in ("PHISH", "PHISHING", "MALICIOUS"):
-                    label, score = "PHISH", 1.0
-                else:
-                    label, score = "LEGIT", 0.0
-    except Exception as e:
-        return JSONResponse(status_code=500, content={"error": str(e)})
-
-    return {"label": label, "score": float(score)}
-
-