cornjob / app.py
srinetwork's picture
Update app.py
f5fef0e verified
Raw History Blame Contribute Delete
4.46 kB
from flask import Flask, render_template, request, jsonify
import requests, time, json, os, threading
from bs4 import BeautifulSoup
from urllib.parse import urljoin
app = Flask(__name__)
DATA_FILE = "/tmp/urls.json"
ADMIN_PASSWORD = os.environ.get("ADMIN_PASSWORD", "admin")
# ---------- STORAGE ----------
if not os.path.exists(DATA_FILE):
with open(DATA_FILE, "w") as f:
json.dump([], f)
def load_urls():
with open(DATA_FILE, "r") as f:
return json.load(f)
def save_urls(data):
with open(DATA_FILE, "w") as f:
json.dump(data, f, indent=2)
# ---------- IMAGE FETCH ----------
def fetch_site_image(url):
try:
r = requests.get(url, timeout=10)
soup = BeautifulSoup(r.text, "html.parser")
# og:image
og = soup.find("meta", property="og:image")
if og and og.get("content"):
return urljoin(url, og["content"])
# favicon
return url.rstrip("/") + "/favicon.ico"
except:
return ""
# ---------- CHECKER ----------
def check_sites():
while True:
sites = load_urls()
now = time.time()
for s in sites:
if now >= s["next_check"]:
s["total_checks"] += 1
up = False
ping = None
try:
r = requests.get(s["url"], timeout=10)
ping = round(r.elapsed.total_seconds() * 1000, 2)
up = r.status_code == 200
except:
pass
if up:
s["up_checks"] += 1
s["status"] = "UP" if up else "DOWN"
s["last_ping"] = ping
s["history"].append({
"time": int(now),
"uptime": 100 if up else 0,
"ping": ping
})
s["history"] = s["history"][-50:]
s["last_check"] = now
s["next_check"] = now + s["interval"]
save_urls(sites)
time.sleep(5)
# ---------- ROUTES ----------
@app.route("/")
def index():
return render_template("index.html")
@app.route("/data")
def data():
sites = load_urls()
now = time.time()
for s in sites:
s["remaining"] = max(0, int(s["next_check"] - now))
total = int(now - s["first_added"])
d, rem = divmod(total, 86400)
h, rem = divmod(rem, 3600)
m, sec = divmod(rem, 60)
s["uptime_str"] = f"{d}d {h}h {m}m {sec}s"
s["uptime_percent"] = round(
(s["up_checks"] / s["total_checks"]) * 100, 2
) if s["total_checks"] else 0
return jsonify(sites)
def auth(req):
return req.form.get("password") == ADMIN_PASSWORD
@app.route("/add", methods=["POST"])
def add():
if not auth(request):
return "Unauthorized", 401
now = time.time()
data = load_urls()
url = request.form["url"]
data.append({
"id": int(now * 1000),
"name": request.form["name"],
"url": url,
"img": fetch_site_image(url),
"interval": int(request.form["interval"]),
"status": "UNKNOWN",
"last_ping": None,
"last_check": 0,
"next_check": now,
"first_added": now,
"total_checks": 0,
"up_checks": 0,
"history": []
})
save_urls(data)
return "OK"
@app.route("/edit/<int:id>", methods=["POST"])
def edit(id):
if not auth(request):
return "Unauthorized", 401
sites = load_urls()
for s in sites:
if s["id"] == id:
old_url = s["url"]
new_url = request.form["url"]
s["name"] = request.form["name"]
s["url"] = new_url
s["interval"] = int(request.form["interval"])
s["img"] = fetch_site_image(new_url)
# 🔥 RESET uptime if URL changed
if old_url != new_url:
now = time.time()
s["first_added"] = now
s["total_checks"] = 0
s["up_checks"] = 0
s["history"] = []
s["last_check"] = 0
s["next_check"] = now
s["status"] = "UNKNOWN"
s["last_ping"] = None
break
save_urls(sites)
return "OK"
# ---------- START ----------
if __name__ == "__main__":
threading.Thread(target=check_sites, daemon=True).start()
app.run(host="0.0.0.0", port=7860)