Perth0603 commited on
Commit
3d59fda
·
verified ·
1 Parent(s): 75f4bd9

Upload app.py

Browse files
Files changed (1) hide show
  1. app.py +26 -117
app.py CHANGED
@@ -1,5 +1,4 @@
1
  import os
2
-
3
  os.environ.setdefault("HOME", "/data")
4
  os.environ.setdefault("XDG_CACHE_HOME", "/data/.cache")
5
  os.environ.setdefault("HF_HOME", "/data/.cache")
@@ -12,145 +11,55 @@ from pydantic import BaseModel
12
  from transformers import AutoTokenizer, AutoModelForSequenceClassification
13
  import torch
14
 
 
15
  MODEL_ID = os.environ.get("MODEL_ID", "Perth0603/phishing-email-mobilebert")
 
 
16
  CACHE_DIR = os.environ.get("HF_CACHE_DIR", "/data/.cache")
17
  os.makedirs(CACHE_DIR, exist_ok=True)
18
 
19
- # Optional overrides:
20
- # - set PHISH_INDEX=0 or 1 to force which logit is "phish"
21
- # - or POSITIVE_CLASS=PHISH/LEGIT to force polarity if the head is single-logit
22
- PHISH_INDEX_ENV = os.environ.get("PHISH_INDEX", "").strip()
23
- POS_CLASS_ENV = os.environ.get("POSITIVE_CLASS", "").strip().upper() # for single-logit heads
24
 
25
- app = FastAPI(title="Phishing Text Classifier", version="1.2.0")
26
 
27
  class PredictPayload(BaseModel):
28
  inputs: str
29
 
 
 
30
  _tokenizer = None
31
  _model = None
32
- _id2label = None
33
- _phish_idx = None # which class id means "phish" for multi-logit heads
34
-
35
-
36
- def _norm(lbl: str) -> str:
37
- return (lbl or "").lower().replace("label_", "").replace("_", "").strip()
38
-
39
-
40
- def _auto_pick_phish_idx(id2label: dict) -> int | None:
41
- # Try to find an explicit phishing-like label
42
- normalized = {int(k): _norm(v) for k, v in id2label.items()}
43
- for i, s in normalized.items():
44
- if any(k in s for k in ("phish", "phishing", "spam", "malicious")):
45
- return int(i)
46
- # Try to infer from legit/ham/safe presence
47
- legit_like = [i for i, s in normalized.items() if any(k in s for k in ("legit", "ham", "safe", "nonphish"))]
48
- if len(legit_like) == 1 and len(id2label) == 2:
49
- return 1 - int(legit_like[0])
50
- return None # unknown
51
-
52
-
53
- def _calibrate_phish_idx(model, tokenizer, num_labels: int) -> int:
54
- # Probe with typical phishing vs legitimate texts to decide which index behaves like "phish"
55
- phish_samples = [
56
- "Urgent: Your account has been suspended. Verify your password at http://login-security-update.com",
57
- "We detected unusual activity. Confirm your identity now or your account will be closed.",
58
- ]
59
- legit_samples = [
60
- "Let's schedule a meeting for tomorrow to discuss the project status.",
61
- "Thank you for your purchase. Here is your receipt for your recent order.",
62
- ]
63
- with torch.no_grad():
64
- p_logits = model(**tokenizer(phish_samples, return_tensors="pt", padding=True, truncation=True, max_length=512)).logits
65
- l_logits = model(**tokenizer(legit_samples, return_tensors="pt", padding=True, truncation=True, max_length=512)).logits
66
- if p_logits.shape[-1] == 1:
67
- # Single-logit: positive class is sigmoid(logit). If POSITIVE_CLASS env says PHISH, use it; else assume PHISH
68
- return 0 # index unused for single-logit; return anything
69
- probs_p = torch.softmax(p_logits, dim=-1).mean(dim=0) # [C]
70
- probs_l = torch.softmax(l_logits, dim=-1).mean(dim=0) # [C]
71
- # Pick the index that is high for phishing samples and low for legit samples
72
- diffs = probs_p - probs_l
73
- return int(torch.argmax(diffs).item())
74
 
75
 
76
  def _load_model():
77
- global _tokenizer, _model, _id2label, _phish_idx
78
- if _tokenizer is not None and _model is not None:
79
- return
80
-
81
- _tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
82
- _model = AutoModelForSequenceClassification.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
83
- _model.eval()
84
-
85
- cfg = _model.config
86
- id2label = getattr(cfg, "id2label", None) or {i: f"LABEL_{i}" for i in range(int(getattr(cfg, "num_labels", 2)))}
87
- _id2label = {int(k): str(v) for k, v in id2label.items()}
88
- num_labels = len(_id2label)
89
-
90
- # 1) Respect explicit env override
91
- if PHISH_INDEX_ENV in ("0", "1"):
92
- _phish_idx = int(PHISH_INDEX_ENV)
93
- else:
94
- # 2) Try to infer from label names
95
- auto_idx = _auto_pick_phish_idx(_id2label) if num_labels >= 2 else None
96
- if auto_idx is not None:
97
- _phish_idx = auto_idx
98
- else:
99
- # 3) Calibrate by probing example texts (multi-logit only)
100
- _phish_idx = _calibrate_phish_idx(_model, _tokenizer, num_labels) if num_labels >= 2 else 0
101
-
102
- # Warm-up
103
- with torch.no_grad():
104
- _ = _model(**_tokenizer(["warm up"], return_tensors="pt")).logits
105
 
