Skip to content

Instantly share code, notes, and snippets.

@theSoberSobber
Last active August 9, 2024 07:32
Show Gist options
  • Save theSoberSobber/bc922d4c5bd2acf1e264240dfcc39433 to your computer and use it in GitHub Desktop.
Save theSoberSobber/bc922d4c5bd2acf1e264240dfcc39433 to your computer and use it in GitHub Desktop.
Termux CronTab Script Codeforces, (crontab -e, */15 * * * * /data/data/com.termux/files/home/monitor/data/main.sh >> /data/data/com.termux/files/home/monitor/data/logs.txt 2>&1 , nohup crond -n &>/dev/null &)
#!/bin/bash
# Function for colored logging
debug() {
local mode="$1"
local log="$2"
local color="$3"
case "$mode" in
fetch) echo -e "\e[1;36m[Fetch]\e[0m $log" ;;
value) echo -e "\e[1;34m[Value]\e[0m $log" ;;
update) echo -e "\e[1;32m[Update]\e[0m $log" ;;
notify) echo -e "\e[1;33m[Notify]\e[0m $log" ;;
*) echo "$log" ;;
esac
}
# Function to extract numeric value from the string
extract_numeric_value() {
local str="$1"
echo "$str" | grep -oP '\d+'
}
# Echo separators and current time
echo "Current Time: $(date)"
echo "---------------------"
# Function to fetch value from URL
fetch_value() {
local url="$1"
curl -s "$url" | grep -oP '(?<=<div class="_UserActivityFrame_counterValue">).*?(?=<\/div>)' | head -n 1
}
# Create 'data' directory if it doesn't exist
if [ ! -d "/data/data/com.termux/files/home/monitor/data" ]; then
mkdir "/data/data/com.termux/files/home/monitor/data"
fi
# Read CSV file and process each entry
while IFS=, read -r entry; do
url="https://codeforces.com/profile/$entry"
debug fetch "Fetching value for $entry" cyan
value=$(fetch_value "$url")
if [ -n "$value" ]; then
debug value "Value for $entry: $value" blue
filename="/data/data/com.termux/files/home/monitor/data/$entry.txt"
stored_value="-1 problem" # Initialize stored value
last_checked_file="/data/data/com.termux/files/home/monitor/data/$entry-lastChecked.txt"
# Initialize last checked variable with current date
last_checked=$(date +"%b %d,%H:%M")
if [ -f "$last_checked_file" ]; then
last_checked=$(cat "$last_checked_file")
fi
if [ -f "$filename" ]; then
stored_value=$(cat "$filename")
fi
if [ "$value" != "$stored_value" ]; then
debug update "Updating file $entry" green
echo "$value" > "$filename"
debug notify "Notifying" yellow
# Extract numeric values for comparison
new_value_numeric=$(extract_numeric_value "$value")
stored_value_numeric=$(extract_numeric_value "$stored_value")
# Calculate change in values
change=$((new_value_numeric - stored_value_numeric))
# Determine sign for change
sign=""
if [ "$change" -gt 0 ]; then
sign="+"
fi
# Construct notification content
notif_content="New value: $value
Previous value: $stored_value
Change: $sign$change
Last Checked: $last_checked"
# Send notification
termux-notification -t "[Codeforces] $entry $sign$change" -c "$notif_content" --priority high
fi
# Update last checked timestamp after each iteration
echo "$(date +"%b %d,%H:%M")" > "$last_checked_file"
else
debug fetch "Skipping $entry - Fetch failed" red
fi
echo "---------------------" # Separator after each entry's processing
sleep 2 # 2-second delay after each fetch
done < /data/data/com.termux/files/home/monitor/input.csv
#input.csv
#a,
#b,
#c
@theSoberSobber
Copy link
Author

Upgraded Script (need ntfy cli)

import requests
import os
from datetime import datetime
import time
import webbrowser
import re

def debug(mode, log, color):
    colors = {
        "fetch": "\033[1;36m[Fetch]\033[0m",
        "value": "\033[1;34m[Value]\033[0m",
        "update": "\033[1;32m[Update]\033[0m",
        "notify": "\033[1;33m[Notify]\033[0m"
    }
    print(f"{colors.get(mode, '')} {log}")

def extract_numeric_value(text):
    return ''.join(filter(str.isdigit, text))

def fetch_value(url):
    response = requests.get(url)
    match = re.search(r'<div class="_UserActivityFrame_counterValue">(.+?)</div>', response.text)
    if match:
        return match.group(1)
    return None

