import os import gzip import json import sqlite3 import hashlib from pathlib import Path from django.core.management.base import BaseCommand from django.conf import settings from django.db import transaction from apps.hadis.models import ( Hadis, HadisCategory, HadisInterpretation, HadisCorrection, HadisDatabaseVersion ) from apps.hadis.serializers.hadis import ( DetailedCorrectionReferenceSerializer, DetailedInterpretationReferenceSerializer, ) class Command(BaseCommand): help = "Export Hadis and Categories into an optimized, compressed SQLite database for offline mobile sync." def add_arguments(self, parser): parser.add_argument( '--notes', type=str, default='', help='Release notes for this SQLite database version' ) parser.add_argument( '--force-version', type=int, default=None, help='Force a specific version number' ) def handle(self, *args, **options): self.stdout.write(self.style.SUCCESS("Starting Hadis SQLite database generation (Single-File Architecture)...")) # 1. Determine destination and version number latest_ver = HadisDatabaseVersion.objects.order_by('-version').first() if options['force_version'] is not None: new_version = options['force_version'] else: new_version = (latest_ver.version + 1) if latest_ver else 1 output_dir = Path(settings.MEDIA_ROOT) / 'hadis_db' output_dir.mkdir(parents=True, exist_ok=True) temp_db_path = output_dir / "temp_hadis_build.sqlite" final_gz_name = "hadis_offline.sqlite.gz" final_gz_path = output_dir / final_gz_name if temp_db_path.exists(): temp_db_path.unlink() # 2. Initialize SQLite database conn = sqlite3.connect(str(temp_db_path)) cursor = conn.cursor() # Optimize SQLite for bulk writes cursor.execute("PRAGMA synchronous = OFF;") cursor.execute("PRAGMA journal_mode = MEMORY;") cursor.execute("PRAGMA encoding = 'UTF-8';") self.stdout.write("Creating SQLite tables and indexes...") self._create_tables(cursor) # 3. Export data categories_count = self._export_categories(cursor) interp_count = self._export_category_interpretations(cursor) hadis_count, corr_count = self._export_hadiths_and_corrections(cursor) conn.commit() # Create indexes after insertion for maximum performance self.stdout.write("Building indexes...") self._create_indexes(cursor) conn.commit() conn.close() # 4. Gzip compress directly to the single target file self.stdout.write(f"Compressing database to single file: {final_gz_name}...") hasher = hashlib.md5() temp_gz_path = output_dir / "temp_hadis_build.sqlite.gz" with open(temp_db_path, 'rb') as f_in: with gzip.open(temp_gz_path, 'wb', compresslevel=9) as f_out: while chunk := f_in.read(1024 * 1024): f_out.write(chunk) hasher.update(chunk) checksum = hasher.hexdigest() temp_size = temp_db_path.stat().st_size # Atomic replacement: replace final_gz_path with temp_gz_path if final_gz_path.exists(): final_gz_path.unlink() temp_gz_path.rename(final_gz_path) file_size = final_gz_path.stat().st_size # Clean up any remaining temp files in output_dir temp_db_path.unlink(missing_ok=True) for extra_file in output_dir.iterdir(): if extra_file.name != final_gz_name: try: extra_file.unlink(missing_ok=True) except OSError: pass # 5. Record version in database (Single Active Record) with transaction.atomic(): # Delete old records to keep DB clean HadisDatabaseVersion.objects.all().delete() db_version = HadisDatabaseVersion.objects.create( version=new_version, file=f"hadis_db/{final_gz_name}", file_size=file_size, checksum_md5=checksum, hadis_count=hadis_count, categories_count=categories_count, is_active=True, notes=options['notes'], ) self.stdout.write(self.style.SUCCESS( f"Successfully updated single Hadis SQLite DB to v{new_version}!\n" f" - Uncompressed size: {temp_size / (1024*1024):.2f} MB\n" f" - Gzipped size: {file_size / (1024*1024):.2f} MB\n" f" - Hadiths: {hadis_count}\n" f" - Categories: {categories_count}\n" f" - Interpretations: {interp_count}\n" f" - Corrections: {corr_count}\n" f" - MD5 Checksum: {checksum}\n" f" - File path: {final_gz_path}" )) def _create_tables(self, cursor): cursor.executescript(""" CREATE TABLE IF NOT EXISTS categories ( id INTEGER PRIMARY KEY, title TEXT, slug TEXT UNIQUE, source_type TEXT, sect_type TEXT ); CREATE TABLE IF NOT EXISTS category_interpretations ( id INTEGER PRIMARY KEY, category_id INTEGER, title TEXT, slug TEXT, narrator TEXT, text TEXT, translation TEXT, references_json TEXT, links_json TEXT, FOREIGN KEY (category_id) REFERENCES categories (id) ); CREATE TABLE IF NOT EXISTS hadiths ( id INTEGER PRIMARY KEY, slug TEXT UNIQUE, category_id INTEGER, category_slug TEXT, title TEXT, title_narrator TEXT, text TEXT, translation_json TEXT, address TEXT, share_link TEXT, hadis_status_id INTEGER, status_title TEXT, status_color TEXT, status_main_color_code TEXT, status_text TEXT, tags_json TEXT, references_json TEXT, reference_images_json TEXT, narrators_json TEXT, explanations_json TEXT, FOREIGN KEY (category_id) REFERENCES categories (id) ); CREATE TABLE IF NOT EXISTS hadith_corrections ( id INTEGER PRIMARY KEY, hadis_id INTEGER, title TEXT, slug TEXT, narrator TEXT, description TEXT, translation TEXT, share_link TEXT, address TEXT, images_json TEXT, references_json TEXT, links_json TEXT, FOREIGN KEY (hadis_id) REFERENCES hadiths (id) ); """) def _create_indexes(self, cursor): cursor.executescript(""" CREATE INDEX IF NOT EXISTS idx_cat_slug ON categories(slug); CREATE INDEX IF NOT EXISTS idx_interp_cat ON category_interpretations(category_id); CREATE INDEX IF NOT EXISTS idx_hadith_cat ON hadiths(category_id); CREATE INDEX IF NOT EXISTS idx_hadith_slug ON hadiths(slug); CREATE INDEX IF NOT EXISTS idx_corr_hadis ON hadith_corrections(hadis_id); """) def _is_empty_localized_data(self, val): if not val: return True if isinstance(val, (list, tuple)): if len(val) == 0: return True for item in val: if isinstance(item, dict): txt = item.get('text') or item.get('title') or item.get('description') or '' if str(txt).strip(): return False elif str(item).strip(): return False return True if isinstance(val, dict): if not val: return True for k, v in val.items(): if str(v).strip(): return False return True if isinstance(val, str) and not val.strip(): return True return False def _export_categories(self, cursor): cats = HadisCategory.objects.all() rows = [ ( c.id, json.dumps(c.title, ensure_ascii=False) if isinstance(c.title, (dict, list)) else (c.title or ''), c.slug, getattr(c, 'source_type', None), getattr(c, 'sect_type', None), ) for c in cats ] cursor.executemany( "INSERT INTO categories (id, title, slug, source_type, sect_type) VALUES (?, ?, ?, ?, ?)", rows ) return len(rows) def _make_full_url(self, url): if not url: return '' if url.startswith('http://') or url.startswith('https://'): return url if not url.startswith('/'): url = '/' + url return f'https://dovodi.newhorizonco.uk{url}' def _serialize_reference(self, ref): book = getattr(ref, 'book_reference', None) author = book.author if book and getattr(book, 'author', None) else None ed = getattr(ref, 'edition', None) vol_obj = getattr(ref, 'book_volume', None) images_list = [] if hasattr(ref, 'images'): for img in ref.images.all(): if img.image: img_url = self._make_full_url(img.image.url) thumb_url = self._make_full_url(img.thumbnail.url if getattr(img, 'thumbnail', None) else img.image.url) images_list.append({ 'id': img.id, 'image': img_url, 'thumbnail': thumb_url, 'priority': getattr(img, 'priority', 0), }) book_dict = { 'id': book.id, 'slug': book.slug, 'title': book.title, 'publisher': getattr(book, 'publisher', '') or '', 'year_of_publication': getattr(book, 'year_of_publication', '') or '', 'number_of_volumes': getattr(book, 'number_of_volumes', None), } if book else None author_dict = { 'id': author.id, 'slug': author.slug, 'name': author.name, } if author else None edition_dict = { 'id': ed.id, 'edition_number': ed.edition_number, 'publisher': getattr(ed, 'publisher', '') or '', 'year_of_publication': getattr(ed, 'year_of_publication', '') or '', 'number_of_volumes': getattr(ed, 'number_of_volumes', None), } if ed else None volume_info = { 'id': vol_obj.id, 'title': vol_obj.title, } if vol_obj else None open_book_url = None if getattr(ref, 'url', None): open_book_url = ref.url elif vol_obj and getattr(vol_obj, 'file', None): open_book_url = self._make_full_url(vol_obj.file.url) elif ed and getattr(ed, 'source_url', None): open_book_url = ed.source_url elif book and getattr(book, 'slug', None): open_book_url = f"/arguments/sources/{book.slug}" parts = [] vol_val = getattr(ref, 'volume', None) or (vol_obj.title if vol_obj else None) if vol_val: v_str = str(vol_val).strip() parts.append(v_str if v_str.startswith("جلد") else f"جلد {v_str}") pg_val = getattr(ref, 'pages', None) if pg_val: p_str = str(pg_val).strip() parts.append(p_str if (p_str.startswith("ص") or p_str.startswith("صفحه")) else f"صفحه {p_str}") num_val = getattr(ref, 'hadith_number', None) if num_val: n_str = str(num_val).strip() parts.append(n_str if (n_str.startswith("ح") or n_str.startswith("حدیث") or n_str.startswith("شماره")) else f"شماره حدیث {n_str}") citation_summary = " • ".join(parts) if parts else "" return { 'id': ref.id, 'book': book_dict, 'author': author_dict, 'edition': edition_dict, 'volume_info': volume_info, 'hadith_number': getattr(ref, 'hadith_number', None), 'volume': vol_val, 'pages': pg_val, 'address': getattr(ref, 'address', None), 'url': getattr(ref, 'url', None), 'open_book_url': open_book_url, 'citation_summary': citation_summary, 'images': images_list, 'images_count': len(images_list), 'book_title': getattr(book, 'title', None) if book else None, 'book_authors': [{'id': author.id, 'name': author.name, 'slug': author.slug}] if author else [], 'edition_info': getattr(ed, 'publisher', None) if ed else None, } def _export_category_interpretations(self, cursor): interps = HadisInterpretation.objects.select_related('category').prefetch_related( 'references__book_reference__author', 'references__images', 'references__edition', 'references__book_volume', ).all() rows = [] for inp in interps: refs_list = [self._serialize_reference(ref) for ref in inp.references.all()] rows.append(( inp.id, inp.category_id, json.dumps(inp.title, ensure_ascii=False) if isinstance(inp.title, (dict, list)) else (inp.title or ''), inp.slug, inp.narrator or '', inp.text or '', json.dumps(inp.translation, ensure_ascii=False) if isinstance(inp.translation, (dict, list)) else (inp.translation or ''), json.dumps(refs_list, ensure_ascii=False), json.dumps(inp.links or [], ensure_ascii=False), )) cursor.executemany( """INSERT INTO category_interpretations (id, category_id, title, slug, narrator, text, translation, references_json, links_json) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?)""", rows ) return len(rows) def _export_hadiths_and_corrections(self, cursor): hadiths = ( Hadis.objects .filter(status=True) .select_related('category', 'hadis_status') .prefetch_related( 'tags', 'references__book_reference__author', 'references__images', 'transmitters__transmitter__reliability', 'transmitters__uncertain_transmitters__reliability', 'transmitters__narrator_layer', 'hadiscorrection_set__references__book_reference__author', 'hadiscorrection_set__references__edition', 'hadiscorrection_set__references__book_volume', 'hadiscorrection_set__references__images', ) .order_by('id') ) hadis_rows = [] correction_rows = [] for h in hadiths: # 1. Status Block status_id = h.hadis_status.id if h.hadis_status else None status_title = h.hadis_status.title if h.hadis_status else None status_color = h.hadis_status.color if h.hadis_status else None status_code = h.hadis_status.main_color_code if h.hadis_status else None # 2. Tags tags_list = [{'id': t.id, 'title': t.title} for t in h.tags.all()] # 3. References & Reference Images refs_list = [] ref_imgs_list = [] for ref in h.references.all(): book = ref.book_reference authors = [{'id': book.author.id, 'name': book.author.name}] if book and book.author else [] refs_list.append({ 'id': ref.id, 'title': book.title if book else None, 'authors': authors, 'share_link': getattr(book, 'share_link', None) if book else None, }) for img in ref.images.all(): ref_imgs_list.append({ 'id': img.id, 'thumbnail': self._make_full_url(img.thumbnail.url if img.thumbnail else None), 'priority': img.priority, }) # 4. Narrators transmitters_data = [] for tr in h.transmitters.all(): t = tr.transmitter rel = t.reliability if t else None transmitters_data.append({ 'id': t.id if t else None, 'name': t.full_name if t else None, 'slug': t.slug if t else None, 'layer_id': tr.narrator_layer_id, 'is_uncertain': tr.is_uncertain, 'reliability': { 'id': rel.id, 'title': rel.title, 'slug': rel.slug, 'color': rel.color, 'main_color_code': rel.main_color_code, } if rel else None, }) # 5. Explanations explanations_data = [] if hasattr(h, 'explanations') and h.explanations: for exp in h.explanations.all(): explanations_data.append({ 'title': exp.title, 'detail': getattr(exp, 'detail', '') }) # Status description fallback: use hadis_status_text if present and non-empty, otherwise fallback to hadis_status.description status_desc_raw = h.hadis_status_text if self._is_empty_localized_data(status_desc_raw) and h.hadis_status: status_desc_raw = h.hadis_status.description status_text_val = json.dumps(status_desc_raw, ensure_ascii=False) if isinstance(status_desc_raw, (dict, list)) else (status_desc_raw or '') hadis_rows.append(( h.id, h.slug, h.category_id, h.category.slug if h.category else None, json.dumps(h.title, ensure_ascii=False) if isinstance(h.title, (dict, list)) else (h.title or ''), json.dumps(h.title_narrator, ensure_ascii=False) if isinstance(h.title_narrator, (dict, list)) else (h.title_narrator or ''), h.text or '', json.dumps(h.translation or [], ensure_ascii=False), json.dumps(h.address, ensure_ascii=False) if isinstance(h.address, (dict, list)) else (h.address or ''), h.share_link or '', status_id, json.dumps(status_title, ensure_ascii=False) if isinstance(status_title, (dict, list)) else (status_title or ''), status_color or '', status_code or '', status_text_val, json.dumps(tags_list, ensure_ascii=False), json.dumps(refs_list, ensure_ascii=False), json.dumps(ref_imgs_list, ensure_ascii=False), json.dumps({'transmitters': transmitters_data}, ensure_ascii=False), json.dumps(explanations_data, ensure_ascii=False), )) # 6. Corrections for corr in h.hadiscorrection_set.all(): corr_imgs = [] corr_refs = [] for cref in corr.references.all(): ref_data = self._serialize_reference(cref) corr_refs.append(ref_data) if ref_data.get('images'): corr_imgs.extend(ref_data['images']) corr_address = corr_refs[0].get('address') or '' if corr_refs else '' correction_rows.append(( corr.id, h.id, json.dumps(corr.title, ensure_ascii=False) if isinstance(corr.title, (dict, list)) else (corr.title or ''), corr.slug, corr.narrator or '', corr.text or '', json.dumps(corr.translation, ensure_ascii=False) if isinstance(corr.translation, (dict, list)) else (corr.translation or ''), corr.share_link or '', corr_address, json.dumps(corr_imgs, ensure_ascii=False), json.dumps(corr_refs, ensure_ascii=False), json.dumps(corr.links or [], ensure_ascii=False), )) cursor.executemany( """INSERT INTO hadiths ( id, slug, category_id, category_slug, title, title_narrator, text, translation_json, address, share_link, hadis_status_id, status_title, status_color, status_main_color_code, status_text, tags_json, references_json, reference_images_json, narrators_json, explanations_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", hadis_rows ) cursor.executemany( """INSERT INTO hadith_corrections ( id, hadis_id, title, slug, narrator, description, translation, share_link, address, images_json, references_json, links_json ) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)""", correction_rows ) return len(hadis_rows), len(correction_rows)