Spaces:
Sleeping
Sleeping
Download app.py from srinetwork/cornjob: direct link, hf CLI and curl.
- Browser
- Download file 4.46 kB
-
https://huggingface.co/spaces/srinetwork/cornjob/resolve/main/app.py
- Command line
-
hf download hf://spaces/srinetwork/cornjob/app.py
-
curl -L -o app.py https://huggingface.co/spaces/srinetwork/cornjob/resolve/main/app.py
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 ---------- | |
| def index(): | |
| return render_template("index.html") | |
| 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 | |
| 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" | |
| 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) |