<<<<<<< HEAD

SMS Campaign Producer / Consumer

A Python async service that reads pending SMS rows from file_campaign_data, publishes them to RabbitMQ (one queue per provider), and dispatches them through the correct gateway — with end-to-end deduplication and safe row deletion.

sms_campaign

SMS Campaign done in python

SMS Campaign Python Processor

Overview

This project is a high-performance asynchronous SMS processing service built in Python that integrates with a Laravel SMS application. It offloads SMS processing from Laravel by using RabbitMQ, allowing campaigns to be processed concurrently while Laravel remains responsive.

The processor consists of three services:

• Producer – Polls pending SMS records from the Laravel MySQL database and publishes them to RabbitMQ.
• Consumer – Consumes SMS messages from RabbitMQ, sends them to the appropriate SMS provider (Safaricom or Airtel), updates campaign statistics, and stores delivery reports.
• DLR Worker – Processes Delivery Receipt (DLR) callbacks asynchronously and updates message statuses.

d71884cbe46dd3eae3e8f1c4b5bcf769936ac822

---

Architecture

<<<<<<< HEAD

MySQL ──► Producer ──► RabbitMQ exchange (sms.campaigns)
                              │
                  ┌───────────┴────────────┐
              sms.safaricom           sms.airtel
                  │                        │
            Consumer(s) ──► Safaricom API / Airtel API

Deduplication layers

Layer	Redis key	When set	Guards against
Producer	sms:produced:<id>	Before publish	Double-publish (e.g. producer crash then restart)
Consumer	sms:consumed:<id>	Before processing	Duplicate delivery from broker (QoS, redelivery)

Both use SET NX (atomic) with a configurable TTL (default 24 h).

Row deletion

Rows are deleted from file_campaign_data after the broker confirms receipt (mandatory=True + publisher confirms channel). If the broker is unavailable the row survives and will be picked up on the next poll cycle. The producer-side Redis key prevents double-publish in that scenario.

Paused campaigns

The SQL query joins campaigns and filters WHERE campaigns.status != 'paused', so paused campaigns are never touched.

Concurrent producers (scale-out)

The query uses SELECT … FOR UPDATE SKIP LOCKED, meaning multiple producer instances can run safely in parallel without processing the same rows.

---

Project structure

