Cette révision appartient à :
+256
-299
@@ -1,300 +1,257 @@
|
|||||||
import os
|
import os
|
||||||
import time
|
import time
|
||||||
import json
|
import json
|
||||||
import re
|
import re
|
||||||
import socket
|
import socket
|
||||||
import requests
|
import requests
|
||||||
import smtplib
|
import smtplib
|
||||||
from email.message import EmailMessage
|
from email.message import EmailMessage
|
||||||
from imapclient import IMAPClient
|
from imapclient import IMAPClient
|
||||||
from requests.auth import HTTPBasicAuth
|
from requests.auth import HTTPBasicAuth
|
||||||
from mail2sms.text_utils import html_to_sms_text
|
from mail2sms.text_utils import html_to_sms_text
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# CONFIGURATION
|
# CONFIGURATION
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
IMAP_SERVER = os.getenv("IMAP_SERVER")
|
IMAP_SERVER = os.getenv("IMAP_SERVER")
|
||||||
IMAP_PORT = int(os.getenv("IMAP_PORT", 993))
|
IMAP_PORT = int(os.getenv("IMAP_PORT", 993))
|
||||||
EMAIL_ACCOUNT = os.getenv("EMAIL_ACCOUNT")
|
EMAIL_ACCOUNT = os.getenv("EMAIL_ACCOUNT")
|
||||||
EMAIL_PASSWORD = os.getenv("EMAIL_PASSWORD")
|
EMAIL_PASSWORD = os.getenv("EMAIL_PASSWORD")
|
||||||
|
|
||||||
SMTP_SERVER = os.getenv("SMTP_SERVER")
|
SMTP_SERVER = os.getenv("SMTP_SERVER")
|
||||||
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
|
SMTP_PORT = int(os.getenv("SMTP_PORT", 587))
|
||||||
|
|
||||||
API_URL = os.getenv("API_URL")
|
API_URL = os.getenv("API_URL")
|
||||||
|
|
||||||
API_USERNAME = os.getenv("API_USERNAME")
|
API_USERNAME = os.getenv("API_USERNAME")
|
||||||
API_PASSWORD = os.getenv("API_PASSWORD")
|
API_PASSWORD = os.getenv("API_PASSWORD")
|
||||||
|
|
||||||
DEFAULT_DEVICE_ID = os.getenv("DEFAULT_DEVICE_ID")
|
DEFAULT_DEVICE_ID = os.getenv("DEFAULT_DEVICE_ID")
|
||||||
DEFAULT_SIM_NUMBER = os.getenv("DEFAULT_SIM_NUMBER")
|
DEFAULT_SIM_NUMBER = os.getenv("DEFAULT_SIM_NUMBER")
|
||||||
|
|
||||||
IDLE_TIMEOUT = 60 * 25 # 25 minutes (safe before typical 30 min timeout)
|
IDLE_TIMEOUT = 60 * 25 # 25 minutes (safe before typical 30 min timeout)
|
||||||
TEST_MODE = os.getenv("TEST_MODE", "false").lower() == "true"
|
TEST_MODE = os.getenv("TEST_MODE", "false").lower() == "true"
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# LOGGING
|
# LOGGING
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
def log(message):
|
def log(message):
|
||||||
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}", flush=True)
|
print(f"[{time.strftime('%Y-%m-%d %H:%M:%S')}] {message}", flush=True)
|
||||||
|
|
||||||
|
|
||||||
|
# ============================================================
|
||||||
def html_to_sms_text(raw_html):
|
# SUBJECT PARAMETER EXTRACTION
|
||||||
"""
|
# ============================================================
|
||||||
Convert HTML content to clean SMS-compatible plain text.
|
|
||||||
- Removes HTML tags
|
def extract_parameters(subject):
|
||||||
- Decodes HTML entities
|
"""
|
||||||
- Normalizes unicode
|
Extract parameters from email subject.
|
||||||
- Removes unsupported SMS characters
|
Returns: to, device_id, sim_number, api_login, api_password
|
||||||
"""
|
"""
|
||||||
|
if not subject:
|
||||||
if not raw_html:
|
return None, None, None, None, None
|
||||||
return ""
|
|
||||||
|
# Regex case-insensitive
|
||||||
# Parse HTML
|
def get_param(name):
|
||||||
soup = BeautifulSoup(raw_html, "html.parser")
|
match = re.search(rf"{name}\s*=\s*([^; ]+)", subject, re.IGNORECASE)
|
||||||
|
return match.group(1) if match else None
|
||||||
# Get text only
|
|
||||||
text = soup.get_text(separator=" ")
|
to = get_param("to")
|
||||||
|
device_id = get_param("deviceId")
|
||||||
# Decode HTML entities ( & etc.)
|
sim_number = get_param("simNumber")
|
||||||
text = html.unescape(text)
|
api_login = get_param("login")
|
||||||
|
api_password = get_param("password")
|
||||||
# Normalize unicode (remove weird accents/combined chars)
|
|
||||||
text = unicodedata.normalize("NFKD", text)
|
return to, device_id, sim_number, api_login, api_password
|
||||||
|
|
||||||
# Remove non GSM basic characters
|
# ============================================================
|
||||||
gsm_basic = (
|
# SMS SEND
|
||||||
"@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ"
|
# ============================================================
|
||||||
+ ' !"#¤%&\'()*+,-./'
|
|
||||||
+ "0123456789:;<=>?"
|
def send_sms(payload, username, password):
|
||||||
+ "ABCDEFGHIJKLMNOPQRSTUVWXYZ"
|
"""
|
||||||
+ "ÄÖÑܧ¿"
|
Send SMS using the API.
|
||||||
+ "abcdefghijklmnopqrstuvwxyz"
|
In TEST_MODE, it returns a fake response.
|
||||||
+ "äöñüà"
|
If username/password are provided, they override the defaults from env.
|
||||||
)
|
"""
|
||||||
|
if TEST_MODE:
|
||||||
cleaned = "".join(c for c in text if c in gsm_basic)
|
log("TEST_MODE enabled - Fake SMS send")
|
||||||
|
class FakeResponse:
|
||||||
# Remove multiple spaces
|
status_code = 202
|
||||||
cleaned = re.sub(r"\s+", " ", cleaned).strip()
|
text = json.dumps({"id": "fake-id", "state": "Sent"})
|
||||||
|
return FakeResponse()
|
||||||
return cleaned
|
|
||||||
|
return requests.post(
|
||||||
# ============================================================
|
API_URL,
|
||||||
# SUBJECT PARAMETER EXTRACTION
|
auth=HTTPBasicAuth(username, password),
|
||||||
# ============================================================
|
headers={"Content-Type": "application/json"},
|
||||||
|
data=json.dumps(payload),
|
||||||
def extract_parameters(subject):
|
timeout=15
|
||||||
"""
|
)
|
||||||
Extract parameters from email subject.
|
# ============================================================
|
||||||
Returns: to, device_id, sim_number, api_login, api_password
|
# STATUS EMAIL
|
||||||
"""
|
# ============================================================
|
||||||
if not subject:
|
|
||||||
return None, None, None, None, None
|
def send_status_email(to_address, phone, status_code, response_text):
|
||||||
|
if TEST_MODE:
|
||||||
# Regex case-insensitive
|
log("TEST_MODE enabled - Fake email send")
|
||||||
def get_param(name):
|
return
|
||||||
match = re.search(rf"{name}\s*=\s*([^; ]+)", subject, re.IGNORECASE)
|
msg = EmailMessage()
|
||||||
return match.group(1) if match else None
|
msg["From"] = EMAIL_ACCOUNT
|
||||||
|
msg["To"] = to_address
|
||||||
to = get_param("to")
|
msg["Subject"] = f"SMS Status for {phone}"
|
||||||
device_id = get_param("deviceId")
|
msg.set_content(
|
||||||
sim_number = get_param("simNumber")
|
f"Destination: {phone}\n"
|
||||||
api_login = get_param("login")
|
f"HTTP Code: {status_code}\n\n"
|
||||||
api_password = get_param("password")
|
f"API Response:\n{response_text}"
|
||||||
|
)
|
||||||
return to, device_id, sim_number, api_login, api_password
|
try:
|
||||||
|
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) as server:
|
||||||
# ============================================================
|
server.starttls()
|
||||||
# SMS SEND
|
server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
|
||||||
# ============================================================
|
server.send_message(msg)
|
||||||
|
log(f"Status email sent to {to_address}")
|
||||||
def send_sms(payload, username, password):
|
except Exception as e:
|
||||||
"""
|
log(f"SMTP error: {e}")
|
||||||
Send SMS using the API.
|
|
||||||
In TEST_MODE, it returns a fake response.
|
# ============================================================
|
||||||
If username/password are provided, they override the defaults from env.
|
# PROCESS EMAILS READING
|
||||||
"""
|
# ============================================================
|
||||||
if TEST_MODE:
|
|
||||||
log("TEST_MODE enabled - Fake SMS send")
|
def process_unseen(server):
|
||||||
class FakeResponse:
|
messages = server.search(['UNSEEN'])
|
||||||
status_code = 202
|
if not messages:
|
||||||
text = json.dumps({"id": "fake-id", "state": "Sent"})
|
return
|
||||||
return FakeResponse()
|
log(f"{len(messages)} new message(s) detected")
|
||||||
|
fetched = server.fetch(messages, ['RFC822'])
|
||||||
return requests.post(
|
for uid, message_data in fetched.items():
|
||||||
API_URL,
|
try:
|
||||||
auth=HTTPBasicAuth(username, password),
|
import email
|
||||||
headers={"Content-Type": "application/json"},
|
msg = email.message_from_bytes(message_data[b'RFC822'])
|
||||||
data=json.dumps(payload),
|
sender = email.utils.parseaddr(msg["From"])[1]
|
||||||
timeout=15
|
subject = msg["Subject"]
|
||||||
)
|
phone_number, subject_device_id, subject_sim_number, api_login, api_password = extract_parameters(subject)
|
||||||
# ============================================================
|
|
||||||
# STATUS EMAIL
|
if not phone_number:
|
||||||
# ============================================================
|
log("No phone number found in subject")
|
||||||
|
server.add_flags(uid, ['\\Seen'])
|
||||||
def send_status_email(to_address, phone, status_code, response_text):
|
continue
|
||||||
if TEST_MODE:
|
device_id = subject_device_id or DEFAULT_DEVICE_ID
|
||||||
log("TEST_MODE enabled - Fake email send")
|
sim_number = subject_sim_number or DEFAULT_SIM_NUMBER
|
||||||
return
|
username = api_login or API_USERNAME
|
||||||
msg = EmailMessage()
|
if not username:
|
||||||
msg["From"] = EMAIL_ACCOUNT
|
log("No API Username found")
|
||||||
msg["To"] = to_address
|
server.add_flags(uid, ['\\Seen'])
|
||||||
msg["Subject"] = f"SMS Status for {phone}"
|
continue
|
||||||
msg.set_content(
|
password = api_password or API_PASSWORD
|
||||||
f"Destination: {phone}\n"
|
if not password:
|
||||||
f"HTTP Code: {status_code}\n\n"
|
log("No API Password found")
|
||||||
f"API Response:\n{response_text}"
|
server.add_flags(uid, ['\\Seen'])
|
||||||
)
|
continue
|
||||||
try:
|
|
||||||
with smtplib.SMTP(SMTP_SERVER, SMTP_PORT, timeout=10) as server:
|
sms_text = ""
|
||||||
server.starttls()
|
|
||||||
server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
|
if msg.is_multipart():
|
||||||
server.send_message(msg)
|
for part in msg.walk():
|
||||||
log(f"Status email sent to {to_address}")
|
content_type = part.get_content_type()
|
||||||
except Exception as e:
|
payload = part.get_payload(decode=True)
|
||||||
log(f"SMTP error: {e}")
|
|
||||||
|
if not payload:
|
||||||
# ============================================================
|
continue
|
||||||
# PROCESS EMAILS READING
|
|
||||||
# ============================================================
|
decoded = payload.decode(errors="ignore")
|
||||||
|
|
||||||
def process_unseen(server):
|
if content_type == "text/plain":
|
||||||
messages = server.search(['UNSEEN'])
|
sms_text = decoded.strip()
|
||||||
if not messages:
|
break
|
||||||
return
|
|
||||||
log(f"{len(messages)} new message(s) detected")
|
elif content_type == "text/html" and not sms_text:
|
||||||
fetched = server.fetch(messages, ['RFC822'])
|
sms_text = html_to_sms_text(decoded)
|
||||||
for uid, message_data in fetched.items():
|
|
||||||
try:
|
else:
|
||||||
import email
|
payload = msg.get_payload(decode=True)
|
||||||
msg = email.message_from_bytes(message_data[b'RFC822'])
|
if payload:
|
||||||
sender = email.utils.parseaddr(msg["From"])[1]
|
decoded = payload.decode(errors="ignore")
|
||||||
subject = msg["Subject"]
|
if msg.get_content_type() == "text/html":
|
||||||
phone_number, subject_device_id, subject_sim_number, api_login, api_password = extract_parameters(subject)
|
sms_text = html_to_sms_text(decoded)
|
||||||
|
else:
|
||||||
if not phone_number:
|
sms_text = decoded.strip()
|
||||||
log("No phone number found in subject")
|
|
||||||
server.add_flags(uid, ['\\Seen'])
|
payload = {
|
||||||
continue
|
"textMessage": {"text": sms_text},
|
||||||
device_id = subject_device_id or DEFAULT_DEVICE_ID
|
"phoneNumbers": [phone_number]
|
||||||
sim_number = subject_sim_number or DEFAULT_SIM_NUMBER
|
}
|
||||||
username = api_login or API_USERNAME
|
if device_id:
|
||||||
if not username:
|
payload["deviceId"] = device_id
|
||||||
log("No API Username found")
|
if sim_number:
|
||||||
server.add_flags(uid, ['\\Seen'])
|
payload["simNumber"] = int(sim_number)
|
||||||
continue
|
log(f"Sending SMS to {phone_number} using deviceId={device_id} simNumber={sim_number}")
|
||||||
password = api_password or API_PASSWORD
|
response = send_sms(payload, username, password)
|
||||||
if not password:
|
log(f"HTTP {response.status_code}")
|
||||||
log("No API Password found")
|
send_status_email(sender, phone_number, response.status_code, response.text)
|
||||||
server.add_flags(uid, ['\\Seen'])
|
if 200 <= response.status_code <= 204:
|
||||||
continue
|
server.delete_messages(uid)
|
||||||
|
log("Email deleted (success)")
|
||||||
sms_text = ""
|
else:
|
||||||
|
server.add_flags(uid, ['\\Seen'])
|
||||||
if msg.is_multipart():
|
log("Email kept (non-success response)")
|
||||||
for part in msg.walk():
|
except Exception as e:
|
||||||
content_type = part.get_content_type()
|
log(f"Processing error: {e}")
|
||||||
payload = part.get_payload(decode=True)
|
server.add_flags(uid, ['\\Seen'])
|
||||||
|
server.expunge()
|
||||||
if not payload:
|
|
||||||
continue
|
# ============================================================
|
||||||
|
# HANDLE IMAP CONNECTION
|
||||||
decoded = payload.decode(errors="ignore")
|
# ============================================================
|
||||||
|
|
||||||
if content_type == "text/plain":
|
def connect():
|
||||||
sms_text = decoded.strip()
|
while True:
|
||||||
break
|
try:
|
||||||
|
log("Connecting to IMAP server...")
|
||||||
elif content_type == "text/html" and not sms_text:
|
server = IMAPClient(IMAP_SERVER, port=IMAP_PORT, ssl=True, timeout=20)
|
||||||
sms_text = html_to_sms_text(decoded)
|
server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
|
||||||
|
server.select_folder("INBOX")
|
||||||
else:
|
log("IMAP connected successfully")
|
||||||
payload = msg.get_payload(decode=True)
|
return server
|
||||||
if payload:
|
except (socket.error, Exception) as e:
|
||||||
decoded = payload.decode(errors="ignore")
|
log(f"IMAP connection failed: {e}")
|
||||||
if msg.get_content_type() == "text/html":
|
time.sleep(10)
|
||||||
sms_text = html_to_sms_text(decoded)
|
|
||||||
else:
|
# ============================================================
|
||||||
sms_text = decoded.strip()
|
# IDLE LOOP
|
||||||
|
# ============================================================
|
||||||
payload = {
|
|
||||||
"textMessage": {"text": sms_text},
|
def idle_loop():
|
||||||
"phoneNumbers": [phone_number]
|
server = connect()
|
||||||
}
|
while True:
|
||||||
if device_id:
|
try:
|
||||||
payload["deviceId"] = device_id
|
log("Entering IDLE mode")
|
||||||
if sim_number:
|
server.idle()
|
||||||
payload["simNumber"] = int(sim_number)
|
responses = server.idle_check(timeout=IDLE_TIMEOUT)
|
||||||
log(f"Sending SMS to {phone_number} using deviceId={device_id} simNumber={sim_number}")
|
server.idle_done()
|
||||||
response = send_sms(payload, username, password)
|
if responses:
|
||||||
log(f"HTTP {response.status_code}")
|
log("New mail event received")
|
||||||
send_status_email(sender, phone_number, response.status_code, response.text)
|
process_unseen(server)
|
||||||
if 200 <= response.status_code <= 204:
|
else:
|
||||||
server.delete_messages(uid)
|
log("Idle timeout reached (keep-alive)")
|
||||||
log("Email deleted (success)")
|
except (socket.error, Exception) as e:
|
||||||
else:
|
log(f"Connection lost: {e}")
|
||||||
server.add_flags(uid, ['\\Seen'])
|
try:
|
||||||
log("Email kept (non-success response)")
|
server.logout()
|
||||||
except Exception as e:
|
except:
|
||||||
log(f"Processing error: {e}")
|
pass
|
||||||
server.add_flags(uid, ['\\Seen'])
|
time.sleep(5)
|
||||||
server.expunge()
|
server = connect()
|
||||||
|
|
||||||
# ============================================================
|
# ============================================================
|
||||||
# HANDLE IMAP CONNECTION
|
# ENTRY POINT
|
||||||
# ============================================================
|
# ============================================================
|
||||||
|
|
||||||
def connect():
|
if __name__ == "__main__":
|
||||||
while True:
|
log("Mail2SMS service started")
|
||||||
try:
|
if TEST_MODE:
|
||||||
log("Connecting to IMAP server...")
|
log("Running in TEST_MODE - no IMAP connection")
|
||||||
server = IMAPClient(IMAP_SERVER, port=IMAP_PORT, ssl=True, timeout=20)
|
else:
|
||||||
server.login(EMAIL_ACCOUNT, EMAIL_PASSWORD)
|
|
||||||
server.select_folder("INBOX")
|
|
||||||
log("IMAP connected successfully")
|
|
||||||
return server
|
|
||||||
except (socket.error, Exception) as e:
|
|
||||||
log(f"IMAP connection failed: {e}")
|
|
||||||
time.sleep(10)
|
|
||||||
|
|
||||||
# ============================================================
|
|
||||||
# IDLE LOOP
|
|
||||||
# ============================================================
|
|
||||||
|
|
||||||
def idle_loop():
|
|
||||||
server = connect()
|
|
||||||
while True:
|
|
||||||
try:
|
|
||||||
log("Entering IDLE mode")
|
|
||||||
server.idle()
|
|
||||||
responses = server.idle_check(timeout=IDLE_TIMEOUT)
|
|
||||||
server.idle_done()
|
|
||||||
if responses:
|
|
||||||
log("New mail event received")
|
|
||||||
process_unseen(server)
|
|
||||||
else:
|
|
||||||
log("Idle timeout reached (keep-alive)")
|
|
||||||
except (socket.error, Exception) as e:
|
|
||||||
log(f"Connection lost: {e}")
|
|
||||||
try:
|
|
||||||
server.logout()
|
|
||||||
except:
|
|
||||||
pass
|
|
||||||
time.sleep(5)
|
|
||||||
server = connect()
|
|
||||||
|
|
||||||
# ============================================================
|
|
||||||
# ENTRY POINT
|
|
||||||
# ============================================================
|
|
||||||
|
|
||||||
if __name__ == "__main__":
|
|
||||||
log("Mail2SMS service started")
|
|
||||||
if TEST_MODE:
|
|
||||||
log("Running in TEST_MODE - no IMAP connection")
|
|
||||||
else:
|
|
||||||
idle_loop()
|
idle_loop()
|
||||||
Référencer dans un nouveau ticket
Bloquer un utilisateur