import os, time, requests from flask import Flask, render_template_string, request, redirect, session, jsonify from flask_sqlalchemy import SQLAlchemy from werkzeug.security import generate_password_hash, check_password_hash from apscheduler.schedulers.background import BackgroundScheduler app = Flask(__name__) app.secret_key = 'supersecretkey' app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///uptimezelarixa.db' db = SQLAlchemy(app) # ✅ FIX: Jalankan di dalam app context agar tidak error with app.app_context(): db.create_all() # Ambil secrets dari environment BOT_TOKEN = os.getenv("BOT_TOKEN") CHAT_ID = os.getenv("CHAT_ID") # ---------- Database ---------- class User(db.Model): id = db.Column(db.Integer, primary_key=True) email = db.Column(db.String(120), unique=True) password = db.Column(db.String(200)) class Monitor(db.Model): id = db.Column(db.Integer, primary_key=True) user_id = db.Column(db.Integer, db.ForeignKey('user.id')) name = db.Column(db.String(100)) url = db.Column(db.String(300)) status = db.Column(db.String(20)) last_check = db.Column(db.String(100)) response_time = db.Column(db.Float) # ---------- Fungsi utama ---------- def send_telegram_alert(name, url): if not BOT_TOKEN or not CHAT_ID: print("⚠️ BOT_TOKEN atau CHAT_ID belum diatur di Secrets!") return text = f"⚠️ [ALERT] {name} ({url}) sedang DOWN!" try: requests.get(f"https://api.telegram.org/bot{BOT_TOKEN}/sendMessage", params={"chat_id": CHAT_ID, "text": text}) except Exception as e: print(f"Gagal kirim notifikasi Telegram: {e}") def check_monitors(): with app.app_context(): monitors = Monitor.query.all() for m in monitors: try: start = time.time() r = requests.get(m.url, timeout=5) m.response_time = round((time.time() - start) * 1000, 2) m.status = "Online" if r.status_code == 200 else f"Error {r.status_code}" except: if m.status != "Offline": send_telegram_alert(m.name, m.url) m.status = "Offline" m.response_time = None m.last_check = time.strftime("%H:%M:%S %d-%m-%Y") db.session.commit() # Scheduler otomatis cek setiap 5 menit scheduler = BackgroundScheduler() scheduler.add_job(check_monitors, 'interval', minutes=5) scheduler.start() # ---------- Routes ---------- @app.route('/') def index(): if 'user_id' not in session: return redirect('/login') user = User.query.get(session['user_id']) monitors = Monitor.query.filter_by(user_id=user.id).all() return render_template_string(TEMPLATE, user=user, monitors=monitors) @app.route('/register', methods=['GET', 'POST']) def register(): if request.method == 'POST': email, pw = request.form['email'], request.form['password'] if User.query.filter_by(email=email).first(): return "Email sudah terdaftar." user = User(email=email, password=generate_password_hash(pw)) db.session.add(user) db.session.commit() return redirect('/login') return render_template_string(REGISTER_TEMPLATE) @app.route('/login', methods=['GET', 'POST']) def login(): if request.method == 'POST': email, pw = request.form['email'], request.form['password'] user = User.query.filter_by(email=email).first() if user and check_password_hash(user.password, pw): session['user_id'] = user.id return redirect('/') return "Login gagal." return render_template_string(LOGIN_TEMPLATE) @app.route('/logout') def logout(): session.pop('user_id', None) return redirect('/login') @app.route('/add', methods=['POST']) def add(): if 'user_id' not in session: return redirect('/login') name = request.form['name'] url = request.form['url'] m = Monitor(user_id=session['user_id'], name=name, url=url, status="Unknown") db.session.add(m) db.session.commit() return redirect('/') @app.route('/check_now', methods=['POST']) def check_now(): check_monitors() return jsonify({"status": "checked"}) # ---------- Template HTML ---------- TEMPLATE = """
Halo {{user.email}} | Logout
| Nama | URL | Status | Respons (ms) | Terakhir Dicek |
|---|---|---|---|---|
| {{m.name}} | {{m.url}} | {{m.status}} | {{m.response_time or '-'}} | {{m.last_check or '-'}} |