Method add

src/crawlee/statistics/_error_tracker.py:45–94

Add an error in the statistics. Args: error: Error to be added to statistics. context: Context used to collect error snapshot.

(
        self,
        error: Exception,
        *,
        context: BasicCrawlingContext | None = None,
    )

        self._errors: ErrorFilenameGroups = defaultdict(lambda: defaultdict(Counter))

    async def add(
        self,
        error: Exception,
        *,
        context: BasicCrawlingContext | None = None,
    ) -> None:
        """Add an error in the statistics.

        Args:
            error: Error to be added to statistics.
            context: Context used to collect error snapshot.
        """
        error_group_name = error.__class__.__name__ if self.show_error_name else None
        error_group_message = self._get_error_message(error)
        new_error_group_message = ''  # In case of wildcard similarity match
        error_group_file_and_line = self._get_file_and_line(error)

        # First two levels are grouped only in case of exact match.
        specific_groups = self._errors[error_group_file_and_line][error_group_name]

        # Lowest level group is matched by similarity.
        if error_group_message in specific_groups:
            # Exact match.
            specific_groups.update([error_group_message])
        else:
            for existing_error_group_message in specific_groups:
                # Add to first group with similar text. Modify text with wildcard characters if necessary.
                if new_error_group_message := self._create_generic_message(
                    existing_error_group_message, error_group_message
                ):
                    # Replace old name.
                    specific_groups[new_error_group_message] = specific_groups.pop(existing_error_group_message)
                    # Increment.
                    specific_groups.update([new_error_group_message])
                    break
            else:
                # No similar message found. Create new group.
                self._errors[error_group_file_and_line][error_group_name].update([error_group_message])

        if (
            self._errors[error_group_file_and_line][error_group_name][new_error_group_message or error_group_message]
            == 1
            and context is not None
        ):
            # Save snapshot only on the first occurrence of the error and only if context and kvs was passed as well.
            await self._capture_error_snapshot(
                error_message=new_error_group_message or error_group_message,
                file_and_line=error_group_file_and_line,
                context=context,
            )

    async def _capture_error_snapshot(
        self, error_message: str, file_and_line: str, context: BasicCrawlingContext