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 if os.path.exists(TOKEN_FILE): try: with open(TOKEN_FILE, "rb") as f: creds = pickle.load(f) except: creds = None if creds and creds.expired and creds.refresh_token: try: creds.refresh(Request()) with open(TOKEN_FILE, "wb") as f: pickle.dump(creds, f) except: creds = None if not creds: raise Exception("❌ token.pkl missing. Upload valid token.") return build("gmail", "v1", credentials=creds) 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)