import os import re import uuid import pytz import psycopg2 import requests import gradio as gr import json import pickle import base64 from google.auth.transport.requests import Request from googleapiclient.discovery import build from google.oauth2.credentials import Credentials from email.mime.text import MIMEText from datetime import datetime, date from dotenv import load_dotenv # ================= LOAD ENV ================= load_dotenv() DB_URL = os.getenv("DB_URL") GMAIL_USER = os.getenv("GMAIL_USER") CRON_SECRET = os.getenv("CRON") HF_URL = os.getenv("HF_URL") LEETCODE_API = "https://leetcode-api-vercel.vercel.app" SCOPES = ["https://www.googleapis.com/auth/gmail.send"] TOKEN_FILE = "token.pkl" # ================= DB ================= def get_db(): return psycopg2.connect(DB_URL, sslmode="require") # ================= GMAIL ================= def get_gmail_service(): creds = None # First, try to load existing token if os.path.exists(TOKEN_FILE): try: with open(TOKEN_FILE, "rb") as f: creds = pickle.load(f) except Exception as e: print(f"⚠ī¸ Failed to load token.pkl: {e}") creds = None # Refresh token if needed if creds and creds.expired and creds.refresh_token: try: print("🔄 Refreshing expired token...") creds.refresh(Request()) # Save refreshed token with open(TOKEN_FILE, "wb") as f: pickle.dump(creds, f) print("✅ Token refreshed and saved") except Exception as e: print(f"❌ Failed to refresh token: {e}") creds = None # If no valid token, try to create from credentials.json if not creds or not creds.valid: creds = create_token_from_credentials() return build("gmail", "v1", credentials=creds) def create_token_from_credentials(): """Create token.pkl from credentials.json using OAuth flow""" credentials_file = "credentials.json" if not os.path.exists(credentials_file): raise Exception(f""" ❌ Neither {TOKEN_FILE} nor {credentials_file} found! Please provide one of: 1. Upload token.pkl (ready-to-use token) 2. Upload credentials.json (will generate token.pkl) To get credentials.json: 1. Go to Google Cloud Console 2. Enable Gmail API 3. Create OAuth 2.0 credentials 4. Download as credentials.json """) try: print(f"📁 Found {credentials_file}, creating token...") from google_auth_oauthlib.flow import InstalledAppFlow # Create flow from credentials flow = InstalledAppFlow.from_client_secrets_file( credentials_file, SCOPES ) # Run local server for OAuth (will open browser) creds = flow.run_local_server( port=8080, prompt='consent', authorization_prompt_message='Please visit this URL to authorize: {url}', success_message='Authorization complete! You can close this window.', open_browser=True ) # Save token for future use with open(TOKEN_FILE, "wb") as f: pickle.dump(creds, f) print(f"✅ Token created and saved to {TOKEN_FILE}") print("ℹī¸ You can now delete credentials.json for security") return creds except ImportError: raise Exception(""" ❌ Missing required package! Please install: pip install google-auth-oauthlib This is needed to create token from credentials.json """) except Exception as e: raise Exception(f""" ❌ Failed to create token from credentials.json: {e} Common issues: 1. Invalid credentials.json format 2. Gmail API not enabled 3. OAuth consent screen not configured 4. Port 8080 already in use Please check your Google Cloud Console settings. """) def validate_gmail_setup(): """Validate Gmail API setup and provide helpful error messages""" try: service = get_gmail_service() # Test API access profile = service.users().getProfile(userId="me").execute() email = profile.get('emailAddress') print(f"✅ Gmail API connected: {email}") return True, email except Exception as e: error_msg = str(e) if "credentials" in error_msg.lower(): return False, "❌ Missing credentials. Upload token.pkl or credentials.json" elif "quota" in error_msg.lower(): return False, "❌ Gmail API quota exceeded. Try again later" elif "permission" in error_msg.lower(): return False, "❌ Insufficient permissions. Re-authorize with full Gmail access" else: return False, f"❌ Gmail API error: {error_msg}" def send_email(to, subject, html): try: service = get_gmail_service() msg = MIMEText(html, "html") msg["To"] = to msg["From"] = GMAIL_USER msg["Subject"] = subject raw = base64.urlsafe_b64encode( msg.as_bytes() ).decode() body = {"raw": raw} service.users().messages().send( userId="me", body=body ).execute() print("✅ Sent:", to) return True except Exception as e: print("❌ Gmail error:", e) return False def create_email_template(title, content, unsubscribe_link, email_type="reminder", problem_link=None, difficulty=None): """Create a beautiful HTML email template""" # Color scheme based on email type colors = { "morning": {"primary": "#4CAF50", "secondary": "#81C784", "bg": "#E8F5E8"}, "afternoon": {"primary": "#FF9800", "secondary": "#FFB74D", "bg": "#FFF3E0"}, "night": {"primary": "#F44336", "secondary": "#EF5350", "bg": "#FFEBEE"}, "verification": {"primary": "#2196F3", "secondary": "#64B5F6", "bg": "#E3F2FD"} } color = colors.get(email_type, colors["morning"]) # Difficulty badge colors difficulty_colors = { "Easy": "#00B04F", "Medium": "#FFA116", "Hard": "#FF375F" } # Handle special cases for verification emails if email_type == "verification": problem_button = f"""
✅ Verify Email
""" tips_section = "" motivation_section = "" difficulty_badge = "" else: # Use provided problem_link or construct from title if problem_link: link_url = problem_link else: link_url = f"https://leetcode.com/problems/{title.lower().replace(' ', '-').replace('.', '')}" # Add difficulty badge if difficulty: diff_color = difficulty_colors.get(difficulty, "#666") difficulty_badge = f"""
{difficulty}
""" else: difficulty_badge = "" problem_button = f"""
🚀 Solve Problem
""" # Dynamic tips based on difficulty if difficulty == "Easy": tips_content = """
  • Focus on understanding the problem clearly
  • Think about the simplest approach first
  • Test with the given examples
  • Consider edge cases like empty inputs
    • """ elif difficulty == "Medium": tips_content = """
  • Break the problem into smaller subproblems
  • Consider multiple approaches (greedy, DP, etc.)
  • Think about time and space complexity
  • Use appropriate data structures
    • """ else: # Hard tips_content = """
  • Study similar problems and patterns
  • Don't rush - take time to understand
  • Consider advanced algorithms and techniques
  • Break it down step by step
    • """ tips_section = f"""

    💡 {difficulty} Problem Tips

    """ motivation_quotes = [ "The expert in anything was once a beginner.", "Every problem is a step forward in your journey.", "Consistency beats perfection every time.", "Code today, conquer tomorrow.", "Small progress is still progress." ] import random quote = random.choice(motivation_quotes) motivation_section = f"""

    "{quote}" đŸ’Ē

    """ return f""" LeetCode Daily Tracker

    🧠 LeetCode Daily Tracker

    Your coding journey companion

    {content} {difficulty_badge}
    {problem_button} {tips_section} {motivation_section}

    Keep coding, keep growing! 🌱

    📊 LeetCode | đŸ’ģ GitHub | {'Unsubscribe' if email_type != 'verification' else ''}

    Š 2024 LeetCode Daily Tracker. Made with ❤ī¸ for coders.

    """ # ================= VALIDATION ================= EMAIL_REGEX = re.compile( r"^[\w\.-]+@[\w\.-]+\.\w+$" ) def valid_email(email): return bool(email and EMAIL_REGEX.match(email)) def valid_leetcode(username): if not username or len(username) < 3: return False try: r = requests.get( f"{LEETCODE_API}/{username}", timeout=8 ) return r.status_code == 200 except: return False # ================= LEETCODE ================= def get_daily_problem(): try: r = requests.get(f"{LEETCODE_API}/daily", timeout=10) r.raise_for_status() d = r.json() print("📡 Daily API response keys:", list(d.keys())) # Handle the current API format if "questionTitle" in d and "titleSlug" in d: title = d["questionTitle"] slug = d["titleSlug"] link = d.get("questionLink", f"https://leetcode.com/problems/{slug}/") difficulty = d.get("difficulty", "Unknown") print(f"✅ Found daily problem: {title} ({difficulty}) - {slug}") return title, slug, link, difficulty # Fallback for older format if "title" in d and "titleSlug" in d: title = d["title"] slug = d["titleSlug"] link = f"https://leetcode.com/problems/{slug}/" difficulty = d.get("difficulty", "Unknown") print(f"✅ Found daily problem (fallback): {title} ({difficulty}) - {slug}") return title, slug, link, difficulty # If we can't find the expected fields, print the response print("❌ Available fields in API response:", list(d.keys())) raise Exception("Could not find title and slug in daily API response") except requests.exceptions.RequestException as e: print(f"❌ Network error calling daily API: {e}") raise Exception(f"Failed to fetch daily problem: {e}") except json.JSONDecodeError as e: print(f"❌ JSON decode error: {e}") raise Exception("Invalid JSON response from daily API") def solved_today(username, slug): try: r = requests.get( f"{LEETCODE_API}/{username}/acSubmission?limit=20", timeout=10 ) if r.status_code != 200: return False d = r.json() if "submission" in d: subs = d["submission"] elif "data" in d: subs = d["data"] elif isinstance(d, list): subs = d else: return False today = date.today() for s in subs: if not isinstance(s, dict): continue if s.get("titleSlug") != slug: continue ts = s.get("timestamp") if not ts: continue solved = datetime.fromtimestamp( int(ts), tz=pytz.utc ).date() if solved == today: return True return False except: return False # ================= SUBSCRIBE ================= def subscribe(username, email, timezone): if not valid_leetcode(username): return "❌ Invalid LeetCode username" if not valid_email(email): return "❌ Invalid email" conn = get_db() cur = conn.cursor() cur.execute(""" SELECT email_verified, verification_token, unsubscribed FROM users WHERE email=%s """, (email,)) row = cur.fetchone() # Existing user logic... if row: verified, token, unsub = row if verified and not unsub: cur.close() conn.close() return "⚠ī¸ Already subscribed" if verified and unsub: cur.execute(""" UPDATE users SET unsubscribed=false, leetcode_username=%s, timezone=%s, last_sent_date=NULL, last_sent_slot=NULL WHERE email=%s """, (username, timezone, email)) conn.commit() cur.close() conn.close() return "✅ Re-subscribed" # Resend verification with enhanced template verification_content = f"""

    Welcome back! 👋

    We're excited to have you on your coding journey again!

    Click the verification button above to activate your daily LeetCode reminders.

    """ html_email = create_email_template( "Email Verification", verification_content, f"{HF_URL}?verify={token}", "verification" ) send_email(email, "🔔 Please verify your email", html_email) cur.close() conn.close() return "📩 Verification re-sent" # New user - remove duplicate code token = uuid.uuid4().hex cur.execute(""" INSERT INTO users( leetcode_username,email,timezone, email_verified,verification_token,unsubscribed ) VALUES(%s,%s,%s,false,%s,false) """, (username, email, timezone, token)) conn.commit() cur.close() conn.close() # Welcome email with enhanced template welcome_content = f"""

    Welcome to the club! 🎉

    You're about to embark on an amazing coding journey with daily LeetCode challenges!

    📅 Your Schedule:

    🌅 9 AM - Daily problem
    🌆 3 PM - Gentle reminder
    🌙 8 PM - Final reminder

    Click the verification button above to start receiving your personalized reminders!

    """ html_email = create_email_template( "Welcome", welcome_content, f"{HF_URL}?verify={token}", "verification" ) send_email(email, "đŸŽ¯ Verify your LeetCode journey!", html_email) return "📩 Verification sent" # ================= VERIFY ================= def verify_user(token): conn = get_db() cur = conn.cursor() cur.execute(""" UPDATE users SET email_verified=true WHERE verification_token=%s AND email_verified=false """, (token,)) ok = cur.rowcount conn.commit() cur.close() conn.close() if ok == 0: return "❌ Invalid link" return "✅ Email verified" # ================= UNSUBSCRIBE ================= def unsubscribe_user(token): conn = get_db() cur = conn.cursor() cur.execute(""" UPDATE users SET unsubscribed=true WHERE verification_token=%s """, (token,)) ok = cur.rowcount conn.commit() cur.close() conn.close() if ok == 0: return "❌ Invalid link" return "✅ Unsubscribed" # ================= URL HANDLER ================= def handle_url(request: gr.Request): try: params = request.query_params if "verify" in params: return verify_user(params["verify"]) if "unsubscribe" in params: return unsubscribe_user(params["unsubscribe"]) return "" except Exception as e: print("URL error:", e) return "" def run_scheduler(secret): if secret != CRON_SECRET: return "❌ Unauthorized" conn = get_db() cur = conn.cursor() cur.execute(""" SELECT id,leetcode_username,email,timezone, last_sent_date,last_sent_slot,verification_token FROM users WHERE email_verified=true AND unsubscribed=false """) users = cur.fetchall() try: title, slug, problem_link, difficulty = get_daily_problem() except Exception as e: cur.close() conn.close() return f"❌ Failed to get daily problem: {e}" now = datetime.now(pytz.utc) sent = 0 for uid, user, mail, tz, last_d, last_s, token in users: try: local = now.astimezone(pytz.timezone(tz)) h = local.hour # Enhanced email content based on time if 8 <= h <= 9: slot = "morning" subject = f"🌅 Today's LeetCode Challenge: {title}" content = f"""

    Good morning, coder! ☀ī¸

    Today's Problem: {title}

    Start your day with a fresh challenge! This {difficulty} problem is perfect for warming up your coding muscles. Take your time to understand the requirements! 🚀

    """ email_type = "morning" elif 14 <= h <= 15: slot = "afternoon" subject = f"⏰ Afternoon Coding Break: {title}" content = f"""

    Time for a coding break! ⚡

    Haven't tackled {title} yet? No worries!

    This {difficulty} problem is waiting for you. Sometimes a fresh afternoon perspective can lead to breakthrough solutions! 💡

    """ email_type = "afternoon" elif 19 <= h <= 20: slot = "night" subject = f"🌙 Last Call: {title}" content = f"""

    Final reminder! đŸ”Ĩ

    {title} ({difficulty}) is still waiting for you!

    Don't let the day end without giving it a try. Even reading through the problem and thinking about approaches counts as progress!

    đŸ’Ē Remember: Consistency beats perfection. Every attempt makes you stronger!

    """ email_type = "night" else: continue today = date.today() # Check if already sent today if last_d == today and last_s == slot: print(f"⏭ī¸ Skipping {mail} - already sent {slot} email today") continue # Check if user already solved the problem if solved_today(user, slug): print(f"✅ {user} already solved {slug} - skipping email") continue # Create beautiful HTML email with all enhancements html_email = create_email_template( title, content, f"{HF_URL}?unsubscribe={token}", email_type, problem_link, difficulty ) ok = send_email(mail, subject, html_email) if not ok: print(f"❌ Failed to send email to {mail}") continue # Update database cur.execute(""" UPDATE users SET last_sent_date=%s, last_sent_slot=%s WHERE id=%s """, (today, slot, uid)) sent += 1 except Exception as e: print(f"❌ Error processing user {user} ({mail}): {e}") continue conn.commit() cur.close() conn.close() return f"✅ Scheduler completed. Sent: {sent} emails" # ================= UI ================= with gr.Blocks( title="LeetCode Notifier", theme=gr.themes.Soft(), css=""" .gradio-container { max-width: 800px !important; margin: auto !important; } """ ) as app: gr.Markdown(""" # đŸ“Ŧ LeetCode Daily Email Notifier Get personalized daily LeetCode problem reminders sent directly to your inbox! Never miss a day of coding practice. """) with gr.Row(): with gr.Column(): u = gr.Textbox( label="🧑‍đŸ’ģ LeetCode Username", placeholder="Enter your LeetCode username", info="We'll verify this username exists on LeetCode" ) m = gr.Textbox( label="📧 Email Address", placeholder="your.email@gmail.com", info="You'll receive a verification email" ) tz = gr.Dropdown( choices=sorted(pytz.all_timezones), value="Asia/Kolkata", label="🌍 Timezone", info="Choose your timezone for proper scheduling" ) with gr.Row(): subscribe_btn = gr.Button("🚀 Subscribe", variant="primary", scale=2) out = gr.Textbox(label="📝 Status", interactive=False, lines=2) subscribe_btn.click(subscribe, [u, m, tz], out) gr.Markdown(""" --- ### ⏰ Email Schedule - **🌅 9:00 AM** - Daily problem notification - **🌆 3:00 PM** - Gentle reminder (if not solved) - **🌙 8:00 PM** - Final reminder (if not solved) ### 🔒 Admin Panel """) with gr.Row(): sec = gr.Textbox( label="🔑 Secret Key", type="password", placeholder="Enter scheduler secret key" ) run_btn = gr.Button("â–ļī¸ Run Scheduler", variant="secondary") run_btn.click(run_scheduler, sec, out) # URL handler for verification/unsubscribe app.load(handle_url, outputs=out) if __name__ == "__main__": app.launch(debug=True)