from flask import Flask, request, jsonify, render_template_string from flask_cors import CORS import os from transformers import AutoTokenizer, AutoModelForSequenceClassification, AutoModelForSeq2SeqLM import torch import logging import io import pandas as pd from google.oauth2 import id_token from google.auth.transport import requests as google_requests import sqlite3 import re from langdetect import detect, DetectorFactory from urllib.parse import urlparse, urlunparse import time from openrouter import OpenRouter import requests # Make langdetect deterministic DetectorFactory.seed = 0 # Configure logging logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) app = Flask(__name__) CORS(app) # Disable HF telemetry os.environ['HF_HUB_DISABLE_TELEMETRY'] = '1' # Globals tokenizer = None model = None translator_tokenizer = None translator_model = None device = torch.device("cuda" if torch.cuda.is_available() else "cpu") # OpenRouter client + model ids (env-configurable placeholders) or_client = None OR_MODEL_1 = os.environ.get("OR_MODEL_1", "or-model-1") OR_MODEL_2 = os.environ.get("OR_MODEL_2", "or-model-2") OR_TIMEOUT = int(os.environ.get("OR_TIMEOUT", "15")) OR_CONSECUTIVE_FAILURES = 0 OR_CB_THRESHOLD = int(os.environ.get("OR_CB_THRESHOLD", "5")) OR_CB_OPEN_SECONDS = int(os.environ.get("OR_CB_OPEN_SECONDS", "300")) OR_LAST_ERROR = None OR_CB_OPEN_UNTIL = 0 # ----------------------- # DB + URL helpers # ----------------------- def get_db_connection(): db_path = os.path.join(os.path.dirname(__file__), "whitelist.db") conn = sqlite3.connect(db_path) # logger.info(f"Whitelist DB path: {db_path}") return conn url_pattern = re.compile(r'https?://[^\s<>"\']+|\b[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)+\b|\S+@\S+') def extract_domains(text): """Return list of domain strings extracted from text (lowercased, no www).""" tokens = url_pattern.findall(text) domains = [] for t in tokens: t = t.strip() # If it's a full URL, parse host, else treat as bare domain/email if t.lower().startswith("http://") or t.lower().startswith("https://") or t.lower().startswith("www."): try: parsed = urlparse(t if "://" in t else "http://" + t) host = parsed.netloc.lower().split(':')[0] if host.startswith("www."): host = host[4:] domains.append(host) except Exception: continue else: # email or bare domain if "@" in t: try: host = t.split("@", 1)[1].lower().split(':')[0] if host.startswith("www."): host = host[4:] domains.append(host) except Exception: continue else: d = t.lower().strip().strip('/') if d.startswith("www."): d = d[4:] if "." in d: domains.append(d.split(':')[0]) # dedupe, keep order return list(dict.fromkeys(domains)) def is_whitelisted(domains): """ Return True if any domain in `domains` is whitelisted. Matching rules: - exact match: domain == whitelist_entry - suffix match: domain endswith '.' + whitelist_entry (so 'docs.google.com' matches 'google.com') """ if not domains: return False # load whitelist into a set (lowercased) conn = get_db_connection() cursor = conn.cursor() try: cursor.execute("SELECT LOWER(domain) FROM whitelist") rows = cursor.fetchall() whitelist_set = {r[0] for r in rows if r and r[0]} finally: conn.close() for domain in domains: d = (domain or "").lower().strip().strip('/') if not d: continue # direct match if d in whitelist_set: return True # suffix match (allow subdomains) for wl in whitelist_set: if d == wl or d.endswith("." + wl): return True return False # ----------------------- # Blacklist helpers # ----------------------- def get_blacklist_connection(): db_path = os.path.join(os.path.dirname(__file__), "blacklist.db") conn = sqlite3.connect(db_path) # logger.info(f"Blacklist DB path: {db_path}") return conn def normalize_full_url(raw_url: str) -> str: """Normalize a URL for matching while preserving query and fragment. (If you prefer to strip query/fragment, revert to your previous version.) """ try: if not raw_url: return "" # ensure scheme for parsing if "://" not in raw_url: raw_url = "http://" + raw_url p = urlparse(raw_url) scheme = p.scheme.lower() netloc = p.netloc.lower().rstrip('/') path = p.path.rstrip('/') query = p.query # preserve query fragment = p.fragment # preserve fragment # Rebuild including query and fragment normalized = urlunparse((scheme, netloc, path, "", query, fragment)) return normalized except Exception: return raw_url.strip().lower().rstrip('/') def extract_urls_and_domains(text: str): """ Return two lists: full_urls (as extracted, preserving query/fragment) and domains (hostnames). Uses `url_pattern` to find tokens. """ found = url_pattern.findall(text) tokens = [] for f in found: if isinstance(f, tuple): t = "".join(f).strip() if t: tokens.append(t) else: tokens.append(f.strip()) full_urls = [] domains = [] for token in tokens: if token.lower().startswith("http://") or token.lower().startswith("https://") or token.lower().startswith("www."): # keep the whole URL (including query + fragment) full_urls.append(token) # extract domain try: parsed = urlparse(token if "://" in token else "http://" + token) host = parsed.netloc.lower().split(':')[0] if host.startswith("www."): host = host[4:] domains.append(host) except Exception: pass else: # bare domain/email if "@" in token: try: domain_part = token.split("@", 1)[1].lower().strip().strip('/') if domain_part.startswith("www."): domain_part = domain_part[4:] domains.append(domain_part.split(':')[0]) except Exception: pass else: tok = token.lower().strip().strip('/') if tok.startswith("www."): tok = tok[4:] if "." in tok: domains.append(tok.split(':')[0]) # debug prints (optional) # logger.info(f"[DEBUG] Input text: {text}") # logger.info(f"[DEBUG] Extracted full URLs: {full_urls}") # logger.info(f"[DEBUG] Extracted domains: {domains}") return list(dict.fromkeys(full_urls)), list(dict.fromkeys(domains)) def is_blacklisted(text: str): """ Returns (bool, reason_str) — True + reason if any URL/domain in text matches blacklist.db. """ try: full_urls, domains = extract_urls_and_domains(text) if not full_urls and not domains: # logger.info("[DEBUG] No URLs or domains extracted.") return False, None conn = get_blacklist_connection() cursor = conn.cursor() # 1) check normalized full URLs for u in full_urls: # logger.info(f"[DEBUG] Checking full URL against DB: {u}") cursor.execute("SELECT url FROM blacklist WHERE LOWER(url) = ?", (u.lower(),)) row = cursor.fetchone() if row: # logger.info(f"[DEBUG] FULL URL matched blacklist → {row[0]}") conn.close() return True, f"Blacklisted URL matched: {u}" # 2) check domain exact matches for d in domains: # logger.info(f"[DEBUG] Checking domain against DB: {d}") cursor.execute("SELECT url FROM blacklist WHERE LOWER(url) = ? OR LOWER(url) = ?", (d.lower(), f"www.{d}".lower())) row = cursor.fetchone() if row: # logger.info(f"[DEBUG] DOMAIN matched blacklist → {row[0]}") conn.close() return True, f"Blacklisted domain matched: {d}" # 3) fallback substring check for d in domains: like_pattern = f"%{d.lower()}%" # logger.info(f"[DEBUG] Checking substring match for domain: {d} ({like_pattern})") cursor.execute("SELECT url FROM blacklist WHERE LOWER(url) LIKE ? LIMIT 1", (like_pattern,)) row = cursor.fetchone() if row: # logger.info(f"[DEBUG] SUBSTRING matched blacklist → {row[0]}") conn.close() return True, f"Blacklisted domain substring matched in URL for: {d}" conn.close() # logger.info("[DEBUG] No blacklist match found.") return False, None except Exception as e: logger.error(f"is_blacklisted error: {e}") return False, None # ----------------------- # URL masking for safe translation # ----------------------- # This pattern matches: # - full URLs starting with http(s):// or www. # - bare domains like example.com # - email addresses URL_PATTERN = re.compile( r'((?:https?://|http://|www\.)\S+|\b[A-Za-z0-9-]+(?:\.[A-Za-z0-9-]+)+\b|\S+@\S+)' ) def mask_urls(text: str): """Replace found URLs/domains/emails with placeholders and return (masked_text, mapping).""" found = URL_PATTERN.findall(text) mapping = {} masked = text # sort by length desc to avoid partial overlaps when replacing for i, original in enumerate(sorted(set(found), key=lambda s: -len(s))): placeholder = f"__URL_{i}__" mapping[placeholder] = original # use replace (safe after sorting by length) masked = masked.replace(original, placeholder) return masked, mapping def restore_placeholders(text: str, mapping: dict): """Restore placeholders back to original strings.""" for placeholder, original in mapping.items(): text = text.replace(placeholder, original) return text # ----------------------- # Model loading & inference # ----------------------- def load_model(): """Load phishing classifier and translation model, move to device.""" global tokenizer, model, translator_tokenizer, translator_model, device try: logger.info("Loading phishing detection model...") phishing_name = "ealvaradob/bert-finetuned-phishing" tokenizer = AutoTokenizer.from_pretrained(phishing_name) model = AutoModelForSequenceClassification.from_pretrained(phishing_name) model.to(device) model.eval() logger.info("Phishing model loaded.") logger.info("Loading Tagalog->English translation model...") trans_name = "Helsinki-NLP/opus-mt-tl-en" translator_tokenizer = AutoTokenizer.from_pretrained(trans_name) translator_model = AutoModelForSeq2SeqLM.from_pretrained(trans_name) translator_model.to(device) translator_model.eval() logger.info("Translation model loaded.") except Exception as e: logger.error(f"Error loading models: {e}") raise def detect_language(text: str) -> str: try: lang = detect(text) return lang except Exception: return "unknown" def translate_tl_to_en(text: str) -> str: """Translate Tagalog (tl) text to English but keep URLs/domains/emails exactly the same.""" global translator_tokenizer, translator_model try: # 1) mask links/domains/emails masked_text, mapping = mask_urls(text) # 2) translate the masked text inputs = translator_tokenizer(masked_text, return_tensors="pt", truncation=True, padding=True) inputs = {k: v.to(device) for k, v in inputs.items()} out_ids = translator_model.generate(**inputs, max_length=512) translated = translator_tokenizer.decode(out_ids[0], skip_special_tokens=True) # 3) restore original URLs/domains/emails translated = restore_placeholders(translated, mapping) return translated except Exception as e: logger.warning(f"Translation failed or error: {e}. Returning original text.") return text def predict_phishing(text: str): """Return (label, confidence_percent).""" global tokenizer, model try: inputs = tokenizer(text, return_tensors="pt", truncation=True, padding=True, max_length=512) inputs = {k: v.to(device) for k, v in inputs.items()} with torch.no_grad(): outputs = model(**inputs) probs = torch.nn.functional.softmax(outputs.logits, dim=1) confidence, pred_class = torch.max(probs, dim=1) label = "Phishing" if pred_class.item() == 1 else "Safe" confidence_percent = round(confidence.item() * 100, 1) return label, confidence_percent except Exception as e: logger.error(f"Error predicting phishing: {e}") return "Error", 0.0 def generate_text(prompt: str, model_id: str, timeout: int = OR_TIMEOUT): """Call the FastAPI explainer service (which has OpenRouter access) instead of direct OpenRouter. This relays the phishing classification request through the explainer microservice. Falls back gracefully on network or parsing errors. """ global OR_CONSECUTIVE_FAILURES, OR_LAST_ERROR, OR_CB_OPEN_UNTIL explainer_url = os.environ.get("EXPLAINER_URL") if not explainer_url: OR_CONSECUTIVE_FAILURES += 1 OR_LAST_ERROR = "Missing EXPLAINER_URL" logger.warning("EXPLAINER_URL not set; cannot reach OpenRouter. Set EXPLAINER_URL env var.") return {"ok": False, "text": None, "error": "Missing EXPLAINER_URL", "status": None} # circuit breaker: short-circuit if open if OR_CB_OPEN_UNTIL and time.time() < OR_CB_OPEN_UNTIL: return {"ok": False, "text": None, "error": "circuit_open", "status": None} try: # Call the FastAPI explainer's /classify endpoint classify_url = explainer_url.rstrip('/') + "/classify" headers = {"Content-Type": "application/json"} body = {"message": prompt, "model_id": model_id} # Pass model_id to explainer logger.info(f"Calling explainer service at: {classify_url} with model: {model_id}") resp = requests.post(classify_url, headers=headers, json=body, timeout=timeout) text = resp.text if resp.status_code != 200: logger.error(f"Explainer service error {resp.status_code}: {text}") OR_CONSECUTIVE_FAILURES += 1 OR_LAST_ERROR = f"status {resp.status_code}" if OR_CONSECUTIVE_FAILURES >= OR_CB_THRESHOLD: OR_CB_OPEN_UNTIL = time.time() + OR_CB_OPEN_SECONDS logger.warning("Explainer circuit opened due to repeated failures") return {"ok": False, "text": text, "error": f"status {resp.status_code}", "status": resp.status_code} # Extract the classification JSON from explainer response j = resp.json() content = j.get("reply") # Explainer returns {"reply": "..."} if not content: logger.error(f"Explainer returned no reply: {j}") OR_CONSECUTIVE_FAILURES += 1 OR_LAST_ERROR = "Empty explainer response" return {"ok": False, "text": None, "error": "Empty explainer response", "status": resp.status_code} # success -> reset failure counter OR_CONSECUTIVE_FAILURES = 0 OR_LAST_ERROR = None logger.info(f"Explainer classification successful: {content}") return {"ok": True, "text": content, "error": None, "status": resp.status_code} except Exception as e: logger.error(f"Explainer request error: {e}") OR_CONSECUTIVE_FAILURES += 1 OR_LAST_ERROR = str(e) if OR_CONSECUTIVE_FAILURES >= OR_CB_THRESHOLD: OR_CB_OPEN_UNTIL = time.time() + OR_CB_OPEN_SECONDS logger.warning("Explainer circuit opened due to repeated exceptions") return {"ok": False, "text": None, "error": str(e), "status": None} def predict_phishing_or(text: str, model_id: str): """Call OpenRouter to classify text. Return (label, confidence_percent, raw_response). Falls back to HF `predict_phishing` on error or parse failure. """ # craft a strict prompt that asks for JSON prompt = ( "Classify the following text as either 'Phishing' or 'Safe'.\n" "Return ONLY a single JSON object with exactly these fields: {\"label\": \"Phishing\"|\"Safe\", \"confidence\": <0-100 float>}\n" "Do not add any other text or explanation.\n\n" f"TEXT:\n{text}" ) resp = generate_text(prompt, model_id) logger.info(f"OpenRouter raw response: {resp}") if not resp.get("ok") or not resp.get("text"): # fallback to HF label, conf = predict_phishing(text) return label, conf, {"fallback": True, "reason": resp.get("error"), "raw": resp.get("text")} # try to parse JSON object from resp['text'] txt = resp.get("text") # extract the first {...} block import json, re m = re.search(r"\{.*\}", txt, re.DOTALL) if not m: label, conf = predict_phishing(text) return label, conf, {"fallback": True, "reason": "no_json", "raw": txt} try: obj = json.loads(m.group(0)) lab = obj.get("label") conf = obj.get("confidence") if isinstance(conf, str): try: conf = float(conf) except Exception: conf = 0.0 if lab and (lab.lower() in ("phishing", "safe")): label = "Phishing" if lab.lower() == "phishing" else "Safe" conf_val = round(float(conf), 1) if conf is not None else 0.0 return label, conf_val, {"fallback": False, "raw": txt} else: raise ValueError("Invalid label") except Exception as e: logger.error(f"Error parsing OpenRouter JSON: {e} -- raw: {txt}") label, conf = predict_phishing(text) return label, conf, {"fallback": True, "reason": str(e), "raw": txt} # ----------------------- # Unified pipeline # ----------------------- def analyze_pipeline(message: str): """Do: domain whitelist (with subdomain match) -> but let exact full-URL blacklist override -> detect lang -> translate if Tagalog -> classify -> return dict""" try: text = (message or "").strip() if not text: return {"error": "empty", "blacklist": False} # extract both full URLs and domains once full_urls, domains = extract_urls_and_domains(text) # --- load whitelist set (lowercased) --- try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT LOWER(domain) FROM whitelist") rows = cursor.fetchall() whitelist_set = {r[0] for r in rows if r and r[0]} conn.close() except Exception as e: logger.warning(f"Whitelist load failed: {e}") whitelist_set = set() # 1) domain whitelist check (allow subdomains) if domains and whitelist_set: for domain in domains: d = (domain or "").lower().strip().strip('/') if not d: continue # direct or subdomain match matched_whitelist = False if d in whitelist_set: matched_whitelist = True else: for wl in whitelist_set: if d == wl or d.endswith("." + wl): matched_whitelist = True break if matched_whitelist: # BEFORE returning Safe, check whether any full URL in the message # is exactly present in the blacklist (normalized). If so, treat as Phishing. if full_urls: try: bconn = get_blacklist_connection() bcur = bconn.cursor() for u in full_urls: norm_u = normalize_full_url(u).lower() bcur.execute("SELECT 1 FROM blacklist WHERE LOWER(url) = ?", (norm_u,)) if bcur.fetchone(): bconn.close() return { "result": "Phishing", "confidence": "100.0%", "message": text, "blacklist": True, "whitelist": False, "detected_lang": None, "translated_text": None } bconn.close() except Exception as e: logger.warning(f"Blacklist-full-url check failed: {e}") # If no full-url blacklist hit, return Safe because domain is whitelisted return { "result": "Safe", "confidence": "100.0%", "message": text, "blacklist": False, "whitelist": True, "detected_lang": None, "translated_text": None } # 2) not whitelisted by domain (or whitelist didn't apply) -> regular blacklist check is_black, _reason = is_blacklisted(text) if is_black: return { "result": "Phishing", "confidence": "100.0%", "message": text, "blacklist": True, "whitelist": False, "detected_lang": None, "translated_text": None } # 3) language detection + optional translate lang = detect_language(text) translated_text = None analysis_text = text if lang in ("tl", "fil"): translated_text = translate_tl_to_en(text) analysis_text = translated_text or text # 4) classification label, conf = predict_phishing(analysis_text) return { "result": label, "confidence": f"{conf}%", "message": text, "blacklist": False, "whitelist": False, "detected_lang": lang, "translated_text": translated_text } except Exception as e: logger.error(f"Pipeline error: {e}") return {"result": "Error", "confidence": "0%", "message": message, "blacklist": False, "whitelist": False} # ----------------------- # Routes # ----------------------- @app.route("/", methods=["GET"]) def home(): return jsonify({ "status": "healthy", "message": "Anti-Phishing Scanner API", "endpoints": { "/analyze": "POST - Analyze text for phishing", "/health": "GET - Health check", "/evaluate": "GET/POST - Upload CSV and evaluate model accuracy" } }) @app.route("/health", methods=["GET"]) def health(): or_configured = bool(os.environ.get("OPENROUTER_API_KEY")) return jsonify({ "status": "healthy", "model_loaded": model is not None, "openrouter": { "configured": or_configured, "client_initialized": or_client is not None, "consecutive_failures": OR_CONSECUTIVE_FAILURES, "circuit_open": bool(OR_CB_OPEN_UNTIL and time.time() < OR_CB_OPEN_UNTIL), "last_error": OR_LAST_ERROR } }) @app.route("/analyze", methods=["POST"]) def analyze(): try: data = request.get_json() if not data or "message" not in data: return jsonify({"error": "Missing 'message' field"}), 400 message = data["message"] if not message or not message.strip(): return jsonify({"error": "Message cannot be empty"}), 400 result = analyze_pipeline(message) return jsonify(result), 200 except Exception as e: logger.error(f"Error in analyze endpoint: {e}") return jsonify({"error": "Internal server error"}), 500 def analyze_pipeline_with_model(message: str, model_id: str): """Same pipeline as analyze_pipeline but uses the specified OpenRouter model for classification. Falls back to HF classifier when OR fails. """ try: text = (message or "").strip() if not text: return {"error": "empty", "blacklist": False} # extract both full URLs and domains once full_urls, domains = extract_urls_and_domains(text) # --- load whitelist set (lowercased) --- try: conn = get_db_connection() cursor = conn.cursor() cursor.execute("SELECT LOWER(domain) FROM whitelist") rows = cursor.fetchall() whitelist_set = {r[0] for r in rows if r and r[0]} conn.close() except Exception as e: logger.warning(f"Whitelist load failed: {e}") whitelist_set = set() # 1) domain whitelist check (allow subdomains) if domains and whitelist_set: for domain in domains: d = (domain or "").lower().strip().strip('/') if not d: continue # direct or subdomain match matched_whitelist = False if d in whitelist_set: matched_whitelist = True else: for wl in whitelist_set: if d == wl or d.endswith("." + wl): matched_whitelist = True break if matched_whitelist: # BEFORE returning Safe, check whether any full URL in the message # is exactly present in the blacklist (normalized). If so, treat as Phishing. if full_urls: try: bconn = get_blacklist_connection() bcur = bconn.cursor() for u in full_urls: norm_u = normalize_full_url(u).lower() bcur.execute("SELECT 1 FROM blacklist WHERE LOWER(url) = ?", (norm_u,)) if bcur.fetchone(): bconn.close() return { "result": "Phishing", "confidence": "100.0%", "message": text, "blacklist": True, "whitelist": False, "detected_lang": None, "translated_text": None, "model_used": model_id } bconn.close() except Exception as e: logger.warning(f"Blacklist-full-url check failed: {e}") # If no full-url blacklist hit, return Safe because domain is whitelisted return { "result": "Safe", "confidence": "100.0%", "message": text, "blacklist": False, "whitelist": True, "detected_lang": None, "translated_text": None, "model_used": model_id } # 2) not whitelisted by domain (or whitelist didn't apply) -> regular blacklist check is_black, _reason = is_blacklisted(text) if is_black: return { "result": "Phishing", "confidence": "100.0%", "message": text, "blacklist": True, "whitelist": False, "detected_lang": None, "translated_text": None, "model_used": model_id } # 3) language detection + optional translate lang = detect_language(text) translated_text = None analysis_text = text if lang in ("tl", "fil"): translated_text = translate_tl_to_en(text) analysis_text = translated_text or text # 4) classification via OpenRouter model (with HF fallback inside) label, conf, raw = predict_phishing_or(analysis_text, model_id) return { "result": label, "confidence": f"{conf}%", "message": text, "blacklist": False, "whitelist": False, "detected_lang": lang, "translated_text": translated_text, "model_used": model_id, "or_raw": raw } except Exception as e: logger.error(f"Pipeline error: {e}") return {"result": "Error", "confidence": "0%", "message": message, "blacklist": False, "whitelist": False} @app.route("/analyze/or1", methods=["POST"]) def analyze_or1(): try: data = request.get_json() if not data or "message" not in data: return jsonify({"error": "Missing 'message' field"}), 400 message = data["message"] if not message or not message.strip(): return jsonify({"error": "Message cannot be empty"}), 400 result = analyze_pipeline_with_model(message, OR_MODEL_1) return jsonify(result), 200 except Exception as e: logger.error(f"Error in analyze/or1 endpoint: {e}") return jsonify({"error": "Internal server error"}), 500 @app.route("/analyze/or2", methods=["POST"]) def analyze_or2(): try: data = request.get_json() if not data or "message" not in data: return jsonify({"error": "Missing 'message' field"}), 400 message = data["message"] if not message or not message.strip(): return jsonify({"error": "Message cannot be empty"}), 400 result = analyze_pipeline_with_model(message, OR_MODEL_2) return jsonify(result), 200 except Exception as e: logger.error(f"Error in analyze/or2 endpoint: {e}") return jsonify({"error": "Internal server error"}), 500 # ============================= # /evaluate (GET form + POST CSV) # ============================= @app.route("/evaluate", methods=["GET", "POST"]) def evaluate(): """Upload a CSV with text+label to compute accuracy, precision, recall, F1""" if request.method == "GET": # Simple HTML form to upload a CSV return render_template_string( """ Model Evaluation

