import tkinter as tk from tkinter import ttk, messagebox import threading import time import os import json import base64 import urllib.request import urllib.parse import urllib.error import http.server import socketserver import subprocess from datetime import datetime
CONFIG_FILE = "config.json" SCOPES = "https://www.googleapis.com/auth/gmail.readonly" REDIRECT_URI = "http://localhost:8080" GMAIL_API = "https://gmail.googleapis.com/gmail/v1" GROQ_API = "https://api.groq.com/openai/v1/chat/completions" GROQ_MODEL = "llama-3.1-8b-instant"
def load_config(): if os.path.exists(CONFIG_FILE): with open(CONFIG_FILE) as f: return json.load(f) return {}
def save_config(data): with open(CONFIG_FILE, "w") as f: json.dump(data, f, indent=2)
def http_get(url, token=None): req = urllib.request.Request(url) if token: req.add_header("Authorization", f"Bearer {token}") with urllib.request.urlopen(req) as r: return json.loads(r.read().decode())
def http_post(url, payload, headers=None): data = json.dumps(payload).encode() req = urllib.request.Request(url, data=data, method="POST") req.add_header("Content-Type", "application/json") if headers: for k, v in headers.items(): req.add_header(k, v) with urllib.request.urlopen(req) as r: return json.loads(r.read().decode())
_oauth_code = None
class OAuthHandler(http.server.BaseHTTPRequestHandler): def do_GET(self): global _oauth_code parsed = urllib.parse.urlparse(self.path) params = urllib.parse.parse_qs(parsed.query) if "code" in params: _oauth_code = params["code"][0] self.send_response(200) self.end_headers() self.wfile.write(b"
") else: self.send_response(400) self.end_headers()def log_message(self, *args):
pass
def exchange_code_for_token(code, client_id, client_secret): payload = { "code": code, "client_id": client_id, "client_secret": client_secret, "redirect_uri": REDIRECT_URI, "grant_type": "authorization_code", } data = urllib.parse.urlencode(payload).encode() req = urllib.request.Request("https://oauth2.googleapis.com/token", data=data, method="POST") with urllib.request.urlopen(req) as r: return json.loads(r.read().decode())
def refresh_access_token(client_id, client_secret, refresh_token): payload = { "client_id": client_id, "client_secret": client_secret, "refresh_token": refresh_token, "grant_type": "refresh_token", } data = urllib.parse.urlencode(payload).encode() req = urllib.request.Request("https://oauth2.googleapis.com/token", data=data, method="POST") with urllib.request.urlopen(req) as r: return json.loads(r.read().decode())
def fetch_unread_emails(token, max_results=5): url = f"{GMAIL_API}/users/me/messages?labelIds=UNREAD&maxResults={max_results}" data = http_get(url, token) messages = data.get("messages", []) emails = [] for msg in messages: detail = http_get(f"{GMAIL_API}/users/me/messages/{msg['id']}?format=full", token) headers = {h["name"]: h["value"] for h in detail["payload"]["headers"]} subject = headers.get("Subject", "(no subject)") sender = headers.get("From", "unknown")
body = ""
payload = detail.get("payload", {})
if "parts" in payload:
for part in payload["parts"]:
if part.get("mimeType") == "text/plain":
raw = part["body"].get("data", "")
body = base64.urlsafe_b64decode(raw + "==").decode("utf-8", errors="ignore")
break
elif "body" in payload:
raw = payload["body"].get("data", "")
body = base64.urlsafe_b64decode(raw + "==").decode("utf-8", errors="ignore")
emails.append({
"id": msg["id"],
"subject": subject,
"sender": sender,
"body": body[:1500],
})
return emails
def classify_email(email, groq_key): prompt = f"""You are a spam detection system. Classify the following email as either SPAM or HAM.
From: {email['sender']} Subject: {email['subject']} Body: {email['body']}
Respond in this exact JSON format: {{"verdict": "SPAM" or "HAM", "confidence": "HIGH" or "MEDIUM" or "LOW", "reason": "one sentence explanation"}}
Only return the JSON, nothing else."""
payload = {
"model": GROQ_MODEL,
"messages": [{"role": "user", "content": prompt}],
"temperature": 0,
}
response = http_post(GROQ_API, payload, headers={
"Authorization": f"Bearer {groq_key}",
"User-Agent": "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36",
"Accept": "application/json",
"Accept-Language": "en-US,en;q=0.9",
"Origin": "https://console.groq.com",
"Referer": "https://console.groq.com/",})
raw = response["choices"][0]["message"]["content"].strip()
raw = raw.replace("```json", "").replace("```", "").strip()
return json.loads(raw)
class SpamDetectorApp(tk.Tk): def init(self): super().init() self.title("Gmail Spam Detector") self.geometry("950x680") self.resizable(True, True) self.config = load_config() self.access_token = None self._build_styles() self._build_ui() self._check_saved_auth()
def _build_styles(self):
self.configure(bg="#1e1e2e")
style = ttk.Style(self)
style.theme_use("clam")
style.configure("TFrame", background="#1e1e2e")
style.configure("Card.TFrame", background="#2a2a3e", relief="flat")
style.configure("TLabel", background="#1e1e2e", foreground="#cdd6f4", font=("Segoe UI", 10))
style.configure("Title.TLabel", background="#1e1e2e", foreground="#cba6f7", font=("Segoe UI", 16, "bold"))
style.configure("Sub.TLabel", background="#1e1e2e", foreground="#a6adc8", font=("Segoe UI", 9))
style.configure("Card.TLabel", background="#2a2a3e", foreground="#cdd6f4", font=("Segoe UI", 10))
style.configure("Primary.TButton", background="#cba6f7", foreground="#1e1e2e",
font=("Segoe UI", 10, "bold"), borderwidth=0, focusthickness=0, padding=8)
style.map("Primary.TButton", background=[("active", "#b4befe")])
style.configure("Secondary.TButton", background="#313244", foreground="#cdd6f4",
font=("Segoe UI", 10), borderwidth=0, focusthickness=0, padding=8)
style.map("Secondary.TButton", background=[("active", "#45475a")])
style.configure("Treeview", background="#2a2a3e", foreground="#cdd6f4",
fieldbackground="#2a2a3e", rowheight=36, borderwidth=0, font=("Segoe UI", 10))
style.configure("Treeview.Heading", background="#313244", foreground="#cba6f7",
font=("Segoe UI", 10, "bold"), borderwidth=0)
style.map("Treeview", background=[("selected", "#45475a")], foreground=[("selected", "#cdd6f4")])
style.configure("TEntry", fieldbackground="#313244", foreground="#cdd6f4",
insertcolor="#cdd6f4", borderwidth=0, font=("Segoe UI", 10))
style.configure("TNotebook", background="#1e1e2e", borderwidth=0)
style.configure("TNotebook.Tab", background="#313244", foreground="#a6adc8",
font=("Segoe UI", 10), padding=[12, 6])
style.map("TNotebook.Tab", background=[("selected", "#2a2a3e")], foreground=[("selected", "#cba6f7")])
style.configure("TProgressbar", troughcolor="#313244", background="#cba6f7", borderwidth=0)
style.configure("TSeparator", background="#45475a")
def _build_ui(self):
self.notebook = ttk.Notebook(self)
self.notebook.pack(fill="both", expand=True, padx=12, pady=12)
self._build_setup_tab()
self._build_inbox_tab()
self._build_about_tab()
def _build_setup_tab(self):
tab = ttk.Frame(self.notebook)
self.notebook.add(tab, text=" ⚙ Setup ")
ttk.Label(tab, text="Gmail Spam Detector", style="Title.TLabel").pack(pady=(30, 4))
ttk.Label(tab, text="Connect your Gmail and Groq API to get started",
style="Sub.TLabel").pack(pady=(0, 24))
card = ttk.Frame(tab, style="Card.TFrame", padding=24)
card.pack(padx=60, fill="x")
ttk.Label(card, text="Google OAuth 2.0", style="Card.TLabel",
font=("Segoe UI", 11, "bold")).grid(row=0, column=0, columnspan=2, sticky="w", pady=(0, 12))
ttk.Label(card, text="Client ID", style="Card.TLabel").grid(row=1, column=0, sticky="w", pady=4)
self.client_id_var = tk.StringVar(value=self.config.get("client_id", ""))
ttk.Entry(card, textvariable=self.client_id_var, width=55).grid(row=1, column=1, padx=(12,0), sticky="ew")
ttk.Label(card, text="Client Secret", style="Card.TLabel").grid(row=2, column=0, sticky="w", pady=4)
self.client_secret_var = tk.StringVar(value=self.config.get("client_secret", ""))
ttk.Entry(card, textvariable=self.client_secret_var, show="*", width=55).grid(row=2, column=1, padx=(12,0), sticky="ew")
ttk.Separator(card, orient="horizontal").grid(row=3, column=0, columnspan=2, sticky="ew", pady=16)
ttk.Label(card, text="Groq API", style="Card.TLabel",
font=("Segoe UI", 11, "bold")).grid(row=4, column=0, columnspan=2, sticky="w", pady=(0,12))
ttk.Label(card, text="Groq API Key", style="Card.TLabel").grid(row=5, column=0, sticky="w", pady=4)
self.groq_key_var = tk.StringVar(value=self.config.get("groq_key", ""))
ttk.Entry(card, textvariable=self.groq_key_var, show="*", width=55).grid(row=5, column=1, padx=(12,0), sticky="ew")
card.columnconfigure(1, weight=1)
btn_row = ttk.Frame(tab)
btn_row.pack(pady=20)
ttk.Button(btn_row, text="Save Config", style="Secondary.TButton",
command=self._save_config).pack(side="left", padx=6)
ttk.Button(btn_row, text="Authenticate with Google", style="Primary.TButton",
command=self._start_oauth).pack(side="left", padx=6)
self.auth_status = ttk.Label(tab, text="", style="Sub.TLabel")
self.auth_status.pack()
help_card = ttk.Frame(tab, style="Card.TFrame", padding=16)
help_card.pack(padx=60, fill="x", pady=(10,0))
ttk.Label(help_card, text="How to get credentials:", style="Card.TLabel",
font=("Segoe UI", 10, "bold")).pack(anchor="w")
steps = [
"1. Go to console.cloud.google.com → Enable Gmail API → Create OAuth Desktop app",
"2. Get Groq API key from console.groq.com (free, no credit card needed)",
]
for s in steps:
ttk.Label(help_card, text=s, style="Sub.TLabel").pack(anchor="w", pady=1)
def _build_inbox_tab(self):
tab = ttk.Frame(self.notebook)
self.notebook.add(tab, text=" 📬 Inbox ")
top = ttk.Frame(tab)
top.pack(fill="x", padx=16, pady=(16, 8))
ttk.Label(top, text="Unread Emails", style="Title.TLabel").pack(side="left")
self.fetch_btn = ttk.Button(top, text="Fetch & Analyse", style="Primary.TButton",
command=self._fetch_and_analyse)
self.fetch_btn.pack(side="right")
self.email_count_label = ttk.Label(top, text="", style="Sub.TLabel")
self.email_count_label.pack(side="right", padx=12)
self.progress = ttk.Progressbar(tab, mode="indeterminate", style="TProgressbar")
self.progress.pack(fill="x", padx=16, pady=(0, 8))
cols = ("sender", "subject", "verdict", "confidence", "time")
self.tree = ttk.Treeview(tab, columns=cols, show="headings", selectmode="browse")
self.tree.heading("sender", text="From")
self.tree.heading("subject", text="Subject")
self.tree.heading("verdict", text="Verdict")
self.tree.heading("confidence", text="Confidence")
self.tree.heading("time", text="Analysed At")
self.tree.column("sender", width=220, anchor="w")
self.tree.column("subject", width=300, anchor="w")
self.tree.column("verdict", width=90, anchor="center")
self.tree.column("confidence", width=100, anchor="center")
self.tree.column("time", width=130, anchor="center")
self.tree.tag_configure("SPAM", foreground="#f38ba8")
self.tree.tag_configure("HAM", foreground="#a6e3a1")
vsb = ttk.Scrollbar(tab, orient="vertical", command=self.tree.yview)
hsb = ttk.Scrollbar(tab, orient="horizontal", command=self.tree.xview)
self.tree.configure(yscrollcommand=vsb.set, xscrollcommand=hsb.set)
self.tree.pack(side="left", fill="both", expand=True, padx=(16,0), pady=(0,4))
vsb.pack(side="left", fill="y", pady=(0,4))
hsb.pack(side="bottom", fill="x", padx=16)
self.tree.bind("<<TreeviewSelect>>", self._on_select)
detail_frame = ttk.Frame(tab, style="Card.TFrame", padding=12)
detail_frame.pack(fill="x", padx=16, pady=(0,12), side="bottom")
ttk.Label(detail_frame, text="Reason", style="Card.TLabel",
font=("Segoe UI", 10, "bold")).pack(anchor="w")
self.reason_text = tk.Text(detail_frame, height=3, wrap="word",
bg="#2a2a3e", fg="#a6adc8", font=("Segoe UI", 9),
relief="flat", borderwidth=0, state="disabled")
self.reason_text.pack(fill="x", expand=False)
self._results_cache = {}
def _build_about_tab(self):
tab = ttk.Frame(self.notebook)
self.notebook.add(tab, text=" ℹ About ")
ttk.Label(tab, text="Gmail Spam Detector", style="Title.TLabel").pack(pady=(40,8))
ttk.Label(tab, text="Powered by Gmail API + Groq (LLaMA 3.1 8B)", style="Sub.TLabel").pack()
ttk.Label(tab, text="Built with Python & Tkinter", style="Sub.TLabel").pack(pady=4)
ttk.Label(tab, text="Your emails are never stored — analysis happens in real-time.",
style="Sub.TLabel").pack(pady=(16,4))
def _save_config(self):
self.config["client_id"] = self.client_id_var.get().strip()
self.config["client_secret"] = self.client_secret_var.get().strip()
self.config["groq_key"] = self.groq_key_var.get().strip()
save_config(self.config)
messagebox.showinfo("Saved", "Configuration saved.")
def _check_saved_auth(self):
if self.config.get("refresh_token"):
self._set_status("✅ Authenticated (saved session)", "#a6e3a1")
self.notebook.select(1)
def _set_status(self, msg, color="#a6adc8"):
self.auth_status.configure(text=msg, foreground=color)
def _start_oauth(self):
cid = self.client_id_var.get().strip()
csec = self.client_secret_var.get().strip()
if not cid or not csec:
messagebox.showerror("Missing", "Enter Client ID and Client Secret first.")
return
self._save_config()
self._set_status("⏳ Waiting for browser authentication…")
threading.Thread(target=self._oauth_flow, args=(cid, csec), daemon=True).start()
def _oauth_flow(self, cid, csec):
global _oauth_code
_oauth_code = None
auth_url = (
"https://accounts.google.com/o/oauth2/v2/auth?"
+ urllib.parse.urlencode({
"client_id": cid,
"redirect_uri": REDIRECT_URI,
"response_type": "code",
"scope": SCOPES,
"access_type": "offline",
"prompt": "consent",
})
)
subprocess.run(["cmd.exe", "/c", "start", auth_url.replace("&", "^&")])
with socketserver.TCPServer(("", 8080), OAuthHandler) as httpd:
httpd.timeout = 120
httpd.handle_request()
if not _oauth_code:
self.after(0, lambda: self._set_status("❌ Auth timed out.", "#f38ba8"))
return
try:
tokens = exchange_code_for_token(_oauth_code, cid, csec)
self.access_token = tokens["access_token"]
self.config["refresh_token"] = tokens.get("refresh_token", "")
save_config(self.config)
self.after(0, lambda: self._set_status("✅ Authenticated successfully!", "#a6e3a1"))
self.after(0, lambda: self.notebook.select(1))
except Exception as e:
self.after(0, lambda: self._set_status(f"❌ Token exchange failed: {e}", "#f38ba8"))
def _ensure_token(self):
if self.access_token:
return True
rt = self.config.get("refresh_token")
cid = self.config.get("client_id")
csec = self.config.get("client_secret")
if rt and cid and csec:
try:
tokens = refresh_access_token(cid, csec, rt)
self.access_token = tokens["access_token"]
return True
except Exception as e:
messagebox.showerror("Token Error", f"Could not refresh token:\n{e}")
return False
messagebox.showwarning("Not Authenticated", "Please authenticate first in the Setup tab.")
return False
def _fetch_and_analyse(self):
if not self._ensure_token():
return
gk = self.groq_key_var.get().strip() or self.config.get("groq_key", "")
if not gk:
messagebox.showerror("Missing", "Enter your Groq API key in Setup.")
return
self.fetch_btn.state(["disabled"])
self.progress.start(10)
self._results_cache.clear()
for row in self.tree.get_children():
self.tree.delete(row)
threading.Thread(target=self._worker, args=(gk,), daemon=True).start()
def _worker(self, groq_key):
try:
emails = fetch_unread_emails(self.access_token, max_results=5)
self.after(0, lambda: self.email_count_label.configure(
text=f"{len(emails)} unread email(s) found"))
for email in emails:
try:
result = classify_email(email, groq_key)
except Exception as e:
result = {"verdict": "ERROR", "confidence": "-", "reason": str(e)}
now = datetime.now().strftime("%H:%M:%S")
self.after(0, self._insert_row, email, result, now)
except urllib.error.HTTPError as e:
if e.code == 401:
self.access_token = None
self.after(0, lambda: messagebox.showerror(
"Auth Error", "Access token expired. Re-authenticate in Setup."))
else:
self.after(0, lambda: messagebox.showerror("API Error", str(e)))
except Exception as e:
self.after(0, lambda: messagebox.showerror("Error", str(e)))
finally:
self.after(0, self._done)
def _insert_row(self, email, result, timestamp):
verdict = result.get("verdict", "?")
tag = verdict if verdict in ("SPAM", "HAM") else ""
iid = self.tree.insert("", "end", tags=(tag,), values=(
email["sender"][:40],
email["subject"][:60],
verdict,
result.get("confidence", "-"),
timestamp,
))
self._results_cache[iid] = result.get("reason", "")
def _done(self):
self.progress.stop()
self.fetch_btn.state(["!disabled"])
def _on_select(self, _):
sel = self.tree.selection()
if sel:
reason = self._results_cache.get(sel[0], "")
text = f"AI Reasoning: {reason}" if reason else "No reason provided."
self.reason_text.configure(state="normal")
self.reason_text.delete("1.0", "end")
self.reason_text.insert("1.0", text)
self.reason_text.configure(state="disabled")
if name == "main": app = SpamDetectorApp() app.mainloop()