python/src/server/api_routes/knowledge/upload.py

import asyncio
import json
import uuid

from fastapi import APIRouter, Depends, File, Form, Header, HTTPException, UploadFile

from src.server.config.logfire_config import get_logger, safe_logfire_error

# TOP-LEVEL IMPORTS for Test Patching
from src.server.services.source_management_service import SourceManagementService
from src.server.services.storage.storage_services import DocumentStorageService
from src.server.utils.document_processing import extract_text_from_document
from src.server.utils.progress.progress_tracker import ProgressTracker

from ...auth.dependencies import requires_permission
from ...auth.permissions import TASK_CREATE

# Domain State
from . import active_crawl_tasks

router = APIRouter()
logger = get_logger(__name__)


@router.post("/documents/upload")
async def upload_document(
    file: UploadFile = File(...),
    knowledge_type: str = Form("technical"),
    tags: str = Form("[]"),
    x_user_role: str | None = Header(None, alias="X-User-Role"),
    current_user: dict = Depends(requires_permission(TASK_CREATE)),
):
    """Upload a document and process it into knowledge chunks. Requires TASK_CREATE."""
    try:
        tag_list = json.loads(tags)
        progress_id = str(uuid.uuid4())

        file_content = await file.read()
        # Capturing metadata early to avoid competition in background task
        filename = file.filename or "uploaded_document"
        content_type = file.content_type or "application/octet-stream"
        file_metadata = {"filename": filename, "content_type": content_type, "size": len(file_content)}

        tracker = ProgressTracker(progress_id)
        await tracker.start({"status": "initializing", "progress": 0, "log": f"Starting upload for {filename}"})

        # Launch background task and store it for tracking (Physical Alignment with domain state)
        task = asyncio.create_task(
            background_upload(
                file_content=file_content,
                file_metadata=file_metadata,
                progress_id=progress_id,
                tags=tag_list,
                knowledge_type=knowledge_type,
                tracker=tracker,
            )
        )

        active_crawl_tasks[progress_id] = task

        return {"status": "success", "progress_id": progress_id, "message": "Upload started in background"}

    except Exception as e:
        safe_logfire_error(f"Upload initialization failed: {e}")
        raise HTTPException(status_code=500, detail=str(e)) from e


async def background_upload(
    file_content: bytes,
    file_metadata: dict,
    progress_id: str,
    tags: list[str],
    knowledge_type: str,
    tracker: ProgressTracker,
):
    """Background task for processing uploaded documents."""
    try:
        # Use top-level imported services
        storage = DocumentStorageService()
        source_manager = SourceManagementService()

        await tracker.update("extracting", 20, "Extracting text...")

        text = extract_text_from_document(file_content, file_metadata["filename"], file_metadata["content_type"])

        if not text:
            raise ValueError("No text could be extracted from the document.")

        await tracker.update("storing", 50, "Storing document...")

        # Simple implementation for seeding/upload
        source_id = f"upload-{file_metadata['filename']}-{uuid.uuid4().hex[:6]}"
        await source_manager.create_source_info(
            source_id=source_id,
            content_sample=text[:200],  # Required argument
            word_count=len(text.split()),
            knowledge_type=knowledge_type,
            tags=tags,
        )

        await storage.store_documents(
            documents=[
                {
                    "source_id": source_id,
                    "content": text,
                    "filename": file_metadata["filename"],  # Physical alignment: must be at top level
                    "metadata": file_metadata,
                }
            ]
        )

        await tracker.complete({"log": "Upload successful!"})
    except Exception as e:
        safe_logfire_error(f"Background upload failed for {file_metadata.get('filename')}: {e}")
        await tracker.update("failed", 0, f"Error: {str(e)}")
    finally:
        if progress_id in active_crawl_tasks:
            del active_crawl_tasks[progress_id]