368 lines
18 KiB
Python
368 lines
18 KiB
Python
"""استخراج و تجمیع شماره‌های موبایل هر مشتری در ``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)"
|
|
))
|
|
|