chore: initial commit
This commit is contained in:
+396
@@ -0,0 +1,396 @@
|
||||
#!/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)
|
||||
Reference in New Issue
Block a user