Upload a CSV to Evaluate Model Accuracy

Expected columns: text (or message) and label (values: phishing/safe or 1/0)


""" ) # POST: handle CSV upload, run evaluation try: if "file" not in request.files: return jsonify({"error": "No file uploaded. Please upload a CSV with 'text' or 'message' and 'label' columns."}), 400 file = request.files["file"] # Read CSV (handle utf-8 gracefully) content = file.stream.read().decode("utf-8", errors="ignore") df = pd.read_csv(io.StringIO(content)) # Determine text column text_col = None if "text" in df.columns: text_col = "text" elif "message" in df.columns: text_col = "message" if text_col is None: return jsonify({"error": "CSV must have a 'text' or 'message' column."}), 400 if "label" not in df.columns: return jsonify({"error": "CSV must have a 'label' column."}), 400 # Normalize labels to 0/1 (0=safe, 1=phishing) def to_int_label(x): if isinstance(x, str): s = x.strip().lower() if s in ("phishing", "spam", "1"): # treat 'spam' as phishing return 1 if s in ("safe", "ham", "0"): return 0 try: v = int(x) return 1 if v == 1 else 0 except Exception: return None texts = df[text_col].astype(str).tolist() labels = [to_int_label(v) for v in df["label"].tolist()] # Filter out rows with invalid labels valid_items = [(t, y) for t, y in zip(texts, labels) if y is not None] if not valid_items: return jsonify({"error": "No valid rows. Ensure 'label' values are 'phishing'/'safe' or 1/0."}), 400 texts_valid, y_true = zip(*valid_items) # Predict # Predict using the unified analyze_pipeline y_pred = [] for txt in texts_valid: result = analyze_pipeline(txt) # returns dict with "result" y_pred.append(1 if result["result"].lower() == "phishing" else 0) # Compute metrics (no sklearn) tp = sum(1 for p, y in zip(y_pred, y_true) if p == 1 and y == 1) tn = sum(1 for p, y in zip(y_pred, y_true) if p == 0 and y == 0) fp = sum(1 for p, y in zip(y_pred, y_true) if p == 1 and y == 0) fn = sum(1 for p, y in zip(y_pred, y_true) if p == 0 and y == 1) total = len(y_true) accuracy = (tp + tn) / total if total else 0.0 precision = tp / (tp + fp) if (tp + fp) else 0.0 recall = tp / (tp + fn) if (tp + fn) else 0.0 f1 = (2 * precision * recall / (precision + recall)) if (precision + recall) else 0.0 skipped = len(texts) - total # Render results page return render_template_string( f""" Evaluation Results

