""" ╔══════════════════════════════════════════════════════════╗ ║ EMAIL JOB TRACKER — Hugging Face Spaces v2.1 ║ ║ SDK: Gradio | LLaMA-3 via Groq | Self-Learn ║ ╚══════════════════════════════════════════════════════════╝ Files needed: app.py ← this file requirements.txt ← gradio / imapclient / pyzmail36 / openai / pandas """ import gradio as gr import imapclient import pyzmail import pandas as pd import json import os import traceback from datetime import datetime from openai import OpenAI # ────────────────────────────────────────────────────────── # CONSTANTS # ────────────────────────────────────────────────────────── IMAP_SERVER = "imap.gmail.com" LEARN_FILE = "/tmp/learned_keywords.json" GMAIL_FOLDERS = [ "INBOX", "[Gmail]/Important", "[Gmail]/Starred", "[Gmail]/Sent Mail", "[Gmail]/Drafts", ] CATEGORY_COLORS = { "Offer": "#10b981", "Interview": "#3b82f6", "New Opportunity": "#8b5cf6", "Rejected": "#f43f5e", "Urgent": "#f97316", "Alerts": "#facc15", "Spam": "#6b7280", "Unknown": "#94a3b8", } CATEGORY_ICONS = { "Offer": "🎉", "Interview": "📅", "New Opportunity": "🚀", "Rejected": "❌", "Urgent": "🔥", "Alerts": "🔔", "Spam": "🗑️", "Unknown": "❓", } # ────────────────────────────────────────────────────────── # SELF-LEARNING # ────────────────────────────────────────────────────────── def load_learnings(): try: if os.path.exists(LEARN_FILE): with open(LEARN_FILE) as f: return json.load(f) except Exception: pass return {} def save_learnings(lk): try: with open(LEARN_FILE, "w") as f: json.dump(lk, f, indent=2) except Exception: pass # ────────────────────────────────────────────────────────── # KEYWORD ENGINE # ────────────────────────────────────────────────────────── def get_base_keywords(): return { "Offer": [ "offer letter","congratulations","selected","employment offer", "welcome aboard","joining","ctc","salary","compensation", "offer rollout","final selection","offer acceptance", ], "Interview": [ "interview","round","technical","hr round","final round", "assessment","test","assignment","panel","discussion", "screening","shortlisted","interview invite", ], "New Opportunity": [ "job opportunity","hiring","opening","vacancy","role opportunity", "we are hiring","position available","jd attached", "job description","career opportunity","looking for candidates", ], "Rejected": [ "regret","unfortunately","not selected","not shortlisted", "rejected","not a fit","position filled","application unsuccessful", ], "Urgent": [ "urgent","immediate","asap","priority","today","tomorrow", "deadline","action required","important","quick response", ], "Alerts": [ "alert","notification","reminder","update","system alert", "security alert","account alert","important update", ], "Spam": [ "sale","discount","offer deal","win","lottery","free", "click here","subscribe","buy now","limited offer", ], } def rule_based_classification(text, lk): text = text.lower() kw = get_base_keywords() for cat in lk: kw.setdefault(cat, []).extend(lk[cat]) scores = {c: sum(1 for w in words if w in text) for c, words in kw.items()} best = max(scores, key=scores.get) return best if scores[best] > 0 else "Unknown" def learn_from_email(text, category, lk): filtered = [w for w in text.lower().split() if len(w) > 6] lk.setdefault(category, []) for w in filtered[:5]: if w not in lk[category]: lk[category].append(w) save_learnings(lk) # ────────────────────────────────────────────────────────── # LLM CLASSIFIER # ────────────────────────────────────────────────────────── def classify_email(subject, body, client, lk): text = (subject + " " + body)[:1500] prompt = f"""Classify this email into exactly ONE of: New Opportunity, Interview, Spam, Unknown, Offer, Rejected, Urgent, Alerts Extract company name, role/position, and interview round if present. Email: {text} Return ONLY valid JSON, no markdown, no backticks: {{ "category": "", "company": "", "role": "", "round": "", "confidence": 0.0 }}""" try: res = client.chat.completions.create( model="llama3-70b-8192", messages=[{"role": "user", "content": prompt}], temperature=0, ) output = res.choices[0].message.content.strip() output = output.replace("```json", "").replace("```", "").strip() result = json.loads(output) if result.get("confidence", 0) > 0.8: learn_from_email(text, result["category"], lk) return result except Exception as e: print(f"[LLM ERROR] {e}") cat = rule_based_classification(text, lk) learn_from_email(text, cat, lk) return {"category": cat, "company": "Unknown", "role": "Unknown", "round": "N/A", "confidence": 0.6} # ────────────────────────────────────────────────────────── # EMAIL FETCH # ────────────────────────────────────────────────────────── def fetch_all_emails(email, password, limit): print(f"[IMAP] Connecting to {IMAP_SERVER} as {email}") mail = imapclient.IMAPClient(IMAP_SERVER, ssl=True) mail.login(email, password) print("[IMAP] Login successful") collected = [] for folder in GMAIL_FOLDERS: try: mail.select_folder(folder, readonly=True) uids = mail.search(["ALL"])[-limit:] print(f"[IMAP] {folder}: found {len(uids)} emails") for uid in uids: try: raw = mail.fetch(uid, ["BODY[]"]) msg = pyzmail.PyzMessage.factory(raw[uid][b"BODY[]"]) subj = msg.get_subject() or "(no subject)" if msg.text_part: body = msg.text_part.get_payload().decode( msg.text_part.charset or "utf-8", errors="replace") elif msg.html_part: body = msg.html_part.get_payload().decode( msg.html_part.charset or "utf-8", errors="replace") else: body = "" collected.append({"folder": folder, "subject": subj, "body": body}) except Exception as e: print(f"[IMAP] UID {uid} error: {e}") continue except Exception as e: print(f"[IMAP] Folder '{folder}' skipped: {e}") continue mail.logout() print(f"[IMAP] Total emails fetched: {len(collected)}") return collected # ────────────────────────────────────────────────────────── # HTML BUILDERS # ────────────────────────────────────────────────────────── def make_summary_cards(counts): cards = "" for cat, cnt in counts.items(): color = CATEGORY_COLORS.get(cat, "#94a3b8") icon = CATEGORY_ICONS.get(cat, "•") cards += f"""
{icon}
{cnt}
{cat}
""" return f'
{cards}
' def make_log_html(log_items): rows = "" for item in log_items: color = CATEGORY_COLORS.get(item["category"], "#94a3b8") icon = CATEGORY_ICONS.get(item["category"], "•") conf_pct = int(item["confidence"] * 100) conf_color = "#10b981" if conf_pct >= 80 else "#facc15" if conf_pct >= 60 else "#f43f5e" folder_short = item["folder"].replace("[Gmail]/", "") rows += f"""
{icon} {item["category"]} [{folder_short}] {item["subject"][:72]} {item["company"]} {conf_pct}%
""" header = """
Category Folder Subject Company Conf.
""" return f"""
{header}
{rows}
""" def make_pipeline_html(tracker_df): if tracker_df is None or tracker_df.empty: return "" sections = "" for cat in tracker_df["Category"].unique(): color = CATEGORY_COLORS.get(cat, "#94a3b8") icon = CATEGORY_ICONS.get(cat, "•") subset = tracker_df[tracker_df["Category"] == cat] items = "" for _, row in subset.iterrows(): round_text = "" if str(row.get("Round", "N/A")) not in ["N/A", "", "None", "nan"]: round_text = f" · {row['Round']}" items += f"""
{row.get("Company","—")}
{row.get("Role","—")}{round_text}
""" sections += f"""
{icon} {cat} ({len(subset)})
{items}
""" return f"""
{sections}
""" def make_error_html(msg): return f"""
❌ Error

