def create_new_bot()

in packages/blueprints/gen-ai-chatbot/static-assets/chatbot-genai-components/backend/python/app/usecases/bot.py [0:0]

• 121 lines of code
• 17 McCabe index (conditional complexity)

def create_new_bot(user_id: str, bot_input: BotInput) -> BotOutput:
    """Create a new bot.
    Bot is created as private and not pinned.
    """
    current_time = get_current_time()
    has_knowledge = bot_input.knowledge and (
        len(bot_input.knowledge.source_urls) > 0
        or len(bot_input.knowledge.sitemap_urls) > 0
        or len(bot_input.knowledge.filenames) > 0
    )
    sync_status: type_sync_status = "QUEUED" if has_knowledge else "SUCCEEDED"

    source_urls = []
    sitemap_urls = []
    filenames = []
    if bot_input.knowledge:
        source_urls = bot_input.knowledge.source_urls
        sitemap_urls = bot_input.knowledge.sitemap_urls

        # Commit changes to S3
        _update_s3_documents_by_diff(
            user_id, bot_input.id, bot_input.knowledge.filenames, []
        )
        # Delete files from upload temp directory
        delete_files_with_prefix_from_s3(
            DOCUMENT_BUCKET, compose_upload_temp_s3_prefix(user_id, bot_input.id)
        )
        filenames = bot_input.knowledge.filenames

    chunk_size = (
        bot_input.embedding_params.chunk_size
        if bot_input.embedding_params
        else DEFAULT_EMBEDDING_CONFIG["chunk_size"]
    )

    chunk_overlap = (
        bot_input.embedding_params.chunk_overlap
        if bot_input.embedding_params
        else DEFAULT_EMBEDDING_CONFIG["chunk_overlap"]
    )

    enable_partition_pdf = (
        bot_input.embedding_params.enable_partition_pdf
        if bot_input.embedding_params
        else DEFAULT_EMBEDDING_CONFIG["enable_partition_pdf"]
    )

    generation_params = (
        bot_input.generation_params.model_dump()
        if bot_input.generation_params
        else DEFAULT_GENERATION_CONFIG
    )

    search_params = (
        bot_input.search_params.model_dump()
        if bot_input.search_params
        else DEFAULT_SEARCH_CONFIG
    )

    agent = (
        AgentModel(
            tools=[
                AgentToolModel(name=t.name, description=t.description)
                for t in [
                    get_tool_by_name(tool_name) for tool_name in bot_input.agent.tools
                ]
            ]
        )
        if bot_input.agent
        else AgentModel(tools=[])
    )

    store_bot(
        user_id,
        BotModel(
            id=bot_input.id,
            title=bot_input.title,
            description=bot_input.description if bot_input.description else "",
            instruction=bot_input.instruction,
            create_time=current_time,
            last_used_time=current_time,
            public_bot_id=None,
            is_pinned=False,
            owner_user_id=user_id,  # Owner is the creator
            embedding_params=EmbeddingParamsModel(
                chunk_size=chunk_size,
                chunk_overlap=chunk_overlap,
                enable_partition_pdf=enable_partition_pdf,
            ),
            generation_params=GenerationParamsModel(**generation_params),  # type: ignore
            search_params=SearchParamsModel(**search_params),
            agent=agent,
            knowledge=KnowledgeModel(
                source_urls=source_urls, sitemap_urls=sitemap_urls, filenames=filenames
            ),
            sync_status=sync_status,
            sync_status_reason="",
            sync_last_exec_id="",
            published_api_stack_name=None,
            published_api_datetime=None,
            published_api_codebuild_id=None,
            display_retrieved_chunks=bot_input.display_retrieved_chunks,
        ),
    )
    return BotOutput(
        id=bot_input.id,
        title=bot_input.title,
        instruction=bot_input.instruction,
        description=bot_input.description if bot_input.description else "",
        create_time=current_time,
        last_used_time=current_time,
        is_public=False,
        is_pinned=False,
        owned=True,
        embedding_params=EmbeddingParams(
            chunk_size=chunk_size,
            chunk_overlap=chunk_overlap,
            enable_partition_pdf=enable_partition_pdf,
        ),
        generation_params=GenerationParams(**generation_params),
        search_params=SearchParams(**search_params),
        agent=Agent(
            tools=[
                AgentTool(name=tool.name, description=tool.description)
                for tool in agent.tools
            ]
        ),
        knowledge=Knowledge(
            source_urls=source_urls, sitemap_urls=sitemap_urls, filenames=filenames
        ),
        sync_status=sync_status,
        sync_status_reason="",
        sync_last_exec_id="",
        display_retrieved_chunks=bot_input.display_retrieved_chunks,
    )