sms_campaign/
├── config/
│   └── settings.py          # Pydantic settings from .env
├── models/
│   ├── db_models.py          # SQLAlchemy ORM (Campaign, SendingServer, FileCampaignData)
│   └── schemas.py            # SmsMessage Pydantic model + dedup key helpers
├── producer/
│   ├── broker.py             # RabbitMQ topology declaration
│   ├── dedup.py              # Redis SET NX deduplication store
│   └── producer.py           # Poll → dedup → publish → delete
├── consumer/
│   ├── consumer.py           # Subscribe → dedup → dispatch → ack/nack
│   ├── safaricom_sender.py   # Safaricom gateway stub
│   └── airtel_sender.py      # Airtel gateway stub
├── run_producer.py           # $ python run_producer.py
├── run_consumer.py           # $ python run_consumer.py
├── docker-compose.yml
├── Dockerfile
└── requirements.txt
=======
```text
Laravel Application
        │
        │ PROCESSOR_PYTHON=true
        ▼
   MySQL Database
        │
        ▼
Python Producer
        │
        ▼
    RabbitMQ Exchange
        │
 ┌──────┴────────┐
 ▼               ▼
Safaricom Queue  Airtel Queue
        │
        ▼
 Python Consumer
        │
        ▼
SMS Provider APIs
        │
        ▼
Delivery Receipts
        │
        ▼
Python DLR Worker
        │
        ▼
Laravel Database
>>>>>>> d71884cbe46dd3eae3e8f1c4b5bcf769936ac822

---

<<<<<<< HEAD

Quick start

1. Install dependencies

=======

Laravel Configuration

This project depends on the Laravel application's .env file.

Rather than maintaining a separate Python configuration, the processor loads configuration directly from Laravel using the --env-file argument.

Enable the Python processor:

PROCESSOR_PYTHON=true

Add the following RabbitMQ configuration if it does not already exist:

RABBITMQ_URL=amqp://guest:guest@localhost:5672/
RABBITMQ_SAFARICOM_QUEUE=sms.safaricom
RABBITMQ_AIRTEL_QUEUE=sms.airtel
RABBITMQ_EXCHANGE=sms.campaigns
RABBITMQ_PREFETCH_COUNT=10

Optional producer tuning:

PRODUCER_BATCH_SIZE=100
PRODUCER_POLL_INTERVAL_SECONDS=5
DEDUP_TTL_SECONDS=86400

The Python services automatically read database credentials, RabbitMQ configuration, API credentials, and other required settings from the specified Laravel .env file.

---

Requirements

• Python 3.11+
• RabbitMQ
• Redis
• MySQL / MariaDB
• Laravel application

---

Installation

Create a virtual environment:

python -m venv .venv

Activate it.

Windows

.venv\Scripts\activate

Linux

source .venv/bin/activate

Install dependencies:

d71884cbe46dd3eae3e8f1c4b5bcf769936ac822

pip install -r requirements.txt

<<<<<<< HEAD

2. Configure

cp .env.example .env
# Edit .env with your DB / RabbitMQ / Redis credentials

3. Run with Docker Compose (easiest)

docker compose up --build
# Scale consumers:
docker compose up --build --scale consumer=3

4. Run manually

# Terminal 1 — producer
python run_producer.py

# Terminal 2 — consumer
python run_consumer.py

---

Database requirements

The sending_servers table must have a provider column with lowercase values safaricom or airtel. Example:

INSERT INTO sending_servers (id, provider, name) VALUES
  (1, 'safaricom', 'Safaricom Kenya'),
  (2, 'airtel',    'Airtel Kenya');

---

Adding a new provider

1. Add a new queue in producer/broker.py (declare_topology).
2. Create consumer/<provider>_sender.py with a send(msg) method.
3. Add a branch in consumer/consumer.py → _dispatch().
4. Update sending_servers.provider values in the DB.

---

Configuration reference

Variable	Default	Description
DB_HOST	localhost	MySQL host
DB_PORT	3306	MySQL port
DB_NAME	sms_db	Database name
DB_USER	root	Database user
DB_PASSWORD	secret	Database password
RABBITMQ_URL	amqp://guest:guest@localhost:5672/	RabbitMQ connection string
RABBITMQ_SAFARICOM_QUEUE	sms.safaricom	Safaricom queue name
RABBITMQ_AIRTEL_QUEUE	sms.airtel	Airtel queue name
RABBITMQ_EXCHANGE	sms.campaigns	Direct exchange name
RABBITMQ_PREFETCH_COUNT	10	Consumer QoS prefetch
REDIS_URL	redis://localhost:6379/0	Redis connection string
DEDUP_TTL_SECONDS	86400	Dedup key TTL (24 h)
PRODUCER_BATCH_SIZE	100	Rows fetched per poll
PRODUCER_POLL_INTERVAL_SECONDS	5	Seconds between polls
=======

---

Running the Services

Start the producer:

python run_producer.py --env-file=C:\laravel\sms\.env

Start the consumer:

python run_consumer.py --env-file=C:\laravel\sms\.env

Start the DLR worker:

python run_dlr_worker.py --env-file=C:\laravel\sms\.env

Each service loads its configuration from the Laravel .env file supplied via the --env-file parameter.

---

Processing Flow

1. Laravel inserts pending campaign records into the database.
2. The Producer polls pending records.
3. Messages are published to RabbitMQ.
4. The Consumer processes messages concurrently.
5. SMS is submitted to the appropriate provider.
6. Campaign statistics are updated.
7. Delivery reports are stored.
8. The DLR Worker processes provider callbacks asynchronously.

---

Features

• Async processing using asyncio
• RabbitMQ message queues
• Redis-based deduplication
• Automatic retries
• Dead Letter Queue (DLQ) support
• Concurrent consumers
• Provider abstraction
• Campaign progress tracking
• Delivery report processing
• Structured logging
• SQLAlchemy async database access

---

Notes

• Ensure PROCESSOR_PYTHON=true is enabled in the Laravel .env; otherwise, Laravel will continue processing campaigns using its native queue workers.
• RabbitMQ, Redis, and MySQL must be running before starting the Python services.
• All three Python services use the same Laravel database and configuration.
• The Python processor is designed to work alongside the Laravel application rather than replacing it.

---

License

This project is intended to integrate with the Laravel SMS platform and follows the licensing terms of the parent application.

d71884cbe46dd3eae3e8f1c4b5bcf769936ac822