72 lines
2.3 KiB
Python
72 lines
2.3 KiB
Python
import os
|
||
import ovh
|
||
from dotenv import load_dotenv
|
||
import unicodedata, re
|
||
|
||
load_dotenv()
|
||
|
||
def normaliser_sms(message: str, prefix: str = "") -> str:
|
||
REPL = {"°": "C", "’": "'", "“": '"', "”": '"', "–": "-", "—": "-", "…": "..."}
|
||
for k, v in REPL.items():
|
||
message = message.replace(k, v)
|
||
message = unicodedata.normalize("NFKD", message).encode("ascii", "ignore").decode("ascii")
|
||
rules = [
|
||
(r"\bTemperature\b", "T"),
|
||
(r"\bTemp[ée]rature\b", "T"),
|
||
(r"\bSeuil\b", "S"),
|
||
(r"\bHeure\b", "H"),
|
||
(r"\s{2,}", " "),
|
||
]
|
||
for pat, repl in rules:
|
||
message = re.sub(pat, repl, message, flags=re.IGNORECASE)
|
||
if prefix:
|
||
message = f"[{prefix}] {message}"
|
||
return message[:160]
|
||
|
||
def envoyer_sms(message: str, lieu: str = ""):
|
||
try:
|
||
client = ovh.Client(
|
||
endpoint=os.getenv("OVH_ENDPOINT"),
|
||
application_key=os.getenv("OVH_APP_KEY"),
|
||
application_secret=os.getenv("OVH_APP_SECRET"),
|
||
consumer_key=os.getenv("OVH_CONSUMER_KEY"),
|
||
)
|
||
|
||
services = client.get('/sms/')
|
||
if not services:
|
||
print("❌ Aucun service SMS OVH trouvé", flush=True)
|
||
return
|
||
|
||
service_name = services[0]
|
||
numero_dest = os.getenv("SMS_RECEIVER")
|
||
sender = os.getenv("OVH_SMS_SENDER")
|
||
|
||
if numero_dest.startswith('+'):
|
||
numero_dest = '00' + numero_dest[1:]
|
||
|
||
if not numero_dest or not numero_dest.isdigit():
|
||
print(f"❌ Numéro de téléphone invalide ou manquant : '{numero_dest}'", flush=True)
|
||
return
|
||
message = normaliser_sms(message, prefix=lieu)
|
||
payload = {
|
||
"sender": sender,
|
||
"receivers": [numero_dest],
|
||
"message": message, # Pas d'encodage ni de nettoyage ici
|
||
"priority": "high",
|
||
"noStopClause": False
|
||
|
||
}
|
||
|
||
print("📤 Requête envoyée à OVH :")
|
||
print(payload)
|
||
|
||
result = client.post(f'/sms/{service_name}/jobs', **payload)
|
||
|
||
print(f"📱 SMS envoyé à {numero_dest} pour {lieu}. Job ID : {result['ids']}", flush=True)
|
||
|
||
except Exception as e:
|
||
print(f"❌ Erreur envoi SMS : {e}", flush=True)
|
||
|
||
if __name__ == "__main__":
|
||
envoyer_sms("Test SMS OVH", lieu="utils_sms")
|