report

@dataclass
class PublishPluginReportInfo:
    """Information about single plugin in publish process."""
    id: str
    name: str
    label: str
    order: float
    filepath: str
    docstring: str | None
    plugin_type: Literal["instance", "context"]
    families: list[str]
    targets: list[str]
    skipped: bool = False
    passed: bool = False
    process_reports: list[PublishProcessReport] = field(
        default_factory=list
    )
    actions_reports: list[PublishPluginActionReport] = field(
        default_factory=list
    )

    def to_data(self, current_plugin_id: str | None = None) -> dict[str, Any]:
        passed = self.passed
        if current_plugin_id == self.id:
            passed = True
        return {
            "id": self.id,
            "name": self.name,
            "label": self.label,
            "order": self.order,
            "filepath": self.filepath,
            "docstring": self.docstring,
            "plugin_type": self.plugin_type,
            "families": self.families,
            "targets": self.targets,
            "instances_data": [r.to_data() for r in self.process_reports],
            "actions_data": [r.to_data() for r in self.actions_reports],
            "skipped": self.skipped,
            "passed": passed,
        }

    @classmethod
    def from_data(cls, data: dict[str, Any]) -> PublishPluginReportInfo:
        return cls(
            id=data["id"],
            name=data["name"],
            label=data["label"],
            order=data["order"],
            filepath=data["filepath"],
            docstring=data.get("docstring"),
            plugin_type=data["plugin_type"],
            families=data["families"],
            targets=data["targets"],
            skipped=data.get("skipped", False),
            passed=data.get("passed", False),
            process_reports=[
                PublishProcessReport.from_data(r)
                for r in data.get("instances_data", [])
            ],
            actions_reports=[
                PublishPluginActionReport.from_data(action_data)
                for action_data in data["actions_data"]
            ],
        )

    @classmethod
    def from_plugin(
        cls, plugin: pyblish.api.Plugin
    ) -> PublishPluginReportInfo:
        label = None
        if hasattr(plugin, "label"):
            label = plugin.label

        if not label:
            label = plugin.__name__

        plugin_type = "instance" if plugin.__instanceEnabled__ else "context"
        # Get docstring
        # NOTE we do care only about docstring from the plugin so we can't
        #   use 'inspect.getdoc' which also looks for docstring in parent
        #   classes.
        docstring = getattr(plugin, "__doc__", None)
        if docstring:
            docstring = inspect.cleandoc(docstring)
        return cls(
            id=plugin.id,
            name=plugin.__name__,
            label=label,
            order=plugin.order,
            filepath=inspect.getfile(plugin),
            docstring=docstring,
            plugin_type=plugin_type,
            families=list(plugin.families),
            targets=list(plugin.targets),
        )

