khatibi/admin_panel/management/commands/sms_worker.py

"""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)