diff --git a/lib/crewai/src/crewai/flow/flow_visualizer.py b/lib/crewai/src/crewai/flow/flow_visualizer.py index d928377e2..d49f2cf34 100644 --- a/lib/crewai/src/crewai/flow/flow_visualizer.py +++ b/lib/crewai/src/crewai/flow/flow_visualizer.py @@ -2,7 +2,7 @@ from __future__ import annotations import os -from typing import TYPE_CHECKING +from typing import TYPE_CHECKING, Any from pyvis.network import Network # type: ignore[import-untyped] @@ -29,7 +29,7 @@ _printer = Printer() class FlowPlot: """Handles the creation and rendering of flow visualization diagrams.""" - def __init__(self, flow: Flow) -> None: + def __init__(self, flow: Flow[Any]) -> None: """ Initialize FlowPlot with a flow object. @@ -136,7 +136,7 @@ class FlowPlot: f"Unexpected error during flow visualization: {e!s}" ) from e finally: - self._cleanup_pyvis_lib() + self._cleanup_pyvis_lib(filename) def _generate_final_html(self, network_html: str) -> str: """ @@ -186,26 +186,33 @@ class FlowPlot: raise IOError(f"Failed to generate visualization HTML: {e!s}") from e @staticmethod - def _cleanup_pyvis_lib() -> None: + def _cleanup_pyvis_lib(filename: str) -> None: """ Clean up the generated lib folder from pyvis. This method safely removes the temporary lib directory created by pyvis - during network visualization generation. + during network visualization generation. The lib folder is created in the + same directory as the output HTML file. + + Parameters + ---------- + filename : str + The output filename (without .html extension) used for the visualization. """ try: - lib_folder = safe_path_join("lib", root=os.getcwd()) - if os.path.exists(lib_folder) and os.path.isdir(lib_folder): - import shutil + import shutil - shutil.rmtree(lib_folder) - except ValueError as e: - _printer.print(f"Error validating lib folder path: {e}", color="red") + output_dir = os.path.dirname(os.path.abspath(filename)) or os.getcwd() + lib_folder = os.path.join(output_dir, "lib") + if os.path.exists(lib_folder) and os.path.isdir(lib_folder): + vis_js = os.path.join(lib_folder, "vis-network.min.js") + if os.path.exists(vis_js): + shutil.rmtree(lib_folder) except Exception as e: _printer.print(f"Error cleaning up lib folder: {e}", color="red") -def plot_flow(flow: Flow, filename: str = "flow_plot") -> None: +def plot_flow(flow: Flow[Any], filename: str = "flow_plot") -> None: """ Convenience function to create and save a flow visualization. diff --git a/lib/crewai/src/crewai/flow/html_template_handler.py b/lib/crewai/src/crewai/flow/html_template_handler.py index 55567393c..9218d1ae9 100644 --- a/lib/crewai/src/crewai/flow/html_template_handler.py +++ b/lib/crewai/src/crewai/flow/html_template_handler.py @@ -1,5 +1,8 @@ +"""HTML template processing and generation for flow visualization diagrams.""" + import base64 import re +from typing import Any from crewai.flow.path_utils import validate_path_exists @@ -7,7 +10,7 @@ from crewai.flow.path_utils import validate_path_exists class HTMLTemplateHandler: """Handles HTML template processing and generation for flow visualization diagrams.""" - def __init__(self, template_path, logo_path): + def __init__(self, template_path: str, logo_path: str) -> None: """ Initialize HTMLTemplateHandler with validated template and logo paths. @@ -29,23 +32,23 @@ class HTMLTemplateHandler: except ValueError as e: raise ValueError(f"Invalid template or logo path: {e}") from e - def read_template(self): + def read_template(self) -> str: """Read and return the HTML template file contents.""" with open(self.template_path, "r", encoding="utf-8") as f: return f.read() - def encode_logo(self): + def encode_logo(self) -> str: """Convert the logo SVG file to base64 encoded string.""" with open(self.logo_path, "rb") as logo_file: logo_svg_data = logo_file.read() return base64.b64encode(logo_svg_data).decode("utf-8") - def extract_body_content(self, html): + def extract_body_content(self, html: str) -> str: """Extract and return content between body tags from HTML string.""" match = re.search("