"""Service Microsoft Teams — envoi de messages via Incoming Webhook. Webhook = simple, pas d'auth Azure AD requise. Format : Adaptive Card (titre + corps) ou simple "text" en fallback. À noter : Teams a déprécié les "Office 365 Connectors" (anciens webhooks chans Teams). Les nouveaux webhooks utilisent "Workflows" (Power Automate) qui acceptent un payload Adaptive Card. Ce service envoie un payload compatible avec les deux. """ import json import logging from datetime import datetime from typing import Dict, Any import requests log = logging.getLogger("patchcenter.teams") HTTP_TIMEOUT = 10 PROXY_URL = None # à override via Settings si besoin def _proxies(): if PROXY_URL: return {"http": PROXY_URL, "https": PROXY_URL} return None def _post(webhook_url: str, payload: Dict[str, Any]) -> Dict[str, Any]: """Poste un payload sur le webhook Teams. Retourne {ok, status, detail}.""" try: r = requests.post(webhook_url, json=payload, timeout=HTTP_TIMEOUT, proxies=_proxies()) ok = (200 <= r.status_code < 300) return { "ok": ok, "status": r.status_code, "detail": (r.text or "")[:500] if not ok else "Message envoyé", } except requests.exceptions.RequestException as e: return {"ok": False, "status": 0, "detail": f"Réseau: {e}"} def _adaptive_card(title: str, body_lines: list, color: str = "good") -> Dict[str, Any]: """Construit un payload Adaptive Card minimal (compatible Workflows). color : 'good' (vert) | 'warning' (orange) | 'attention' (rouge) | 'default'.""" facts = [] body_blocks = [ {"type": "TextBlock", "text": title, "weight": "Bolder", "size": "Medium", "color": color, "wrap": True} ] for ln in body_lines: if isinstance(ln, dict) and "title" in ln and "value" in ln: facts.append({"title": ln["title"], "value": str(ln["value"])}) else: body_blocks.append({"type": "TextBlock", "text": str(ln), "wrap": True, "spacing": "Small"}) if facts: body_blocks.append({"type": "FactSet", "facts": facts}) body_blocks.append({"type": "TextBlock", "text": f"_{datetime.now().strftime('%Y-%m-%d %H:%M')} — PatchCenter_", "size": "Small", "isSubtle": True, "spacing": "Medium"}) return { "type": "message", "attachments": [{ "contentType": "application/vnd.microsoft.card.adaptive", "contentUrl": None, "content": { "$schema": "http://adaptivecards.io/schemas/adaptive-card.json", "type": "AdaptiveCard", "version": "1.4", "body": body_blocks, } }] } def resolve_channel_for_server(db, server_id: int) -> Dict[str, Any]: """Résout le canal Teams à utiliser pour un serveur donné. Priorité : 1. servers.teams_channel_id (override par serveur) 2. applications.teams_channel_id (canal de l'app rattachée) 3. teams_channels.is_default = true (canal global) Retourne {ok, channel_id, name, webhook_url, source} ou {ok: False, msg}.""" from sqlalchemy import text as sqlt if not server_id: return {"ok": False, "msg": "server_id manquant"} # 1) Override par serveur row = db.execute(sqlt(""" SELECT tc.id, tc.name, tc.webhook_url, 'server' AS source FROM servers s JOIN teams_channels tc ON tc.id = s.teams_channel_id WHERE s.id = :sid AND tc.is_active = true """), {"sid": server_id}).fetchone() if row: return {"ok": True, "channel_id": row.id, "name": row.name, "webhook_url": row.webhook_url, "source": row.source} # 2) Canal de l'application row = db.execute(sqlt(""" SELECT tc.id, tc.name, tc.webhook_url, 'application' AS source FROM servers s JOIN applications a ON a.id = s.application_id JOIN teams_channels tc ON tc.id = a.teams_channel_id WHERE s.id = :sid AND tc.is_active = true """), {"sid": server_id}).fetchone() if row: return {"ok": True, "channel_id": row.id, "name": row.name, "webhook_url": row.webhook_url, "source": row.source} # 3) Canal défaut global row = db.execute(sqlt(""" SELECT id, name, webhook_url, 'default' AS source FROM teams_channels WHERE is_default = true AND is_active = true ORDER BY id LIMIT 1 """)).fetchone() if row: return {"ok": True, "channel_id": row.id, "name": row.name, "webhook_url": row.webhook_url, "source": row.source} return {"ok": False, "msg": "Aucun canal Teams configuré (ni serveur, ni app, ni défaut)"} def send_test_message(webhook_url: str, channel_name: str, sender: str) -> Dict[str, Any]: """Envoie un message de test pour valider la config du webhook.""" payload = _adaptive_card( title=f"✅ Test PatchCenter — canal '{channel_name}'", body_lines=[ "Ce message confirme que le webhook Teams est correctement configuré.", {"title": "Envoyé par", "value": sender}, {"title": "Source", "value": "PatchCenter > Settings > Teams"}, ], color="good", ) return _post(webhook_url, payload) def send_intervention_start(webhook_url: str, hostname: str, application: str, intervenant: str, planned_at: str = None) -> Dict[str, Any]: """Annonce le DÉBUT d'une intervention de patching.""" body = [ {"title": "Serveur", "value": hostname}, {"title": "Application", "value": application or "—"}, {"title": "Intervenant", "value": intervenant or "—"}, ] if planned_at: body.append({"title": "Prévu", "value": planned_at}) payload = _adaptive_card( title=f"🟧 Début intervention patching — {hostname}", body_lines=body, color="warning", ) return _post(webhook_url, payload) def send_intervention_end(webhook_url: str, hostname: str, application: str, intervenant: str, status: str = "ok") -> Dict[str, Any]: """Annonce la FIN d'une intervention. status = 'ok' | 'ko'.""" color = "good" if status == "ok" else "attention" icon = "✅" if status == "ok" else "❌" payload = _adaptive_card( title=f"{icon} Fin intervention patching — {hostname}", body_lines=[ {"title": "Serveur", "value": hostname}, {"title": "Application", "value": application or "—"}, {"title": "Intervenant", "value": intervenant or "—"}, {"title": "Statut technique", "value": status.upper()}, "_(En attente de validation par le responsable applicatif)_", ], color=color, ) return _post(webhook_url, payload) def send_planning_reminder(webhook_url: str, hostname: str, application: str, jour: str, heure: str, intervenant: str) -> Dict[str, Any]: """Rappel de planning (envoyable du jeudi/vendredi précédent jusqu'au jour J).""" payload = _adaptive_card( title=f"🗓 Rappel planning patching — {hostname}", body_lines=[ {"title": "Serveur", "value": hostname}, {"title": "Application", "value": application or "—"}, {"title": "Date prévue", "value": jour or "—"}, {"title": "Heure prévue", "value": heure or "—"}, {"title": "Intervenant", "value": intervenant or "—"}, "_Merci de confirmer la disponibilité applicative._", ], color="default", ) return _post(webhook_url, payload)