leetcode_beeminder.py

import requests
import datetime
import os
import sys
from typing import Optional, Dict
import json
import logging

# Create custom loggers for stdout and stderr
class InfoLogger(logging.StreamHandler):
    def __init__(self):
        logging.StreamHandler.__init__(self, sys.stdout)
    
    def emit(self, record):
        if record.levelno <= logging.INFO:
            super().emit(record)

class ErrorLogger(logging.StreamHandler):
    def __init__(self):
        logging.StreamHandler.__init__(self, sys.stderr)
    
    def emit(self, record):
        if record.levelno > logging.INFO:
            super().emit(record)

# Configure logging with separate handlers
logger = logging.getLogger(__name__)
logger.setLevel(logging.INFO)

# Create formatters
formatter = logging.Formatter('%(asctime)s - %(levelname)s: %(message)s', datefmt='%Y-%m-%d %H:%M:%S')

# Create and add handlers
info_handler = InfoLogger()
info_handler.setLevel(logging.INFO)
info_handler.setFormatter(formatter)

error_handler = ErrorLogger()
error_handler.setLevel(logging.WARNING)
error_handler.setFormatter(formatter)

# Add handlers to logger
logger.addHandler(info_handler)
logger.addHandler(error_handler)

class LeetCodeBeeminderTracker:
    def __init__(self, leetcode_username: str, beeminder_username: str, beeminder_auth_token: str, goal_name: str):
        self.leetcode_username = leetcode_username
        self.beeminder_username = beeminder_username
        self.beeminder_auth_token = beeminder_auth_token
        self.goal_name = goal_name
        
        # API endpoints
        self.leetcode_api = f"https://leetcode-api-faisalshohag.vercel.app/{leetcode_username}"
        self.beeminder_api = f"https://www.beeminder.com/api/v1/users/{beeminder_username}/goals/{goal_name}/datapoints.json"
        
        logger.info(f"Initialized tracker for LeetCode user: {leetcode_username}")
        logger.info(f"Beeminder goal: {beeminder_username}/{goal_name}")
    
    def get_leetcode_submissions(self) -> Optional[Dict]:
        """Fetch LeetCode submissions using the custom API"""
        logger.info(f"\nFetching LeetCode data from: {self.leetcode_api}")
        try:
            response = requests.get(self.leetcode_api)
            response.raise_for_status()
            data = response.json()
            logger.info("Successfully fetched LeetCode data:")
            logger.info(f"Total solved: {data.get('totalSolved', 0)}")
            logger.info(f"Recent submissions: {len(data.get('recentSubmissions', []))}")
            return data
        except requests.exceptions.RequestException as e:
            logger.error(f"Error fetching LeetCode data: {e}")
            if hasattr(e.response, 'text'):
                logger.error(f"Response text: {e.response.text}")
            return None

    def submit_to_beeminder(self, date: str, value: int, comment: str) -> bool:
        """Submit a datapoint to Beeminder"""
        params = {
            'auth_token': self.beeminder_auth_token,
            'value': value,
            'comment': comment,
            'timestamp': int(datetime.datetime.strptime(date, '%Y-%m-%d').timestamp()),
            'requestid': f"leetcode_{date}"  # Prevents duplicate submissions
        }
        
        logger.info(f"\nSubmitting to Beeminder for date {date}:")
        logger.info(f"Value: {value}")
        logger.info(f"Comment: {comment}")
        logger.info(f"API URL: {self.beeminder_api}")
        
        try:
            response = requests.post(self.beeminder_api, params=params)
            response.raise_for_status()
            result = response.json()
            logger.info("Beeminder response:")
            logger.info(json.dumps(result, indent=2))
            return True
        except requests.exceptions.RequestException as e:
            logger.error(f"Error submitting to Beeminder: {e}")
            if hasattr(e, 'response') and e.response is not None:
                logger.error(f"Response status code: {e.response.status_code}")
                logger.error(f"Response text: {e.response.text}")
            return False

    def update_tracking(self, days_to_check: int = 7):
        """Main function to update Beeminder with LeetCode progress"""
        # Get LeetCode submission data
        submission_data = self.get_leetcode_submissions()
        if not submission_data:
            return
        
        # Get the submission calendar which contains daily activity
        submission_calendar = submission_data.get('submissionCalendar', {})
        logger.info(f"\nSubmission calendar data:")
        logger.info(json.dumps(submission_calendar, indent=2))
        
        # Get dates to check (up to N days back)
        check_range = [
            datetime.date.today() - datetime.timedelta(days=x)
            for x in range(days_to_check)
        ]
        
        for date in check_range:
            # logger.info(f"\nProcessing date: {date}")
            # Convert date to Unix timestamp (seconds) for lookup
            timestamp = int(datetime.datetime.combine(date, datetime.time.min).timestamp())
            # logger.info(f"Looking for timestamp: {timestamp}")
            
            # Calculate day start and end timestamps
            day_start = timestamp
            day_end = timestamp + 86400  # Add 24 hours in seconds
            
            # Check for any submissions within this day's range
            day_submissions = 0
            for submit_time_str, count in submission_calendar.items():
                submit_time = int(submit_time_str)
                if day_start <= submit_time < day_end:
                    day_submissions = count
                    break
            
            # logger.info(f"Submissions found for this day: {day_submissions}")
            
            if day_submissions > 0:
                # Get unique accepted submissions for this day from recentSubmissions
                unique_problems = set(
                    sub['titleSlug'] for sub in submission_data.get('recentSubmissions', [])
                    if (
                        int(sub['timestamp']) >= timestamp and
                        int(sub['timestamp']) < timestamp + 86400 and  # Within 24 hours
                        sub['statusDisplay'] == 'Accepted'
                    )
                )
                
                num_unique_solved = len(unique_problems)
                
                if num_unique_solved > 0:  # Only log days with accepted submissions
                    success = self.submit_to_beeminder(
                        date.strftime('%Y-%m-%d'),
                        num_unique_solved,  # Track number of unique problems solved
                        f"Solved {num_unique_solved} different problems"
                    )
                    if success:
                        logger.info(f"Successfully logged activity for {date.strftime('%Y-%m-%d')} ({num_unique_solved} unique problems solved)")
                    else:
                        logger.info(f"Failed to log to Beeminder for {date.strftime('%Y-%m-%d')}")
            # else:
                # logger.info("No submissions found for this day, skipping")

def main():
    # Load configuration from environment variables
    config = {
        'leetcode_username': os.getenv('LEETCODE_USERNAME'),
        'beeminder_username': os.getenv('BEEMINDER_USERNAME'),
        'beeminder_auth_token': os.getenv('BEEMINDER_AUTH_TOKEN'),
        'goal_name': os.getenv('BEEMINDER_GOAL_NAME', 'leetcode')
    }
    
    logger.info("\nConfiguration:")
    for key, value in config.items():
        if key == 'beeminder_auth_token':
            logger.info(f"{key}: {'*' * len(value) if value else 'Not set'}")
        else:
            logger.info(f"{key}: {value or 'Not set'}")
    
    # Validate configuration
    missing_configs = [k for k, v in config.items() if not v]
    if missing_configs:
        logger.error(f"\nMissing required configuration: {', '.join(missing_configs)}")
        return
    
    tracker = LeetCodeBeeminderTracker(**config)
    tracker.update_tracking()

if __name__ == "__main__":
    main()