106
 
107
  @app.get("/")
108
  def root():
109
- return {
110
- "status": "ok",
111
- "model": MODEL_ID,
112
- "id2label": _id2label,
113
- "phish_index": _phish_idx,
114
- "note": "Set PHISH_INDEX=0|1 to override mapping if needed.",
115
- }
116
 
117
 
118
  @app.post("/predict")
119
- def predict(payload: PredictPayload, debug: bool = False):
120
  try:
121
  _load_model()
122
  with torch.no_grad():
123
  inputs = _tokenizer([payload.inputs], return_tensors="pt", truncation=True, max_length=512)
124
- logits = _model(**inputs).logits # [1, C] or [1, 1]
125
-
126
- if logits.shape[-1] == 1:
127
- # Single-logit: sigmoid gives probability of the model's POSITIVE class.
128
- p_pos = torch.sigmoid(logits[0, 0]).item()
129
- # If POSITIVE_CLASS says PHISH, treat positive as PHISH; else treat positive as LEGIT.
130
- positive_is_phish = (POS_CLASS_ENV == "PHISH") or (POS_CLASS_ENV == "")
131
-
132
- phish_proba = p_pos if positive_is_phish else (1.0 - p_pos)
133
- probs = torch.tensor([1.0 - phish_proba, phish_proba]) # for debugging
134
- else:
135
- probs = torch.softmax(logits, dim=-1)[0]
136
- phish_proba = float(probs[int(_phish_idx)])
137
-
138
- label = "PHISH" if phish_proba >= 0.5 else "LEGIT"
139
- frontend_label = "Phishing" if label == "PHISH" else "Legitimate"
140
-
141
- out = {
142
- "label": label,
143
- "frontend_label": frontend_label,
144
- "phishing_probability": float(phish_proba),
145
- }
146
- if debug:
147
- out.update({
148
- "id2label": _id2label,
149
- "phish_index": int(_phish_idx),
150
- "logits": [float(x) for x in logits[0]],
151
- "probs": [float(x) for x in (probs if isinstance(probs, torch.Tensor) else torch.tensor(probs))],
152
- })
153
- return out
154
-
155
  except Exception as e:
156
- return JSONResponse(status_code=500, content={"error": str(e)})
 
 
 
 
 
 
 
 
1
  import os
 
2
  os.environ.setdefault("HOME", "/data")
3
  os.environ.setdefault("XDG_CACHE_HOME", "/data/.cache")
4
  os.environ.setdefault("HF_HOME", "/data/.cache")
 
11
  from transformers import AutoTokenizer, AutoModelForSequenceClassification
12
  import torch
13
 
14
+
15
  MODEL_ID = os.environ.get("MODEL_ID", "Perth0603/phishing-email-mobilebert")
16
+
17
+ # Ensure writable cache directory for HF/torch inside Spaces Docker
18
  CACHE_DIR = os.environ.get("HF_CACHE_DIR", "/data/.cache")
19
  os.makedirs(CACHE_DIR, exist_ok=True)
20
 
21
+ app = FastAPI(title="Phishing Text Classifier", version="1.0.0")
 
 
 
 
22
 
 
23
 
24
  class PredictPayload(BaseModel):
25
  inputs: str
26
 
27
+
28
+ # Lazy singletons for model/tokenizer
29
  _tokenizer = None
30
  _model = None
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
31
 
32
 
33
  def _load_model():
34
+ global _tokenizer, _model
35
+ if _tokenizer is None or _model is None:
36
+ _tokenizer = AutoTokenizer.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
37
+ _model = AutoModelForSequenceClassification.from_pretrained(MODEL_ID, cache_dir=CACHE_DIR)
38
+ # Warm-up
39
+ with torch.no_grad():
40
+ _ = _model(**_tokenizer(["warm up"], return_tensors="pt")).logits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
41
 
42
 
43
  @app.get("/")
44
  def root():
45
+ return {"status": "ok", "model": MODEL_ID}
 
 
 
 
 
 
46
 
47
 
48
  @app.post("/predict")
49
+ def predict(payload: PredictPayload):
50
  try:
51
  _load_model()
52
  with torch.no_grad():
53
  inputs = _tokenizer([payload.inputs], return_tensors="pt", truncation=True, max_length=512)
54
+ logits = _model(**inputs).logits
55
+ probs = torch.softmax(logits, dim=-1)[0]
56
+ score, idx = torch.max(probs, dim=0)
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
57
  except Exception as e:
58
+ return JSONResponse(status_code=500, content={"error": str(e)})
59
+
60
+ # Map common ids to labels (kept generic; your config also has these)
61
+ id2label = {0: "LEGIT", 1: "PHISH"}
62
+ label = id2label.get(int(idx), str(int(idx)))
63
+ return {"label": label, "score": float(score)}
64
+
65
+