class PublishReportMaker:
    """Report for single publishing process.

    Report keeps current state of publishing and currently processed plugin.
    """
    def __init__(
        self,
        publish_plugins: list[pyblish.api.Plugin] | None = None,
        crashed_filepaths: dict[str, str] | None = None,
        blocking_crashed_paths: set[str] | None = None,
    ) -> None:
        if publish_plugins is None:
            publish_plugins = []

        if crashed_filepaths is None:
            crashed_filepaths = {}

        if blocking_crashed_paths is None:
            blocking_crashed_paths = set()

        # Make sure plugins are sorted
        # TODO try to comment this to find out if it is necessary...
        publish_plugins.sort(key=lambda p: p.order)

        # Track information if instances have been updated since
        #   last report generation.
        self._instances_updated: bool = False

        # Current plugin id which is being processed.
        self._current_plugin_id: str | None = None

        self._plugin_info_by_id: dict[str, PublishPluginReportInfo] = {}
        self._all_instances_by_id: dict[str, pyblish.api.Instance] = {}
        self._report = PublishReport(
            crashed_filepaths=crashed_filepaths,
            blocking_crashed_paths=list(blocking_crashed_paths),
        )
        self._prepare_publish_plugin_items(publish_plugins)

    @property
    def report_version(self) -> str:
        return self._report.version

    def get_current_plugin_id(self) -> str | None:
        return self._current_plugin_id

    current_plugin_id = property(get_current_plugin_id)

    def reset(
        self,
        publish_plugins: list[pyblish.api.Plugin] | None = None,
        crashed_filepaths: dict[str, str] | None = None,
        blocking_crashed_paths: set[str] | None = None,
    ) -> None:
        """Reset report and clear all data."""
        if publish_plugins is None:
            publish_plugins = []

        if crashed_filepaths is None:
            crashed_filepaths = {}

        if blocking_crashed_paths is None:
            blocking_crashed_paths = set()

        self._instances_updated = False
        self._current_plugin_id = None
        self._plugin_info_by_id = {}
        self._all_instances_by_id = {}

        self._report = PublishReport(
            crashed_filepaths=crashed_filepaths,
            blocking_crashed_paths=list(blocking_crashed_paths),
        )

        self._prepare_publish_plugin_items(publish_plugins)

    def reset_with_discover_results(
        self,
        creator_discover_result: DiscoverResult,
        convertor_discover_result: DiscoverResult,
        publish_discover_result: DiscoverResult,
        blocking_crashed_paths: set[str],
    ) -> None:
        """Reset report and set discover results."""

        crashed_filepaths = {}
        for report in (
            creator_discover_result,
            convertor_discover_result,
            publish_discover_result,
        ):
            items = report.crashed_file_paths.items()
            for filepath, exc_info in items:
                crashed_filepaths[filepath] = "".join(
                    traceback.format_exception(*exc_info)
                )

        self.reset(
            publish_discover_result.plugins,
            crashed_filepaths,
            blocking_crashed_paths,
        )

    def add_plugin_iter(
        self, plugin_id: str, context: pyblish.api.Context
    ) -> None:
        """Add report about single iteration of plugin."""
        self.update_publish_instances(context)
        context_label = context.data.get("label")
        if context_label:
            self._report.context.label = context_label

        self._current_plugin_id = plugin_id

    def set_plugin_passed(self, plugin_id: str) -> None:
        plugin_info = self._plugin_info_by_id[plugin_id]
        plugin_info.passed = True

    def set_plugin_skipped(self, plugin_id: str) -> None:
        """Set that current plugin has been skipped."""
        plugin_info = self._plugin_info_by_id[plugin_id]
        plugin_info.skipped = True

    def add_result(self, plugin_id: str, result: dict[str, Any]) -> None:
        """Handle result of one plugin and it's instance."""
        plugin_info = self._plugin_info_by_id[plugin_id]
        plugin_info.process_reports.append(
            PublishProcessReport.from_result(result)
        )

    def add_action_result(
        self, action: pyblish.api.Action, result: dict[str, Any]
    ) -> None:
        """Add result of single action."""
        plugin = result["plugin"]

        plugin_info = self._plugin_info_by_id[plugin.id]

        action_name = action.__name__
        action_label = action.label or action_name
        plugin_info.actions_reports.append(PublishPluginActionReport(
            success=result["success"],
            name=action_name,
            label=action_label,
            logs=ReportLog.log_items_from_result(result),
        ))

    def get_report(self) -> PublishReport:
        self._update_instances()
        return self._report

    def get_report_data(self) -> dict[str, Any]:
        self._update_instances()
        return self._report.to_data(
            current_plugin_id=self._current_plugin_id,
        )

    def update_publish_instances(
        self, context: pyblish.api.Context
    ) -> None:
        """Report data with all details of current state."""
        for instance in context:
            if instance.id not in self._all_instances_by_id:
                self._instances_updated = False
                self._all_instances_by_id[instance.id] = instance

    def _update_instances(self) -> None:
        if self._instances_updated:
            return

        self._report.instances_by_id = {
            instance.id: PublishInstanceInfo.from_instance(instance)
            for instance in self._all_instances_by_id.values()
        }
        self._instances_updated = True

    def _prepare_publish_plugin_items(
        self, publish_plugins: list[pyblish.api.Plugin]
    ) -> None:
        for plugin in publish_plugins:
            if plugin.id in self._plugin_info_by_id:
                # A plugin would be processed more than once.
                #   What can cause it:
                #   - there is a bug in iteration logic
                #   - plugin class is imported into multiple files
                #       - this can happen even with base classes from
                #            'pyblish'
                raise ValueError(
                    f"Plugin '{plugin}' is already stored"
                )

            plugin_info = PublishPluginReportInfo.from_plugin(plugin)
            self._plugin_info_by_id[plugin.id] = plugin_info
            self._report.plugins_info.append(plugin_info)