Flow

def generate(
    self,
    merged_result: ExportResult,
    is_async: bool,
    after_run: str,
    **kwargs: Any,
) -> str:
    """Generate the complete flow notebook content.

    Parameters
    ----------
    merged_result : ExportResult
        The merged export result containing all content.
    is_async : bool
        Whether to generate async conten
    after_run : str
        Additional content to add after the main flow execution.
    **kwargs : Any
        Additional keyword arguments for the generator.

    Returns
    -------
    str
        The complete flow notebook content.

    Raises
    ------
    ExporterContentError
        If there is no content to export.
    """
    # 1. Generate header
    header = self.get_header(merged_result)

    # 2. Generate imports
    imports_section = merged_result.get_content_by_position(
        ExportPosition.IMPORTS
    )

    # 3. Generate content sections
    tools_section = merged_result.get_content_by_position(
        ExportPosition.TOOLS
    )
    models_section = merged_result.get_content_by_position(
        ExportPosition.MODELS
    )
    agents_section = merged_result.get_content_by_position(
        ExportPosition.AGENTS,
        # Skip agent arguments (should already be there)
        skip_agent_arguments=True,
    )
    chats_content = merged_result.get_content_by_position(
        ExportPosition.CHATS
    )
    if not chats_content:
        raise ExporterContentError(
            "No content to export. Please ensure that the flow has chats."
        )
    after_chats = merged_result.get_content_by_position(
        ExportPosition.BOTTOM
    )

    main, call_main, execution_block = self._get_execution_content(
        chats_content=chats_content,
        is_async=is_async,
        after_run=after_run,
        for_notebook=self.config.for_notebook,
    )

    # 5. Combine everything
    everything: list[str] = [header]
    if imports_section:
        comment = get_comment(
            "Imports",
            for_notebook=self.config.for_notebook,
        )
        everything.append(comment)
        everything.append(
            "\n".join([entry.content for entry in imports_section])
        )
    if tools_section:
        comment = get_comment(
            "Tools",
            for_notebook=self.config.for_notebook,
        )
        everything.append(comment)
        everything.append(
            "\n".join([entry.content for entry in tools_section]) + "\n"
        )
    if models_section:
        comment = get_comment(
            "Models",
            for_notebook=self.config.for_notebook,
        )
        everything.append(comment)
        everything.append(
            "\n\n".join([entry.content for entry in models_section]) + "\n"
        )
    if agents_section:
        comment = get_comment(
            "Agents",
            for_notebook=self.config.for_notebook,
        )
        if self.config.for_notebook:
            comment += "# pyright: reportUnnecessaryIsInstance=false\n"
        everything.append(comment)
        everything.append(
            "\n\n".join([entry.content for entry in agents_section]) + "\n"
        )
    everything.append(main)
    if after_chats:
        everything.append(
            "\n".join([entry.content for entry in after_chats])
        )
    everything.append(call_main)
    if execution_block:
        everything.append(execution_block)

    return "\n".join(everything)

def get_header(self, merged_result: ExportResult) -> str:
    """Get or generate the header for the script.

    Parameters
    ----------
    merged_result : ExportResult
        The merged export result containing all content.

    Returns
    -------
    str
        The header content.
    """
    from_result = merged_result.get_content_by_position(ExportPosition.TOP)
    if not from_result:
        return generate_header(
            name=self.config.name,
            description=self.config.description,
            requirements=self.config.requirements,
            tags=self.config.tags,
            for_notebook=self.config.for_notebook,
        )
    header_string = "\n".join(content.content for content in from_result)
    while not header_string.endswith("\n\n"):
        header_string += "\n"
    return header_string

