diff --git a/scripts/deduplicate_category_titles.py b/scripts/deduplicate_category_titles.py new file mode 100644 index 0000000..9e080d7 --- /dev/null +++ b/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)