{msg}
""" def make_info_html(msg, color="#facc15"): return f"""
{msg}
""" # ────────────────────────────────────────────────────────── # MAIN PIPELINE # ────────────────────────────────────────────────────────── def run_pipeline(email, password, groq_key, limit, progress=gr.Progress(track_tqdm=True)): # ── Validation ── if not email.strip() or not password.strip() or not groq_key.strip(): yield make_error_html("Please fill in all three credential fields."), \ "", "", None, None return # ── Init ── client = OpenAI(api_key=groq_key.strip(), base_url="https://api.groq.com/openai/v1") lk = load_learnings() yield make_info_html("🔌 Connecting to Gmail IMAP…"), "", "", None, None # ── Fetch ── try: emails = fetch_all_emails(email.strip(), password.strip(), int(limit)) except Exception as e: tb = traceback.format_exc() print(f"[FETCH ERROR]\n{tb}") # Show a friendly + technical message err_detail = str(e) tip = "" if "AUTHENTICATIONFAILED" in err_detail or "Invalid credentials" in err_detail: tip = "

Fix: Wrong email or App Password. Make sure you generated an App Password (not your real password) at myaccount.google.com/apppasswords." elif "IMAP access is disabled" in err_detail: tip = "

Fix: Enable IMAP in Gmail → Settings → See all settings → Forwarding and POP/IMAP → Enable IMAP." elif "timed out" in err_detail.lower() or "connect" in err_detail.lower(): tip = "

Fix: Network timeout — HF Spaces may be blocking outbound IMAP (port 993). Try running locally instead." yield make_error_html(f"{err_detail}{tip}"), "", "", None, None return if not emails: yield make_info_html("⚠️ No emails found in any folder.", "#f97316"), "", "", None, None return # ── Classify ── records, log_items = [], [] total = len(emails) for i, mail in enumerate(emails): progress((i + 1) / total) result = classify_email(mail["subject"], mail["body"], client, lk) rec = { "Timestamp": datetime.now().strftime("%Y-%m-%d %H:%M"), "Folder": mail["folder"], "Company": result.get("company", "Unknown"), "Role": result.get("role", "Unknown"), "Category": result.get("category", "Unknown"), "Round": result.get("round", "N/A"), "Confidence": round(result.get("confidence", 0.5), 2), "Subject": mail["subject"], } records.append(rec) log_items.append({ "folder": mail["folder"], "subject": mail["subject"], "category": rec["Category"], "company": rec["Company"], "confidence": rec["Confidence"], }) # stream update every 5 emails if (i + 1) % 5 == 0 or (i + 1) == total: df_p = pd.DataFrame(records) yield ( make_summary_cards(df_p["Category"].value_counts()), make_log_html(log_items), "", None, None, ) # ── Save & Final ── df = pd.DataFrame(records) tracker = df.groupby(["Company", "Role"]).last().reset_index() df.to_csv("/tmp/email_log.csv", index=False) tracker.to_csv("/tmp/job_tracker.csv", index=False) print(f"[DONE] {len(records)} emails classified, {len(tracker)} unique jobs tracked") yield ( make_summary_cards(tracker["Category"].value_counts()), make_log_html(log_items), make_pipeline_html(tracker), "/tmp/email_log.csv", "/tmp/job_tracker.csv", ) # ────────────────────────────────────────────────────────── # GRADIO CSS # ────────────────────────────────────────────────────────── CSS = """ @import url('https://fonts.googleapis.com/css2?family=JetBrains+Mono:wght@300;400;600;800&family=DM+Sans:wght@300;400;500;700&display=swap'); *, *::before, *::after { box-sizing: border-box; } body, .gradio-container { background: #010409 !important; color: #e6edf3 !important; font-family: 'DM Sans', sans-serif !important; } /* Tabs */ .tab-nav { background: #0d1117 !important; border-bottom: 1px solid #21262d !important; } .tab-nav button { color: #8b949e !important; font-family: 'DM Sans', sans-serif !important; font-size: .85rem !important; } .tab-nav button.selected { color: #e6edf3 !important; border-bottom: 2px solid #58a6ff !important; background: transparent !important; } /* Labels */ label > span { color: #8b949e !important; font-size: .68rem !important; letter-spacing: .12em !important; text-transform: uppercase !important; font-family: 'JetBrains Mono', monospace !important; } /* Inputs */ input[type=text], input[type=password], textarea { background: #0d1117 !important; border: 1px solid #30363d !important; color: #e6edf3 !important; border-radius: 6px !important; font-family: 'DM Sans', sans-serif !important; font-size: .88rem !important; } input[type=text]:focus, input[type=password]:focus { border-color: #58a6ff !important; box-shadow: 0 0 0 3px #58a6ff18 !important; } /* Slider */ input[type=range] { accent-color: #58a6ff !important; } /* Buttons */ .gr-button { font-family: 'DM Sans', sans-serif !important; border-radius: 6px !important; font-weight: 600 !important; font-size: .88rem !important; transition: all .18s !important; } .gr-button-primary { background: #238636 !important; border: 1px solid #2ea043 !important; color: #fff !important; } .gr-button-primary:hover { background: #2ea043 !important; transform: translateY(-1px) !important; box-shadow: 0 4px 16px #2ea04344 !important; } .gr-button-secondary { background: #21262d !important; border: 1px solid #30363d !important; color: #c9d1d9 !important; } .gr-button-secondary:hover { background: #30363d !important; } /* Panels */ .gr-panel, .gr-box, .gr-form { background: #0d1117 !important; border: 1px solid #21262d !important; border-radius: 10px !important; } /* Dataframe */ .gr-dataframe table { background: #0d1117 !important; font-family: 'JetBrains Mono', monospace !important; font-size: .78rem !important; } .gr-dataframe th { background: #161b22 !important; color: #8b949e !important; letter-spacing: .06em !important; text-transform: uppercase !important; } .gr-dataframe td { color: #c9d1d9 !important; border-color: #21262d !important; } .gr-dataframe tr:hover td { background: #161b22 !important; } /* Scrollbar */ ::-webkit-scrollbar { width: 5px; height: 5px; } ::-webkit-scrollbar-track { background: #0d1117; } ::-webkit-scrollbar-thumb { background: #30363d; border-radius: 3px; } ::-webkit-scrollbar-thumb:hover { background: #484f58; } """ # ────────────────────────────────────────────────────────── # STATIC HTML BLOCKS # ────────────────────────────────────────────────────────── HEADER_HTML = """
📬
LLaMA-3 · Groq · Self-Learning · Gradio

Email Job Tracker

Fetch → classify → track your entire job pipeline automatically

v2.1 HF SPACES
""" CRED_HELP = """
🔐 Setup Help
• Use a Gmail App Password — not your real password
• Requires 2FA ON → Generate App Password
• Enable IMAP: Gmail → Settings → Forwarding & POP/IMAP → Enable IMAP
• Free Groq key → console.groq.com
• Credentials are never stored
""" GUIDE_HTML = """

🚀 Quick Start

  1. Enable 2-Step Verification on your Google account
  2. Generate a Gmail App Password (16 chars — enter without spaces)
  3. Enable IMAP: Gmail → ⚙️ Settings → See all settings → Forwarding and POP/IMAP → Enable IMAP → Save
  4. Get a free API key from Groq Console
  5. Enter credentials in Run Agent tab and click Run

🏷️ Categories

Category Trigger Keywords
🎉 OfferOffer letter, CTC, Welcome aboard, Salary
📅 InterviewInvite, Round, Assessment, Screening, Shortlisted
🚀 New OpportunityHiring, Opening, JD Attached, Vacancy
❌ RejectedRegret, Unfortunately, Not selected, Not a fit
🔥 UrgentASAP, Deadline, Action Required, Priority
🔔 AlertsSecurity Alert, Notification, System Update
🗑️ SpamSale, Lottery, Buy Now, Free, Subscribe

🧠 Self-Learning

Every classification with >80% confidence teaches the system new keywords for that category, saved to /tmp/learned_keywords.json. Accuracy improves the more emails you process.

🛠️ Troubleshooting

Error Fix
AUTHENTICATIONFAILEDWrong App Password or email — regenerate App Password
IMAP access disabledEnable IMAP in Gmail settings (step 3 above)
Connection timeoutHF Spaces may block port 993 — run locally with python app.py
Groq errorCheck API key at console.groq.com — falls back to rule-based
""" EMPTY_LOG = """
░░ no data yet — run the agent ░░
""" def sec(label, color="#58a6ff"): return (f'
{label}
') # ────────────────────────────────────────────────────────── # GRADIO LAYOUT # ────────────────────────────────────────────────────────── with gr.Blocks(css=CSS, title="📬 Email Job Tracker") as demo: gr.HTML(HEADER_HTML) with gr.Tabs(): # ══ TAB 1: RUN AGENT ══════════════════════════ with gr.TabItem("⚡ Run Agent"): with gr.Row(equal_height=False): with gr.Column(scale=1, min_width=300): gr.HTML(sec("Credentials")) email_in = gr.Textbox(label="Gmail Address", placeholder="you@gmail.com") pass_in = gr.Textbox(label="Gmail App Password", type="password", placeholder="xxxx xxxx xxxx xxxx") groq_in = gr.Textbox(label="Groq API Key", type="password", placeholder="gsk_...") gr.HTML(sec("Settings", "#8b949e")) limit_in = gr.Slider(label="Emails per folder", minimum=5, maximum=50, value=20, step=5) with gr.Row(): run_btn = gr.Button("⚡ Run Agent", variant="primary", size="lg") clear_btn = gr.Button("✕ Clear", variant="secondary", size="lg") gr.HTML(CRED_HELP) with gr.Column(scale=2): gr.HTML(sec("Summary")) summary_out = gr.HTML(value=EMPTY_LOG) gr.HTML(sec("Live Log")) log_out = gr.HTML(value=EMPTY_LOG) gr.HTML(sec("Pipeline View")) pipeline_out = gr.HTML(value=EMPTY_LOG) gr.HTML(sec("Downloads", "#8b949e")) with gr.Row(): file_log = gr.File(label="📄 email_log.csv") file_tracker = gr.File(label="📊 job_tracker.csv") # ══ TAB 2: TRACKER TABLE ══════════════════════ with gr.TabItem("📊 Tracker Table"): gr.HTML(sec("Latest status per company / role")) table_out = gr.Dataframe( headers=["Company","Role","Category","Round","Confidence","Subject"], interactive=False, wrap=False, ) # ══ TAB 3: GUIDE ══════════════════════════════ with gr.TabItem("📖 Guide"): gr.HTML(GUIDE_HTML) # ── EVENT HANDLERS ────────────────────────────── def do_run(email, password, groq_key, limit, progress=gr.Progress(track_tqdm=True)): last = None for result in run_pipeline(email, password, groq_key, limit, progress): last = result yield result[0], result[1], result[2], result[3], result[4], gr.update() if last is None: return # populate tracker table try: df = pd.read_csv("/tmp/job_tracker.csv") cols = [c for c in ["Company","Role","Category","Round","Confidence","Subject"] if c in df.columns] yield last[0], last[1], last[2], last[3], last[4], df[cols] except Exception: yield last[0], last[1], last[2], last[3], last[4], gr.update() def do_clear(): return EMPTY_LOG, EMPTY_LOG, EMPTY_LOG, None, None, None run_btn.click( fn=do_run, inputs=[email_in, pass_in, groq_in, limit_in], outputs=[summary_out, log_out, pipeline_out, file_log, file_tracker, table_out], ) clear_btn.click( fn=do_clear, outputs=[summary_out, log_out, pipeline_out, file_log, file_tracker, table_out], ) if __name__ == "__main__": demo.launch(ssr_mode=False)