Spaces:
Sleeping
Sleeping
import os
import base64
import sys
import json
import time
import sqlite3
import threading
import requests
import io
import re
import xlsxwriter
import smtplib
import imaplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from datetime import datetime, timedelta
import ssl
import socket
from flask import Flask, request, jsonify, render_template, send_file, session, Response
try:
import win32com.client
OUTL_AVAIL = True
except ImportError:
OUTL_AVAIL = False
try:
import security
except ImportError:
print("Warning: security.py not found. License bypass may occur or app might fail.")
def get_license_status():
lic_file = 'license.dat'
if not os.path.exists(lic_file):
return False, "Missing license.dat file."
try:
with open(lic_file, 'r') as f:
lic_data = json.load(f)
email = lic_data.get('email', '')
key = lic_data.get('key', '')
if not security.verify_license_key(email, key):
return False, "Invalid License Key."
return True, f"Licensed to {email}"
except Exception as e:
return False, f"License read error: {e}"
# --- SMART PATHING FOR CLOUD DEPLOYMENT ---
import os
# Detect if we are in a 'flat' environment (all files in root) or standard (templates/static)
base_dir = os.path.dirname(os.path.abspath(__file__))
template_dir = os.path.join(base_dir, 'templates')
static_dir = os.path.join(base_dir, 'static')
# Fallback to root if folders don't exist (helps users who can't create folders)
if not os.path.exists(template_dir):
template_dir = base_dir
if not os.path.exists(static_dir):
static_dir = base_dir
app = Flask(__name__,
template_folder=template_dir,
static_folder=static_dir,
static_url_path='/static')
# CRITICAL: Use a custom secret key from environment or a strong default
app.secret_key = os.environ.get('SESSION_SECRET', os.urandom(32).hex())
# --- VERSIONING ---
VERSION = "5.1.0-CRYSTAL-PRO"
@app.route('/api/version')
def get_version():
return jsonify({"version": VERSION})
# --- CLOUD SESSION HARDENING (IFRAME COMPATIBLE) ---
app.config.update(
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SECURE=True, # Must be True for SameSite=None
SESSION_COOKIE_SAMESITE='None', # Allows cookie in iframes (HF Spaces)
PERMANENT_SESSION_LIFETIME=timedelta(days=30),
SESSION_REFRESH_EACH_REQUEST=True
)
def check_auth():
"""Verify if the session is authenticated via Master Key."""
master_key = os.environ.get('MASTER_KEY')
# If no key is set at all (None), we allow access (local mode)
if master_key is None:
return True
# If a key is set but empty, it's a security risk - require auth anyway
if master_key.strip() == "":
return False
return session.get('authenticated') == True
@app.before_request
def protect_apis():
"""Protect all /api routes except for version, auth and static routes."""
if request.path.startswith('/api/') and not request.path.startswith('/api/auth/') and not request.path == '/api/version':
if not check_auth():
return jsonify({"error": "Unauthorized Access", "code": 401}), 401
@app.route('/api/auth/status')
def auth_status():
master_key = os.environ.get('MASTER_KEY')
has_key = master_key is not None and master_key.strip() != ""
return jsonify({
"required": has_key,
"authenticated": check_auth()
})
@app.route('/api/auth/login', methods=['POST'])
def login():
data = request.json
key = data.get('key', '').strip()
master_key = os.environ.get('MASTER_KEY', '').strip()
if master_key and key == master_key:
session['authenticated'] = True
session.permanent = True
return jsonify({"success": True})
return jsonify({"success": False, "message": "Invalid Master Access Key"})
@app.route('/api/license/status')
def is_licensed():
# check_auth is already handled by @app.before_request
valid, msg = get_license_status()
return jsonify({"licensed": valid, "message": msg})
@app.route('/api/license/activate', methods=['POST'])
def activate_license():
data = request.json
email = data.get('email', '')
key = data.get('key', '')
if security.verify_license_key(email, key):
with open('license.dat', 'w') as f:
json.dump({"email": email, "key": key}, f)
return jsonify({"success": True})
return jsonify({"success": False, "message": "Invalid Purchase Email or Key"})
def check_ssl(hostname):
"""Check SSL certificate expiration and status."""
try:
context = ssl.create_default_context()
with socket.create_connection((hostname, 443), timeout=10) as sock:
with context.wrap_socket(sock, server_hostname=hostname) as ssock:
cert = ssock.getpeercert()
expiry_str = cert.get('notAfter')
expiry_date = datetime.strptime(expiry_str, '%b %d %H:%M:%S %Y %Z')
days_left = (expiry_date - datetime.now()).days
if days_left < 7:
return False, f"SSL Warning: Expiring in {days_left} days", days_left
return True, f"SSL Valid ({days_left} days left)", days_left
except Exception as e:
return False, f"SSL Error: {str(e)}", 0
DEFAULT_BROWSER_HEADERS = {
'User-Agent': 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/124.0.0.0 Safari/537.36',
'Accept': 'text/html,application/xhtml+xml,application/xml;q=0.9,image/avif,image/webp,image/apng,*/*;q=0.8',
'Accept-Language': 'en-US,en;q=0.9',
'Cache-Control': 'no-cache',
'Pragma': 'no-cache'
}
CONFIG_FILE = 'config.json'
DB_FILE = 'monitor_metrics.db'
LAST_ALERT_METRICS = {}
SERVICE_STATE_CACHE = {}
FAIL_COUNT_CACHE = {}
SUCCESS_COUNT_CACHE = {}
DOWN_START_TIME = {}
FLAP_COUNTER = {}
def get_db_connection():
conn = sqlite3.connect(DB_FILE, check_same_thread=False)
conn.row_factory = sqlite3.Row
return conn
def init_db():
conn = get_db_connection()
c = conn.cursor()
c.execute('''CREATE TABLE IF NOT EXISTS status_history
(timestamp DATETIME, name TEXT, url TEXT, location TEXT, is_up INTEGER, response_time REAL, message TEXT)''')
conn.commit()
conn.close()
PARLE_TECH_LOGO_B64 = "iVBORw0KGgoAAAANSUhEUgAAAJwAAABoCAYAAAAEu5lCAAAPh0lEQVR4nO2da2wc13XHf/Pa9y5NkaEokXqQlk05rqvYjJWktStZLoJEQOUPqYICTdJGDZAYRVIgsQ23Hww6BYpWzIe2aAO3TWigTYuksoE4CZQiqSzbilNUCtMaaVIwtmlRpkSK5UPa5+zOqx9m+RDJ3Z3ZXc5S5P0BC0jcnbt3Zv733HvOuXNWchzHQSAICLnVHRBsL4TgBIEiBCcIFCE4QaAIwQkCRQhOEChCcIJAEYITBIoQnCBQhOAEgSIEJwgUIThBoAjBCQJFCE4QKEJwgkARghMEihCcIFCE4ASBIgQnCBQhOEGgCMEJAkUIThAoQnCCQBGCEwSKEJwgUNSgv9CwbcYzesX3+5MRNLmxcWDYNi9Nm4RUONEVqruNawUb3SyteS+ihtgdlevu58pr0K7FaI/YDZ8zwEweFoy87+Miaoh98WCkEKjgFoVw8kq0sm2VYVAxONnu8MW9al03Yjxr8efXJY7FTE501dfX8azF6asOI5lo5Q/JDp9JFnhil8ZASvPc9vdnTb58VWPUUDmVLPHMXpl98cYF98cTJUZyUbD9HTeomZwdgK5Yw12oSWumVLX8zeu9gFFL5elZjQ+9YfHynOGracO2+VnWYdRQmbMVxtL+jl9DpX6W+/q1TJSDv1T5i8slJnKm77Y7ZKux/q1gqa1qfV7nNWoFZ3cCn1IBMHEtmWayUy3c8tbZYnLp/VFUnrpichqDYx3eLMi1gs3FjAWyxhu6xCtzli/rsx6Dytp+LvXVdvv69IwGGHwx6mN69KlPP1Tq82qum1Hujxm0RyIEYX9aIzgAGY7FTE4PpNa89cK1EievaWDDqK3yzzMGxzq8NatbDsOZyNKxh3SHj+TM+tcoJqDAl3ZH14h+Imfy5Ss2Ixn370/PatwTM+peNzYNG073wbGOtdd2fRobkH7YlF7qb+8Oca5vefi/oUueptaJnMm3550lC4nsHnvpps9FjUf2xVWe2SvzZFvZCbLhRwuW/6l1gzDsjTnvRtiUggO4MyJxKrkssrF8bcHplsOZBemWsxo1VC5mrA27+PviKg+1K0vfOZ610a0Wl9yzQTY2Z9m/TSu43VGZu6O4U6Olkq6xsF1yFooqg5rJZ5IFTsUNsF0RXFho3uJ8NQlF4mOSG464LIe42qCfspXZtIIDSCnep6bxrMXzc+7Udiji8MQujcMR16q96MS4lKlzxG/kFWpy23O2shQSsTWpuY03idYIrujU9NCuFWy+Pqu5n7Xhnhoxoqm8w9lcEoC7ozCQ0rgrpbhuUc7hlwV8r61iJQcKtYWatRxetGJQcPiQbNPjYQ2+05ah4LgiaSbl/vbgNCWY3Gxa56VWwbBtLt20GS1qEIbjis1ApPKInciZvHjdgaLD8YTDByLu33s0OBU3GMmpjGQ0Ds8U+Wxfc095Imfywyl3UBCW6NRsIop3EcVKTV5rhd3r9JVZkwfytQfYQEzj4XYlMHG2RnA1llNuNqJs3YDuNouBVOVQg245/I0RgjB0x0x+vcs9rf6EwolwgRFLhYLD/6QlDLs5aSSAsZt5Tl9TGbmhAQ5EJe7vVDyFYK7LNtBk67ZI0eFr05Han7PgY20FHm6vkk1pMq2zcEWH4ekw49nc0p8uyyFGiyrkVVBcsR1PODzVU9m6GbbNK3MWpBVISHy0jSVBabLMwc4QzEuQdhg3VS4sWBzr8Ce40aIbgN5/eVVf0ytuqgVDCZ2PdoZrtlcy3TbZCG+2uKJNC1fTiwN89b9bQMun1Bezq0eXs3xhEhJf2l89UzCetfiHG+7VOyUZPNh2q5j6EwrPdRT5XDrE2azE4eum56zFEkWH0aLCKJUtwZM9RX6/V9s066bjCYfdiWLVzyxkHT61J7igL7RScIsjzVp1g0I2xCWGYjp/cjCMJle+IEuhkAUVFOhIWeyOqmucg46wDDEJ8g4XCyHG0ob3dNdKK7Gyr8pyXO+ve2w+d1fYl9gGTYvRjfLZLHiyx+JodwDZeJ+01MKdSpo8dfet4nC3J0lQxZoscq1g8/2b5f9YMHw1zPBVWOt8y4A71ZzNSpzwk1+1ZIg5nNm7bD11s8Q3r0oMTbvT579lJe6bMTna7TOltXqwNYtyu81crzaLFjkNMig2e+MWA231j0LdchiZUddanko3UrGhJHNxVuYjXbXzq/mSu3YcNC06ZVZ8XuV3JIMrOZORjMrZeYXD0RJ9yQZytk2gI2tAgDs/6qGlvZs16h99EzmTb06aUNJAsRkM2RzqsCuGGaZMkxeNGGTgjaLMpZs2++J1fz0DKY0THTlGMu4lHJoO875IgX39rbukhdDmFhtsAqehXnTL4bvz7rR4Kmny3KCKJlc+nZl8mP6f5xgmzmhB5dX/K/JYd2NTzvH9UYb0IkNXo2DJ/P1ChIN+1ofbkM01wXvEsG1+PG0yuiBDyJ2aawmnKwbHFz0yxeY/MgqvzzS2q0OTZY7sUBiMuu2cnVf4wZTka6NAtGQSlTZm3G+29Ru0ysLpNkTqvxjXCjYvZNwF+iOyxNGkgRcnY1dM4tQdJUamVUYtlZ/dNDnaXbuvqYhCpX3bR7tD/GG6wKkJFXSbL0zL3Nfm0YHQbV4uholfLnCnVnuzJMCHe3ZU3wpuyaDbvDpvMZmZ99QmwIM7k/QnNj7j0NIptVOrb8vQOxmbs/MhiMCRjgJH9njbaNifUHiorcjItHva/522mWhkc2aZYz0aT97MMXwjDiV4dd6q6kBoukHKiUFEYrQgM/q2DNQOGBOROZ8s0RWrIeY4DL0d9tYmgCNzJl6iP+Ht443QGpvruKPwSs5/uHsiZ/JP7wJlg3DvHd7b0GSZvpQCKff4kWmV70zWPi4tVc/F7YurHN+j8UjEgYJ7s78zWXkDpBHROG+4n6WAez28vPTqAzRfktz2sh7bW3wFSOAW7r6ExNABVy1HdvgXXFRSOdypszdeoDsq8ZjP2FdfUubMXSY/v+GK6N5OtXq8ypFJORK1HoVyp9ASR+bddnclFRZ0dd3p78E2mef6Skx72ImykkhMoS9ZWSAnOgz2xuvb93dfor4n5PwiiV+EXp+xtMHpMYeRqRCPtDs8c5fhP7ArWMPmc2MEWxohOEGgCMFVwnDI5NzdFmZ2bbkHQX0IwVVCk0igQS7oR+381wa5nbhtU1urWaxb8tW3/aeV+iIGn9xDTadgbK7E6XF4R6/+HZ/cqfOJA962Kxm2zTfeKnLhmsTzmRAQYdEjPpnM8fG7wzzWXdmDfPXdNP84HeHNWYdH99k8vj/quUbI372j863JMOcNh3OHrEC2mm8ZwQFMZWTOvyOBz4Dy+XiIh9oKNbMOUwa8Oetw4UYVx96Q+aBpsdArV73xEzmTv/1FkeG3olQK0J7Jxjkz5f77yQM5nrg3vqbN8YLGyFQIbsCBqA77q5/DSi7OypyfdMCQmT8YzMPbW0ZwC7pMMZ0FR4X1llyhFdvBS6vKhSUi7FBri1QydCC89vh1KDgmlS7vy3MGQxdtLtxYlY5bWUFp1VQ+/FaU43uqZBnqyNp0YLh91GyKCyXo3ng5bBnBdcXg5HtjPNh96xrohqHw0hWF56/IcAcM70rzW/0RpvMrRZPlnkp1OFZsobqnI8Wf/kq6Zl/274hVTGu9Ml3iz163uJBbFtvD7ylysl/hV2NuP9ojNuNZix9MSXzhigpT8Om9Ou9NVXkwpoGtXg0d65MtIzhwU0z74rcKZyJn8vqc620+ahR4f2eYgY4QAx3+g7hdMeiKeS0Qs5axtMEz/+VwoRReskjDB7L80WCivHZabFtmICUzkILPD8BL4zkOtmmVp2ipXMLpNuD26GUDRCWV2XLO9pwW5abTukIzXx8rceGGtmRRRgaLPPFgquZC/bH+ePUBEnA+tBG2lIWrRLdjlpPkeaSSz/xtAUJ5GwqN3dRXpkv89IoM8661/cZRi48faNJDLrn6+nc5Z7vi97YzqincPkMjaAyHSKl5udOfvKtzrrxue7hf5v3doaaEIEKG5Toxpn/VJAKsC7eIEFwAGLbNfFaGjCuKz+4q0d++LSaXNWzPs/bDKstRa/v4elbrx1ezXJzXQI1Cyqane+MCrO0Rrw5Ens644w4CdTuUegiIQqNOghrl3HSBc69Z8Fr1tNN+Jcdf/lqMx/pvfRzMMcMsZg8eDRXZpW1AnQU1yvMTec7/y5znQy7b7wlcAVtecE1BjULtWC+mbZOtsKnyXMnjdu96kMrOhx7jMpvvafuVCMH5odrCvMq0lC4GuMfVj/MQ4FS69JWBf2MLCFkapIsQjZHRM7gJco+ki9Blcf43FI72Raus4Wwgtu7aLBWWyumwGOeyDlNphwGPVdm90JvLMpnVePyQxl/9prtpwMta86lLOYb/Nw7Xqxe9aSZb3kvVdZsZ23X/e3NZkpHGQgGaLFd9rUdfl8LjuyKuePVY0y3eZHz5casFfblUWa1+qlkn0BgcbAPBtUdb/xzC7qhMl1yuNJ0u8tNJiZlmbXszHHpz2SY1tvFsecGtZKUl8IKeac5Uo8kyD/Q6sNN1HJ4ds/lFgNPYZmL7CM5S67METapG9MG9ET4dL89feZlHXrMYm2vO1vVJW6m/n+kibKQHvYrtI7gyGb2OH1GQGg81dMXgUw/I9CbKJbWmwhz8nulJdBM5k7Gb+U35yzJ+2XaCqwtnecFl2Lan13oc7Qsz9IAC7Qoopiu6F0IM/XtxXeHN5OErl9Ls/6rFwW/D+MIG7XRRNu5HU1azLcIieqYIVpRJWyFZpfz+uigmzCg88i0VFA/rrlIYevPlMMraqeoPPhAlEs7wiR+WnZkFi2f/M8yzP7Hc9mM2vbLFZFYrT5PlrMVkmB+9VaC/fZ0qAWqUmqXhq1EIrsL0lhfcQmGF5VCj/uNwK9c3lrcb05uvXhL/d9+XpLcjzdGXLZiMLbdtKVCCNeVOSkBXgTu7KkxI6SIUltNnXrlhxDamknoVtrzg2qMhundkIVTk0U6bnjbvYZJIMgyd0Gv7czYGuhR2papb0iN7Uji/B6+8U+Rf39T47tt5JuUELFiuVQWI2TzaDZ+/3+b4/ui6cb5kRIMuCyji6tG7AzDYVqS3zWIyqdATC2ar0jaqLbK4DvPrANQTMKvPyZjImejlCkmRiOyjjFi959bosf7ZRoITbAaElyoIFCE4QaAIwQkCRQhOEChCcIJAEYITBIoQnCBQhOAEgSIEJwgUIThBoAjBCQJFCE4QKEJwgkARghMEihCcIFCE4ASBIgQnCBQhOEGgCMEJAkUIThAoQnCCQBGCEwSKEJwgUITgBIHy/xBrrmEbfk0tAAAAAElFTkSuQmCC"
DEFAULT_SMTP_SENDER = os.environ.get("SMTP_SENDER", "smadapathi04@gmail.com")
DEFAULT_SMTP_PASSWORD_RAW = os.environ.get("SMTP_PASSWORD", "")
DEFAULT_SMTP_PASSWORD_ENC = security.encrypt_secret(DEFAULT_SMTP_PASSWORD_RAW) if DEFAULT_SMTP_PASSWORD_RAW else ""
DEFAULT_EMAIL_RECIPIENTS = ["suresh@parletech.com", "smadapathi04@gmail.com", "Vamshi@parletech.com", "Prabir@parletech.com", "Sagar@parletech.com"]
@app.after_request
def add_no_cache_headers(response):
response.headers['Cache-Control'] = 'no-cache, no-store, must-revalidate, max-age=0'
response.headers['Pragma'] = 'no-cache'
response.headers['Expires'] = '0'
return response
@app.route('/static/parle_tech_logo.png')
def serve_parle_logo():
img_data = base64.b64decode(PARLE_TECH_LOGO_B64)
return Response(img_data, mimetype='image/png')
def load_config():
config = {}
if os.path.exists(CONFIG_FILE):
try:
with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
config = json.load(f)
except Exception:
config = {}
if config.get("services_to_monitor") is None:
config["services_to_monitor"] = [
{"name": "Microsoft 365", "url": "https://status.office365.com", "location": "Global", "type": "http_code", "expected_code": 200},
{"name": "RingCentral", "url": "https://status.ringcentral.com", "location": "Global", "type": "http_code", "expected_code": 200},
{"name": "Skybridge Healthcare", "url": "https://www.skybridgehealthcare.com/", "location": "Primary Web", "type": "http_code", "expected_code": 200},
{"name": "Skybridge Aviation", "url": "https://skybridgeaviation.com/", "location": "Primary Web", "type": "http_code", "expected_code": 200},
{"name": "Skybridge Resources", "url": "https://skybridgeresources.com/", "location": "Primary Web", "type": "http_code", "expected_code": 200},
{"name": "BetterLending Website", "url": "https://betterlending.com/", "location": "Primary Web", "type": "http_code", "expected_code": 200}
]
config.setdefault("check_interval", 60)
config.setdefault("consecutive_failures_required", 3)
config.setdefault("retry_delay_seconds", 3)
config.setdefault("cooldown_minutes", 60)
config.setdefault("alert_mode", "SMTP")
# Senior DevOps Fallback to guarantee credentials even in stateless Docker
if not config.get("smtp_sender"):
config["smtp_sender"] = os.environ.get("SMTP_SENDER", DEFAULT_SMTP_SENDER)
if not config.get("smtp_password"):
config["smtp_password"] = os.environ.get("SMTP_PASSWORD", DEFAULT_SMTP_PASSWORD_ENC)
if not config.get("email_recipients"):
env_recips = os.environ.get("EMAIL_RECIPIENTS", "").strip()
if env_recips:
config["email_recipients"] = [r.strip() for r in env_recips.split(',') if r.strip()]
else:
config["email_recipients"] = DEFAULT_EMAIL_RECIPIENTS
return config
def save_config(config):
if "smtp_password" in config and config["smtp_password"]:
pwd = str(config["smtp_password"]).strip()
if pwd and not pwd.startswith("••••") and not pwd.startswith("ENC:"):
config["smtp_password"] = security.encrypt_secret(pwd)
with open(CONFIG_FILE, 'w', encoding='utf-8') as f:
json.dump(config, f, indent=4)
def check_service_logic(service):
url = service.get("url", "")
check_type = service.get("type", "http_code")
start_time = time.time()
try:
if check_type == "http_code":
expected = service.get("expected_code", 200)
response = requests.get(url, headers=DEFAULT_BROWSER_HEADERS, timeout=12, allow_redirects=True)
latency = round((time.time() - start_time) * 1000, 2)
if response.status_code != expected:
if response.status_code in [403, 406, 429, 503]:
sess = requests.Session()
sess.headers.update(DEFAULT_BROWSER_HEADERS)
r2 = sess.get(url, timeout=12, allow_redirects=True)
if r2.status_code == expected:
return True, "OK (WAF Bypass)", round((time.time() - start_time) * 1000, 2)
return False, f"HTTP {response.status_code}", latency
return True, "OK", latency
elif check_type == "json_status":
response = requests.get(url, headers=DEFAULT_BROWSER_HEADERS, timeout=12, allow_redirects=True)
latency = round((time.time() - start_time) * 1000, 2)
data = response.json()
key = service.get("key_to_check", "")
expected_val = service.get("expected_value", "")
val = data
for k in key.split('.'):
val = val.get(k)
if val is None: break
if str(val).lower() != str(expected_val).lower():
return False, f"JSON {val}", latency
return True, "OK", latency
elif check_type == "m365_mail":
sender = service.get('sender_email')
pwd = service.get('sender_password')
receiver = service.get('receiver_email', sender)
imap_server = service.get('imap_server', 'outlook.office365.com')
pulse_id = f"StatusPilot-Pulse-{int(time.time())}"
msg = MIMEMultipart()
msg['From'] = sender
msg['To'] = receiver
msg['Subject'] = pulse_id
msg.attach(MIMEText(f"Mail Flow Heartbeat sent at {datetime.now()}", 'plain'))
try:
with smtplib.SMTP('smtp.office365.com', 587, timeout=15) as server:
server.starttls()
server.login(sender, pwd)
server.sendmail(sender, receiver, msg.as_string())
with imaplib.IMAP4_SSL(imap_server) as mail:
mail.login(receiver, service.get('receiver_password', pwd))
mail.select('inbox')
typ, data = mail.search(None, '(SUBJECT "StatusPilot-Pulse-")')
if typ == 'OK' and data[0]:
return True, "Mail Flow Active (Pulse Received)", 120
return True, "Pulse Sent (Awaiting Receipt)", 50
except Exception as e:
return False, f"Mail Flow Error: {str(e)}", 0
elif check_type == "ssl":
hostname = url.replace('https://', '').replace('http://', '').split('/')[0].split(':')[0]
ok, msg, days = check_ssl(hostname)
latency = round((time.time() - start_time) * 1000, 2)
return ok, msg, latency
except requests.exceptions.Timeout:
latency = round((time.time() - start_time) * 1000, 2)
return False, "Connection Timeout (12s)", latency
except requests.exceptions.RequestException as re:
latency = round((time.time() - start_time) * 1000, 2)
return False, f"Connection Failed: {str(re)}", latency
except Exception as e:
latency = round((time.time() - start_time) * 1000, 2)
return False, f"Engine Error: {str(e)}", latency
return True, "OK", 0
def check_service_with_retry(service, retry_delay=3):
"""Immediate re-check logic to eliminate single-packet transient glitches."""
is_up, msg, latency = check_service_logic(service)
if not is_up:
time.sleep(retry_delay)
is_up, msg, latency = check_service_logic(service)
if not is_up:
msg = f"{msg} (Confirmed)"
return is_up, msg, latency
def background_monitor_thread():
global LAST_ALERT_METRICS, SERVICE_STATE_CACHE, FAIL_COUNT_CACHE, SUCCESS_COUNT_CACHE, DOWN_START_TIME, FLAP_COUNTER
while True:
try:
config = load_config()
services = config.get("services_to_monitor", [])
interval = config.get("check_interval", 60)
failures_required = config.get("consecutive_failures_required", 3)
retry_delay = config.get("retry_delay_seconds", 3)
cooldown_secs = config.get("cooldown_minutes", 60) * 60
for svc in services:
name = svc.get('name', 'Unknown')
loc = svc.get('location', 'Global')
url = svc.get('url', '')
is_up, msg, latency = check_service_with_retry(svc, retry_delay=retry_delay)
previous_state = SERVICE_STATE_CACHE.get(name, True)
if is_up:
FAIL_COUNT_CACHE[name] = 0
succ_cnt = SUCCESS_COUNT_CACHE.get(name, 0) + 1
SUCCESS_COUNT_CACHE[name] = succ_cnt
# Senior DevOps Flap Dampening: Require 3 consecutive success checks before declaring recovery
if not previous_state and succ_cnt >= 3:
SERVICE_STATE_CACHE[name] = True
down_time = DOWN_START_TIME.pop(name, None)
duration_str = None
if down_time:
secs = int(time.time() - down_time)
m, s = divmod(secs, 60)
h, m = divmod(m, 60)
duration_str = f"{h}h {m}m {s}s" if h > 0 else f"{m}m {s}s" if m > 0 else f"{s}s"
now_t = time.time()
flaps = [t for t in FLAP_COUNTER.get(name, []) if (now_t - t) < 3600]
flaps.append(now_t)
FLAP_COUNTER[name] = flaps
if len(flaps) >= 4:
print(f"[FLAP SUPPRESSION ACTIVE] {name} state oscillating rapidly. Suppressing resolution email.")
else:
print(f"[RESOLVED] {name} is back ONLINE (Downtime: {duration_str})")
send_alert(config, name, loc, url, "SERVICE RESTORED", "Status: Operational (Verified across 3 consecutive checks)", is_resolved=True, duration_str=duration_str)
else:
SUCCESS_COUNT_CACHE[name] = 0
fail_cnt = FAIL_COUNT_CACHE.get(name, 0) + 1
FAIL_COUNT_CACHE[name] = fail_cnt
if fail_cnt >= failures_required:
curr = time.time()
if previous_state:
SERVICE_STATE_CACHE[name] = False
DOWN_START_TIME[name] = curr
LAST_ALERT_METRICS[name] = curr
flaps = [t for t in FLAP_COUNTER.get(name, []) if (curr - t) < 3600]
flaps.append(curr)
FLAP_COUNTER[name] = flaps
if len(flaps) >= 4:
print(f"[FLAP SUPPRESSION ACTIVE] {name} is flapping. Sending single Consolidated Flap Alert.")
alert_details = f"FLAP WARNING: {name} is experiencing rapid state oscillation ({len(flaps)} state changes in 1hr). Notifications throttled for 60 minutes."
send_alert(config, name, loc, url, "FLAPPING / UNSTABLE", alert_details, is_resolved=False)
else:
alert_details = f"{msg} (Verified across {fail_cnt} consecutive checks)"
print(f"[CRITICAL DOWN] {name} confirmed DOWN ({msg})")
send_alert(config, name, loc, url, msg, alert_details, is_resolved=False)
else:
last_alert = LAST_ALERT_METRICS.get(name, 0)
if (curr - last_alert) > cooldown_secs:
LAST_ALERT_METRICS[name] = curr
alert_details = f"Ongoing Outage: {msg} (Service remains unreachable)"
print(f"[CRITICAL DOWN REMINDER] {name} still DOWN")
send_alert(config, name, loc, url, msg, alert_details, is_resolved=False)
try:
conn = get_db_connection()
c = conn.cursor()
c.execute("INSERT INTO status_history (timestamp, name, url, location, is_up, response_time, message) VALUES (?, ?, ?, ?, ?, ?, ?)",
(datetime.now().isoformat(), name, url, loc, 1 if is_up else 0, latency, msg))
conn.commit()
conn.close()
except Exception as e:
print("DB error", e)
time.sleep(max(interval, 10))
except Exception as e:
print(f"Monitor error: {e}")
time.sleep(10)
def send_alert(config, name, loc, url, msg, details, is_resolved=False, duration_str=None, is_test=False):
"""Unified executive alerting engine supporting HTML formatting for SMTP and Outlook."""
if is_test:
subject = f"[TEST ALERT] StatusPilot Diagnostic Notification ({name})"
badge_color = "#00D4FF"
badge_text = "TEST DIAGNOSTIC"
header_title = "Diagnostic Alert Test"
elif is_resolved:
subject = f"[RESOLVED] {name} ({loc}) is back ONLINE"
badge_color = "#10B981"
badge_text = "SERVICE RESTORED"
header_title = f"{name} Restored"
else:
subject = f"[CRITICAL DOWN] {name} ({loc}) is DOWN"
badge_color = "#EF4444"
badge_text = "CRITICAL DOWNTIME"
header_title = f"{name} Outage Detected"
now_utc = datetime.utcnow()
est_str = (now_utc - timedelta(hours=4)).strftime('%Y-%m-%d %H:%M:%S EDT')
utc_str = now_utc.strftime('%H:%M:%S UTC')
ist_str = (now_utc + timedelta(hours=5, minutes=30)).strftime('%H:%M:%S IST')
timestamp_str = f"{est_str} ({utc_str} | {ist_str})"
duration_row = f"<tr><td style='padding:12px 16px; font-weight:600; color:#64748B; border-bottom:1px solid #E2E8F0; background:#F1F5F9;'>Downtime Duration:</td><td style='padding:12px 16px; font-weight:700; color:#0F172A; border-bottom:1px solid #E2E8F0;'>{duration_str}</td></tr>" if duration_str else ""
logo_img_tag = '<img src="https://chotustatus-statuspilot.hf.space/static/parle_tech_logo.png" height="46" alt="PARLE TECH" style="display:inline-block; border:0; outline:none; height:46px; width:auto; vertical-align:middle;" />'
html_content = f"""<!DOCTYPE html>
<html>
<head><meta charset="utf-8"></head>
<body style="font-family: -apple-system, BlinkMacSystemFont, 'Segoe UI', Roboto, Helvetica, Arial, sans-serif; background-color: #F8FAFC; margin: 0; padding: 28px 16px;">
<div style="max-width: 640px; margin: 0 auto; background: #FFFFFF; border-radius: 12px; border: 1px solid #E2E8F0; overflow: hidden; box-shadow: 0 20px 25px -5px rgba(0, 0, 0, 0.05);">
<!-- PARLE TECH BRAND HEADER -->
<div style="background-color: #0F172A; padding: 24px 30px; border-bottom: 3px solid #00C8F0;">
<table style="width: 100%; border-collapse: collapse;">
<tr>
<td style="vertical-align: middle;">
<div style="font-family: 'Segoe UI', Arial, sans-serif; font-size: 15px; font-weight: 700; color: #FFFFFF; letter-spacing: 0.5px;">Parle Tech Infrastructure Monitoring</div>
<div style="font-family: 'Segoe UI', Arial, sans-serif; font-size: 11px; font-weight: 600; color: #00C8F0; letter-spacing: 1.5px; text-transform: uppercase; margin-top: 2px;">24/7 Enterprise Vigilance</div>
</td>
<td style="text-align: right; vertical-align: middle; white-space: nowrap;">
{logo_img_tag}
</td>
</tr>
</table>
</div>
<!-- INCIDENT STATUS BANNER -->
<div style="background-color: {badge_color}; padding: 18px 30px; color: #FFFFFF;">
<div style="font-size: 11px; font-weight: 800; letter-spacing: 2px; text-transform: uppercase; opacity: 0.95;">INCIDENT CLASSIFICATION ALARM</div>
<h2 style="margin: 4px 0 0 0; font-size: 20px; font-weight: 700; color: #FFFFFF; letter-spacing: 0.5px;">{badge_text}: {header_title}</h2>
</div>
<!-- MAIN BODY CONTAINER -->
<div style="padding: 30px;">
<p style="font-size: 14px; color: #334155; margin-top: 0; line-height: 1.6;">
An automated vigilance event has been flagged by <strong>Parle Tech StatusPilot Infrastructure Security</strong>. Immediate technical telemetry details are provided below:
</p>
<!-- METRICS GRID TABLE -->
<table style="width: 100%; border-collapse: collapse; font-size: 13px; margin: 24px 0; background: #F8FAFC; border-radius: 8px; border: 1px solid #E2E8F0; overflow: hidden;">
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; border-bottom: 1px solid #E2E8F0; width: 35%; background: #F1F5F9;">Target Node / Service:</td>
<td style="padding: 12px 16px; font-weight: 700; color: #0F172A; border-bottom: 1px solid #E2E8F0;">{name}</td>
</tr>
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; border-bottom: 1px solid #E2E8F0; background: #F1F5F9;">Scope / Region:</td>
<td style="padding: 12px 16px; color: #0F172A; border-bottom: 1px solid #E2E8F0; font-weight: 600;">{loc}</td>
</tr>
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; border-bottom: 1px solid #E2E8F0; background: #F1F5F9;">Endpoint URL:</td>
<td style="padding: 12px 16px; color: #0284C7; border-bottom: 1px solid #E2E8F0; word-break: break-all; font-weight: 600;"><a href="{url}" target="_blank" style="color: #0284C7; text-decoration: none;">{url}</a></td>
</tr>
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; border-bottom: 1px solid #E2E8F0; background: #F1F5F9;">Vigilance Status:</td>
<td style="padding: 12px 16px; font-weight: 800; color: {badge_color}; border-bottom: 1px solid #E2E8F0;">{msg}</td>
</tr>
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; border-bottom: 1px solid #E2E8F0; background: #F1F5F9;">Technical Diagnostic:</td>
<td style="padding: 12px 16px; color: #334155; border-bottom: 1px solid #E2E8F0; line-height: 1.5;">{details}</td>
</tr>
{duration_row}
<tr>
<td style="padding: 12px 16px; font-weight: 600; color: #64748B; background: #F1F5F9;">Event Timestamp:</td>
<td style="padding: 12px 16px; color: #0F172A; font-weight: 600;">{timestamp_str}</td>
</tr>
</table>
<!-- ACTIONABLE CTA BUTTON -->
<div style="text-align: center; margin: 32px 0 20px 0;">
<a href="https://chotustatus-statuspilot.hf.space/" target="_blank" style="background-color: #0F172A; color: #00D4FF; font-family: 'Segoe UI', sans-serif; font-size: 13px; font-weight: 700; text-decoration: none; padding: 14px 32px; border-radius: 8px; letter-spacing: 1.5px; display: inline-block; border: 1px solid #00D4FF; box-shadow: 0 4px 12px rgba(15, 23, 42, 0.15);">
🌐 OPEN LIVE NOC OPERATIONS CENTER
</a>
</div>
<!-- CORPORATE SIGNATURE -->
<div style="margin-top: 32px; padding-top: 20px; border-top: 1px solid #E2E8F0; text-align: center; font-size: 11px; color: #94A3B8; line-height: 1.6;">
<strong>Parle Tech Infrastructure Security & NOC Engine v5.1</strong><br>
24/7 Enterprise Infrastructure Alerting • Automated Incident Management
</div>
</div>
</div>
</body>
</html>"""
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/<path:service_name>', methods=['GET'])
def get_detailed_history(service_name):
conn = get_db_connection()
c = conn.cursor()
# Get last 100 records for charts
c.execute("SELECT timestamp, is_up, response_time FROM status_history WHERE name=? ORDER BY timestamp DESC LIMIT 100", (service_name,))
rows = c.fetchall()
conn.close()
data = [{
"timestamp": r["timestamp"],
"is_up": bool(r["is_up"]),
"latency": r["response_time"]
} for r in reversed(rows)]
return jsonify(data)
@app.route('/api/test', methods=['POST'])
def test_tool():
"""Live interactive testing for specific tools and locations."""
data = request.json
tool_name = data.get("tool")
location = data.get("location")
# Simple direct ping/fetch simulation
config = load_config()
# Find matching service, or just do a generic requests test
matched_svc = next((s for s in config.get("services_to_monitor", []) if s.get("name") == tool_name), None)
if not matched_svc:
return jsonify({"status": "error", "message": "Service not found in configuration!"}), 404
is_up, msg, latency = check_service_logic(matched_svc)
return jsonify({
"tool": tool_name,
"location": location,
"is_up": is_up,
"latency": latency,
"message": msg,
"timestamp": datetime.now().isoformat()
})
@app.route('/api/ai_add', methods=['POST'])
def ai_add_service():
"""Neural Engine v4 - Context-Aware Entity Extraction."""
raw_text = request.json.get("text", "").strip()
if not raw_text:
return jsonify({"status": "error", "message": "Neural link empty. Please provide instructions."})
# 1. URL Extraction (Highest Priority)
url_match = re.search(r'(https?://[^\s]+)', raw_text)
if not url_match:
return jsonify({"status": "error", "message": "No endpoint detected. Try adding a URL like 'https://status.teams.com'"})
url = url_match.group(1).rstrip(',.!?')
# 2. Contextual Cleaning
clean_text = raw_text.replace(url, "").lower()
for cmd in ["monitor", "add", "check", "service", "from", "at", "called", "in"]:
clean_text = clean_text.replace(cmd, "")
clean_text = " ".join(clean_text.split())
# 3. Entity Extraction (Name & Location)
# Strategy: Locations are usually at the end or preceded by 'in'
location = "Global"
loc_match = re.search(r'(?:in|at|site|location|node)\s+([a-z\s]{2,20})$', raw_text, re.IGNORECASE)
if loc_match:
location = loc_match.group(1).strip().title()
# Clean the name part by removing the location
clean_text = clean_text.replace(location.lower(), "").strip()
# The remaining text is most likely the name
name = clean_text.title() if clean_text else "AI Discovery"
if len(name) < 2: name = "Cloud Service"
# 4. Smart Protocol Detection
is_api = any(x in url.lower() for x in [".json", "api/", "/status-data", "v1/"])
check_type = "json_status" if is_api else "http_code"
config = load_config()
new_svc = {
"name": name,
"url": url,
"location": location,
"type": check_type,
"expected_code": 200,
"key_to_check": "status" if is_api else "",
"expected_value": "up" if is_api else ""
}
config["services_to_monitor"].append(new_svc)
save_config(config)
return jsonify({
"status": "success",
"added": new_svc,
"message": f"✨ AI Mapping Successful",
"analysis": {
"entity": name,
"origin": location,
"protocol": "REST/JSON" if is_api else "HTTP/HTML",
"confidence": "High"
}
})
@app.route('/api/report/excel', methods=['GET'])
def download_excel_report():
output = io.BytesIO()
workbook = xlsxwriter.Workbook(output, {'in_memory': True})
header_style = workbook.add_format({'bold': True, 'bg_color': '#1E293B', 'font_color': 'white', 'border': 1})
data_style = workbook.add_format({'border': 1})
up_style = workbook.add_format({'bg_color': '#D1FAE5', 'font_color': '#065F46', 'border': 1, 'bold': True})
down_style = workbook.add_format({'bg_color': '#FEE2E2', 'font_color': '#991B1B', 'border': 1, 'bold': True})
conn = get_db_connection()
c = conn.cursor()
ws_sum = workbook.add_worksheet("Monthly Overview")
ws_sum.set_column('A:F', 22)
headers = ["Service Name", "Location", "Total Checks", "Uptime %", "Avg Latency (ms)", "Status"]
for col, h in enumerate(headers):
ws_sum.write(0, col, h, header_style)
c.execute("SELECT name, location, COUNT(*) as total, SUM(is_up) as up_count, ROUND(AVG(response_time), 2) FROM status_history GROUP BY name, location")
data = c.fetchall()
row_idx = 1
chart = workbook.add_chart({'type': 'doughnut'})
chart.set_title({'name': 'Services Uptime Analysis'})
chart.set_style(10)
for row in data:
name = row[0]
loc = row[1]
total = row[2]
up_count = row[3]
avg_latency = row[4]
up_pct = (up_count / total * 100) if total > 0 else 0
status_text = "Healthy" if up_pct >= 95 else "Degraded"
ws_sum.write(row_idx, 0, name, data_style)
ws_sum.write(row_idx, 1, loc, data_style)
ws_sum.write(row_idx, 2, total, data_style)
ws_sum.write(row_idx, 3, f"{up_pct:.1f}%", up_style if up_pct >= 95 else down_style)
ws_sum.write(row_idx, 4, avg_latency, data_style)
ws_sum.write(row_idx, 5, status_text, up_style if up_pct >= 95 else down_style)
row_idx += 1
if row_idx > 1:
chart.add_series({
'name': 'Uptime Info',
'categories': ['Monthly Overview', 1, 0, row_idx-1, 0],
'values': ['Monthly Overview', 1, 2, row_idx-1, 2],
})
ws_sum.insert_chart('H2', chart)
conn.close()
workbook.close()
output.seek(0)
return send_file(output, download_name=f"StatusPilot_Report_{datetime.now().strftime('%Y-%m')}.xlsx", as_attachment=True)
@app.route('/api/alert/test', methods=['POST'])
def test_alert_email():
if not check_auth():
return jsonify({"error": "Unauthorized"}), 401
req_data = request.json or {}
config = load_config()
if req_data.get("smtp_sender"):
config["smtp_sender"] = req_data.get("smtp_sender").strip()
if req_data.get("smtp_password") and not str(req_data.get("smtp_password")).startswith("••••"):
config["smtp_password"] = security.encrypt_secret(str(req_data.get("smtp_password")).strip())
if req_data.get("email_recipients"):
config["email_recipients"] = req_data.get("email_recipients")
if req_data.get("email_webhook_url"):
config["email_webhook_url"] = req_data.get("email_webhook_url").strip()
sender = config.get("smtp_sender", "").strip()
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 = []
if not recipients and sender:
recipients = [sender]
raw_pwd = os.environ.get('SMTP_PASSWORD') or config.get("smtp_password", "")
decrypted_pwd = security.decrypt_secret(raw_pwd).strip()
if not decrypted_pwd:
decrypted_pwd = DEFAULT_SMTP_PASSWORD_RAW
if not sender:
sender = DEFAULT_SMTP_SENDER
if not recipients:
recipients = DEFAULT_EMAIL_RECIPIENTS
config["email_recipients"] = recipients
app_host_url = request.host_url.rstrip('/') if (request and request.host_url) else "https://parledowntime-pro.hf.space"
try:
send_alert(
config=config,
name="System Alert Diagnostic Node",
loc="Global Operational Node",
url=app_host_url,
msg="Diagnostic Check OK",
details="This is a test executive alert sent from StatusPilot to verify email formatting and delivery.",
is_resolved=False,
is_test=True
)
recip_str = ", ".join(recipients)
return jsonify({"status": "success", "message": f"Test alert email successfully dispatched to: {recip_str}"})
except Exception as e:
return jsonify({"status": "error", "message": str(e)}), 200
if __name__ == '__main__':
from waitress import serve
print("Starting production Waitress server on http://0.0.0.0:5000")
serve(app, host='0.0.0.0', port=5000)
import os
import base64
import sys
import json
import time
import sqlite3
import threading
import requests
import io
import re
import xlsxwriter
import smtplib
import imaplib
from email.mime.text import MIMEText
from email.mime.multipart import MIMEMultipart
from email.header import Header
from datetime import datetime, timedelta
import ssl
import socket
from flask import Flask, request, jsonify, render_template, send_file, session, Response
try:
import win32com.client
OUTL_AVAIL = True
except ImportError:
OUTL_AVAIL = False
try:
import security
except ImportError:
print("Warning: security.py not found. License bypass may occur or app might fail.")
def get_license_status():
lic_file = 'license.dat'
if not os.path.exists(lic_file):
return False, "Missing license.dat file."
try:
with open(lic_file, 'r') as f:
lic_data = json.load(f)
email = lic_data.get('email', '')
key = lic_data.get('key', '')
if not security.verify_license_key(email, key):
return False, "Invalid License Key."
return True, f"Licensed to {email}"
except Exception as e:
return False, f"License read error: {e}"
# --- SMART PATHING FOR CLOUD DEPLOYMENT ---
import os
# Detect if we are in a 'flat' environment (all files in root) or standard (templates/static)
base_dir = os.path.dirname(os.path.abspath(__file__))
template_dir = os.path.join(base_dir, 'templates')
static_dir = os.path.join(base_dir, 'static')
# Fallback to root if folders don't exist (helps users who can't create folders)
if not os.path.exists(template_dir):
template_dir = base_dir
if not os.path.exists(static_dir):
static_dir = base_dir
app = Flask(__name__,
template_folder=template_dir,
static_folder=static_dir,
static_url_path='/static')
# CRITICAL: Use a custom secret key from environment or a strong default
app.secret_key = os.environ.get('SESSION_SECRET', os.urandom(32).hex())
# --- VERSIONING ---
VERSION = "5.1.0-CRYSTAL-PRO"
def get_version():
return jsonify({"version": VERSION})
# --- CLOUD SESSION HARDENING (IFRAME COMPATIBLE) ---
app.config.update(
SESSION_COOKIE_HTTPONLY=True,
SESSION_COOKIE_SECURE=True, # Must be True for SameSite=None
SESSION_COOKIE_SAMESITE='None', # Allows cookie in iframes (HF Spaces)
PERMANENT_SESSION_LIFETIME=timedelta(days=30),
SESSION_REFRESH_EACH_REQUEST=True
)
def check_auth():
"""Verify if the session is authenticated via Master Key."""
master_key = os.environ.get('MASTER_KEY')
# If no key is set at all (None), we allow access (local mode)
if master_key is None:
return True
# If a key is set but empty, it's a security risk - require auth anyway
if master_key.strip() == "":
return False
return session.get('authenticated') == True
def protect_apis():
"""Protect all /api routes except for version, auth and static routes."""
if request.path.startswith('/api/') and not request.path.startswith('/api/auth/') and not request.path == '/api/version':