khatibi/admin_panel/management/commands/gt_esh.py

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)