Dovodi_Backend/scripts/update_readable_slugs.py

#!/usr/bin/env python3
"""
High-Performance Script to Regenerate and Update Slugs for:
- HadisCategory
- Hadis
- Transmitters

Features:
- Human-readable slugs generated from actual multilingual titles and names
- Fast in-memory deduplication and uniqueness guarantees
- Safe two-phase atomic update using raw SQL temporary assignment + bulk update
- Supports --dry-run flag for inspection before applying
"""

import os
import sys
import re
import time
import argparse
from pathlib import Path
from django.utils.text import slugify

# Force UTF-8 stdout for Windows consoles
if hasattr(sys.stdout, "reconfigure"):
    sys.stdout.reconfigure(encoding="utf-8")

BASE_DIR = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(BASE_DIR))
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'config.settings.base')

import django
django.setup()

from django.db import transaction, connection
from apps.hadis.models import HadisCategory, Hadis, Transmitters

DUMMY_NAMES = {
    "al-imam-al-hafiz-abu-isa-muhammad-ibn-isa-ibn-sawrah-ibn-musa-ibn-al-dahhak-al-sulami-al-tirmidhi",
    "al-imam-al-hafiz-abu-isa-muhammad",
    "الإمام الحافظ أبو عيسى محمد بن عيسى بن سَوْرَة بن موسى بن الضحاک السُّلَمي الترمذي",
    "اسم-عربی",
}


def extract_best_text(json_val, preferred_langs=('en', 'ru', 'fa', 'ar')):
    """Extract the best text from a multilingual JSONField (list of dicts)."""
    if not json_val:
        return ""
    if isinstance(json_val, str):
        return json_val.strip()
    if isinstance(json_val, list):
        for lang in preferred_langs:
            for item in json_val:
                if isinstance(item, dict) and item.get('language_code') == lang:
                    val = item.get('text') or item.get('title') or item.get('value') or item.get('name')
                    if val and str(val).strip():
                        return str(val).strip()
        for item in json_val:
            if isinstance(item, dict):
                val = item.get('text') or item.get('title') or item.get('value') or item.get('name')
                if val and str(val).strip():
                    return str(val).strip()
            elif isinstance(item, str) and item.strip():
                return item.strip()
    return ""


def extract_transmitter_text(transmitter):
    """Extract best specific name for a transmitter, avoiding dummy/repeated texts."""
    candidates = []
    for field_val in (transmitter.full_name, transmitter.known_as, transmitter.nickname):
        if not field_val or not isinstance(field_val, list):
            continue
        for lang in ('ru', 'fa', 'en', 'ar'):
            for item in field_val:
                if isinstance(item, dict) and item.get('language_code') == lang:
                    val = (item.get('text') or item.get('title') or '').strip()
                    if val and val not in DUMMY_NAMES and not val.startswith("313"):
                        candidates.append(val)

    if candidates:
        return candidates[0]
        
    return extract_best_text(transmitter.full_name) or extract_best_text(transmitter.known_as) or ""


def clean_slug_text(text: str, max_len: int = 80) -> str:
    """Clean and slugify text while preserving readability and length constraints."""
    if not text:
        return ""
    cleaned = re.sub(r'[\(\)\[\]\{\}\<\>\'\"\`\:\;\,\.\?\!\@\#\$\%\^\&\*\+\=\|\/\\]+', ' ', text)
    slug = slugify(cleaned, allow_unicode=True).strip('-').lower()
    if len(slug) > max_len:
        slug = slug[:max_len].rstrip('-')
    return slug


def generate_unique_slug(base_slug: str, used_slugs: set, max_len: int = 90) -> str:
    """Ensure slug uniqueness using an in-memory set with counter suffix."""
    if not base_slug:
        base_slug = "item"
    
    candidate = base_slug
    counter = 1
    
    while candidate in used_slugs:
        counter += 1
        suffix = f"-{counter}"
        avail_len = max_len - len(suffix)
        candidate = f"{base_slug[:avail_len].rstrip('-')}{suffix}"
        
    used_slugs.add(candidate)
    return candidate


