Flying docs (#200)

* Adds basic procedure logic

* Adds alembic version

* Adds max_file_size

* Adds archived column

* Adds new env var id column

* comment

* Redo of alembic file

* changes route

* PR comments

* Adds cascade logic to existing alembic revision

* Submodule update

Author: JWittmeyer

alembic/versions/889ef4df126f_adds_tmp_file_indicators.py

@@ -0,0 +1,44 @@
+"""Adds tmp file indicators
+
+Revision ID: 889ef4df126f
+Revises: 4861b97fcd5d
+Create Date: 2024-04-10 14:31:06.210778
+
+"""
+from alembic import op
+import sqlalchemy as sa
+from sqlalchemy.dialects import postgresql
+
+# revision identifiers, used by Alembic.
+revision = '889ef4df126f'
+down_revision = '4861b97fcd5d'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.add_column('conversation', sa.Column('has_tmp_files', sa.Boolean(), nullable=True), schema='cognition')
+    op.add_column('conversation', sa.Column('archived', sa.Boolean(), nullable=True), schema='cognition')
+    op.add_column('project', sa.Column('allow_file_upload', sa.Boolean(), nullable=True), schema='cognition')
+    op.add_column('project', sa.Column('max_file_size_mb', sa.Float(), nullable=True), schema='cognition')
+    op.add_column('project', sa.Column('open_ai_env_var_id', postgresql.UUID(as_uuid=True), nullable=True), schema='cognition')
+    op.create_index(op.f('ix_cognition_project_open_ai_env_var_id'), 'project', ['open_ai_env_var_id'], unique=False, schema='cognition')
+    op.create_foreign_key(None, 'project', 'environment_variable', ['open_ai_env_var_id'], ['id'], source_schema='cognition', referent_schema='cognition', ondelete='SET NULL')    
+    op.drop_constraint('message_strategy_id_fkey', 'message', schema='cognition', type_='foreignkey')
+    op.create_foreign_key('message_strategy_id_fkey', 'message', 'strategy', ['strategy_id'], ['id'], source_schema='cognition', referent_schema='cognition', ondelete='SET NULL')
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_constraint(None, 'project', schema='cognition', type_='foreignkey')
+    op.drop_index(op.f('ix_cognition_project_open_ai_env_var_id'), table_name='project', schema='cognition')
+    op.drop_column('project', 'open_ai_env_var_id', schema='cognition')
+    op.drop_column('project', 'max_file_size_mb', schema='cognition')
+    op.drop_column('project', 'allow_file_upload', schema='cognition')
+    op.drop_column('conversation', 'archived', schema='cognition')
+    op.drop_column('conversation', 'has_tmp_files', schema='cognition')
+    op.drop_constraint('message_strategy_id_fkey', 'message', schema='cognition', type_='foreignkey')
+    op.create_foreign_key('message_strategy_id_fkey', 'message', 'strategy', ['strategy_id'], ['id'], source_schema='cognition', referent_schema='cognition', ondelete='CASCADE')
+    # ### end Alembic commands ###

api/transfer.py

@@ -37,6 +37,7 @@
 from submodules.model.enums import NotificationType
 from submodules.model.models import UploadTask
 from util import daemon, notification
+from controller.transfer.cognition.minio_upload import handle_cognition_file_upload
 
 from controller.task_queue import manager as task_queue_manager
 from submodules.model.enums import TaskType, RecordTokenizationScope
@@ -51,12 +52,18 @@ async def post(self, request) -> PlainTextResponse:
         data = await request.json()
         file_path = data["Key"]
 
-        if len(file_path.split("/")) != 4:
+        parts = file_path.split("/")
+
+        if parts[1] == "_cognition":
+            handle_cognition_file_upload(parts)
+            return PlainTextResponse("OK")
+
+        if len(parts) != 4:
             # We need handling for lf execution notification here.
             # ATM we have a different path of handling in util/payload_scheduler.py update_records method
             return PlainTextResponse("OK")
 
-        org_id, project_id, upload_task_id, file_name = file_path.split("/")
+        org_id, project_id, upload_task_id, file_name = parts
         if len(project_id) != 36:
             return PlainTextResponse("OK")
         if upload_task_id == "download":

controller/task_queue/handler/parse_cognition_tmp_file.py

@@ -0,0 +1,41 @@
+from typing import Any, Dict, Tuple, Callable
+import os
+import submodules.s3.controller as s3
+
+import requests
+from submodules.model.business_objects import (
+    task_queue as task_queue_db_bo,
+    general,
+)
+
+BASE_URI = os.getenv("COGNITION_GATEWAY")
+
+
+def get_task_functions() -> Tuple[Callable, Callable, int]:
+    return __start_task, __check_finished, 1
+
+
+def __start_task(task: Dict[str, Any]) -> bool:
+    # check task still relevant
+    task_db_obj = task_queue_db_bo.get(task["id"])
+    if task_db_obj is None or task_db_obj.is_active:
+        return False
+
+    task_db_obj.is_active = True
+    general.commit()
+
+    action = task["task_info"]
+    conversation_id = action["conversation_id"]
+    cognition_project_id = action["cognition_project_id"]
+    requests.post(
+        f"{BASE_URI}/api/v1/converters/internal/projects/{cognition_project_id}/conversation/{conversation_id}/parse-tmp-file",
+        json={"minio_path": action["minio_path"], "bucket": action["bucket"]},
+    )
+    return True
+
+
+def __check_finished(task: Dict[str, Any]) -> bool:
+
+    action = task["task_info"]
+
+    return not s3.object_exists(action["bucket"], action["minio_path"])

controller/task_queue/manager.py

@@ -17,6 +17,7 @@
     attribute_calculation as attribute_calculation_handler,
     task_queue as task_queue_handler,
     markdown_file as markdown_file_handler,
+    parse_cognition_tmp_file as cognition_tmp_file,
 )
 from .util import if_task_queue_send_websocket
 
