Spaces:
Paused
Paused
File size: 8,790 Bytes
ece3e8e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 | import imaplib
import email
from email.header import decode_header
import re
import time
def get_latest_otp(bot_email, bot_password, search_term, timeout=60):
"""
Connects to Gmail, waits for a new email containing the search_term (e.g. 'workday' or 'netflix'),
extracts a 6-digit OTP code or a verification link. It will leave the email in your inbox.
"""
print(f"[IMAP] Waiting up to {timeout} seconds for an email related to '{search_term}'...")
start_time = time.time()
while time.time() - start_time < timeout:
try:
mail = imaplib.IMAP4_SSL("imap.gmail.com")
mail.login(bot_email, bot_password)
mail.select("inbox")
# Search for ONLY UNSEEN emails since we are keeping them in the inbox now
status, messages = mail.search(None, "UNSEEN")
email_ids = messages[0].split()
if email_ids:
# Check from newest to oldest
for e_id in reversed(email_ids):
res, msg_data = mail.fetch(e_id, "(RFC822)")
for response_part in msg_data:
if isinstance(response_part, tuple):
msg = email.message_from_bytes(response_part[1])
# Decode subject
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else "utf-8")
sender = msg.get("From")
# If this email is from the company we are waiting for
if search_term.lower() in subject.lower() or search_term.lower() in sender.lower():
print(f"[IMAP] Found email: {subject} from {sender}")
# Extract Body
body = ""
if msg.is_multipart():
for part in msg.walk():
content_type = part.get_content_type()
if content_type == "text/plain" or content_type == "text/html":
try:
body = part.get_payload(decode=True).decode()
except:
pass
else:
body = msg.get_payload(decode=True).decode()
# Clean HTML tags if present to make regex more accurate
# We can do a simple replace or just rely on regex, but removing tags helps.
clean_body = re.sub(r'<[^>]+>', ' ', body)
# 1. Look for a 4 to 8 digit OTP code (some companies use 4, 6, or 8 digits)
otp_match = re.search(r'\b\d{4,8}\b', clean_body)
if otp_match:
code = otp_match.group(0)
print(f"[IMAP] Extracted OTP Code: {code}")
# Mark as read instead of deleting (optional, but safe)
mail.store(e_id, '+FLAGS', '\\Seen')
mail.logout()
return {"type": "code", "value": code}
# 2. Look for a verification link
link_match = re.search(r'href=[\'"]?([^\'" >]+)', body)
if link_match:
link = link_match.group(1)
if "verify" in link.lower() or "confirm" in link.lower():
print(f"[IMAP] Extracted Verification Link: {link}")
mail.store(e_id, '+FLAGS', '\\Seen')
mail.logout()
return {"type": "link", "value": link}
print("[IMAP] Found email but no OTP or Link inside.")
mail.logout()
except Exception as e:
print(f"[IMAP] Error reading email: {e}")
# Wait 5 seconds before checking again
time.sleep(5)
print("[IMAP] Timeout reached. No email arrived.")
return None
def scan_for_interview_invites():
import os
from dotenv import load_dotenv
load_dotenv(override=True)
bot_email = os.getenv("BOT_EMAIL")
bot_password = os.getenv("BOT_PASSWORD")
if not bot_email or not bot_password:
raise Exception("BOT_EMAIL or BOT_PASSWORD not set in .env")
mail = imaplib.IMAP4_SSL("imap.gmail.com")
mail.login(bot_email, bot_password)
mail.select("inbox")
# Search for UNSEEN emails
status, messages = mail.search(None, "UNSEEN")
email_ids = messages[0].split()
results = []
if not email_ids:
mail.logout()
return results
for e_id in reversed(email_ids): # Process newest first
res, msg_data = mail.fetch(e_id, "(RFC822)")
for response_part in msg_data:
if isinstance(response_part, tuple):
msg = email.message_from_bytes(response_part[1])
subject, encoding = decode_header(msg["Subject"])[0]
if isinstance(subject, bytes):
subject = subject.decode(encoding if encoding else "utf-8")
sender = msg.get("From", "")
subj_lower = subject.lower()
# We are looking for interviews OR rejections
is_interview = any(k in subj_lower for k in ["interview", "next steps", "assessment", "calendly", "invitation"])
is_rejection = any(k in subj_lower for k in ["update on your application", "unfortunately", "not moving forward", "status of your application"])
if is_interview or is_rejection:
body = ""
if msg.is_multipart():
for part in msg.walk():
if part.get_content_type() == "text/plain":
try: body = part.get_payload(decode=True).decode()
except: pass
break
else:
try: body = msg.get_payload(decode=True).decode()
except: pass
if not body: continue
# Determine status
if is_interview: new_status = "Interview"
elif "unfortunately" in body.lower() or "not moving forward" in body.lower(): new_status = "Rejected"
else: continue # False positive
# Extract company name using Gemini
from main import get_gemini_client
client = get_gemini_client()
company_name = sender.split("@")[-1].split(".")[0].title() # fallback
if client:
try:
prompt = f"Extract the company name from this email. Reply ONLY with the company name, nothing else. Email From: {sender}\nSubject: {subject}\nBody: {body[:1000]}"
resp = client.models.generate_content(model="gemini-2.5-flash", contents=prompt)
company_name = resp.text.strip()
except: pass
results.append({"company": company_name, "status": new_status, "subject": subject})
# Update Notion CRM
from bot_features import sync_to_notion
sync_to_notion("", f"Update from email: {subject}", new_status, client, override_company=company_name)
mail.store(e_id, '+FLAGS', '\\Seen')
# Limit to 20 emails per scan
if len(results) >= 20: break
mail.logout()
return results
|