feat(hadis): add category title deduplication script

scripts/deduplicate_category_titles.py

@@ -0,0 +1,144 @@
+import os
+import sys
+import copy
+import argparse
+from collections import defaultdict
+
+# Configure UTF-8 for Windows console
+sys.stdout.reconfigure(encoding="utf-8")
+sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__))))
+os.environ.setdefault("DJANGO_SETTINGS_MODULE", "config.settings.develop")
+
+import django
+django.setup()
+
+from django.db import transaction
+from apps.hadis.models import HadisCategory
+from utils.text_normalizer import build_search_blob
+
+
+def deduplicate_category_titles(dry_run=True):
+    print("=" * 70)
+    print("           HADIS CATEGORY TITLE DEDUPLICATION SCRIPT           ")
+    print(f" Mode: {'DRY RUN (Simulation only, no DB changes)' if dry_run else 'EXECUTE (Applying changes to DB)'}")
+    print("=" * 70)
+
+    categories = list(HadisCategory.objects.all().order_by("id"))
+    total_categories = len(categories)
+    print(f"\nLoaded {total_categories} categories from database.\n")
+
+    # Map: lang -> normalized_title -> list of occurrences
+    # each occurrence: {"cat_id": cat.id, "item_idx": idx, "original_val": val, "cat_obj": cat}
+    lang_title_map = defaultdict(lambda: defaultdict(list))
+
+    for cat in categories:
+        if not isinstance(cat.title, list):
+            continue
+        for idx, item in enumerate(cat.title):
+            if not isinstance(item, dict):
+                continue
+            l_code = item.get("language_code")
+            val = item.get("title") or item.get("text") or ""
+            val = val.strip()
+            if l_code and val:
+                lang_title_map[l_code][val.lower()].append({
+                    "cat_id": cat.id,
+                    "item_idx": idx,
+                    "original_val": val,
+                    "cat_obj": cat,
+                })
+
+    # Prepare changes
+    # cat_id -> {"cat_obj": cat, "title_copy": [...], "changes": [{"lang": l, "old": o, "new": n}]}
+    pending_updates = {}
+
+    for cat in categories:
+        pending_updates[cat.id] = {
+            "cat_obj": cat,
+            "title_copy": copy.deepcopy(cat.title) if isinstance(cat.title, list) else cat.title,
+            "changes": [],
+        }
+
+    total_duplicate_groups = 0
+
+    print("--- Scanning for duplicate titles by language ---")
+    for l_code in sorted(lang_title_map.keys()):
+        title_dict = lang_title_map[l_code]
+        dup_groups = {k: v for k, v in title_dict.items() if len(v) > 1}
+        print(f"  Language '{l_code}': {len(dup_groups)} duplicate title groups")
+        total_duplicate_groups += len(dup_groups)
+
+        for lower_val, occ_list in dup_groups.items():
+            # Sort occurrences deterministically by category ID
+            occ_list.sort(key=lambda x: x["cat_id"])
+            for rank, occ in enumerate(occ_list, start=1):
+                cat_id = occ["cat_id"]
+                idx = occ["item_idx"]
+                orig_val = occ["original_val"]
+                new_val = f"{orig_val} {rank}"
+
+                entry = pending_updates[cat_id]
+                title_item = entry["title_copy"][idx]
+
+                if "title" in title_item:
+                    title_item["title"] = new_val
+                if "text" in title_item:
+                    title_item["text"] = new_val
+                if "value" in title_item:
+                    title_item["value"] = new_val
+
+                entry["changes"].append({
+                    "lang": l_code,
+                    "old": orig_val,
+                    "new": new_val,
+                })
+
+    # Filter only categories that actually have changes
+    cats_to_update = [info for info in pending_updates.values() if len(info["changes"]) > 0]
+
+    print(f"\nFound {len(cats_to_update)} categories out of {total_categories} requiring title deduplication.")
+
+    if not cats_to_update:
+        print("✅ No duplicate category titles found. Everything is already unique.")
+        return
+
+    print("\n--- Detailed List of Updates ---")
+    for info in cats_to_update:
+        cat = info["cat_obj"]
+        changes_str = ", ".join(f"[{c['lang']}] '{c['old']}' -> '{c['new']}'" for c in info["changes"])
+        print(f"  Category #{cat.id} (slug: {cat.slug}, source: {cat.source_type}): {changes_str}")
+
+    if dry_run:
+        print("\n" + "=" * 70)
+        print("ℹ️ DRY RUN COMPLETED. No database changes were made.")
+        print("To apply changes to the database, run with: python scripts/deduplicate_category_titles.py --execute")
+        print("=" * 70)
+        return
+
+    print("\nApplying updates to database...")
+    with transaction.atomic():
+        updated_count = 0
+        for info in cats_to_update:
+            cat = info["cat_obj"]
+            cat.title = info["title_copy"]
+            cat.normalized_text = build_search_blob(cat.title, cat.description)
+            cat.save(update_fields=["title", "normalized_text"])
+            updated_count += 1
+
+    print("\n" + "=" * 70)
+    print(f"✅ SUCCESS: Successfully updated {updated_count} categories in database.")
+    print("All duplicate titles have been resolved with incremental numbers.")
+    print("=" * 70)
+
+
+if __name__ == "__main__":
+    parser = argparse.ArgumentParser(description="Deduplicate HadisCategory titles across languages.")
+    parser.add_argument(
+        "--execute",
+        "--apply",
+        action="store_true",
+        dest="execute",
+        help="Apply changes to the database (default is dry-run).",
+    )
+    args = parser.parse_args()
+    deduplicate_category_titles(dry_run=not args.execute)