92 lines
3 KiB
Python
92 lines
3 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""Start an integration watchdog only when its configuration is usable."""
|
||
|
|
|
||
|
|
import os
|
||
|
|
import sys
|
||
|
|
|
||
|
|
|
||
|
|
def integration_is_configured(name):
|
||
|
|
import pymysql
|
||
|
|
from sqlalchemy.engine import make_url
|
||
|
|
|
||
|
|
database_url = make_url(os.environ['DATABASE_URL'])
|
||
|
|
connection = pymysql.connect(
|
||
|
|
host=database_url.host,
|
||
|
|
port=database_url.port or 3306,
|
||
|
|
user=database_url.username,
|
||
|
|
password=database_url.password,
|
||
|
|
database=database_url.database,
|
||
|
|
connect_timeout=5,
|
||
|
|
)
|
||
|
|
checks = {
|
||
|
|
"outlook": """
|
||
|
|
SELECT EXISTS(SELECT 1 FROM gmao_context WHERE auto_interpret_enabled = 1)
|
||
|
|
AND EXISTS(SELECT 1 FROM outlook_accounts WHERE is_active = 1)
|
||
|
|
""",
|
||
|
|
"ent": """
|
||
|
|
SELECT EXISTS(SELECT 1 FROM gmao_context WHERE auto_interpret_enabled = 1)
|
||
|
|
AND EXISTS(SELECT 1 FROM ent_credentials WHERE is_active = 1)
|
||
|
|
""",
|
||
|
|
"pronote": """
|
||
|
|
SELECT EXISTS(SELECT 1 FROM gmao_context WHERE pronote_sync_enabled = 1)
|
||
|
|
AND EXISTS(
|
||
|
|
SELECT 1 FROM pronote_session
|
||
|
|
WHERE session_token IS NOT NULL AND session_token != ''
|
||
|
|
)
|
||
|
|
""",
|
||
|
|
"yeastar": """
|
||
|
|
SELECT username_encrypted, password_encrypted
|
||
|
|
FROM yeastar_config
|
||
|
|
WHERE is_active = 1 AND base_url != ''
|
||
|
|
AND EXISTS(SELECT 1 FROM yeastar_dnd_config WHERE is_enabled = 1)
|
||
|
|
LIMIT 1
|
||
|
|
""",
|
||
|
|
}
|
||
|
|
if name not in checks:
|
||
|
|
raise ValueError(f"Integration inconnue: {name}")
|
||
|
|
try:
|
||
|
|
with connection.cursor() as cursor:
|
||
|
|
cursor.execute(checks[name])
|
||
|
|
result = cursor.fetchone()
|
||
|
|
if name != "yeastar":
|
||
|
|
return bool(result and result[0])
|
||
|
|
if not result:
|
||
|
|
return False
|
||
|
|
from cryptography.fernet import Fernet
|
||
|
|
|
||
|
|
key = os.environ.get('YEASTAR_ENCRYPTION_KEY') or os.environ.get(
|
||
|
|
'OUTLOOK_ENCRYPTION_KEY'
|
||
|
|
)
|
||
|
|
if not key:
|
||
|
|
return False
|
||
|
|
try:
|
||
|
|
cipher = Fernet(key.encode())
|
||
|
|
return bool(cipher.decrypt(result[0]) and cipher.decrypt(result[1]))
|
||
|
|
except Exception:
|
||
|
|
return False
|
||
|
|
finally:
|
||
|
|
connection.close()
|
||
|
|
|
||
|
|
|
||
|
|
def main():
|
||
|
|
if len(sys.argv) != 3:
|
||
|
|
raise SystemExit("usage: watchdog_gate.py INTEGRATION SCRIPT")
|
||
|
|
|
||
|
|
integration, script = sys.argv[1:]
|
||
|
|
try:
|
||
|
|
configured = integration_is_configured(integration)
|
||
|
|
except Exception as exc:
|
||
|
|
print(f"[{integration}] watchdog non lance: configuration illisible ({exc})")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
if not configured:
|
||
|
|
print(f"[{integration}] watchdog non lance: integration non configuree")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
print(f"[{integration}] configuration valide, lancement de {script}")
|
||
|
|
os.execv(sys.executable, [sys.executable, script])
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
raise SystemExit(main())
|