def orchestrate(self) -> ExportResult:
    """Orchestrate the export process.

    Returns
    -------
    ExportResult
        The result of the export process,
        containing the generated script and any additional metadata.
    """
    results: list[ExportResult] = []
    agent_arguments: dict[str, list[str]] = {}

    # 1. Tools first (needed by agents)
    if self.waldiez.tools:
        self.logger.info("Exporting tools ...")
        tools_result = self._get_tools_exporter().export()
        # Extract tool arguments for agents
        tool_arguments = self._extract_agent_arguments_from_result(
            tools_result
        )
        # Merge tool arguments into agent arguments
        self._merge_agent_arguments(
            source=tool_arguments,
            target=agent_arguments,
        )
        results.append(tools_result)
        self.logger.debug("Exported %s", tools_result)

    # 2. Models second (needed by agents)
    if self.waldiez.models:
        self.logger.info("Exporting models ...")
        models_result = self._get_models_exporter().export()
        # Extract model arguments for agents
        model_arguments = self._extract_agent_arguments_from_result(
            models_result
        )
        # Merge model arguments into agent arguments
        self._merge_agent_arguments(
            source=model_arguments,
            target=agent_arguments,
        )
        results.append(models_result)
        self.logger.debug("Exported %s", models_result)
    # 3. Chats third (agents might need agent chat registrations)
    # we always have at least one chat (already validated in Waldiez init)
    self.logger.info("Exporting chats ...")
    chats_result = self._get_chats_exporter().export()
    self.logger.debug("Exported %s", chats_result)
    # Extract chat arguments for agents
    chat_arguments = self._extract_agent_arguments_from_result(chats_result)
    # Merge chat arguments into agent arguments
    self._merge_agent_arguments(
        source=chat_arguments,
        target=agent_arguments,
    )
    results.append(chats_result)

    # 4. Agents last
    # we always have at least one agent (already validated in Waldiez init)
    agent_results = self._export_all_agents(agent_arguments)
    results.extend(agent_results)

    # 5. Merge everything
    merger = ContentMerger(self.context)
    merged_result = merger.merge_results(results)
    # Check for issues
    stats = merger.get_merge_statistics()
    if stats.conflicts_found:
        self.logger.info(
            "Resolved %d merge conflicts", len(stats.conflicts_found)
        )
    self.logger.debug("Merged result: %s", merged_result)
    return self._finalize_export(merged_result)

def get_after_run_content(self) -> str:
    """Get the content to be executed after the main flow run.

    Returns
    -------
    str
        The content to be executed after the main flow run.
    """
    return get_after_run_content(
        waldiez=self.waldiez,
        agent_names=self.agent_names,
        tabs=1,
    )

def merge_results(self, results: list[ExportResult]) -> ExportResult:
    """Merge multiple export results into one.

    Parameters
    ----------
    results : list[ExportResult]
        The export results to merge

    Returns
    -------
    ExportResult
        The merged export result
    """
    if not results:
        return ExportResult()

    if len(results) == 1:
        return results[0]

    # Initialize statistics
    self.statistics = MergeStatistics(total_results=len(results))

    # Create merged result
    merged = ExportResult()

    # 1. Merge imports with intelligent deduplication
    merged.imports = self._merge_imports(results)

    # 2. Merge positioned content with proper ordering
    merged.positioned_content = self._merge_positioned_content(results)

    # 3. Merge environment variables with conflict detection
    merged.environment_variables = self._merge_environment_variables(
        results
    )

    # 4. Merge validation results
    merged.validation_result = self._merge_validation_results(results)

    # 5. Handle main content (typically empty for merged results)
    merged.main_content = self._merge_main_content(results)

    return merged

def get_merge_statistics(self) -> MergeStatistics:
    """Get statistics about the last merge operation.

    Returns
    -------
    MergeStatistics
        Statistics about the merge operation
    """
    return self.statistics

def generate_main_content(self) -> str:
    """Generate the main content of the export.

    Returns
    -------
    str
        The final executable script or notebook content.
    """
    orchestrator = ExportOrchestrator(
        waldiez=self.waldiez,
        context=self.context,
    )
    merged_result = orchestrator.orchestrate()
    after_run = orchestrator.get_after_run_content()
    generator = FileGenerator(
        context=self.context,
    )
    return generator.generate(
        merged_result=merged_result,
        is_async=self.waldiez.is_async,
        after_run=after_run,
    )

