import time import logging import contextlib from django.core.management.base import BaseCommand from django.db import transaction, connection from django.db.models import Min, Max from admin_panel.models import ProductInStore # مدل انبار پروژه شما logger = logging.getLogger('management_commands') class Command(BaseCommand): help = '🚀 همگام‌سازی ساختاریافته فیلد movement_type (مقاوم در برابر بن‌بست بر اساس مستندات SQLite3)' def add_arguments(self, parser): parser.add_argument( '--dry-run', action='store_true', help='اجرای شبیه‌سازی بدون ثبت تغییرات نهایی در پایگاه‌داده' ) parser.add_argument( '--chunk-size', type=int, default=1000, # کاهش حجم بسته به ۱۰۰۰ جهت سازگاری حداکثری با RAM کانتینرها و قفل SQLite help='تعداد ردیف‌هایی که در هر دسته پردازش می‌شوند (پیش‌فرض ۱۰۰۰)' ) def handle(self, *args, **options): dry_run = options['dry_run'] chunk_size = options['chunk_size'] self.stdout.write(self.style.MIGRATE_HEADING("==================================================")) self.stdout.write(self.style.MIGRATE_HEADING("⚙️ شروع فرآیند مهاجرت ترکیبی داده‌های انبار (Hybrid Memory Phase)")) if dry_run: self.stdout.write(self.style.WARNING("⚠️ [حالت شبیه‌سازی - DRY RUN] هیچ دیتایی تغییر نخواهد کرد.")) self.stdout.write(self.style.MIGRATE_HEADING("==================================================")) # اعمال پیکربندی سخت‌افزاری اتصال بر اساس استاندارد رسمی SQLite3 if connection.vendor == 'sqlite': with connection.cursor() as cursor: cursor.execute('PRAGMA busy_timeout=60000;') # انتظار صبورانه کانتینرها تا ۶۰ ثانیه cursor.execute('PRAGMA synchronous=NORMAL;') metrics = { 'PROMOTION': 0, 'WARRANTY': 0, 'TRANSFER': 0, 'INITIAL': 0, 'DIRECT_IN': 0, 'DIRECT_OUT': 0, 'SALE': 0, 'TOTAL_UPDATED': 0 } start_time = time.time() try: bounds = ProductInStore.objects.aggregate(min_id=Min('id'), max_id=Max('id')) min_id = bounds['min_id'] or 0 max_id = bounds['max_id'] or 0 if not max_id: self.stdout.write(self.style.WARNING("ℹ️ هیچ تراکنشی در جدول انبار یافت نشد.")) return total_rows_to_scan = max_id - min_id + 1 self.stdout.write(self.style.NOTICE(f"🔍 محدوده شناسه‌ها: از {min_id} تا {max_id} ({total_rows_to_scan:,} ردیف)")) if dry_run: with transaction.atomic(): self._execute_hybrid_migration(min_id, max_id, chunk_size, total_rows_to_scan, metrics, dry_run=True) transaction.set_rollback(True) self.stdout.write(self.style.WARNING("\n🍁 [شبیه‌سازی کامل شد] تغییرات با موفقیت لغو شد.")) else: self._execute_by_immediate_chunks(min_id, max_id, chunk_size, total_rows_to_scan, metrics) # نمایش داشبورد نهایی حسابرسی مانیتورینگ داکر duration = round(time.time() - start_time, 2) self.stdout.write(self.style.SUCCESS(f"\n✅ عملیات با موفقیت در {duration} ثانیه به پایان رسید.")) self.stdout.write(self.style.MIGRATE_LABEL("\n📊 گزارش نهایی و تفکیکی کل دیتابیس زنده شما:")) self.stdout.write(f" 🔹 قطعات گارانتی (WARRANTY): {metrics['WARRANTY']:,} ردیف") self.stdout.write(f" 🔹 هدایای تبلیغاتی (PROMOTION): {metrics['PROMOTION']:,} ردیف") self.stdout.write(f" 🔹 انتقال بین انبارها (TRANSFER): {metrics['TRANSFER']:,} ردیف") self.stdout.write(f" 🔹 موجودی اولیه (INITIAL): {metrics['INITIAL']:,} ردیف") self.stdout.write(f" 🔹 ورود دستی (DIRECT_IN): {metrics['DIRECT_IN']:,} ردیف") self.stdout.write(f" 🔹 خروج دستی (DIRECT_OUT): {metrics['DIRECT_OUT']:,} ردیف") self.stdout.write(f" 🔹 فروش عادی و پاکسازی‌ها (SALE): {metrics['SALE']:,} ردیف") self.stdout.write(self.style.MIGRATE_HEADING("==================================================")) except Exception as e: self.stdout.write(self.style.ERROR(f"\n❌ خطای بحرانی در سطح سیستم: {str(e)}")) logger.error(f"Critical error during sync_inventory migration: {str(e)}", exc_info=True) def _execute_by_immediate_chunks(self, min_id, max_id, chunk_size, total_rows_to_scan, metrics): """اجرای حلقه‌های زنده با قفل سخت‌افزاری IMMEDIATE جهت جلوگیری از خطای قفل دیتابیس""" current_id = min_id while current_id <= max_id: end_id = current_id + chunk_size - 1 # بر اساس مستندات SQLite، اجرای دستی BEGIN IMMEDIATE از تعارض قفل کانتینرهای دیگر ملوگیری می‌کند with connection.cursor() as cursor: cursor.execute('BEGIN IMMEDIATE;') try: # فراخوانی هسته اصلی مهاجرت برای بسته جاری self._process_single_chunk(current_id, end_id, metrics) cursor.execute('COMMIT;') except Exception as e: cursor.execute('ROLLBACK;') raise e progress = min(100, int(((current_id - min_id) / total_rows_to_scan) * 100)) self.stdout.write(f"⏳ در حال پردازش دسته‌ها... پیشرفت: {progress}% (شناسه {current_id} تا {end_id})") # ایجاد یک پنجره تنفس ۱۰ میلی‌ثانیه‌ای واقعی جهت فرار از Lock Starvation کانتینرهای Gunicorn time.sleep(0.01) current_id += chunk_size def _execute_hybrid_migration(self, min_id, max_id, chunk_size, total_rows_to_scan, metrics, dry_run): """متد کمکی صرفاً برای حالت شبیه‌سازی (Dry Run) بدون کامیت زنده""" current_id = min_id while current_id <= max_id: end_id = current_id + chunk_size - 1 self._process_single_chunk(current_id, end_id, metrics) current_id += chunk_size def _process_single_chunk(self, start_id, end_id, metrics): """هسته پردازش هیبریدی: یک بار خواندن از دیتابیس، پردازش در رم، یک بار نوشتن""" # ۱. دریافت ردیف‌های این دسته بر اساس مرز ایندکس کلید اصلی (فوق سریع) records = ProductInStore.objects.filter(id__range=(start_id, end_id)).only('id', 'desc', 'factor_id') # آرایه‌های تفکیک‌شده در حافظه موقت (RAM پایتون) updates = { 'PROMOTION': [], 'WARRANTY': [], 'TRANSFER': [], 'INITIAL': [], 'DIRECT_IN': [], 'DIRECT_OUT': [], 'SALE': [] } # ۲. پردازش الگوهای متنی با سرعت مایکروثانیه‌ای در سطح پایتون (بدون درگیر کردن دیسک سخت) for rec in records: desc_text = rec.desc or '' if 'اشانتیون' in desc_text: updates['PROMOTION'].append(rec.id) metrics['PROMOTION'] += 1 elif 'گارانتی' in desc_text: updates['WARRANTY'].append(rec.id) metrics['WARRANTY'] += 1 elif 'انتقال' in desc_text: updates['TRANSFER'].append(rec.id) metrics['TRANSFER'] += 1 elif 'موجودی اولیه' in desc_text: updates['INITIAL'].append(rec.id) metrics['INITIAL'] += 1 elif 'ورود:' in desc_text and not rec.factor_id: updates['DIRECT_IN'].append(rec.id) metrics['DIRECT_IN'] += 1 elif 'خروج:' in desc_text and not rec.factor_id: updates['DIRECT_OUT'].append(rec.id) metrics['DIRECT_OUT'] += 1 else: # قانون پوشش همه‌جانبه (حالت فروش عادی یا پاکسازی نهایی) updates['SALE'].append(rec.id) metrics['SALE'] += 1 # ۳. اعمال تغییرات نهایی در سطح دیتابیس (فقط ردیف‌هایی که در رم تفکیک شدند تغییر می‌کنند) for m_type, ids_list in updates.items(): if ids_list: ProductInStore.objects.filter(id__in=ids_list).update(movement_type=m_type)