def monitor_codeforces_profiles(input_file, data_dir):
    os.makedirs(data_dir, exist_ok=True)

    with open(input_file, 'r') as f:
        for entry in f:
            entry = entry.strip()
            url = f"https://codeforces.com/profile/{entry}"
            debug("fetch", f"Fetching value for {entry}", "cyan")
            value = fetch_value(url)

            if value:
                debug("value", f"Value for {entry}: {value}", "blue")
                filename = os.path.join(data_dir, f"{entry}.txt")
                stored_value = "-1 problem"

                if os.path.exists(filename):
                    with open(filename, 'r') as f:
                        stored_value = f.read().strip()

                if value != stored_value:
                    debug("update", f"Updating file {entry}", "green")
                    with open(filename, 'w') as f:
                        f.write(value)

                    new_value_numeric = int(extract_numeric_value(value))
                    stored_value_numeric = int(extract_numeric_value(stored_value))
                    change = new_value_numeric - stored_value_numeric

                    if change > 0:
                        # Fetch problem names and URLs for the change
                        problem_data = get_problem_data(entry, change)
                        for problem_name, problem_url in problem_data:
                            debug("notify", f"Notifying about {problem_name} solved by {entry}", "yellow")
                            send_notification(entry, f"{problem_name}", problem_url)
                        if len(problem_data) < change:
                            debug("notify", f"Could not find enough distinct problems solved by {entry}", "yellow")
                            send_notification(f"Too Many Problems {entry}!!", f"Could not find all {change} distinct problems", None)
                    elif change < 0:
                        debug("notify", f"Ignoring negative change for {entry}", "yellow")
                    else:
                        debug("notify", f"Ignoring large change (> 20) for {entry}", "yellow")
            else:
                debug("fetch", f"Skipping {entry} - Fetch failed", "red")

            time.sleep(2)

def get_problem_data(handle, change):
    url = f"https://codeforces.com/api/user.status?handle={handle}&from=1&count=100"
    response = requests.get(url)
    data = response.json()
    problem_data = []
    for submission in data["result"]:
        if submission["verdict"] == "OK":
            problem_name = submission["problem"]["name"]
            problem_url = f"https://codeforces.com/contest/{submission['contestId']}/problem/{submission['problem']['index']}"
            problem_data.append((problem_name, problem_url))
        if len(problem_data) == change:
            break
    return list(set(problem_data))

def send_notification(title, message, problem_url):
    cmd = f'ntfy publish --title "{title}" --priority high --tags "work,important" --message "{message}" CHANNEL NAME HERE'
    if problem_url:
        cmd+=f' --click="{problem_url}"'
        os.system(cmd)
    else:
        os.system(cmd)

def open_browser(problem_url):
    webbrowser.open(problem_url)

if __name__ == "__main__":
    print("Current Time:", datetime.now())
    print("---------------------")
    monitor_codeforces_profiles(
        "input.csv",
        "data"
    )

@theSoberSobber
Copy link
Author

sudo nano /etc/systemd/system/codeforces-monitor.service

Contents:

[Unit]
Description=Codeforces Profile Monitor
After=network.target

[Service]
Type=simple
User=meso
WorkingDirectory=/home/meso/monitor
ExecStart=/usr/bin/python3 /home/meso/monitor/main.py
Restart=on-failure

[Install]
WantedBy=multi-user.target

sudo nano /etc/systemd/system/codeforces-monitor.timer

Contents:

[Unit]
Description=Run Codeforces Monitor every 15 minutes

[Timer]
OnBootSec=5min
OnUnitActiveSec=15min
Unit=codeforces-monitor.service

[Install]
WantedBy=timers.target

sudo systemctl enable codeforces-monitor.timer
sudo systemctl start codeforces-monitor.timer

Check Status and Next Trigger Time:

systemctl status codeforces-monitor.timer

@theSoberSobber
Copy link
Author

untested but has rate limit

import requests
import os
from datetime import datetime, timedelta
import time
import webbrowser
import re

def debug(mode, log, color):
    colors = {
        "fetch": "\033[1;36m[Fetch]\033[0m",
        "value": "\033[1;34m[Value]\033[0m",
        "update": "\033[1;32m[Update]\033[0m",
        "notify": "\033[1;33m[Notify]\033[0m",
        "error": "\033[1;31m[Error]\033[0m"
    }
    print(f"{colors.get(mode, '')} {log}")

def extract_numeric_value(text):
    return ''.join(filter(str.isdigit, text))

def fetch_value(url):
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        match = re.search(r'<div class="_UserActivityFrame_counterValue">(.+?)</div>', response.text)
        if match:
            return match.group(1)
        return None
    except requests.exceptions.RequestException as e:
        debug("error", f"Request failed for {url}: {e}", "red")
        return None

