Underground-Digital/Workflow-Engine

1import datetime2import logging3import time4 5import click6from celery import shared_task7from werkzeug.exceptions import NotFound8 9from core.indexing_runner import DocumentIsPausedError, IndexingRunner10from core.rag.extractor.notion_extractor import NotionExtractor11from core.rag.index_processor.index_processor_factory import IndexProcessorFactory12from extensions.ext_database import db13from models.dataset import Dataset, Document, DocumentSegment14from models.source import DataSourceOauthBinding15 16 17@shared_task(queue="dataset")18def document_indexing_sync_task(dataset_id: str, document_id: str):19    """20    Async update document21    :param dataset_id:22    :param document_id:23 24    Usage: document_indexing_sync_task.delay(dataset_id, document_id)25    """26    logging.info(click.style("Start sync document: {}".format(document_id), fg="green"))27    start_at = time.perf_counter()28 29    document = db.session.query(Document).filter(Document.id == document_id, Document.dataset_id == dataset_id).first()30 31    if not document:32        raise NotFound("Document not found")33 34    data_source_info = document.data_source_info_dict35    if document.data_source_type == "notion_import":36        if (37            not data_source_info38            or "notion_page_id" not in data_source_info39            or "notion_workspace_id" not in data_source_info40        ):41            raise ValueError("no notion page found")42        workspace_id = data_source_info["notion_workspace_id"]43        page_id = data_source_info["notion_page_id"]44        page_type = data_source_info["type"]45        page_edited_time = data_source_info["last_edited_time"]46        data_source_binding = DataSourceOauthBinding.query.filter(47            db.and_(48                DataSourceOauthBinding.tenant_id == document.tenant_id,49                DataSourceOauthBinding.provider == "notion",50                DataSourceOauthBinding.disabled == False,51                DataSourceOauthBinding.source_info["workspace_id"] == f'"{workspace_id}"',52            )53        ).first()54        if not data_source_binding:55            raise ValueError("Data source binding not found.")56 57        loader = NotionExtractor(58            notion_workspace_id=workspace_id,59            notion_obj_id=page_id,60            notion_page_type=page_type,61            notion_access_token=data_source_binding.access_token,62            tenant_id=document.tenant_id,63        )64 65        last_edited_time = loader.get_notion_last_edited_time()66 67        # check the page is updated68        if last_edited_time != page_edited_time:69            document.indexing_status = "parsing"70            document.processing_started_at = datetime.datetime.now(datetime.timezone.utc).replace(tzinfo=None)71            db.session.commit()72 73            # delete all document segment and index74            try:75                dataset = db.session.query(Dataset).filter(Dataset.id == dataset_id).first()76                if not dataset:77                    raise Exception("Dataset not found")78                index_type = document.doc_form79                index_processor = IndexProcessorFactory(index_type).init_index_processor()80 81                segments = db.session.query(DocumentSegment).filter(DocumentSegment.document_id == document_id).all()82                index_node_ids = [segment.index_node_id for segment in segments]83 84                # delete from vector index85                index_processor.clean(dataset, index_node_ids)86 87                for segment in segments:88                    db.session.delete(segment)89 90                end_at = time.perf_counter()91                logging.info(92                    click.style(93                        "Cleaned document when document update data source or process rule: {} latency: {}".format(94                            document_id, end_at - start_at95                        ),96                        fg="green",97                    )98                )99            except Exception:100                logging.exception("Cleaned document when document update data source or process rule failed")101 102            try:103                indexing_runner = IndexingRunner()104                indexing_runner.run([document])105                end_at = time.perf_counter()106                logging.info(107                    click.style("update document: {} latency: {}".format(document.id, end_at - start_at), fg="green")108                )109            except DocumentIsPausedError as ex:110                logging.info(click.style(str(ex), fg="yellow"))111            except Exception:112                pass113