fix mobile and names

admin_panel/management/commands/fix_customer_names.py

@@ -0,0 +1,484 @@
+"""افزودن قطعیِ پیشوند «آقای/خانم» به ابتدای نام مشتریان، بر پایهٔ فایل نام‌های ایرانی.
+
+سیاست‌ها (کاملاً محافظه‌کارانه):
+
+* **پیش‌فرض بدون نوشتن است** (dry-run). نوشتن فقط با ``--apply`` ممکن است.
+* برای اجرای واقعی، علاوه بر ``--apply``، تأیید صریح ``--confirm-production`` لازم است.
+* فقط فیلد ``Customer.name_family`` و آن هم تنها در موارد قطعی تغییر می‌کند.
+* هیچ جنسیتی ذخیره نمی‌شود؛ هیچ فیلدی به مدل مشتری اضافه نشده است.
+
+اجرا:
+
+    python manage.py fix_customer_names --dry-run
+    python manage.py fix_customer_names --apply --confirm-production
+
+ادامهٔ job متوقف‌شده (پیش‌فرض: خودکار آخرین job ناتمام با همان hash ادامه می‌یابد):
+
+    python manage.py fix_customer_names --apply --confirm-production --resume <JOB_ID>
+
+شروع عمدی job جدید (مثلاً پس از تکمیل job قبلی یا تغییر فایل):
+
+    python manage.py fix_customer_names --apply --confirm-production --new-job
+
+جزئیات ایمنی:
+
+* فایل JSON در هر اجرا فقط یک بار خوانده و parse می‌شود و هرگز نوشته نمی‌شود.
+* تغییرات نام و checkpoint هر batch در یک تراکنش اتمیک روی همان دیتابیس ثبت می‌شوند.
+* دامنهٔ job با بیشترین pk مشتریان در لحظهٔ آغاز قفل می‌شود؛ مشتریان جدیدتر وارد نمی‌شوند.
+* قفل بر پایهٔ heartbeat است و پس از crash (پس از ``LOCK_TTL``) قابل بازیابی است.
+* خطای «database is locked» با backoff نمایی + jitter و تعداد محدود retry می‌شود.
+* فرمان هیچ migration خودکاری اجرا نمی‌کند.
+"""
+import random
+import time
+from datetime import timedelta
+from pathlib import Path
+
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+from django.db import DEFAULT_DB_ALIAS, connections, transaction
+from django.db.models import Max
+from django.db.utils import OperationalError
+from django.utils import timezone
+
+from admin_panel.models import Customer, CustomerNameFixJob
+from admin_panel.name_fix import (ALGORITHM_VERSION, DatasetError,
+                                  decide_prefix, load_name_index)
+
+DEFAULT_JSON_FILENAME = "iranianNamesDataset.json"
+
+# مدت اعتبار قفل/heartbeat. batchها کوچک‌اند، پس این مقدار سخاوتمندانه است.
+LOCK_TTL = timedelta(minutes=15)
+MAX_DB_RETRIES = 4
+BACKOFF_BASE_SECONDS = 0.5
+
+# کلیدهای آماری که در ``stats`` مدل job ذخیره و بین اجراها منتقل می‌شوند.
+STATS_KEYS = (
+    "scanned", "changed", "unchanged", "single_word", "has_prefix",
+    "empty_invalid", "unknown_name", "ambiguous_gender", "batches",
+)
+
+# نگاشت «دلیل تصمیم» به «کلید آماری».
+REASON_TO_STAT = {
+    "male": "changed",
+    "female": "changed",
+    "single_word": "single_word",
+    "has_prefix": "has_prefix",
+    "empty": "empty_invalid",
+    "unknown_name": "unknown_name",
+    "ambiguous_gender": "ambiguous_gender",
+}
+
+
+class Command(BaseCommand):
+    help = (
+        "افزودن پیشوند «آقای/خانم» به نام مشتریان بر پایهٔ فایل نام‌های ایرانی؛ "
+        "پیش‌فرض dry-run است و نوشتن نیازمند --apply --confirm-production."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--json-path", default=str(Path(settings.BASE_DIR) / DEFAULT_JSON_FILENAME),
+            help=f"مسیر فایل دادهٔ نام‌ها (پیش‌فرض: {DEFAULT_JSON_FILENAME} در ریشهٔ پروژه).",
+        )
+        parser.add_argument(
+            "--batch-size", type=int, default=200,
+            help="تعداد مشتریان در هر batch (پیش‌فرض ۲۰۰).",
+        )
+        parser.add_argument(
+            "--dry-run", action="store_true",
+            help="صریحاً حالت شبیه‌سازی (پیش‌فرض همین است؛ همراه --apply مجاز نیست).",
+        )
+        parser.add_argument(
+            "--apply", action="store_true",
+            help="نوشتن واقعی در دیتابیس. بدون این گزینه فقط گزارش (dry-run) تولید می‌شود.",
+        )
+        parser.add_argument(
+            "--confirm-production", action="store_true",
+            help="تأیید صریح اپراتور برای نوشتن واقعی؛ همراه --apply الزامی است.",
+        )
+        parser.add_argument(
+            "--resume", type=int, default=None,
+            help="ادامهٔ یک job مشخص با شناسه (pk) آن.",
+        )
+        parser.add_argument(
+            "--new-job", action="store_true",
+            help="شروع عمدی یک job جدید (حتی اگر job ناتمام یا تکمیل‌شده‌ای وجود داشته باشد).",
+        )
+        parser.add_argument(
+            "--samples", type=int, default=5,
+            help="تعداد نمونهٔ «قبل/بعد» برای نمایش در گزارش (پیش‌فرض ۵).",
+        )
+
+    # ------------------------------------------------------------------ main
+    def handle(self, *args, **options):
+        alias = DEFAULT_DB_ALIAS
+        batch_size = max(1, options["batch_size"])
+        samples_wanted = max(0, options["samples"])
+        apply_changes = options["apply"]
+
+        json_path = self._resolve_json_path(options["json_path"])
+        self._announce_target(alias)
+
+        try:
+            name_index = load_name_index(json_path)
+        except DatasetError as exc:
+            raise CommandError(f"فایل داده رد شد و هیچ تغییری اعمال نشد: {exc}")
+        self._print_dataset_report(name_index)
+
+        if options["dry_run"] and apply_changes:
+            raise CommandError("--dry-run و --apply با هم ناسازگارند.")
+
+        if not apply_changes:
+            counters, samples = self._run_readonly(
+                name_index, alias, batch_size, samples_wanted)
+            self._print_summary(counters, samples, dry_run=True, job=None)
+            return
+
+        if not options["confirm_production"]:
+            raise CommandError(
+                "برای نوشتن واقعی، هم --apply و هم --confirm-production لازم است. "
+                "بدون --apply اجرا حالت dry-run است."
+            )
+        self._ensure_job_table(alias)
+
+        job, resumed = self._claim_job(
+            alias, name_index, batch_size, options["resume"], options["new_job"])
+        counters = self._counters_from(job.stats)
+        note = "ادامهٔ job" if resumed else "job جدید"
+        self.stdout.write(self.style.MIGRATE_HEADING(
+            f"job#{job.pk} {note}؛ از pk>{job.last_processed_pk} تا pk<={job.initial_max_pk} "
+            f"(batch={batch_size})"
+        ))
+
+        try:
+            counters, samples = self._apply(
+                name_index, alias, batch_size, job, counters, samples_wanted)
+        except KeyboardInterrupt:
+            self._finalize(alias, job, counters, CustomerNameFixJob.Status.PAUSED,
+                           note="توقف با Ctrl+C؛ قابل ادامه با --resume")
+            self.stdout.write(self.style.WARNING(
+                f"job#{job.pk} متوقف شد؛ برای ادامه: --apply --confirm-production --resume {job.pk}"
+            ))
+            return
+        except Exception as exc:  # noqa: BLE001 - هر خطای غیرمنتظره: توقف امن
+            self._finalize(alias, job, counters, CustomerNameFixJob.Status.FAILED,
+                           note=f"خطا: {type(exc).__name__}")
+            raise CommandError(
+                f"اجرای job#{job.pk} با خطا متوقف شد و batch جاری rollback شد: {exc}"
+            )
+
+        self._finalize(alias, job, counters, CustomerNameFixJob.Status.COMPLETED,
+                       note="تکمیل شد")
+        self._print_summary(counters, samples, dry_run=False, job=job)
+
+    # -------------------------------------------------------------- helpers
+    def _resolve_json_path(self, raw_path):
+        if not raw_path:
+            raise CommandError("مسیر فایل داده مشخص نشده است.")
+        path = Path(raw_path).expanduser()
+        if not path.is_file():
+            raise CommandError(f"فایل داده پیدا نشد: {path}")
+        return path
+
+    def _announce_target(self, alias):
+        """مقصد دیتابیس را بدون افشای credential نمایش می‌دهد."""
+        conf = settings.DATABASES.get(alias, {})
+        engine = conf.get("ENGINE", "?")
+        name = str(conf.get("NAME", ""))
+        settings_module = getattr(settings, "SETTINGS_MODULE", "")
+        self.stdout.write(f"مقصد دیتابیس: alias={alias} engine={engine}")
+        self.stdout.write(f"  NAME: {name}")
+        if "production_settings" in settings_module:
+            self.stdout.write(self.style.WARNING(
+                "  توجه: ماژول تنظیمات production فعال است."))
+        self.stdout.write("  (فقط نام/مسیر دیتابیس چاپ می‌شود؛ credential چاپ نمی‌شود.)")
+
+    def _print_dataset_report(self, name_index):
+        rep = name_index.report
+        self.stdout.write(self.style.MIGRATE_HEADING("اعتبارسنجی فایل داده"))
+        self.stdout.write(f"  مسیر: {name_index.path}")
+        self.stdout.write(f"  SHA-256: {name_index.sha256}")
+        self.stdout.write(
+            f"  رکورد کل: {rep['records_total']} | کلید قابل‌استفاده: {rep['usable_keys']} | "
+            f"کلید غیرقابل‌استفاده: {rep['unusable_keys']}"
+        )
+        self.stdout.write(
+            f"  ردشده → malformed: {rep['rejected_malformed']} | "
+            f"نام خالی: {rep['rejected_empty_name']} | "
+            f"جنسیت نامعتبر: {rep['rejected_unknown_gender']}"
+        )
+        self.stdout.write(
+            f"  نام تکراری: {rep['duplicate_keys']} "
+            f"(هم‌جنسیت: {rep['duplicate_same_gender']} | متناقض: {rep['duplicate_conflict']})"
+        )
+
+    def _ensure_job_table(self, alias):
+        table = CustomerNameFixJob._meta.db_table
+        tables = connections[alias].introspection.table_names()
+        if table not in tables:
+            raise CommandError(
+                f"جدول «{table}» وجود ندارد. ابتدا migration را اعمال کن: "
+                f"python manage.py migrate"
+            )
+
+    def _empty_counters(self):
+        return {key: 0 for key in STATS_KEYS}
+
+    def _counters_from(self, stats):
+        counters = self._empty_counters()
+        if isinstance(stats, dict):
+            for key in STATS_KEYS:
+                value = stats.get(key)
+                if isinstance(value, int):
+                    counters[key] = value
+        return counters
+
+    def _tally(self, counters, decision):
+        counters["scanned"] += 1
+        counters[REASON_TO_STAT[decision.reason]] += 1
+
+    def _finish_counters(self, counters):
+        counters["unchanged"] = counters["scanned"] - counters["changed"]
+        return counters
+
+    def _collect_sample(self, samples, raw_name, new_name, wanted):
+        if wanted and len(samples) < wanted:
+            samples.append((raw_name, new_name))
+
+    # ------------------------------------------------------------ execution
+    def _run_readonly(self, name_index, alias, batch_size, samples_wanted):
+        """اسکن فقط-خواندنی همهٔ مشتریان؛ هیچ نوشتنی (نه نام، نه job) رخ نمی‌دهد."""
+        counters = self._empty_counters()
+        samples = []
+        qs = (Customer.objects.using(alias)
+              .only("id", "name_family").order_by("pk"))
+        in_batch = 0
+        for customer in qs.iterator(chunk_size=batch_size):
+            decision = decide_prefix(customer.name_family, name_index)
+            self._tally(counters, decision)
+            if decision.new_name is not None:
+                self._collect_sample(samples, customer.name_family,
+                                     decision.new_name, samples_wanted)
+            in_batch += 1
+            if in_batch >= batch_size:
+                counters["batches"] += 1
+                in_batch = 0
+        if in_batch:
+            counters["batches"] += 1
+        return self._finish_counters(counters), samples
+
+    def _apply(self, name_index, alias, batch_size, job, totals, samples_wanted):
+        """پردازش batchبه‌batch با checkpoint پایدار بر پایهٔ primary key."""
+        samples = []
+        max_pk = job.initial_max_pk or 0
+        last_pk = job.last_processed_pk or 0
+
+        while last_pk < max_pk:
+            rows = list(
+                Customer.objects.using(alias)
+                .filter(pk__gt=last_pk, pk__lte=max_pk)
+                .order_by("pk").only("id", "name_family")[:batch_size]
+            )
+            if not rows:
+                break
+
+            batch_counters = self._empty_counters()
+            changes = []
+            for customer in rows:
+                decision = decide_prefix(customer.name_family, name_index)
+                self._tally(batch_counters, decision)
+                if decision.new_name is not None:
+                    # فقط فیلد نام تغییر می‌کند؛ سایر فیلدها دست‌نخورده می‌مانند.
+                    changes.append(Customer(pk=customer.pk, name_family=decision.new_name))
+                    self._collect_sample(samples, customer.name_family,
+                                         decision.new_name, samples_wanted)
+
+            new_last_pk = rows[-1].pk
+            batch_counters["batches"] = 1
+            self._commit_batch_with_retry(
+                alias, changes, new_last_pk, batch_counters, job, totals)
+            last_pk = new_last_pk
+
+        return self._finish_counters(totals), samples
+
+    def _commit_batch(self, alias, changes, new_last_pk, batch_counters, job, totals):
+        """تغییرات نام + checkpoint + heartbeat را در یک تراکنش اتمیک ثبت می‌کند."""
+        merged = dict(totals)
+        for key in STATS_KEYS:
+            merged[key] = merged[key] + batch_counters[key]
+        merged = self._finish_counters(merged)
+        now = timezone.now()
+        with transaction.atomic(using=alias):
+            if changes:
+                Customer.objects.using(alias).bulk_update(changes, ["name_family"])
+            CustomerNameFixJob.objects.using(alias).filter(pk=job.pk).update(
+                last_processed_pk=new_last_pk,
+                stats=merged,
+                heartbeat_at=now,
+                updated_at=now,
+                status=CustomerNameFixJob.Status.RUNNING,
+            )
+        # فقط پس از commit موفق، آمار در حافظه به‌روز می‌شود (retry آمار را دوباره اضافه نمی‌کند).
+        totals.update(merged)
+
+    def _commit_batch_with_retry(self, alias, changes, new_last_pk,
+                                 batch_counters, job, totals):
+        for attempt in range(MAX_DB_RETRIES):
+            try:
+                self._commit_batch(alias, changes, new_last_pk,
+                                   batch_counters, job, totals)
+                return
+            except OperationalError as exc:
+                locked = "locked" in str(exc).lower()
+                if not locked or attempt == MAX_DB_RETRIES - 1:
+                    raise
+                delay = BACKOFF_BASE_SECONDS * (2 ** attempt)
+                delay += random.uniform(0, delay / 2)   # jitter
+                self.stderr.write(
+                    f"دیتابیس قفل شد؛ تلاش {attempt + 2}/{MAX_DB_RETRIES} پس از {delay:.2f}s…"
+                )
+                time.sleep(delay)
+
+
+    # ------------------------------------------------------------ job / lock
+    def _claim_job(self, alias, name_index, batch_size, resume_id, force_new):
+        """قفل را می‌گیرد و job مناسب (جدید یا قابل‌ادامه) را RETURN می‌کند.
+
+        خروجی: ``(job, resumed)``. در صورت اجرای هم‌زمان یا تناقض hash،
+        ``CommandError`` پرتاب می‌شود و هیچ تغییری اعمال نمی‌گردد.
+        """
+        Job = CustomerNameFixJob
+        now = timezone.now()
+        digest = name_index.sha256
+
+        active = (Job.objects.using(alias)
+                  .filter(status=Job.Status.RUNNING).order_by("-pk").first())
+        if active is not None:
+            fresh = (active.heartbeat_at is not None
+                     and (now - active.heartbeat_at) < LOCK_TTL)
+            if fresh and active.pk != resume_id:
+                raise CommandError(
+                    f"job#{active.pk} همین حالا در حال اجراست (heartbeat تازه). "
+                    "اجرای هم‌زمان مجاز نیست."
+                )
+            if not fresh:
+                # قفل کهنه (crash/قطع برق): آزادسازی کنترل‌شده
+                Job.objects.using(alias).filter(pk=active.pk).update(
+                    status=Job.Status.FAILED, finished_at=now, updated_at=now,
+                    note="قفل کهنه؛ آزادسازی خودکار پس از crash")
+                self.stderr.write(
+                    f"job#{active.pk} ناتمام رها شده بود و اکنون به‌عنوان failed آزاد شد.")
+
+        if resume_id is not None:
+            job = Job.objects.using(alias).filter(pk=resume_id).first()
+            if job is None:
+                raise CommandError(f"job#{resume_id} پیدا نشد.")
+            self._check_hash(job, digest)
+            if job.status == Job.Status.COMPLETED:
+                raise CommandError(
+                    f"job#{job.pk} قبلاً تکمیل شده است؛ برای اجرای دوباره --new-job بده.")
+            resumed = True
+        elif force_new:
+            job = self._create_job(alias, digest, batch_size, now)
+            resumed = False
+        else:
+            job = (Job.objects.using(alias)
+                   .filter(source_hash=digest, algorithm_version=ALGORITHM_VERSION)
+                   .exclude(status=Job.Status.COMPLETED).order_by("-pk").first())
+            if job is not None:
+                resumed = True
+            else:
+                done = Job.objects.using(alias).filter(
+                    source_hash=digest, algorithm_version=ALGORITHM_VERSION,
+                    status=Job.Status.COMPLETED).exists()
+                if done:
+                    raise CommandError(
+                        "روی همین فایل یک job تکمیل‌شده وجود دارد؛ برای اجرای دوبارهٔ "
+                        "کنترل‌شده --new-job بده."
+                    )
+                job = self._create_job(alias, digest, batch_size, now)
+                resumed = False
+
+        Job.objects.using(alias).filter(pk=job.pk).update(
+            status=Job.Status.RUNNING, heartbeat_at=now, updated_at=now)
+        job.refresh_from_db(using=alias)
+        return job, resumed
+
+    def _check_hash(self, job, digest):
+        if job.source_hash != digest:
+            raise CommandError(
+                f"hash فایل ورودی با job#{job.pk} نمی‌خواند؛ ادامه رد شد و هیچ تغییری "
+                "اعمال نمی‌شود. برای فایل جدید یک job مستقل اجرا کن."
+            )
+        if job.algorithm_version != ALGORITHM_VERSION:
+            raise CommandError(
+                f"نسخهٔ الگوریتم job#{job.pk} ({job.algorithm_version}) با نسخهٔ فعلی "
+                f"({ALGORITHM_VERSION}) نمی‌خواند؛ ادامه رد شد."
+            )
+
+    def _create_job(self, alias, digest, batch_size, now):
+        Job = CustomerNameFixJob
+        initial_max_pk = (Customer.objects.using(alias)
+                          .aggregate(m=Max("id"))["m"]) or 0
+        return Job.objects.using(alias).create(
+            source_hash=digest,
+            algorithm_version=ALGORITHM_VERSION,
+            status=Job.Status.RUNNING,
+            batch_size=batch_size,
+            initial_max_pk=initial_max_pk,
+            last_processed_pk=0,
+            stats=self._finish_counters(self._empty_counters()),
+            heartbeat_at=now,
+            note="",
+        )
+
+
+    def _finalize(self, alias, job, counters, status, note):
+        """وضعیت نهایی/میانی job را با روش اتمیک ثبت می‌کند."""
+        job.refresh_from_db(using=alias)
+        now = timezone.now()
+        persisted = self._finish_counters(dict(counters))
+        finished = status in (CustomerNameFixJob.Status.COMPLETED,
+                              CustomerNameFixJob.Status.FAILED)
+        finished_at = now if finished else job.finished_at
+        with transaction.atomic(using=alias):
+            CustomerNameFixJob.objects.using(alias).filter(pk=job.pk).update(
+                status=status, stats=persisted, updated_at=now, heartbeat_at=now,
+                finished_at=finished_at, note=note[:255])
+        job.status = status
+        job.stats = persisted
+        job.finished_at = finished_at
+
+    # -------------------------------------------------------------- summary
+    def _print_summary(self, counters, samples, dry_run, job):
+        if dry_run:
+            head = "گزارش dry-run — هیچ تغییری نوشته نشد"
+        else:
+            head = f"گزارش اجرای واقعی job#{job.pk} ({job.status})"
+        self.stdout.write(self.style.MIGRATE_HEADING(head))
+        self.stdout.write(f"  مشتریان بررسی‌شده: {counters['scanned']}")
+        self.stdout.write(f"  قابل اصلاح / اصلاح‌شده: {counters['changed']}")
+        self.stdout.write(f"  بدون تغییر: {counters['unchanged']}")
+        self.stdout.write(
+            f"  تک‌کلمه‌ای: {counters['single_word']} | "
+            f"دارای پیشوند: {counters['has_prefix']} | "
+            f"خالی/نامعتبر: {counters['empty_invalid']}"
+        )
+        self.stdout.write(
+            f"  ناشناخته در JSON: {counters['unknown_name']} | "
+            f"جنسیت مبهم/نامعتبر: {counters['ambiguous_gender']}"
+        )
+        self.stdout.write(f"  batchهای تکمیل‌شده: {counters['batches']}")
+        if job is not None:
+            self.stdout.write(
+                f"  شناسهٔ job: {job.pk} | شروع: {job.started_at} | پایان: {job.finished_at}"
+            )
+        if samples:
+            self.stdout.write("  نمونهٔ تغییرات (قبل → بعد):")
+            for before, after in samples:
+                self.stdout.write(f"    «{before}» → «{after}»")
+        if dry_run:
+            self.stdout.write(self.style.WARNING(
+                "برای اجرای واقعی: --apply --confirm-production"))
+

