'
html_content = f"""
"""
plain_body = f"[{badge_text}]\nService: {name}\nLocation: {loc}\nURL: {url}\nStatus: {msg}\nDetails: {details}\nTime: {timestamp_str}"
if duration_str:
plain_body += f"\nDowntime Duration: {duration_str}"
alert_mode = config.get("alert_mode", "SMTP")
raw_recipients = config.get("email_recipients", [])
if isinstance(raw_recipients, str):
recipients = [r.strip() for r in raw_recipients.split(',') if r.strip()]
elif isinstance(raw_recipients, list):
recipients = [r.strip() for r in raw_recipients if isinstance(r, str) and r.strip()]
else:
recipients = []
sender = config.get("smtp_sender", "").strip() or DEFAULT_SMTP_SENDER
raw_pwd = os.environ.get('SMTP_PASSWORD') or config.get("smtp_password", "")
password = security.decrypt_secret(raw_pwd).strip()
if not password:
password = DEFAULT_SMTP_PASSWORD_RAW
if not recipients:
recipients = config.get("email_recipients", []) or DEFAULT_EMAIL_RECIPIENTS
if not recipients and sender:
recipients = [sender]
if alert_mode == "SMTP":
if not sender:
sender = DEFAULT_SMTP_SENDER
smtp_host = config.get("smtp_server", "").strip()
if not smtp_host:
if any(dom in sender.lower() for dom in ["@outlook.", "@hotmail.", "@live.", "@office365."]):
smtp_host = "smtp.office365.com"
else:
smtp_host = "smtp.gmail.com"
# Remove spaces in Google/Microsoft App Passwords (e.g. "abcd efgh ijkl mnop" -> "abcdefghijklmnop")
clean_pwd = (password or DEFAULT_SMTP_PASSWORD_RAW).replace(" ", "")
connected = False
last_error = None
# Custom Webhook / Google Apps Script (HTTPS Port 443 - works on Hugging Face)
webhook_url = os.environ.get("EMAIL_WEBHOOK_URL") or config.get("email_webhook_url", "").strip()
if webhook_url and not connected:
try:
all_webhook_sent = True
for recipient in recipients:
payload = {
"sender": sender,
"senderName": "StatusPilot NOC",
"recipient": recipient,
"subject": subject,
"body": plain_body,
"htmlBody": html_content
}
w_res = requests.post(webhook_url, json=payload, timeout=20, allow_redirects=True)
print(f"Webhook response for {recipient}: status={w_res.status_code} body={w_res.text[:200]}")
if w_res.status_code not in [200, 201, 202] and "OK" not in w_res.text.upper() and "SUCCESS" not in w_res.text.upper():
all_webhook_sent = False
if all_webhook_sent:
connected = True
print(f"Alert successfully dispatched via HTTPS Webhook to all recipients: {recipients}")
except Exception as ew:
last_error = ew
print(f"Webhook dispatch failed: {ew}")
# Try Standard Gmail Port 465 (SMTP_SSL)
if not connected:
try:
ssl_context = ssl.create_default_context()
from email.utils import formataddr
display_from = formataddr(("StatusPilot NOC", sender))
with smtplib.SMTP_SSL(smtp_host, 465, context=ssl_context, timeout=12) as server:
server.login(sender, clean_pwd)
for recipient in recipients:
mime_msg = MIMEMultipart("alternative")
mime_msg['From'] = display_from
mime_msg['To'] = recipient
mime_msg['Subject'] = subject
mime_msg.attach(MIMEText(plain_body, 'plain'))
mime_msg.attach(MIMEText(html_content, 'html'))
server.sendmail(sender, recipient, mime_msg.as_string())
connected = True
print(f"Alert successfully dispatched via SMTPS Port 465 to {recipients}")
except Exception as e1:
last_error = e1
print(f"Port 465 SMTPS failed: {e1}. Trying Port 587 STARTTLS...")
# Try Standard Gmail Port 587 (STARTTLS)
if not connected:
try:
from email.utils import formataddr
display_from = formataddr(("StatusPilot NOC", sender))
with smtplib.SMTP(smtp_host, 587, timeout=12) as server:
server.starttls()
server.login(sender, clean_pwd)
for recipient in recipients:
mime_msg = MIMEMultipart("alternative")
mime_msg['From'] = display_from
mime_msg['To'] = recipient
mime_msg['Subject'] = subject
mime_msg.attach(MIMEText(plain_body, 'plain'))
mime_msg.attach(MIMEText(html_content, 'html'))
server.sendmail(sender, recipient, mime_msg.as_string())
connected = True
print(f"Alert successfully dispatched via STARTTLS Port 587 to {recipients}")
except Exception as e2:
last_error = e2
print(f"Port 587 STARTTLS failed: {e2}")
if not connected and is_test:
raise RuntimeError(f"SMTP Delivery Error ({smtp_host}): {last_error}")
elif alert_mode == "Outlook" and OUTL_AVAIL:
try:
outlook = win32com.client.Dispatch("Outlook.Application")
mail = outlook.CreateItem(0)
mail.Subject = subject
mail.HTMLBody = html_content
if recipients:
mail.To = "; ".join(recipients)
mail.Send()
print("Outlook HTML alert sent.")
except Exception as e:
print(f"Outlook Alert Error: {e}")
if is_test:
raise RuntimeError(f"Outlook App Dispatch Error: {str(e)}")
# Start thread automatically
init_db()
if not os.path.exists(CONFIG_FILE):
save_config(load_config())
threading.Thread(target=background_monitor_thread, daemon=True).start()
# --- WEB ROUTES ---
@app.route('/')
def serve_dashboard():
"""Forcing a new template name to bypass persistent cloud caching."""
return render_template('dashboard_v6.html', logo_b64=PARLE_TECH_LOGO_B64)
@app.route('/api/config/delete', methods=['POST'])
def delete_config_service():
"""Expert-grade service decommissioning via POST for cloud compatibility."""
if not check_auth():
return jsonify({"error": "Unauthorized"}), 401
name = request.json.get('name')
print(f"DECOMMISSION REQUEST: {name}")
config = load_config()
original_count = len(config.get('services_to_monitor', []))
config['services_to_monitor'] = [s for s in config['services_to_monitor'] if s.get('name') != name]
new_count = len(config['services_to_monitor'])
save_config(config)
if original_count == new_count:
return jsonify({"status": "error", "message": f"Service '{name}' not found."}), 404
print(f"Successfully decommissioned: {name}")
return jsonify({"status": "success", "deleted": name})
@app.route('/api/config', methods=['GET', 'POST'])
def handle_config():
if request.method == 'POST':
new_config = request.json or {}
current_config = load_config()
new_pwd = str(new_config.get("smtp_password", "")).strip()
if not new_pwd or new_pwd.startswith("••••"):
new_config["smtp_password"] = current_config.get("smtp_password", "")
else:
new_config["smtp_password"] = security.encrypt_secret(new_pwd)
save_config(new_config)
return jsonify({"status": "success"})
cfg = load_config()
if cfg.get("smtp_password"):
cfg["smtp_password"] = "••••••••••••"
return jsonify(cfg)
@app.route('/api/clear_ai', methods=['POST'])
def clear_ai_services():
config = load_config()
# Robust filter: Removes anything starting with 'Ai Service', 'New Service', or 'Cloud Service'
patterns = ["Ai Service", "New Service", "Cloud Service", "Https", "Global"]
config['services_to_monitor'] = [s for s in config['services_to_monitor'] if not any(p in s['name'] for p in patterns)]
save_config(config)
return jsonify({"status": "success"})
@app.route('/api/status', methods=['GET'])
def get_status():
config = load_config()
services = config.get("services_to_monitor", [])
conn = get_db_connection()
c = conn.cursor()
results = []
for svc in services:
name = svc.get('name')
# Get latest status
c.execute("SELECT is_up, response_time, message, timestamp FROM status_history WHERE name=? ORDER BY timestamp DESC LIMIT 1", (name,))
row = c.fetchone()
# Get last 30 checks for timeline
c.execute("SELECT is_up FROM status_history WHERE name=? ORDER BY timestamp DESC LIMIT 30", (name,))
history_rows = c.fetchall()
timeline = [bool(h["is_up"]) for h in reversed(history_rows)]
if not row:
is_up, msg, latency = check_service_logic(svc)
out_row = {"is_up": 1 if is_up else 0, "response_time": latency, "message": msg, "timestamp": datetime.now().isoformat()}
else:
out_row = {"is_up": row["is_up"], "response_time": row["response_time"], "message": row["message"], "timestamp": row["timestamp"]}
results.append({
"name": name,
"url": svc.get("url"),
"location": svc.get("location", "Global"),
"type": svc.get("type"),
"is_up": bool(out_row["is_up"]),
"latency": out_row["response_time"],
"message": out_row["message"],
"last_check": out_row["timestamp"],
"timeline": timeline
})
conn.close()
return jsonify(results)
@app.route('/api/summary', methods=['GET'])
def get_summary():
conn = get_db_connection()
c = conn.cursor()
# Total Monitors
config = load_config()
services = config.get("services_to_monitor", [])
total_monitors = len(services)
# Global Uptime % (Last 24 hours)
yesterday = (datetime.now() - timedelta(days=1)).isoformat()
c.execute("SELECT AVG(is_up) * 100 FROM status_history WHERE timestamp > ?", (yesterday,))
uptime_pct = c.fetchone()[0] or 100.0
# Total Incidents (Count of UNIQUE services that are CURRENTLY down)
incidents = 0
for svc in services:
name = svc.get('name')
c.execute("SELECT is_up FROM status_history WHERE name=? ORDER BY timestamp DESC LIMIT 1", (name,))
row = c.fetchone()
if row and row[0] == 0:
incidents += 1
# Avg Latency
c.execute("SELECT AVG(response_time) FROM status_history WHERE timestamp > ?", (yesterday,))
avg_latency = c.fetchone()[0] or 0
conn.close()
return jsonify({
"total_monitors": total_monitors,
"uptime_pct": round(uptime_pct, 2),
"incidents": incidents,
"avg_latency": round(avg_latency, 2)
})
@app.route('/api/admin/purge_stale', methods=['POST'])
def purge_stale_data():
"""Purge history records for services no longer in the current configuration."""
if not check_auth():
return jsonify({"error": "Unauthorized"}), 401
config = load_config()
active_names = [s.get('name') for s in config.get('services_to_monitor', [])]
conn = get_db_connection()
c = conn.cursor()
# 1. Delete records for non-existent services
placeholders = ','.join(['?'] * len(active_names))
if active_names:
c.execute(f"DELETE FROM status_history WHERE name NOT IN ({placeholders})", active_names)
else:
c.execute("DELETE FROM status_history")
# 2. Also clear very old data (older than 7 days) to keep DB light
one_week_ago = (datetime.now() - timedelta(days=7)).isoformat()
c.execute("DELETE FROM status_history WHERE timestamp < ?", (one_week_ago,))
count = conn.total_changes
conn.commit()
conn.close()
return jsonify({"status": "success", "purged_records": count})
@app.route('/api/history/