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