sms

admin_panel/management/commands/sms_worker.py

@@ -0,0 +1,179 @@
+"""worker تخلیه‌کنندهٔ صف پیامک.
+
+در یک پردازش جدا اجرا می‌شود تا کندی/خطای سامانه پیامک به برنامه اصلی ضربه نزند.
+"""
+import logging
+import signal
+import time
+from datetime import timedelta
+
+from django.core.management.base import BaseCommand
+from django.db.models import Q
+from django.utils import timezone
+
+from admin_panel.models import SMSLog, SMSProviderConfig, SMSQueue
+from integrations.sms.factory import get_sms_provider
+
+logger = logging.getLogger(__name__)
+
+
+class Command(BaseCommand):
+    help = "تخلیهٔ صف پیامک و ارسال از طریق سرویس‌دهندهٔ تنظیم‌شده."
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+        self._stop = False
+
+    def add_arguments(self, parser):
+        parser.add_argument("--once", action="store_true", help="یک دور پردازش و خروج")
+        parser.add_argument("--sleep", type=float, default=3.0, help="فاصلهٔ بین دورها (ثانیه)")
+        parser.add_argument(
+            "--batch", type=int, default=None,
+            help="تعداد پیامک در هر دور (پیش‌فرض: max_per_run از تنظیمات پنل)",
+        )
+
+    def handle(self, *args, **options):
+        signal.signal(signal.SIGTERM, self._handle_signal)
+        signal.signal(signal.SIGINT, self._handle_signal)
+        once = options["once"]
+        sleep_seconds = options["sleep"]
+        batch = options["batch"]
+
+        self._emit("worker پیامک آغاز شد.", style=self.style.SUCCESS)
+        while not self._stop:
+            # یک دور خراب (خطای دیتابیس، provider، …) نباید کل worker را از پا دربیاورد؛
+            # در غیر این‌صورت صف تا ری‌استارت دستی راکد می‌ماند.
+            try:
+                processed = self._drain_batch(batch)
+            except Exception:
+                logger.exception("خطای غیرمنتظره در یک دور از worker پیامک؛ دور بعد ادامه می‌دهد.")
+                processed = 0
+            if once:
+                break
+            if processed == 0:
+                self._sleep_interruptible(sleep_seconds)
+        self._emit("worker پیامک متوقف شد.", style=self.style.SUCCESS)
+
+    def _handle_signal(self, signum, frame):
+        self._stop = True
+
+    def _emit(self, text, *, style=None, err=False):
+        """نوشتن خروجی با تحمل کدگذاری محدود کنسول (مثلاً cp1252 در ویندوز).
+
+        چاپ، عملیات جانبی است: اگر کنسول نتواند متن را بنویسد نباید ارسال‌ها را
+        متوقف یا «ناموفق» کند. اول متن کامل، بعد نسخهٔ ASCII-شده، و در نهایت سکوت.
+        """
+        stream = self.stderr if err else self.stdout
+        style = style or (lambda value: value)
+        for payload in (text, text.encode("ascii", "replace").decode("ascii")):
+            try:
+                stream.write(style(payload))
+                return
+            except (UnicodeEncodeError, ValueError, OSError):
+                continue
+
+    def _drain_batch(self, batch):
+        config = SMSProviderConfig.load()
+        if not config.is_enabled:
+            logger.info("ارسال پیامک از تنظیمات غیرفعال است؛ صف دست‌نخورده می‌ماند.")
+            return 0
+
+        if not batch:
+            batch = config.max_per_run or 20
+        interval = config.send_interval_seconds or 0
+
+        now = timezone.now()
+        # تنها پیامک‌هایی که مدیر صراحتاً تأیید کرده ارسال می‌شوند؛ موارد «در
+        # انتظار تأیید» دست‌نخورده در صف بازبینی می‌مانند.
+        candidates = SMSQueue.objects.filter(
+            status=SMSQueue.Status.APPROVED,
+        ).filter(
+            Q(next_retry_at__isnull=True) | Q(next_retry_at__lte=now),
+        ).order_by("created_at")[:batch]
+
+        try:
+            provider = get_sms_provider(config)
+        except Exception:
+            logger.exception("ساخت provider پیامک ناموفق بود؛ این دور رد شد.")
+            return 0
+
+        processed = 0
+        for item in candidates:
+            if self._stop:
+                break
+            # claim اتمیک مثبت‌گرا (SQLite از select_for_update پشتیبانی نمی‌کند).
+            claimed = SMSQueue.objects.filter(
+                pk=item.pk, status=SMSQueue.Status.APPROVED,
+            ).update(status=SMSQueue.Status.PROCESSING)
+            if not claimed:
+                continue
+            try:
+                row = SMSQueue.objects.get(pk=item.pk)
+            except SMSQueue.DoesNotExist:
+                # رکورد بین انتخاب و claim حذف شده (مثلاً پاک‌سازی دستی پنل).
+                continue
+            # فاصلهٔ بین پیامک‌ها: پیش از ارسال مورد بعدی صبر می‌کنیم (نه بعد از آخرین).
+            if processed and interval:
+                self._sleep_interruptible(interval)
+                if self._stop:
+                    # مورد claim‌شده را به صف برگردان تا از دست نرود.
+                    SMSQueue.objects.filter(
+                        pk=row.pk, status=SMSQueue.Status.PROCESSING,
+                    ).update(status=SMSQueue.Status.APPROVED)
+                    break
+            processed += 1
+            self._send(provider, row)
+        return processed
+
+    def _sleep_interruptible(self, seconds):
+        """خواب قابل‌قطع تا سیگنال توقف سریع پاسخ داده شود."""
+        remaining = float(seconds)
+        while remaining > 0 and not self._stop:
+            step = min(0.5, remaining)
+            time.sleep(step)
+            remaining -= step
+
+    def _send(self, provider, row):
+        try:
+            result = provider.send_message(row.phone_number, row.message)
+        except Exception as exc:  # provider نباید worker را از پا دربیاورد
+            logger.exception("استثنای غیرمنتظره در ارسال پیامک #%s", row.pk)
+            result = None
+            error = str(exc)
+        else:
+            error = getattr(result, "error", "") or ""
+            result = result if getattr(result, "success", False) else None
+
+        now = timezone.now()
+        row.attempts = (row.attempts or 0) + 1
+        if result is not None:
+            row.status = SMSQueue.Status.SENT
+            row.sent_at = now
+            row.provider_id = getattr(result, "provider_id", "") or ""
+            row.provider_code = getattr(result, "provider_code", "") or ""
+            # شناسه (smsid) داریم ⇒ اهل رهگیری دلیوری؛ دستور sms_delivery پر می‌کند.
+            row.delivery_status = (
+                SMSQueue.DeliveryStatus.PENDING if row.provider_id else ""
+            )
+            row.last_error = ""
+            row.save()
+            SMSLog.objects.create(
+                queue=row, attempt_no=row.attempts, success=True,
+                provider_id=row.provider_id, provider_code=row.provider_code,
+            )
+            self._emit(f"  ارسال موفق #{row.pk} -> {row.phone_number}")
+            return
+
+        row.last_error = error[:2000]
+        if row.attempts >= (row.max_attempts or 5):
+            row.status = SMSQueue.Status.FAILED
+        else:
+            # همچنان تأییدشده می‌ماند تا تلاش خودکار بعدی ادامه یابد.
+            row.status = SMSQueue.Status.APPROVED
+            backoff = min(3600, 30 * (2 ** row.attempts))
+            row.next_retry_at = now + timedelta(seconds=backoff)
+        row.save()
+        SMSLog.objects.create(
+            queue=row, attempt_no=row.attempts, success=False, error=row.last_error,
+        )
+        self._emit(f"  ارسال ناموفق #{row.pk} (تلاش {row.attempts}): {row.last_error[:120]}", err=True)