def monitor_codeforces_profiles(input_file, data_dir):
    os.makedirs(data_dir, exist_ok=True)

    notification_count = 0
    last_notification_time = datetime.min

    with open(input_file, 'r') as f:
        for entry in f:
            entry = entry.strip()
            url = f"https://codeforces.com/profile/{entry}"
            debug("fetch", f"Fetching value for {entry}", "cyan")
            value = fetch_value(url)

            if value:
                debug("value", f"Value for {entry}: {value}", "blue")
                filename = os.path.join(data_dir, f"{entry}.txt")
                stored_value = "-1 problem"

                if os.path.exists(filename):
                    with open(filename, 'r') as f:
                        stored_value = f.read().strip()

                if value != stored_value:
                    debug("update", f"Updating file {entry}", "green")
                    with open(filename, 'w') as f:
                        f.write(value)

                    new_value_numeric = int(extract_numeric_value(value))
                    stored_value_numeric = int(extract_numeric_value(stored_value))
                    change = new_value_numeric - stored_value_numeric

                    if change > 0:
                        problem_data = get_problem_data(entry, change)
                        for problem_name, problem_url, submission_url in problem_data:
                            debug("notify", f"Notifying about {problem_name} solved by {entry}", "yellow")
                            if rate_limit_notifications(notification_count, last_notification_time):
                                send_notification(entry, f"{problem_name}", problem_url, submission_url)
                                notification_count += 1
                                last_notification_time = datetime.now()

                        if len(problem_data) < change:
                            debug("notify", f"Could not find enough distinct problems solved by {entry}", "yellow")
                            if rate_limit_notifications(notification_count, last_notification_time):
                                send_notification(f"Too Many Problems {entry}!!", f"Could not find all {change} distinct problems")
                                notification_count += 1
                                last_notification_time = datetime.now()
                    elif change < 0:
                        debug("notify", f"Ignoring negative change for {entry}", "yellow")
                    else:
                        debug("notify", f"Ignoring large change (> 20) for {entry}", "yellow")
            else:
                debug("fetch", f"Skipping {entry} - Fetch failed", "red")

            time.sleep(2)

def get_problem_data(handle, change):
    url = f"https://codeforces.com/api/user.status?handle={handle}&from=1&count=100"
    try:
        response = requests.get(url, timeout=10)
        response.raise_for_status()
        data = response.json()
    except requests.exceptions.RequestException as e:
        debug("error", f"Failed to fetch problem data for {handle}: {e}", "red")
        return []
    except ValueError as e:
        debug("error", f"Failed to parse JSON for {handle}: {e}", "red")
        return []

    problem_data = []
    for submission in data.get("result", []):
        if submission["verdict"] == "OK":
            problem_name = submission["problem"]["name"]
            problem_url = f"https://codeforces.com/contest/{submission['contestId']}/problem/{submission['problem']['index']}"
            submission_url = f"https://codeforces.com/contest/{submission['contestId']}/submission/{submission['id']}"
            problem_data.append((problem_name, problem_url, submission_url))
        if len(problem_data) == change:
            break
    return list(set(problem_data))

def rate_limit_notifications(notification_count, last_notification_time):
    """
    Rate limit the number of notifications sent.
    """
    max_notifications = 5  # Max notifications allowed before adding a longer delay
    short_delay = 2  # Delay in seconds between each notification
    long_delay = 6  # Delay in seconds after hitting the max notification count

    current_time = datetime.now()
    if notification_count >= max_notifications:
        time_since_last_notification = (current_time - last_notification_time).total_seconds()
        if time_since_last_notification < long_delay:
            debug("notify", f"Rate limiting in effect. Waiting for {long_delay - time_since_last_notification:.2f} seconds.", "red")
            time.sleep(long_delay - time_since_last_notification)
        notification_count = 0  # Reset counter after the longer delay

    time.sleep(short_delay)
    return True

def send_notification(title, message, problem_url=None, submission_url=None):
    cmd = f'ntfy publish --title "{title}" --priority high --tags "work,important" --message "{message}"'
    if problem_url and submission_url:
        cmd += f' --click="{problem_url}" --actions "view, Open, {problem_url}; view, Open Submission, {submission_url}"'
    cmd += " codeforces_grind_fr_fr"
    os.system(cmd)

def open_browser(problem_url):
    webbrowser.open(problem_url)

if __name__ == "__main__":
    print("Current Time:", datetime.now())
    print("---------------------")
    monitor_codeforces_profiles(
        "/home/meso/monitor/input.csv",
        "/home/meso/monitor/data"
    )

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment