Underground-Digital/Workflow-Engine

1import datetime2import logging3import time4 5import click6from celery import shared_task7from werkzeug.exceptions import NotFound8 9from core.rag.index_processor.index_processor_factory import IndexProcessorFactory10from core.rag.models.document import Document11from extensions.ext_database import db12from extensions.ext_redis import redis_client13from models.dataset import Document as DatasetDocument14from models.dataset import DocumentSegment15 16 17@shared_task(queue="dataset")18def add_document_to_index_task(dataset_document_id: str):19    """20    Async Add document to index21    :param document_id:22 23    Usage: add_document_to_index.delay(document_id)24    """25    logging.info(click.style("Start add document to index: {}".format(dataset_document_id), fg="green"))26    start_at = time.perf_counter()27 28    dataset_document = db.session.query(DatasetDocument).filter(DatasetDocument.id == dataset_document_id).first()29    if not dataset_document:30        raise NotFound("Document not found")31 32    if dataset_document.indexing_status != "completed":33        return34 35    indexing_cache_key = "document_{}_indexing".format(dataset_document.id)36 37    try:38        segments = (39            db.session.query(DocumentSegment)40            .filter(DocumentSegment.document_id == dataset_document.id, DocumentSegment.enabled == True)41            .order_by(DocumentSegment.position.asc())42            .all()43        )44 45        documents = []46        for segment in segments:47            document = Document(48                page_content=segment.content,49                metadata={50                    "doc_id": segment.index_node_id,51                    "doc_hash": segment.index_node_hash,52                    "document_id": segment.document_id,53                    "dataset_id": segment.dataset_id,54                },55            )56 57            documents.append(document)58 59        dataset = dataset_document.dataset60 61        if not dataset:62            raise Exception("Document has no dataset")63 64        index_type = dataset.doc_form65        index_processor = IndexProcessorFactory(index_type).init_index_processor()66        index_processor.load(dataset, documents)67 68        end_at = time.perf_counter()69        logging.info(70            click.style(71                "Document added to index: {} latency: {}".format(dataset_document.id, end_at - start_at), fg="green"72            )73        )74    except Exception as e:75        logging.exception("add document to index failed")76        dataset_document.enabled = False77        dataset_document.disabled_at = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)78        dataset_document.status = "error"79        dataset_document.error = str(e)80        db.session.commit()81    finally:82        redis_client.delete(indexing_cache_key)83