From 6c4cc467189561fb4232a0fd194cf822631ac643 Mon Sep 17 00:00:00 2001 From: henrikek Date: Tue, 13 Oct 2026 18:29:58 +0200 Subject: [PATCH] Improved indexing progress feedback --- ESSArch_Core/cli/commands/search.py | 10 ---- ESSArch_Core/ip/models.py | 52 +++++++++++++++++++-- ESSArch_Core/ip/tasks.py | 6 ++- ESSArch_Core/search/ingest.py | 71 +++++++++++++++++++++-------- 4 files changed, 104 insertions(+), 35 deletions(-) diff --git a/ESSArch_Core/cli/commands/search.py b/ESSArch_Core/cli/commands/search.py index 8e35aad06..fb454b97d 100644 --- a/ESSArch_Core/cli/commands/search.py +++ b/ESSArch_Core/cli/commands/search.py @@ -85,16 +85,6 @@ def rebuild(indexes, batch_size, remove_stale, do_not_delete_old, index_file_con For the document index, file content extraction is disabled by default. Use --index-file-content to enable content extraction from the original files during the rebuild. - - :param indexes: Names of the indices to rebuild. Multiple indices can - be specified. - :param batch_size: Number of documents to process in each batch. - :param remove_stale: Whether to remove indexed documents that no longer - exist in the database. - :param do_not_delete_old: Whether to keep the existing index before - rebuilding. - :param index_file_content: Whether to extract and index content from - original files in the document index. """ indexes = get_indexes(indexes) diff --git a/ESSArch_Core/ip/models.py b/ESSArch_Core/ip/models.py index 87fd174de..916f32913 100644 --- a/ESSArch_Core/ip/models.py +++ b/ESSArch_Core/ip/models.py @@ -2804,6 +2804,7 @@ def write_to_search_index(self, task): :param task: The task associated with the indexing operation """ logger = logging.getLogger('essarch.ip') + task.set_progress(1, total=100) srcdir = self.object_path ct_profile = self.get_profile('content_type') indexed_files = [] @@ -2812,6 +2813,7 @@ def write_to_search_index(self, task): profile_type = self.get_package_type_display().lower() index_cits = self.get_profile_data(profile_type).get('index_cits', True) + total_cits_imported = 0 if ct_profile is not None and (index_cits is True or index_cits == 'True'): cts = self.get_content_type_file() if cts is not None: @@ -2822,8 +2824,11 @@ def write_to_search_index(self, task): except KeyError: logger.exception('No content type importer specified in profile') raise - ct_importer = get_importer(ct_importer_name)(task) + ct_importer = get_importer(ct_importer_name)(task.get_processtask()) indexed_files = ct_importer.import_content(cts, ip=self) + if indexed_files is not None: + total_cits_imported = len(indexed_files) + task.create_success_event("Content type specification imported") else: err = "Content type specification file not found" logger.error('{err}: {path}'.format(err=err, path=cts)) @@ -2835,8 +2840,12 @@ def write_to_search_index(self, task): except KeyError: logger.exception('No content type importer specified in profile') raise - ct_importer = get_importer(ct_importer_name)(task) + ct_importer = get_importer(ct_importer_name)(task.get_processtask()) indexed_files = ct_importer.import_content(cts, ip=self) + if indexed_files is not None: + total_cits_imported = len(indexed_files) + task.create_success_event("Content type specification imported") + task.set_progress(2, total=100) # Profile settings control file indexing and content extraction. index_files = self.get_profile_data(profile_type).get('index_files', True) @@ -2844,6 +2853,10 @@ def write_to_search_index(self, task): # Reserved for optionally reusing content from existing Elasticsearch documents. index_files_content_reuse_if_exists = False + + total_documents_to_index = 0 + total_dirs = 0 + if index_files is True or index_files == 'True': group = None try: @@ -2857,6 +2870,7 @@ def write_to_search_index(self, task): for d in dirs: src = os.path.join(root, d) index_path(self, src, group=group, index_file_content=index_files_content) + total_dirs += 1 for f in files: src = os.path.join(root, f) @@ -2878,16 +2892,22 @@ def write_to_search_index(self, task): file_ids.append(str(tag_version.pk)) file_paths.append((src, tag_version)) + task.set_progress(3, total=100) existing_content = {} if index_files_content_reuse_if_exists: + task.set_progress(4, total=100) existing_content = get_existing_file_content(file_ids) + task.set_progress(5, total=100) documents = [] + document_paths = {} + total_files_to_process = len(file_paths) - for src, tag_version in file_paths: + for processed_files, (src, tag_version) in enumerate(file_paths, start=1): content = None if index_files_content_reuse_if_exists: content = existing_content.get(str(tag_version.pk)) + doc, _ = index_document( tag_version, src, @@ -2902,13 +2922,37 @@ def write_to_search_index(self, task): group.add_object(tag_version) documents.append(doc) + document_paths[str(doc.meta.id)] = src + + if total_files_to_process: + # Reserve 10–70% for file processing and content extraction. + progress = 10 + int(processed_files / total_files_to_process * 60) + task.set_progress(progress, total=100) # Index all collected file documents in a single bulk operation # instead of sending an individual Elasticsearch request per file. - bulk_index_documents(documents) + total_documents_to_index = len(documents) + + def report_progress(processed, succeeded, failed): + if total_documents_to_index: + progress = 70 + int(processed / total_documents_to_index * 30) + task.set_progress(progress, total=100) + + succeeded, failed = bulk_index_documents( + documents, + progress_callback=report_progress, + document_paths=document_paths, + ) + + task.create_success_event( + f"Indexed {succeeded} files successfully; " + f"{failed} files failed" + ) InformationPackageDocument.from_obj(self).save() + return total_documents_to_index, total_dirs, total_cits_imported + def get_cached_storage_object(self): cache_method = self.policy.cache_storage return self.storage.get( diff --git a/ESSArch_Core/ip/tasks.py b/ESSArch_Core/ip/tasks.py index 1d73b9c1f..7e9fd4b2a 100644 --- a/ESSArch_Core/ip/tasks.py +++ b/ESSArch_Core/ip/tasks.py @@ -582,10 +582,12 @@ def PreserveInformationPackage(self, storage_method_pk, temp_path=None): pretty_mb_per_sec(mb_per_sec)) -@app.task(bind=True) +@app.task(bind=True, event_type=50770) def WriteInformationPackageToSearchIndex(self): ip = self.get_information_package() - ip.write_to_search_index(self.get_processtask()) + files_count, dirs_count, cits_count = ip.write_to_search_index(self) + + return f"Indexed {files_count} files, {dirs_count} dirs and {cits_count} cits" @app.task(bind=True) diff --git a/ESSArch_Core/search/ingest.py b/ESSArch_Core/search/ingest.py index 1693b94ba..11145f901 100644 --- a/ESSArch_Core/search/ingest.py +++ b/ESSArch_Core/search/ingest.py @@ -4,7 +4,7 @@ from django.conf import settings from elasticsearch import ConnectionError, RequestError, TransportError -from elasticsearch.helpers import streaming_bulk +from elasticsearch.helpers import BulkIndexError, streaming_bulk from elasticsearch_dsl.connections import get_connection from ESSArch_Core.fixity.format import FormatIdentifier @@ -24,10 +24,54 @@ format_identifier = FormatIdentifier() -def bulk_index_documents(documents, batch_size=100): +def bulk_index_documents(documents, batch_size=100, progress_callback=None, document_paths=None): client = get_connection('default') + logger = logging.getLogger('essarch.search.ingest') + document_paths = document_paths or {} actions = [] + succeeded = 0 + failed = 0 + processed = 0 + + def index_batch(batch): + nonlocal succeeded, failed, processed + + try: + for success, info in streaming_bulk( + client, + batch, + chunk_size=batch_size, + raise_on_error=True, + ): + if success: + succeeded += 1 + else: + failed += 1 + logger.error( + 'Failed to bulk index document: %s', info + ) + + processed += 1 + + if progress_callback: + progress_callback(processed, succeeded, failed) + + except BulkIndexError as exc: + for error in exc.errors: + operation = next(iter(error)) + details = error[operation] + document_id = details.get('_id') + path = document_paths.get(document_id, '') + + logger.error( + 'Failed to index file "%s" (document ID: %s): %s', + path, + document_id, + details.get('error'), + ) + + raise for doc in documents: actions.append({ @@ -37,26 +81,13 @@ def bulk_index_documents(documents, batch_size=100): }) if len(actions) >= batch_size: - for success, info in streaming_bulk( - client, - actions, - chunk_size=batch_size, - ): - if not success: - logger = logging.getLogger('essarch.search.ingest') - logger.error('Failed to bulk index document: %s', info) - + index_batch(actions) actions = [] if actions: - for success, info in streaming_bulk( - client, - actions, - chunk_size=batch_size, - ): - if not success: - logger = logging.getLogger('essarch.search.ingest') - logger.error('Failed to bulk index document: %s', info) + index_batch(actions) + + return succeeded, failed def index_document(tag_version, filepath, index_file_content=True, existing_content=None, save=True): @@ -256,6 +287,8 @@ def index_path(ip, path, parent=None, group=None, index_file_content=True, logger.debug('indexing {}'.format(path)) + doc = None + if index: if isfile: doc, tag_version = index_document(tag_version, path, index_file_content,