working version of document ingestion

Author: poornimaramesh

core_backend/app/__init__.py

@@ -16,6 +16,7 @@
     contents,
     dashboard,
     data_api,
+    docmuncher,
     question_answer,
     tags,
     urgency_detection,
@@ -161,6 +162,7 @@ def create_app() -> FastAPI:
     app.include_router(contents.router)
     app.include_router(dashboard.router)
     app.include_router(data_api.router)
+    app.include_router(docmuncher.router)
     app.include_router(question_answer.router)
     app.include_router(tags.router)
     app.include_router(urgency_detection.router)

core_backend/app/docmuncher/__init__.py

@@ -0,0 +1,15 @@
+"""Package initialization for the FastAPI application.
+
+This module imports and exposes key components required for API routing, including the
+main FastAPI router and metadata tags used for API documentation.
+
+Exports:
+    - `router`: The main FastAPI APIRouter instance containing all route definitions.
+    - `TAG_METADATA`: Metadata describing API tags for better documentation.
+
+These components can be imported directly from the package for use in the application.
+"""
+
+from .routers import TAG_METADATA, router
+
+__all__ = ["router", "TAG_METADATA"]

core_backend/app/docmuncher/dependencies.py

@@ -149,28 +149,34 @@ async def convert_markdown_chunks_to_cards(
         If the conversion fails.
     """
     for header_split in md_header_splits:
-        try:
-            card = ContentCreate(
-                content_metadata=header_split.metadata,
-                context_text=header_split.page_content,
-                context_title="--".join(
+        num_sub_chunks = int(len(header_split.page_content) / 2000 + 1)
+        for i in range(num_sub_chunks):
+            try:
+                title = "--".join(
                     [str(v) for v in header_split.metadata.values()]
                     + [header_split.page_content[:10]]
-                ),
-                context_tags=[tag_id],
-            )
-            await save_content_to_db(
-                asession=asession,
-                content=card,
-                exclude_archived=True,
-                workspace_id=workspace_id,
-            )
-        except Exception as e:
-            # TODO: this is a dumb way to handle errors in card creation
-            raise HTTPException(
-                status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
-                detail=f"Failed to process PDF file: {e}",
-            ) from e
+                )
+                metadata = header_split.metadata
+                metadata["sub_chunk"] = i
+
+                card = ContentCreate(
+                    content_text=header_split.page_content[i * 2000 : (i + 1) * 2000],
+                    content_title=title,
+                    content_metadata=metadata,
+                    context_tags=[tag_id],
+                )
+                await save_content_to_db(
+                    asession=asession,
+                    content=card,
+                    exclude_archived=True,
+                    workspace_id=workspace_id,
+                )
+            except Exception as e:
+                # TODO: this is a dumb way to handle errors in card creation
+                raise HTTPException(
+                    status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+                    detail=f"Failed to process PDF file: {e}",
+                ) from e
     return {"detail": "Cards saved successfully"}
 
 
@@ -207,15 +213,18 @@ async def process_pdf_file(
     """
     # Update redis state operations
     redis = request.app.state.redis
-    job_status = redis.get(task_id)
-    if job_status is None:
+    job_status = await redis.get(task_id)
+    if not job_status:
         raise HTTPException(
             status_code=status.HTTP_404_NOT_FOUND,
             detail="Job not found",
         )
-    job_status = DocUploadResponse.model_validate(job_status)
-    job_status.status = DocStatusEnum.in_progress
-    redis.set(task_id, job_status.model_dump_json())
+
+    job_status_pydantic = DocUploadResponse.model_validate(
+        json.loads(job_status.decode("utf-8"))
+    )
+    job_status_pydantic.status = DocStatusEnum.in_progress
+    await redis.set(task_id, job_status_pydantic.model_dump_json())
 
     # Process PDF
     try:
@@ -229,14 +238,14 @@ async def process_pdf_file(
         )
 
     except Exception as e:
-        job_status.status = DocStatusEnum.failed
-        redis.set(task_id, job_status.model_dump_json())
+        job_status_pydantic.status = DocStatusEnum.failed
+        await redis.set(task_id, job_status_pydantic.model_dump_json())
         raise HTTPException(
             status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
             detail=f"Failed to process PDF file: {e}",
         ) from e
-    finally:
-        job_status.status = DocStatusEnum.success
-        redis.set(task_id, job_status.model_dump_json())
+
+    job_status_pydantic.status = DocStatusEnum.success
+    await redis.set(task_id, job_status_pydantic.model_dump_json())
 
     return job_status

core_backend/app/docmuncher/routers.py

@@ -3,7 +3,6 @@
 
 from fastapi import (
     APIRouter,
-    BackgroundTasks,
     Depends,
     File,
     HTTPException,
@@ -34,18 +33,18 @@
     "description": "_Requires user login._ Document management to create content",
 }
 
-router = APIRouter(prefix="/content", tags=[TAG_METADATA["name"]])
+router = APIRouter(prefix="/docmuncher", tags=[TAG_METADATA["name"]])
 logger = setup_logger()
 
 
 @router.post("/upload", response_model=DocUploadResponse)
 async def upload_document(
     request: Request,
+    # background_tasks: BackgroundTasks,
     file: Annotated[UploadFile, File(...)],
     calling_user_db: Annotated[UserDB, Depends(get_current_user)],
     workspace_name: Annotated[str, Depends(get_current_workspace_name)],
     asession: AsyncSession = Depends(get_async_session),
-    background_tasks: BackgroundTasks = Depends(),
 ) -> DocUploadResponse:
     """Upload document to create content.
 
@@ -119,26 +118,34 @@ async def upload_document(
         created_datetime_utc=created_datetime_utc,
         status=DocStatusEnum.not_started,
     )
-    redis.set(task_id, task_status.model_dump_json())
+    await redis.set(task_id, task_status.model_dump_json())
 
     # Start background task
-    background_tasks.add_task(
-        process_pdf_file,
+    await process_pdf_file(
         request=request,
         task_id=task_id,
         file=file,
         tag_id=tag_db.tag_id,
         workspace_id=workspace_db.workspace_id,
         asession=asession,
     )
+    # background_tasks.add_task(
+    #     process_pdf_file,
+    #     request=request,
+    #     task_id=task_id,
+    #     file=file,
+    #     tag_id=tag_db.tag_id,
+    #     workspace_id=workspace_db.workspace_id,
+    #     asession=asession,
+    # )
 
     return task_status
 
 
 @router.get("/status", response_model=DocUploadResponse)
 async def get_doc_ingestion_status(
     request: Request,
-    ingestion_job_id: str,
+    ingestion_job_id: int,
     calling_user_db: Annotated[UserDB, Depends(get_current_user)],
     workspace_name: Annotated[str, Depends(get_current_workspace_name)],
     asession: AsyncSession = Depends(get_async_session),

core_backend/app/docmuncher/schemas.py

@@ -16,7 +16,6 @@ class DocStatusEnum(str, Enum):
 class DocUploadResponse(BaseModel):
     """Pydantic model for document upload response."""
 
-    doc_id: int
     doc_name: str
     ingestion_job_id: int
     created_datetime_utc: datetime