"""Flask app and routes.""" import os from datetime import datetime, timedelta, timezone from flask import Flask, redirect, render_template, request, url_for from app import models def _parse_report_dates(from_ts, to_ts, preset): """Parse from/to dates, applying preset if given. Returns (from_ts, to_ts, from_display, to_display).""" now = datetime.now(timezone.utc) if preset == "24h": to_ts = now.isoformat() from_ts = (now - timedelta(hours=24)).isoformat() elif preset == "7d": to_ts = now.isoformat() from_ts = (now - timedelta(days=7)).isoformat() elif preset == "30d": to_ts = now.isoformat() from_ts = (now - timedelta(days=30)).isoformat() if from_ts and len(from_ts) == 10: from_ts = from_ts + "T00:00:00" if to_ts and len(to_ts) == 10: to_ts = to_ts + "T23:59:59.999999" from_display = from_ts[:16] if from_ts else "" to_display = to_ts[:16] if to_ts else "" return from_ts, to_ts, from_display, to_display ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) app = Flask( __name__, template_folder=os.path.join(ROOT, "templates"), static_folder=os.path.join(ROOT, "static"), ) VERSION = os.environ.get("VERSION", "dev") @app.route("/") def dashboard(): services = models.list_services() return render_template("dashboard.html", services=services, version=VERSION) @app.route("/api/services", methods=["GET"]) def api_list_services(): services = models.list_services() return {"services": services} @app.route("/api/services", methods=["POST"]) def api_add_service(): data = request.get_json(silent=True) if data is None and request.form: data = request.form.to_dict() if not data: return {"error": "Request body must be valid JSON (Content-Type: application/json) or form data"}, 415 name = data.get("name", "").strip() target = data.get("target", "").strip() protocol = (data.get("protocol") or "https").lower().strip() if not name or not target: return {"error": "name and target are required"}, 400 if protocol not in ("http", "https", "tcp"): return {"error": "protocol must be http, https, or tcp"}, 400 interval = int(data.get("interval_seconds", 60)) interval = max(10, min(3600, interval)) try: sid = models.add_service(name, target, protocol, interval) if request.accept_mimetypes.best == "application/json" or request.is_json: return {"id": sid, "name": name, "target": target, "protocol": protocol} return redirect(url_for("dashboard")) except Exception as e: return {"error": str(e)}, 500 @app.route("/api/services/", methods=["DELETE"]) def api_delete_service(service_id): if models.delete_service(service_id): return {"deleted": service_id} return {"error": "Not found"}, 404 @app.route("/api/services/", methods=["PATCH"]) def api_update_service(service_id): svc = models.get_service(service_id) if not svc: return {"error": "Not found"}, 404 data = request.get_json(silent=True) if data is None and request.form: data = request.form.to_dict() if not data: return {"error": "Request body must be valid JSON or form data"}, 415 updates = {} if "name" in data and data["name"] is not None: updates["name"] = str(data["name"]).strip() if "target" in data and data["target"] is not None: updates["target"] = str(data["target"]).strip() if "protocol" in data and data["protocol"] is not None: p = str(data["protocol"]).lower().strip() if p not in ("http", "https", "tcp"): return {"error": "protocol must be http, https, or tcp"}, 400 updates["protocol"] = p if "interval_seconds" in data and data["interval_seconds"] is not None: updates["interval_seconds"] = max(10, min(3600, int(data["interval_seconds"]))) if not updates: return {"id": service_id, **dict(svc)} if updates.get("name") == "" or updates.get("target") == "": return {"error": "name and target cannot be empty"}, 400 if models.update_service(service_id, **updates): updated = models.get_service(service_id) return dict(updated) return {"error": "Update failed"}, 500 @app.route("/api/services/") def api_get_service(service_id): svc = models.get_service(service_id) if not svc: return {"error": "Not found"}, 404 checks = models.get_checks(service_id, limit=50) return {"service": dict(svc), "checks": checks} @app.route("/api/services//edit") def edit_service(service_id): svc = models.get_service(service_id) if not svc: return "Service not found", 404 return render_template("edit.html", service=dict(svc), version=VERSION) @app.route("/api/services//report") def report(service_id): svc = models.get_service(service_id) if not svc: return "Service not found", 404 from_ts = request.args.get("from") or None to_ts = request.args.get("to") or None preset = request.args.get("preset") from_ts, to_ts, from_display, to_display = _parse_report_dates(from_ts, to_ts, preset) status_filter = request.args.get("status") search = request.args.get("search", "").strip() or None stats = models.get_report_stats(service_id, from_ts=from_ts, to_ts=to_ts) checks = models.get_checks(service_id, limit=100, from_ts=from_ts, to_ts=to_ts, status_filter=status_filter, search=search) chart_checks = models.get_checks(service_id, limit=200, from_ts=from_ts, to_ts=to_ts) period_label = _format_period_label(from_display, to_display) if (from_ts or to_ts) else None return render_template( "report.html", service=dict(svc), stats=stats, checks=checks, chart_checks=chart_checks, version=VERSION, from_date=from_display, to_date=to_display, period_label=period_label, preset=preset, status_filter=status_filter or "", search=search or "", ) def _format_period_label(from_display, to_display): """Format period for display.""" if from_display and to_display: return f"{from_display} to {to_display}" if from_display: return f"From {from_display}" if to_display: return f"Until {to_display}" return None @app.route("/api/services//history") def api_history(service_id): svc = models.get_service(service_id) if not svc: return {"error": "Not found"}, 404 limit = min(500, int(request.args.get("limit", 100))) from_ts = request.args.get("from") or None to_ts = request.args.get("to") or None if from_ts and len(from_ts) == 10: from_ts = from_ts + "T00:00:00" if to_ts and len(to_ts) == 10: to_ts = to_ts + "T23:59:59.999999" history = models.get_history(service_id, limit=limit, from_ts=from_ts, to_ts=to_ts) return {"history": history} @app.route("/api/services//stats") def api_report_stats(service_id): """JSON report stats with optional from/to query params for date range.""" svc = models.get_service(service_id) if not svc: return {"error": "Not found"}, 404 from_ts = request.args.get("from") or None to_ts = request.args.get("to") or None if from_ts and len(from_ts) == 10: from_ts = from_ts + "T00:00:00" if to_ts and len(to_ts) == 10: to_ts = to_ts + "T23:59:59.999999" stats = models.get_report_stats(service_id, from_ts=from_ts, to_ts=to_ts) return {"service": dict(svc), "stats": stats}