async def _process_attachments(  # noqa: C901
        self,
        client: httpx.AsyncClient,
        payload: Dict,
        message_id: str,
        thread_id: str,
        breadcrumbs: List[Breadcrumb],
    ) -> AsyncGenerator[GmailAttachmentEntity, None]:
        """Process message attachments using the standard file processing pipeline."""
        logger.info(f"Processing attachments for message {message_id}")

        # Function to recursively find attachments
        async def find_attachments(part, depth=0):
            indent = "  " * depth
            mime_type = part.get("mimeType", "")
            filename = part.get("filename", "")
            part_id = part.get("partId", "unknown")

            part_info = (
                f"{indent}Checking part {part_id} at depth {depth}: "
                f"mime_type={mime_type}, filename={filename}"
            )
            logger.debug(part_info)
            body = part.get("body", {})

            # If part has filename and is not an inline image or text part, treat as attachment
            if (
                filename
                and mime_type not in ("text/plain", "text/html")
                and not (mime_type.startswith("image/") and not filename)
            ):
                attachment_id = body.get("attachmentId")
                attachment_info = (
                    f"{indent}Found potential attachment: {filename} ({mime_type}), "
                    f"attachment_id: {attachment_id}"
                )
                logger.info(attachment_info)

                # Skip if no attachment ID (might be inline content)
                if not attachment_id:
                    logger.debug(f"{indent}Skipping part with filename but no attachment ID")
                    return

                # Get full attachment data
                logger.info(f"{indent}Fetching attachment data for attachment_id: {attachment_id}")
                attachment_url = (
                    f"https://gmail.googleapis.com/gmail/v1/users/me/messages/"
                    f"{message_id}/attachments/{attachment_id}"
                )
                try:
                    attachment_data = await self._get_with_auth(client, attachment_url)
                    keys_info = (
                        f"{indent}Attachment data received with keys: "
                        f"{list(attachment_data.keys())}"
                    )
                    logger.debug(keys_info)
                    size = attachment_data.get("size", 0)
                    logger.info(f"{indent}Attachment size: {size} bytes")

                    # Create a dummy download URL (required by FileEntity)
                    # We'll actually use the content directly, but this satisfies the schema
                    dummy_download_url = f"gmail://attachment/{message_id}/{attachment_id}"

                    # Create file entity
                    # Prefix to ensure uniqueness
                    file_entity = GmailAttachmentEntity(
                        entity_id=f"attach_{message_id}_{attachment_id}",
                        breadcrumbs=breadcrumbs,
                        file_id=attachment_id,
                        name=filename,
                        mime_type=mime_type,
                        size=size,
                        total_size=size,
                        download_url=dummy_download_url,  # Required by FileEntity
                        message_id=message_id,
                        attachment_id=attachment_id,
                        thread_id=thread_id,
                    )

                    # Get base64 data
                    base64_data = attachment_data.get("data", "")
                    if not base64_data:
                        logger.warning(f"{indent}No data found for attachment {filename}")
                        return

                    # Decode the base64 data
                    binary_data = base64.urlsafe_b64decode(base64_data)

                    # Create a memory stream from the decoded data
                    async def content_stream():
                        yield binary_data

                    # Process using the BaseSource method (now abstracted)
                    logger.info(
                        f"{indent}Processing file entity for {filename} with direct content stream"
                    )
                    processed_entity = await self.process_file_entity_with_content(
                        file_entity=file_entity,
                        content_stream=content_stream(),
                        metadata={"source": "gmail", "message_id": message_id},
                    )

                    if processed_entity:
                        logger.info(f"{indent}Yielding processed attachment: {filename}")
                        yield processed_entity
                    else:
                        logger.warning(f"{indent}Processing failed for attachment: {filename}")

                except Exception as e:
                    logger.error(f"{indent}Error processing attachment {attachment_id}: {str(e)}")

            # Recursively process parts for multipart messages
            if part.get("parts"):
                sub_parts = part.get("parts", [])
                logger.debug(f"{indent}Part has {len(sub_parts)} sub-parts, recursing...")
                for sub_part in sub_parts:
                    async for attachment in find_attachments(sub_part, depth + 1):
                        yield attachment

        # Start processing from the top-level payload
        logger.info(f"Starting attachment search from top-level payload for message {message_id}")
        attachment_count = 0
        async for attachment in find_attachments(payload):
            attachment_count += 1
            yield attachment

        completion_msg = (
            f"Attachment processing complete for message {message_id}: "
            f"found {attachment_count} attachments"
        )
        logger.info(completion_msg)
Process message attachments using the standard file processing pipeline.
backend/airweave/platform/sources/gmail.py