Evaluation Results

Samples Tested: {total}

Accuracy: {accuracy:.4f}

Precision: {precision:.4f}

Recall: {recall:.4f}

F1 Score: {f1:.4f}

TP: {tp} • TN: {tn} • FP: {fp} • FN: {fn} • Skipped rows: {skipped}

← Run another test """ ) except Exception as e: logger.error(f"Error in evaluate endpoint: {e}") return jsonify({"error": "Evaluation failed"}), 500 # ----------------------- # Google Auth callback # ----------------------- CLIENT_ID = "1039434473386-fm6bjua9mmpb6umhf6lur1klb806urce.apps.googleusercontent.com" @app.route("/google/auth/callback", methods=["POST"]) def google_login(): """Handle Google login""" try: data = request.get_json() token = data.get("credential") if not token: return jsonify({"success": False, "message": "Missing token"}), 400 # Verify token with Google idinfo = id_token.verify_oauth2_token(token, google_requests.Request(), CLIENT_ID) # Extract user info user_info = { "email": idinfo.get("email"), "name": idinfo.get("name"), "picture": idinfo.get("picture") } # Here you can create a user in your DB or session return jsonify({"success": True, "user": user_info}) except ValueError: return jsonify({"success": False, "message": "Invalid token"}), 400 except Exception as e: logger.error(f"Google login error: {e}") return jsonify({"success": False, "message": "Internal server error"}), 500 # ----------------------- # Error handlers # ----------------------- @app.errorhandler(404) def not_found(error): return jsonify({"error": "Endpoint not found"}), 404 @app.errorhandler(500) def internal_error(error): return jsonify({"error": "Internal server error"}), 500 # ----------------------- # Startup # ----------------------- load_model() try: api_key = os.environ.get("OPENROUTER_API_KEY") if api_key: try: or_client = OpenRouter(api_key=api_key) logger.info("OpenRouter client initialized.") except Exception as e: logger.warning(f"Failed to initialize OpenRouter client object: {e}") else: logger.warning("OPENROUTER_API_KEY not set; OpenRouter endpoints will still attempt HTTP calls but client object is not created.") except Exception as e: logger.warning(f"OpenRouter init warning: {e}")