Last active
August 9, 2024 07:32
-
-
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 &)
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
#!/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 |
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
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
Upgraded Script (need ntfy cli)