67 lines
1.5 KiB
Python
67 lines
1.5 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Health check script for Gadgetbridge MQTT integration
|
|
"""
|
|
|
|
import os
|
|
import sqlite3
|
|
import paho.mqtt.client as mqtt
|
|
from datetime import datetime, timedelta
|
|
|
|
|
|
def check_database():
|
|
"""Check if Gadgetbridge database is accessible"""
|
|
db_path = os.getenv("GADGETBRIDGE_DB_PATH", "/data/Gadgetbridge")
|
|
|
|
if not os.path.exists(db_path):
|
|
return False
|
|
|
|
try:
|
|
conn = sqlite3.connect(db_path)
|
|
cursor = conn.cursor()
|
|
cursor.execute(
|
|
"SELECT name FROM sqlite_master WHERE type='table' AND name='MI_BAND_ACTIVITY_SAMPLE'"
|
|
)
|
|
result = cursor.fetchone()
|
|
conn.close()
|
|
return result is not None
|
|
except:
|
|
return False
|
|
|
|
|
|
def check_mqtt_connection():
|
|
"""Check MQTT broker connectivity"""
|
|
try:
|
|
client = mqtt.Client("healthcheck")
|
|
if os.getenv("MQTT_USERNAME"):
|
|
client.username_pw_set(
|
|
os.getenv("MQTT_USERNAME"), os.getenv("MQTT_PASSWORD")
|
|
)
|
|
|
|
client.connect(
|
|
os.getenv("MQTT_BROKER", "localhost"),
|
|
int(os.getenv("MQTT_PORT", "1883")),
|
|
10,
|
|
)
|
|
client.disconnect()
|
|
return True
|
|
except:
|
|
return False
|
|
|
|
|
|
def main():
|
|
"""Main health check"""
|
|
db_ok = check_database()
|
|
mqtt_ok = check_mqtt_connection()
|
|
|
|
if db_ok and mqtt_ok:
|
|
print("Health check passed")
|
|
exit(0)
|
|
else:
|
|
print(f"Health check failed - DB: {db_ok}, MQTT: {mqtt_ok}")
|
|
exit(1)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|