admin_panel/management/commands/normalize_customer_sms_phones.py

@@ -0,0 +1,367 @@
+"""استخراج و تجمیع شماره‌های موبایل هر مشتری در ``Customer.sms_primary_phones``.
+
+منابع بررسی (به همین ترتیب اولویت برای حذف تکراری‌ها):
+``phone1`` → ``phone2`` → ``phone3`` → رکوردهای ``PhoneNumber`` بر اساس ``id``.
+
+**امنیت داده:** فیلدهای مبدأ، سایر مشخصات مشتری و رکوردهای ``PhoneNumber`` هرگز
+ویرایش/حذف نمی‌شوند. تنها ستونی که نوشته می‌شود ``sms_primary_phones`` است.
+
+**قالب ذخیره‌شده:** ``+989xxxxxxxxx`` — همان قالب متعارف پروژه که
+``admin_panel.sms.phones.stored_primary_normalized`` انتظار دارد و تمام تست‌های
+موجود (``tests/test_sms.py``، ``tests/test_customers.py``) روی آن قفل شده‌اند.
+اعتبارسنجی ورودی با استاندارد ۱۱ رقمی ``09xxxxxxxxx`` انجام می‌شود، ولی مقدارِ
+ذخیره‌شده همان ``+98…`` است؛ در غیر این‌صورت تطبیق «شمارهٔ اصلی» در پنل و
+``resolved_recipients`` هیچ‌وقت نمی‌گیرد و داده بی‌اثر می‌شود.
+
+**تضمین‌ها:**
+
+* پیش‌فرض dry-run است؛ نوشتن فقط با ``--apply``.
+* هر batch در ``transaction.atomic`` اختصاصی؛ خطا ⇒ فقط همان batch برمی‌گردد.
+* Keyset pagination (``id__gt``) + checkpoint فایل پس از commit موفق هر batch.
+* رکوردِ بدون تغییر نوشته نمی‌شود؛ مقدار موجودِ غیرخالی هرگز با «خالی» پاک نمی‌شود.
+* سرریز طول (> ``max_length``) ⇒ رکورد دست‌نخورده + ثبت در گزارش.
+
+اجرا:
+
+    python manage.py normalize_customer_sms_phones --dry-run
+    python manage.py normalize_customer_sms_phones --apply
+    python manage.py normalize_customer_sms_phones --apply --resume
+    python manage.py normalize_customer_sms_phones --apply --start-id 5000
+    python manage.py normalize_customer_sms_phones --apply --reset-checkpoint
+"""
+import json
+import os
+from pathlib import Path
+
+from django.conf import settings
+from django.core.management.base import BaseCommand, CommandError
+from django.db import transaction
+from persiantools import digits
+
+from admin_panel.models import Customer
+from admin_panel.sms.phones import normalize_primary_values
+from integrations.sms.normalize import normalize_mobile
+
+DEFAULT_BATCH_SIZE = 100
+DEFAULT_CHECKPOINT_NAME = ".sms_normalization_checkpoint.json"
+
+# مرز طول از خود مدل خوانده می‌شود تا با تغییر مدل هم‌راستا بماند.
+FIELD_MAX_LENGTH = Customer._meta.get_field("sms_primary_phones").max_length
+LEGACY_FIELDS = ("phone1", "phone2", "phone3")
+
+# حداکثر تعداد شناسهٔ سرریز که در گزارش چاپ می‌شود (بقیه فقط شمرده می‌شوند).
+OVERFLOW_SAMPLE = 10
+
+# ارقام «پیش‌شمارهٔ ثابت»: 021، 031، 051، 071، … (برای گزارش، نه تصمیم داده).
+LANDLINE_SECOND_DIGITS = set("12345678")
+
+
+def digit_string(raw):
+    """فقط ارقامِ یک مقدار ورودی، با تبدیل ارقام فارسی/عربی به انگلیسی."""
+    if raw is None:
+        return ""
+    text = digits.fa_to_en(digits.ar_to_fa(str(raw).strip()))
+    return "".join(ch for ch in text if ch.isdigit())
+
+
+def is_landline(raw):
+    """آیا مقدار ورودی شکل شمارهٔ ثابت دارد؟ (فقط برای تفکیک در گزارش)
+
+    شامل شمارهٔ ثابت محلی ۸رقمی (مثل ``57387608``) و شمارهٔ ثابت با پیش‌شماره
+    (مثل ``02112345678`` / ``03133445566``) می‌شود — همان قاعدهٔ مستندشدهٔ پروژه
+    در ``fix_customer_phones.md``.
+    """
+    text = digit_string(raw)
+    if len(text) == 8:
+        return True
+    if len(text) not in (10, 11) or not text.startswith("0"):
+        return False
+    return text[1] in LANDLINE_SECOND_DIGITS
+
+
+class Command(BaseCommand):
+    help = (
+        "استخراج و تجمیع شماره‌های موبایل معتبر هر مشتری در sms_primary_phones "
+        "(پیش‌فرض dry-run؛ اتمیکِ batch به batch با checkpoint قابل ادامه)."
+    )
+
+    def add_arguments(self, parser):
+        parser.add_argument(
+            "--batch-size", type=int, default=DEFAULT_BATCH_SIZE,
+            help=f"تعداد مشتریان در هر batch (پیش‌فرض {DEFAULT_BATCH_SIZE}).",
+        )
+        parser.add_argument(
+            "--dry-run", action="store_true",
+            help="صریحاً حالت شبیه‌سازی (پیش‌فرض همین است؛ همراه --apply مجاز نیست).",
+        )
+        parser.add_argument(
+            "--apply", action="store_true",
+            help="نوشتن واقعی در دیتابیس. بدون این گزینه فقط شبیه‌سازی می‌شود.",
+        )
+        parser.add_argument(
+            "--resume", action="store_true",
+            help="ادامه از آخرین شناسهٔ ثبت‌شده در فایل وضعیت.",
+        )
+        parser.add_argument(
+            "--start-id", type=int, default=None,
+            help="شروع دستی از یک شناسه (مشتریان با id بزرگ‌تر پردازش می‌شوند). "
+                 "بر --resume اولویت دارد.",
+        )
+        parser.add_argument(
+            "--reset-checkpoint", action="store_true",
+            help="پاک‌کردن فایل وضعیت و شروع از ابتدا (در dry-run نادیده گرفته می‌شود).",
+        )
+        parser.add_argument(
+            "--checkpoint", default=None,
+            help=f"مسیر فایل وضعیت (پیش‌فرض: {DEFAULT_CHECKPOINT_NAME} در ریشهٔ پروژه).",
+        )
+
+    # ------------------------------------------------------------------ main
+    def handle(self, *args, **options):
+        batch_size = max(1, options["batch_size"])
+        apply_changes = options["apply"]
+        checkpoint_path = Path(
+            options["checkpoint"] or (Path(settings.BASE_DIR) / DEFAULT_CHECKPOINT_NAME)
+        )
+
+        self.stdout.write(
+            f"حالت: {'اجرای واقعی (--apply)' if apply_changes else 'شبیه‌سازی (dry-run)'} | "
+            f"batch={batch_size} | فایل وضعیت: {checkpoint_path}"
+        )
+        if options["dry_run"] and apply_changes:
+            raise CommandError("--dry-run و --apply با هم ناسازگارند.")
+        self._handle_reset(checkpoint_path, apply_changes, options["reset_checkpoint"])
+        last_id, source = self._start_point(checkpoint_path, options)
+        self.stdout.write(f"شروع از مشتریان با id > {last_id} ({source})")
+
+        counters = {
+            "scanned": 0, "updated": 0, "unchanged": 0, "kept_nonempty": 0,
+            "filtered_landline": 0, "filtered_invalid": 0, "duplicates": 0,
+            "overflow": 0, "batches": 0,
+        }
+        overflow_ids = []
+
+        try:
+            last_id = self._run(
+                last_id, batch_size, apply_changes, checkpoint_path,
+                counters, overflow_ids)
+        except KeyboardInterrupt:
+            self.stdout.write(self.style.WARNING(
+                f"متوقف شد (Ctrl+C). آخرین شناسهٔ ثبت‌شده: {last_id}. "
+                "برای ادامه: --apply --resume"
+            ))
+            return
+        except Exception as exc:  # noqa: BLE001 - توقف امن با گزارش روشن
+            raise CommandError(
+                f"اجرا متوقف شد؛ batch جاری rollback شد و آخرین شناسهٔ معتبر "
+                f"{last_id} است: {exc}"
+            )
+
+        self._print_report(counters, overflow_ids, apply_changes, last_id)
+
+    # ------------------------------------------------------------ start/lock
+    def _handle_reset(self, checkpoint_path, apply_changes, reset_requested):
+        if not reset_requested:
+            return
+        if not checkpoint_path.exists():
+            self.stdout.write("فایل وضعیت وجود نداشت؛ چیزی برای پاک‌کردن نبود.")
+            return
+        if not apply_changes:
+            self.stdout.write("--reset-checkpoint در شبیه‌سازی نادیده گرفته شد (فایل وضعیت دست‌نخورده).")
+            return
+        try:
+            checkpoint_path.unlink()
+        except OSError as exc:
+            raise CommandError(f"فایل وضعیت پاک نشد ({checkpoint_path}): {exc}")
+        self.stdout.write(f"فایل وضعیت پاک شد: {checkpoint_path}")
+
+    def _start_point(self, checkpoint_path, options):
+        """(last_id, توضیح منبع) — اولویت: --start-id سپس --resume سپس ابتدا."""
+        if options["start_id"] is not None:
+            return max(0, options["start_id"]), "--start-id"
+        if options["resume"]:
+            if not checkpoint_path.exists():
+                raise CommandError(
+                    f"--resume داده شد ولی فایل وضعیت وجود ندارد: {checkpoint_path}"
+                )
+            data = self._read_checkpoint(checkpoint_path)
+            value = data.get("last_id")
+            if not isinstance(value, int) or value < 0:
+                raise CommandError(
+                    f"فایل وضعیت معتبر نیست (last_id={value!r}): {checkpoint_path}"
+                )
+            return value, "--resume"
+        return 0, "ابتدا"
+
+    def _read_checkpoint(self, path):
+        try:
+            data = json.loads(path.read_text(encoding="utf-8"))
+        except (OSError, ValueError) as exc:
+            raise CommandError(f"فایل وضعیت خوانده نشد ({path}): {exc}")
+        if not isinstance(data, dict):
+            raise CommandError(f"فایل وضعیت ساختار معتبر ندارد ({path}).")
+        return data
+
+    def _write_checkpoint(self, path, last_id, batch_size, counters):
+        """نوشتن اتمیک فایل وضعیت (tmp + os.replace) با آخرین شناسهٔ موفق."""
+        payload = {
+            "last_id": last_id,
+            "batch_size": batch_size,
+            "applied": True,
+            "scanned": counters["scanned"],
+            "updated": counters["updated"],
+            "overflow": counters["overflow"],
+            "batches": counters["batches"],
+        }
+        tmp = path.with_name(path.name + ".tmp")
+        try:
+            tmp.write_text(
+                json.dumps(payload, ensure_ascii=False, indent=2), encoding="utf-8")
+            os.replace(tmp, path)
+        except OSError as exc:
+            raise CommandError(f"فایل وضعیت نوشته نشد ({path}): {exc}")
+
+    # ---------------------------------------------------------------- batch
+    def _run(self, last_id, batch_size, apply_changes, checkpoint_path,
+             counters, overflow_ids):
+        """حلقهٔ اصلی: هر batch یک تراکنش اتمیک + checkpoint پس از commit موفق."""
+        max_id = (Customer.objects.order_by("-id")
+                  .values_list("id", flat=True).first()) or 0
+        if last_id >= max_id:
+            self.stdout.write("مشتری‌ای در دامنهٔ پردازش باقی نمانده است.")
+            return last_id
+
+        while last_id < max_id:
+            rows = list(
+                Customer.objects
+                .filter(id__gt=last_id)
+                .order_by("id")
+                .prefetch_related("phones")            # جلوگیری از N+1
+                .only("id", "phone1", "phone2", "phone3", "sms_primary_phones")
+                [:batch_size]
+            )
+            if not rows:
+                break
+
+            pending = []
+            for customer in rows:
+                counters["scanned"] += 1
+                value = self._target_value(customer, counters)
+                current = (customer.sms_primary_phones or "").strip()
+                if value is None:
+                    overflow_ids.append(customer.id)    # سرریز: دست‌نخورده
+                    continue
+                if not value:
+                    # مقدار محاسبه‌شده خالی است: مقدار موجودِ مدیر پاک نمی‌شود.
+                    if current:
+                        counters["kept_nonempty"] += 1
+                    continue
+                if value == current:
+                    counters["unchanged"] += 1
+                    continue
+                pending.append((customer, value))
+
+            if apply_changes and pending:
+                # یک تراکنش کوتاه برای همین batch: خطا ⇒ فقط همین batch برمی‌گردد.
+                with transaction.atomic():
+                    Customer.objects.bulk_update(
+                        [self._stage(customer, value) for customer, value in pending],
+                        ["sms_primary_phones"])
+            counters["updated"] += len(pending)
+
+            new_last_id = rows[-1].id
+            counters["batches"] += 1
+            if apply_changes:
+                self._write_checkpoint(
+                    checkpoint_path, new_last_id, batch_size, counters)
+            last_id = new_last_id
+            self.stdout.write(
+                f"  batch {counters['batches']}: last_id={last_id} | "
+                f"بررسی‌شده={counters['scanned']} | "
+                f"{'به‌روزشده' if apply_changes else 'قابل‌به‌روزرسانی'}={counters['updated']} | "
+                f"سرریز={counters['overflow']}"
+            )
+        return last_id
+
+
+    # -------------------------------------------------------------- helpers
+    def _stage(self, customer, value):
+        """مقدار نهایی را روی نمونه می‌گذارد و همان را برمی‌گرداند (برای bulk_update)."""
+        customer.sms_primary_phones = value
+        return customer
+
+    def _target_value(self, customer, counters):
+        """مقدار نهایی ``sms_primary_phones`` برای یک مشتری.
+
+        خروجی: رشتهٔ نهایی، ``""`` (شمارهٔ معتبری نیست) یا ``None`` (سرریز > max_length).
+        """
+        raw_values = [getattr(customer, field, "") for field in LEGACY_FIELDS]
+        numbers = sorted(customer.phones.all(), key=lambda row: row.id)
+        raw_values.extend(row.phone_number for row in numbers)
+
+        self._count_filters(raw_values, counters)
+        value = normalize_primary_values(raw_values)
+        if not value:
+            return ""
+        if len(value) > FIELD_MAX_LENGTH:
+            counters["overflow"] += 1
+            return None
+        return value
+
+    def _count_filters(self, raw_values, counters):
+        """شمارش شماره‌های فیلترشده (ثابت/نامعتبر) و تکراری‌های حذف‌شده."""
+        seen = set()
+        duplicates = 0
+        for raw in raw_values:
+            if raw is None or not str(raw).strip():
+                continue
+            number = normalize_mobile(raw)
+            if not number:
+                if is_landline(raw):
+                    counters["filtered_landline"] += 1
+                else:
+                    counters["filtered_invalid"] += 1
+                continue
+            if number in seen:
+                duplicates += 1
+            else:
+                seen.add(number)
+        counters["duplicates"] += duplicates
+
+    # -------------------------------------------------------------- report
+    def _print_report(self, counters, overflow_ids, apply_changes, last_id):
+        head = ("گزارش شبیه‌سازی (dry-run) — هیچ تغییری ذخیره نشد"
+                if not apply_changes else "گزارش اجرای واقعی")
+        self.stdout.write(self.style.MIGRATE_HEADING(head))
+        self.stdout.write(f"  مشتریان بررسی‌شده: {counters['scanned']}")
+        self.stdout.write(
+            f"  مشتریان {'به‌روزشده' if apply_changes else 'قابل‌به‌روزرسانی'}: "
+            f"{counters['updated']}"
+        )
+        self.stdout.write(f"  بدون تغییر: {counters['unchanged']}")
+        self.stdout.write(
+            f"  شماره‌های فیلترشده: ثابت={counters['filtered_landline']} | "
+            f"نامعتبر={counters['filtered_invalid']} | "
+            f"جمع={counters['filtered_landline'] + counters['filtered_invalid']}"
+        )
+        self.stdout.write(f"  شماره‌های تکراری حذف‌شده: {counters['duplicates']}")
+        self.stdout.write(
+            f"  سرریز طول (>{FIELD_MAX_LENGTH} کاراکتر، دست‌نخورده): {counters['overflow']}"
+        )
+        if overflow_ids:
+            sample = ", ".join(str(pk) for pk in overflow_ids[:OVERFLOW_SAMPLE])
+            more = len(overflow_ids) - OVERFLOW_SAMPLE
+            tail = f" و {more} مورد دیگر" if more > 0 else ""
+            self.stdout.write(f"    شناسه‌های سرریز: {sample}{tail}")
+        if counters["kept_nonempty"]:
+            self.stdout.write(
+                f"  مقدار موجودِ غیرخالی که محافظت شد (شمارهٔ معتبری پیدا نشد): "
+                f"{counters['kept_nonempty']}"
+            )
+        self.stdout.write(f"  batchهای تکمیل‌شده: {counters['batches']}")
+        self.stdout.write(f"  آخرین شناسهٔ پردازش‌شده: {last_id}")
+        if not apply_changes:
+            self.stdout.write(self.style.WARNING(
+                "برای اجرای واقعی: --apply (و در صورت توقف قبلی: --apply --resume)"
+            ))
+