import threading
import time
import requests
import os
import datetime
from flask import Flask, jsonify, request
from wsgiref.simple_server import make_server
app = Flask(__name__)
# --- CONFIGURATION ---
URL_FILE = 'urls.txt'
SELF_PING_URL = "https://alvin3y1-ping.hf.space/"
INTERVAL_SECONDS = 4 * 3600 # 4 hours
# --- GLOBAL STATE ---
monitoring_state = {}
scan_state = {
'is_scanning': False,
'total': 0,
'completed': 0,
'last_scan_time': 'Never'
}
state_lock = threading.Lock()
# --- FRONTEND TEMPLATE ---
# Served as a static string to keep the project to one file.
HTML_TEMPLATE = """
Service Status
"""
# --- LOGIC ---
def get_all_urls():
urls = [SELF_PING_URL]
if os.path.exists(URL_FILE):
with open(URL_FILE, 'r') as f:
urls.extend([l.strip() for l in f if l.strip()])
return list(set(urls))
def check_single_url(url):
timestamp = datetime.datetime.now().strftime("%H:%M:%S")
try:
r = requests.get(url, timeout=10)
res = {'status': 'UP' if r.status_code < 400 else 'DOWN', 'code': r.status_code, 'time': timestamp, 'msg': 'OK' if r.status_code < 400 else 'Error'}
except Exception as e:
res = {'status': 'DOWN', 'code': 'ERR', 'time': timestamp, 'msg': str(e)[:20]}
with state_lock:
monitoring_state[url] = res
def perform_full_scan():
urls = get_all_urls()
with state_lock:
scan_state.update({'is_scanning': True, 'total': len(urls), 'completed': 0})
for url in urls:
check_single_url(url)
with state_lock:
scan_state['completed'] += 1
with state_lock:
scan_state.update({'is_scanning': False, 'last_scan_time': datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")})
def background_worker():
while True:
perform_full_scan()
time.sleep(INTERVAL_SECONDS)
# --- ROUTES ---
@app.route('/')
def home(): return HTML_TEMPLATE
@app.route('/api/status')
def get_status():
with state_lock:
down = [{'url': u, **s} for u, s in monitoring_state.items() if s['status'] == 'DOWN']
return jsonify({'scan': scan_state, 'down_urls': down})
@app.route('/api/scan', methods=['POST'])
def scan():
threading.Thread(target=perform_full_scan, daemon=True).start()
return jsonify({"status": "started"})
@app.route('/api/retry', methods=['POST'])
def retry():
threading.Thread(target=check_single_url, args=(request.json['url'],), daemon=True).start()
return jsonify({"status": "retrying"})
if __name__ == '__main__':
threading.Thread(target=background_worker, daemon=True).start()
print("Server starting on http://0.0.0.0:7860")
make_server('0.0.0.0', 7860, app).serve_forever()