def update_categories_slugs(dry_run=False):
    print("\n📂 [1/3] Processing HadisCategory...")
    categories = list(HadisCategory.objects.all().order_by('id'))
    used_slugs = set()
    sample_changes = []

    for cat in categories:
        raw_text = extract_best_text(cat.title, ('en', 'ru', 'fa', 'ar')) or extract_best_text(cat.description)
        base_slug = clean_slug_text(raw_text, max_len=75)
        if not base_slug:
            base_slug = f"category-{cat.source_type or 'item'}-{cat.id}"
            
        new_slug = generate_unique_slug(base_slug, used_slugs)
        
        if len(sample_changes) < 5 and cat.slug != new_slug:
            sample_changes.append((cat.id, cat.slug, new_slug, raw_text))
            
        cat.slug = new_slug

    print(f"  Total categories: {len(categories)}")
    for item_id, old_s, new_s, title in sample_changes:
        print(f"  • ID {item_id:4d} | '{title[:30]}' -> {old_s} ➔ {new_s}")

    if not dry_run:
        with transaction.atomic():
            with connection.cursor() as cursor:
                cursor.execute("UPDATE hadis_hadiscategory SET slug = CONCAT('tmp-c-', id);")
            HadisCategory.objects.bulk_update(categories, ['slug'], batch_size=1000)
        print("  ✅ All categories updated successfully.")


def update_hadis_slugs(dry_run=False):
    print("\n📜 [2/3] Processing Hadis...")
    hadiths = list(Hadis.objects.all().order_by('id'))
    used_slugs = set()
    sample_changes = []

    for h in hadiths:
        raw_text = extract_best_text(h.title, ('en', 'ru', 'fa', 'ar')) or extract_best_text(h.translation) or extract_best_text(h.hadis_status_text)
        base_slug = clean_slug_text(raw_text, max_len=75)
        
        if not base_slug or (base_slug.isdigit() and len(base_slug) <= 3):
            if h.number:
                base_slug = f"hadis-{h.number}" + (f"-{base_slug}" if base_slug else "")
            else:
                base_slug = f"hadis-{h.id}"
                
        new_slug = generate_unique_slug(base_slug, used_slugs)
        
        if len(sample_changes) < 5 and h.slug != new_slug:
            sample_changes.append((h.id, h.slug, new_slug, raw_text or str(h.number)))
            
        h.slug = new_slug

    print(f"  Total hadiths: {len(hadiths)}")
    for item_id, old_s, new_s, title in sample_changes:
        print(f"  • ID {item_id:4d} | '{title[:30]}' -> {old_s} ➔ {new_s}")

    if not dry_run:
        with transaction.atomic():
            with connection.cursor() as cursor:
                cursor.execute("UPDATE hadis_hadis SET slug = CONCAT('tmp-h-', id);")
            Hadis.objects.bulk_update(hadiths, ['slug'], batch_size=1000)
        print("  ✅ All hadiths updated successfully.")


def update_transmitters_slugs(dry_run=False):
    print("\n👤 [3/3] Processing Transmitters...")
    transmitters = list(Transmitters.objects.all().order_by('id'))
    used_slugs = set()
    sample_changes = []

    for tr in transmitters:
        raw_text = extract_transmitter_text(tr)
        base_slug = clean_slug_text(raw_text, max_len=75)
        
        if not base_slug:
            base_slug = f"transmitter-{tr.id}"
            
        new_slug = generate_unique_slug(base_slug, used_slugs)
        
        if len(sample_changes) < 5 and tr.slug != new_slug:
            sample_changes.append((tr.id, tr.slug, new_slug, raw_text))
            
        tr.slug = new_slug

    print(f"  Total transmitters: {len(transmitters)}")
    for item_id, old_s, new_s, name in sample_changes:
        print(f"  • ID {item_id:4d} | '{name[:30]}' -> {old_s} ➔ {new_s}")

    if not dry_run:
        with transaction.atomic():
            with connection.cursor() as cursor:
                cursor.execute("UPDATE hadis_transmitters SET slug = CONCAT('tmp-t-', id);")
            Transmitters.objects.bulk_update(transmitters, ['slug'], batch_size=1000)
        print("  ✅ All transmitters updated successfully.")


def main():
    parser = argparse.ArgumentParser(description="Update readable slugs for Categories, Hadiths, and Transmitters.")
    parser.add_argument("--dry-run", action="store_true", help="Simulate slug generation without saving to database.")
    args = parser.parse_args()

    start_time = time.time()
    mode_text = "DRY RUN (Preview Only)" if args.dry_run else "EXECUTE (Database Update)"
    print(f"\n🚀 Starting Slug Optimization [{mode_text}]...")
    print("=" * 70)

    update_categories_slugs(dry_run=args.dry_run)
    update_hadis_slugs(dry_run=args.dry_run)
    update_transmitters_slugs(dry_run=args.dry_run)

    duration = time.time() - start_time
    print("\n" + "=" * 70)
    print(f"✨ Completed in {duration:.2f} seconds!")


if __name__ == "__main__":
    main()