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

import asyncio
import uuid

from fastapi import APIRouter, Depends, Header, HTTPException

from src.server.api_routes.knowledge.schemas import CrawlRequest
from src.server.config.logfire_config import safe_logfire_error, safe_logfire_info
from src.server.services.knowledge.knowledge_item_service import KnowledgeItemService
from src.server.utils import get_supabase_client
from src.server.utils.progress.progress_tracker import ProgressTracker

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

# CENTRAL TASK REGISTRY - Defined here to allow PEP8 compliant top-level imports in Facade
active_crawl_tasks: dict[str, asyncio.Task] = {}

router: APIRouter = APIRouter()


@router.get("/crawl-progress/{progress_id}")
async def get_crawl_progress(progress_id: str):
    """Get the current progress of a crawl or refresh operation. Public within system."""
    try:
        tracker = ProgressTracker(progress_id)
        status = tracker.get_state()
        if not status:
            raise HTTPException(status_code=404, detail="Progress ID not found")
        return status
    except HTTPException:
        raise
    except Exception as e:
        safe_logfire_error(f"Failed to get crawl progress {progress_id} | error={str(e)}")
        raise HTTPException(status_code=500, detail=str(e)) from e


@router.post("/knowledge-items/stop/{progress_id}")
async def stop_crawl_operation(progress_id: str, current_user: dict = Depends(requires_permission(TASK_CREATE))):
    """Stop an active crawl or refresh operation. Requires TASK_CREATE."""
    from src.server.services.crawling import unregister_orchestration

    try:
        safe_logfire_info(f"Stop crawl requested | progress_id={progress_id}")
        if progress_id in active_crawl_tasks:
            task = active_crawl_tasks[progress_id]
            task.cancel()
            del active_crawl_tasks[progress_id]

            tracker = ProgressTracker(progress_id)
            await tracker.update(status="stopped", progress=100, log="Operation stopped by user")
            return {"success": True, "message": f"Operation {progress_id} stopped"}
        else:
            unregister_orchestration(progress_id)
            return {"success": True, "message": f"Stop signal sent to {progress_id}"}
    except Exception as e:
        safe_logfire_error(f"Failed to stop operation {progress_id} | error={str(e)}")
        raise HTTPException(status_code=500, detail=str(e)) from e


@router.post("/knowledge-items/{source_id}/refresh")
async def refresh_knowledge_item(
    source_id: str,
    x_user_role: str | None = Header(None, alias="X-User-Role"),
    current_user: dict = Depends(requires_permission(TASK_CREATE)),
):
    """Refresh an existing knowledge item by re-crawling its source. Requires TASK_CREATE."""
    # LATE IMPORT to share the same physical registry with other modules
    from src.server.services.rbac_service import RBACService

    try:
        service = KnowledgeItemService(get_supabase_client())
        success, res = await service.get_item(source_id)

        if not success or not res.get("item"):
            raise HTTPException(status_code=404, detail=f"Knowledge item {source_id} not found")

        existing_item = res["item"]
        metadata = existing_item.get("metadata", {})
        url = existing_item.get("source_url")

        if not url:
            raise HTTPException(status_code=400, detail="Item has no source URL to refresh")

        rbac_service = RBACService()
        constraints = rbac_service.get_crawler_constraints(x_user_role)
        max_depth = min(metadata.get("max_depth", 2), constraints["max_depth"])

        progress_id = str(uuid.uuid4())
        tracker = ProgressTracker(progress_id)
        await tracker.update(
            status="initializing",
            progress=0,
            log=f"Starting refresh for {url} (Role: {x_user_role or 'unknown'})",
            source_id=source_id,
            operation="refresh",
            crawl_type="refresh",
        )

        from src.server.services.crawler_manager import get_crawler
        from src.server.services.crawling import CrawlOrchestrationService

        crawler = get_crawler()
        crawl_service = CrawlOrchestrationService(crawler=crawler, supabase_client=get_supabase_client())
        crawl_service.set_progress_id(progress_id)

        request_dict = {
            "url": url,
            "knowledge_type": metadata.get("knowledge_type", "technical"),
            "tags": metadata.get("tags", []),
            "max_depth": max_depth,
            "is_refresh": True,
            "source_id": source_id,
        }

        task = asyncio.create_task(crawl_service.orchestrate_crawl(request_dict))
        active_crawl_tasks[progress_id] = task
        safe_logfire_info(f"Refresh task created | progress_id={progress_id} | url={url}")

        return {"success": True, "progressId": progress_id, "message": "Refresh started"}

    except HTTPException:
        raise
    except Exception as e:
        safe_logfire_error(f"Failed to refresh item {source_id} | error={str(e)}")
        raise HTTPException(status_code=500, detail=str(e)) from e


@router.post("/knowledge-items/crawl")
async def crawl_knowledge_item(
    request: CrawlRequest,
    x_user_role: str | None = Header(None, alias="X-User-Role"),
    current_user: dict = Depends(requires_permission(TASK_CREATE)),
):
    """Start a new web crawl to populate the knowledge base. Requires TASK_CREATE."""
    # LATE IMPORT to share the same physical registry with other modules
    from src.server.services.rbac_service import RBACService

    try:
        rbac_service = RBACService()
        constraints = rbac_service.get_crawler_constraints(x_user_role)
        max_depth = min(request.max_depth, constraints["max_depth"])

        progress_id = str(uuid.uuid4())
        tracker = ProgressTracker(progress_id)
        await tracker.update(
            status="initializing",
            progress=0,
            log=f"Initializing crawl for {request.url}...",
            operation="crawl",
            crawl_type="new",
        )

        from src.server.services.crawler_manager import get_crawler
        from src.server.services.crawling import CrawlOrchestrationService

        crawler = get_crawler()
        orchestration_service = CrawlOrchestrationService(crawler, get_supabase_client())
        orchestration_service.set_progress_id(progress_id)

        request_dict = {
            "url": str(request.url),
            "knowledge_type": request.knowledge_type,
            "tags": request.tags,
            "max_depth": max_depth,
            "max_concurrent": constraints.get("max_concurrent", 3),
        }

        task = asyncio.create_task(orchestration_service.orchestrate_crawl(request_dict))
        active_crawl_tasks[progress_id] = task

        return {
            "success": True,
            "progressId": progress_id,
            "message": "Crawling started",
            "estimatedDuration": "3-5 minutes",
        }
    except Exception as e:
        safe_logfire_error(f"Failed to start crawl | error={str(e)}")
        raise HTTPException(status_code=500, detail=str(e)) from e