@staticmethod
def generate(
    content: str,
    is_async: bool,
    for_notebook: bool,
    cache_seed: int | None,
    after_run: str,
) -> str:
    """Generate the complete flow script content.

    Parameters
    ----------
    content : str
        The content of the chats to be included in the main function.
    is_async : bool
        Whether to generate async content.
    for_notebook : bool
        Whether the export is intended for a notebook environment.
    cache_seed : str | int | None
        The cache seed to use for flow chat if any.
    after_run : str, optional
        Additional content to add after the main chat execution,
        by default ""

    Returns
    -------
    str
        The complete flow script content.
    """
    main_function = ExecutionGenerator.generate_main_function(
        content=content,
        is_async=is_async,
        cache_seed=cache_seed,
        after_run=after_run,
        for_notebook=for_notebook,
    )
    call_main_function = ExecutionGenerator.generate_call_main_function(
        is_async=is_async,
        for_notebook=for_notebook,
    )
    if not for_notebook:
        execution_block = ExecutionGenerator.generate_execution_block(
            is_async=is_async,
        )
    else:
        execution_block = ""
    return (
        "\n".join([main_function, call_main_function, execution_block])
        + "\n"
    )

@staticmethod
def generate_main_function(
    content: str,
    is_async: bool,
    cache_seed: int | None,
    after_run: str,
    for_notebook: bool,
) -> str:
    """Generate the main function for the flow script.

    Parameters
    ----------
    content : str
        The content of the chats to be included in the main function.
    is_async : bool
        Whether to generate async content
    cache_seed : str | int | None
        The cache seed to use for flow chat if any
    after_run : str
        Additional content to add after the main chat execution.
    for_notebook : bool
        Whether the export is intended for a notebook environment.

    Returns
    -------
    str
        The complete main function content.
    """
    if content.startswith("\n"):
        content = content[1:]
    flow_content = "\n\n"
    comment = get_comment(
        "Start chatting",
        for_notebook=for_notebook,
    )
    flow_content += f"{comment}\n"
    if is_async:
        flow_content += "async "

    flow_content += f"def main() -> {RETURN_TYPE_HINT}:\n"
    flow_content += f"    {main_doc_string()}\n"
    space = "    "
    if cache_seed is not None:
        flow_content += (
            f"    with Cache.disk(cache_seed={cache_seed}"
            ") as cache:  # pyright: ignore\n"
        )
        space = f"{space}    "
    flow_content += f"{content}" + "\n"
    if is_async:
        flow_content += f"{space}await stop_logging()"
    else:
        flow_content += f"{space}stop_logging()"
    flow_content += "\n"
    if after_run:
        flow_content += after_run + "\n"
    if cache_seed is not None:
        space = space[4:]
    flow_content += f"{space}return results\n"
    return flow_content

@staticmethod
def generate_call_main_function(is_async: bool, for_notebook: bool) -> str:
    """Generate the call_main function for the flow script.

    Parameters
    ----------
    is_async : bool
        Whether to generate async content
    for_notebook : bool
        Whether the export is intended for a notebook environment.

    Returns
    -------
    str
        The complete call_main function content.
    """
    content = "\n"
    if for_notebook:
        if is_async:
            return "# %%\nawait main()\n"
        return "# %%\nmain()\n"
    if is_async:
        content += "async def call_main() -> None:\n"
    else:
        content += "def call_main() -> None:\n"
    content += '    """Run the main function and print the results."""\n'
    content += f"    results: {RETURN_TYPE_HINT} = "
    if is_async:
        content += "await "
    content += "main()\n"
    content += "    if isinstance(results, dict):\n"
    content += "        # order by key\n"
    content += "        ordered_results = dict(sorted(results.items()))\n"
    content += "        for _, result in ordered_results.items():\n"
    content += "            pprint(asdict(result))\n"
    content += "    else:\n"
    content += "        if not isinstance(results, list):\n"
    content += "            results = [results]\n"
    content += "        for result in results:\n"
    content += "            pprint(asdict(result))\n"
    content += "\n"
    return content

@staticmethod
def generate_execution_block(is_async: bool) -> str:
    """Generate the execution block for the main function.

    Parameters
    ----------
    is_async : bool
        Whether to generate async content

    Returns
    -------
    str
        The complete if __name__ == "__main__": block content
    """
    comment = get_comment(
        "Let's go!",
        for_notebook=False,
    )
    content = 'if __name__ == "__main__":\n'
    content += f"    {comment}"
    if is_async:
        content += "    anyio.run(call_main)\n"
    else:
        content += "    call_main()\n"
    return content