@@ -90,6 +91,8 @@ def get_task_function_by_type(task_type: str) -> Tuple[Callable, Callable, int]:
         return task_queue_handler.get_task_functions()
     if task_type == enums.TaskType.PARSE_MARKDOWN_FILE.value:
         return markdown_file_handler.get_task_functions()
+    if task_type == enums.TaskType.PARSE_COGNITION_TMP_FILE.value:
+        return cognition_tmp_file.get_task_functions()
     raise ValueError(f"Task type {task_type} not supported yet")
 
 

controller/transfer/cognition/minio_upload.py

@@ -0,0 +1,35 @@
+from typing import List
+from submodules.model.cognition_objects import project as cognition_project
+from submodules.model.cognition_objects import conversation
+from submodules.model.enums import TaskType
+from controller.task_queue import manager as task_queue_manager
+
+
+def handle_cognition_file_upload(path_parts: List[str]):
+
+    if path_parts[1] != "_cognition":
+        return
+
+    if path_parts[3] == "chat_tmp_files" and path_parts[5] == "queued":
+        cognition_project_id = path_parts[2]
+        conversation_id = path_parts[4]
+        cognition_prj = cognition_project.get(cognition_project_id)
+        if not cognition_prj:
+            return
+        project_id = str(cognition_prj.refinery_references_project_id)
+        conversation_item = conversation.get(cognition_project_id, conversation_id)
+        if not conversation_item:
+            return
+
+        task_queue_manager.add_task(
+            project_id,
+            TaskType.PARSE_COGNITION_TMP_FILE,
+            conversation_item.created_by,
+            {
+                "cognition_project_id": cognition_project_id,
+                "conversation_id": conversation_id,
+                "minio_path": "/".join(path_parts[1:]),
+                "bucket": path_parts[0],
+            },
+            True,  # not sure if prio is right here as the prio tasks should only take < 1 min but waiting for the normal queue will take ages depending on the queue
+        )