485 lines
23 KiB
Python
485 lines
23 KiB
Python
"""افزودن قطعیِ پیشوند «آقای/خانم» به ابتدای نام مشتریان، بر پایهٔ فایل نام‌های ایرانی.
|
|
|
|
سیاست‌ها (کاملاً محافظه‌کارانه):
|
|
|
|
* **پیش‌فرض بدون نوشتن است** (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"))
|
|
|