397 lines
14 KiB
Python
397 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
IMFestival SMS Worker
|
|
---------------------
|
|
Runs as a Flask HTTP service on port 5000.
|
|
|
|
POST /send { "job_id": N }
|
|
Sends all messages for an existing send_jobs row and writes delivery_log entries.
|
|
Called by the PHP API immediately after a send_job is created.
|
|
|
|
GET /health
|
|
Returns {"status": "ok"} — used by Docker health checks.
|
|
|
|
Background scheduler thread
|
|
Wakes every 60 s, resolves due scheduled_rules, creates send_job rows, then
|
|
calls the same send logic used by /send.
|
|
"""
|
|
|
|
import os
|
|
import time
|
|
import logging
|
|
from datetime import datetime, timedelta
|
|
from threading import Thread
|
|
|
|
import requests as http_requests
|
|
import mysql.connector
|
|
from flask import Flask, request, jsonify
|
|
|
|
logging.basicConfig(
|
|
level=logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(message)s",
|
|
)
|
|
log = logging.getLogger(__name__)
|
|
|
|
app = Flask(__name__)
|
|
|
|
CLICKATELL_URL = "https://platform.clickatell.com/messages"
|
|
|
|
DB_CONFIG = {
|
|
"host": os.environ.get("MYSQL_HOST", "db"),
|
|
"user": os.environ.get("MYSQL_USER", "imf_user"),
|
|
"password": os.environ.get("MYSQL_PASSWORD", "password"),
|
|
"database": os.environ.get("MYSQL_DATABASE", "imf_sms"),
|
|
}
|
|
|
|
|
|
# -- Database helpers ---------------------------------------------------------
|
|
|
|
def get_db():
|
|
return mysql.connector.connect(**DB_CONFIG)
|
|
|
|
|
|
def get_api_key(cursor) -> str | None:
|
|
cursor.execute(
|
|
"SELECT setting_value FROM `system` WHERE setting_name = 'clickatell_api_key'"
|
|
)
|
|
row = cursor.fetchone()
|
|
return row["setting_value"] if row and row["setting_value"] else None
|
|
|
|
|
|
# -- SMS character handling --------------------------------------------------
|
|
|
|
# Characters that ARE in the GSM-7 basic + extended character set.
|
|
# Anything outside this set forces UCS-2 (Unicode) encoding → 70 chars/part.
|
|
_GSM7 = set(
|
|
"@£$¥èéùìòÇ\nØø\rÅåΔ_ΦΓΛΩΠΨΣΘΞ\x1b"
|
|
"ÆæßÉ !\"#¤%&'()*+,-./0123456789:;<=>?"
|
|
"¡ABCDEFGHIJKLMNOPQRSTUVWXYZÄÖÑܧ"
|
|
"¿abcdefghijklmnopqrstuvwxyzäöñüà"
|
|
# Extended (each costs 2 GSM-7 chars)
|
|
"€[]{}\\^|~"
|
|
)
|
|
|
|
_NORMALISE_MAP = str.maketrans({
|
|
"\u2018": "'", # left single quotation mark
|
|
"\u2019": "'", # right single quotation mark / apostrophe
|
|
"\u201c": '"', # left double quotation mark
|
|
"\u201d": '"', # right double quotation mark
|
|
"\u2013": "-", # en dash
|
|
"\u2014": "-", # em dash
|
|
"\u2026": "...", # horizontal ellipsis
|
|
"\u00a0": " ", # non-breaking space
|
|
"\u2022": "*", # bullet
|
|
})
|
|
|
|
def normalise_gsm7(text: str) -> str:
|
|
"""Replace common non-GSM-7 characters with safe ASCII equivalents."""
|
|
return text.translate(_NORMALISE_MAP)
|
|
|
|
def sms_length(text: str) -> tuple[int, bool]:
|
|
"""
|
|
Returns (effective_length, is_unicode).
|
|
Extended GSM-7 characters (€[]{}\\^|~) count as 2 chars each.
|
|
If any character is outside GSM-7, the whole message is Unicode (UCS-2).
|
|
"""
|
|
if any(c not in _GSM7 for c in text):
|
|
return len(text), True
|
|
extended = set("€[]{}\\^|~")
|
|
return sum(2 if c in extended else 1 for c in text), False
|
|
|
|
|
|
# -- Clickatell dispatch ------------------------------------------------------
|
|
|
|
def clickatell_send(api_key: str, mobile: str, message: str) -> tuple[bool, str]:
|
|
headers = {
|
|
"Content-Type": "application/json",
|
|
"Accept": "application/json",
|
|
"Authorization": api_key,
|
|
}
|
|
try:
|
|
r = http_requests.post(
|
|
CLICKATELL_URL,
|
|
json={"content": message, "to": [mobile]},
|
|
headers=headers,
|
|
timeout=15,
|
|
)
|
|
return r.status_code == 202, r.text
|
|
except Exception as exc:
|
|
return False, str(exc)
|
|
|
|
|
|
# -- Core send logic ----------------------------------------------------------
|
|
|
|
def execute_job(job_id: int) -> dict:
|
|
"""
|
|
Send messages for the given send_jobs row, write delivery_log entries, and
|
|
update the job status. Returns {"sent": N, "failed": N} or {"error": "..."}.
|
|
"""
|
|
conn = get_db()
|
|
cursor = conn.cursor(dictionary=True)
|
|
try:
|
|
cursor.execute(
|
|
"SELECT sj.id, sj.recipient_scope, sj.workshop_session_id, "
|
|
" COALESCE(sj.custom_body, t.body) AS message_body "
|
|
"FROM send_jobs sj "
|
|
"LEFT JOIN sms_templates t ON t.id = sj.template_id "
|
|
"WHERE sj.id = %s",
|
|
(job_id,),
|
|
)
|
|
job = cursor.fetchone()
|
|
if not job:
|
|
return {"error": f"Job {job_id} not found"}
|
|
|
|
api_key = get_api_key(cursor)
|
|
if not api_key:
|
|
cursor.execute(
|
|
"UPDATE send_jobs SET status='failed' WHERE id = %s", (job_id,)
|
|
)
|
|
conn.commit()
|
|
return {"error": "Clickatell API key not configured"}
|
|
|
|
cursor.execute(
|
|
"SELECT setting_value FROM `system` WHERE setting_name = 'sandbox_mode'"
|
|
)
|
|
sandbox_row = cursor.fetchone()
|
|
is_sandbox = 1 if sandbox_row and sandbox_row["setting_value"] == "1" else 0
|
|
if is_sandbox:
|
|
log.info("Job %d: sandbox mode ON — sends will be tagged as sandbox", job_id)
|
|
|
|
if job["recipient_scope"] == "all":
|
|
cursor.execute(
|
|
"SELECT a.id, a.first_name, a.mobile_number, "
|
|
" (SELECT ws2.workshop_name FROM registrations r2 "
|
|
" JOIN workshop_sessions ws2 ON ws2.id = r2.workshop_session_id "
|
|
" WHERE r2.attendee_id = a.id ORDER BY ws2.workshop_time LIMIT 1) AS workshop_name, "
|
|
" (SELECT ws2.workshop_time FROM registrations r2 "
|
|
" JOIN workshop_sessions ws2 ON ws2.id = r2.workshop_session_id "
|
|
" WHERE r2.attendee_id = a.id ORDER BY ws2.workshop_time LIMIT 1) AS workshop_time "
|
|
"FROM attendees a"
|
|
)
|
|
else:
|
|
cursor.execute(
|
|
"SELECT a.id, a.first_name, a.mobile_number, "
|
|
" ws.workshop_name, ws.workshop_time "
|
|
"FROM attendees a "
|
|
"JOIN registrations r ON r.attendee_id = a.id "
|
|
"JOIN workshop_sessions ws ON ws.id = r.workshop_session_id "
|
|
"WHERE r.workshop_session_id = %s",
|
|
(job["workshop_session_id"],),
|
|
)
|
|
recipients = cursor.fetchall()
|
|
|
|
if not recipients:
|
|
cursor.execute(
|
|
"UPDATE send_jobs SET status='completed', sent_at=NOW() WHERE id = %s",
|
|
(job_id,),
|
|
)
|
|
conn.commit()
|
|
return {"sent": 0, "failed": 0}
|
|
|
|
cursor.execute(
|
|
"UPDATE send_jobs SET status='sending', sent_at=NOW() WHERE id = %s",
|
|
(job_id,),
|
|
)
|
|
conn.commit()
|
|
|
|
sent = failed = 0
|
|
message = job["message_body"] or ""
|
|
|
|
for recipient in recipients:
|
|
ws_name = recipient.get("workshop_name") or ""
|
|
ws_time_raw = recipient.get("workshop_time")
|
|
if ws_time_raw and hasattr(ws_time_raw, "strftime"):
|
|
ws_time = ws_time_raw.strftime("%H:%M")
|
|
elif ws_time_raw:
|
|
ws_time = str(ws_time_raw)[11:16] # extract HH:MM from datetime string
|
|
else:
|
|
ws_time = ""
|
|
msg = (
|
|
message
|
|
.replace("{name}", recipient["first_name"] or "")
|
|
.replace("{first name}", recipient["first_name"] or "")
|
|
.replace("{first_name}", recipient["first_name"] or "")
|
|
.replace("{workshop}", ws_name)
|
|
.replace("{workshop_time}", ws_time)
|
|
)
|
|
msg = normalise_gsm7(msg)
|
|
length, is_unicode = sms_length(msg)
|
|
limit = 70 if is_unicode else 160
|
|
if length > limit:
|
|
log.warning(
|
|
"Message to %s is %d chars (%s encoding, limit %d) — truncating",
|
|
recipient["mobile_number"], length,
|
|
"Unicode" if is_unicode else "GSM-7", limit,
|
|
)
|
|
msg = msg[:limit]
|
|
success, response_text = clickatell_send(api_key, recipient["mobile_number"], msg)
|
|
status = "sent" if success else "failed"
|
|
error = None if success else response_text[:500]
|
|
|
|
cursor.execute(
|
|
"INSERT INTO delivery_log "
|
|
" (send_job_id, attendee_id, mobile_number, status, sent_at, error_message, is_sandbox) "
|
|
"VALUES (%s, %s, %s, %s, NOW(), %s, %s)",
|
|
(job_id, recipient["id"], recipient["mobile_number"], status, error, is_sandbox),
|
|
)
|
|
conn.commit()
|
|
|
|
if success:
|
|
sent += 1
|
|
log.info("Sent -> %s", recipient["mobile_number"])
|
|
else:
|
|
failed += 1
|
|
log.warning("Failed -> %s : %s", recipient["mobile_number"], response_text[:120])
|
|
|
|
final_status = "completed" if sent > 0 else "failed"
|
|
cursor.execute(
|
|
"UPDATE send_jobs SET status=%s, sent_at=NOW() WHERE id = %s",
|
|
(final_status, job_id),
|
|
)
|
|
conn.commit()
|
|
log.info("Job %d complete -- sent=%d failed=%d", job_id, sent, failed)
|
|
return {"sent": sent, "failed": failed}
|
|
|
|
except Exception as exc:
|
|
log.exception("execute_job error for job %d", job_id)
|
|
try:
|
|
cursor.execute(
|
|
"UPDATE send_jobs SET status='failed' WHERE id = %s", (job_id,)
|
|
)
|
|
conn.commit()
|
|
except Exception:
|
|
pass
|
|
return {"error": str(exc)}
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
|
|
# -- HTTP endpoints -----------------------------------------------------------
|
|
|
|
@app.route("/send", methods=["POST"])
|
|
def send_endpoint():
|
|
data = request.get_json(force=True) or {}
|
|
job_id = data.get("job_id")
|
|
if not job_id:
|
|
return jsonify({"error": "job_id required"}), 400
|
|
|
|
result = execute_job(int(job_id))
|
|
if "error" in result:
|
|
return jsonify(result), 500
|
|
return jsonify(result)
|
|
|
|
|
|
@app.route("/health", methods=["GET"])
|
|
def health():
|
|
return jsonify({"status": "ok"})
|
|
|
|
|
|
# -- Scheduler ----------------------------------------------------------------
|
|
|
|
def run_scheduler():
|
|
"""Check for due scheduled_rules and fire any that have not run yet."""
|
|
conn = get_db()
|
|
cursor = conn.cursor(dictionary=True)
|
|
try:
|
|
now = datetime.now()
|
|
|
|
cursor.execute(
|
|
"SELECT id, template_id, mode, offset_value, offset_unit, specific_datetime "
|
|
"FROM scheduled_rules"
|
|
)
|
|
rules = cursor.fetchall()
|
|
|
|
for rule in rules:
|
|
|
|
if rule["mode"] == "specific":
|
|
fire_time = rule["specific_datetime"]
|
|
if isinstance(fire_time, str):
|
|
fire_time = datetime.fromisoformat(fire_time)
|
|
if now < fire_time:
|
|
log.info("Rule %d (specific): not yet due — fires at %s", rule["id"], fire_time)
|
|
continue
|
|
|
|
# Already executed?
|
|
cursor.execute(
|
|
"SELECT id FROM send_jobs "
|
|
"WHERE scheduled_rule_id = %s AND workshop_session_id IS NULL LIMIT 1",
|
|
(rule["id"],),
|
|
)
|
|
if cursor.fetchone():
|
|
continue
|
|
|
|
cursor.execute(
|
|
"INSERT INTO send_jobs "
|
|
" (template_id, recipient_scope, scheduled_rule_id, status) "
|
|
"VALUES (%s, 'all', %s, 'pending')",
|
|
(rule["template_id"], rule["id"]),
|
|
)
|
|
job_id = cursor.lastrowid
|
|
conn.commit()
|
|
log.info("Firing specific rule %d -> job %d", rule["id"], job_id)
|
|
execute_job(job_id)
|
|
|
|
elif rule["mode"] == "relative":
|
|
unit_map = {"minutes": "minutes", "hours": "hours", "days": "days"}
|
|
offset = timedelta(
|
|
**{unit_map.get(rule["offset_unit"], "minutes"): rule["offset_value"]}
|
|
)
|
|
# Apply rule to every workshop session individually
|
|
cursor.execute("SELECT id, workshop_time FROM workshop_sessions")
|
|
for session in cursor.fetchall():
|
|
ws_time = session["workshop_time"]
|
|
if isinstance(ws_time, str):
|
|
ws_time = datetime.fromisoformat(ws_time)
|
|
fire_time = ws_time - offset
|
|
if now < fire_time:
|
|
log.debug("Rule %d session %d: not yet due — fires at %s", rule["id"], session["id"], fire_time)
|
|
continue
|
|
|
|
# Already executed for this (rule, session) pair?
|
|
cursor.execute(
|
|
"SELECT id FROM send_jobs "
|
|
"WHERE scheduled_rule_id = %s AND workshop_session_id = %s LIMIT 1",
|
|
(rule["id"], session["id"]),
|
|
)
|
|
if cursor.fetchone():
|
|
continue
|
|
|
|
cursor.execute(
|
|
"INSERT INTO send_jobs "
|
|
" (template_id, recipient_scope, workshop_session_id, "
|
|
" scheduled_rule_id, status) "
|
|
"VALUES (%s, 'workshop', %s, %s, 'pending')",
|
|
(rule["template_id"], session["id"], rule["id"]),
|
|
)
|
|
job_id = cursor.lastrowid
|
|
conn.commit()
|
|
log.info(
|
|
"Firing relative rule %d for session %d -> job %d",
|
|
rule["id"], session["id"], job_id,
|
|
)
|
|
execute_job(job_id)
|
|
|
|
except Exception as exc:
|
|
log.exception("Scheduler error: %s", exc)
|
|
finally:
|
|
cursor.close()
|
|
conn.close()
|
|
|
|
|
|
def scheduler_loop():
|
|
log.info("Scheduler started -- polling every 60 s")
|
|
# Brief startup delay to let MariaDB finish initialising
|
|
time.sleep(5)
|
|
while True:
|
|
log.info("Scheduler tick")
|
|
run_scheduler()
|
|
time.sleep(60)
|
|
|
|
|
|
# -- Entry point --------------------------------------------------------------
|
|
|
|
if __name__ == "__main__":
|
|
Thread(target=scheduler_loop, daemon=True).start()
|
|
app.run(host="0.0.0.0", port=5000, debug=False)
|