scrapy/scrapy/extensions/feedexport.py

"""
Feed Exports extension

See documentation in docs/topics/feed-exports.rst
"""

from __future__ import annotations

import logging
import re
import sys
import warnings
from datetime import datetime, timezone
from pathlib import Path, PureWindowsPath
from tempfile import NamedTemporaryFile
from typing import (
    IO,
    TYPE_CHECKING,
    Any,
    Callable,
    Dict,
    Iterable,
    List,
    Optional,
    Protocol,
    Tuple,
    Type,
    TypeVar,
    Union,
    cast,
)
from urllib.parse import unquote, urlparse

from twisted.internet.defer import Deferred, DeferredList, maybeDeferred
from twisted.internet.threads import deferToThread
from w3lib.url import file_uri_to_path
from zope.interface import Interface, implementer

from scrapy import Spider, signals
from scrapy.exceptions import NotConfigured, ScrapyDeprecationWarning
from scrapy.extensions.postprocessing import PostProcessingManager
from scrapy.settings import Settings
from scrapy.utils.boto import is_botocore_available
from scrapy.utils.conf import feed_complete_default_values_from_settings
from scrapy.utils.defer import maybe_deferred_to_future
from scrapy.utils.deprecate import create_deprecated_class
from scrapy.utils.ftp import ftp_store_file
from scrapy.utils.log import failure_to_exc_info
from scrapy.utils.misc import build_from_crawler, load_object
from scrapy.utils.python import without_none_values

if TYPE_CHECKING:
    from _typeshed import OpenBinaryMode
    from twisted.python.failure import Failure

    # typing.Self requires Python 3.11
    from typing_extensions import Self

    from scrapy.crawler import Crawler
    from scrapy.exporters import BaseItemExporter
    from scrapy.settings import BaseSettings

try:
    import boto3  # noqa: F401

    IS_BOTO3_AVAILABLE = True
except ImportError:
    IS_BOTO3_AVAILABLE = False


logger = logging.getLogger(__name__)

UriParamsCallableT = Callable[[Dict[str, Any], Spider], Optional[Dict[str, Any]]]

_StorageT = TypeVar("_StorageT", bound="FeedStorageProtocol")


def build_storage(
    builder: Callable[..., _StorageT],
    uri: str,
    *args: Any,
    feed_options: Optional[Dict[str, Any]] = None,
    preargs: Iterable[Any] = (),
    **kwargs: Any,
) -> _StorageT:
    kwargs["feed_options"] = feed_options
    return builder(*preargs, uri, *args, **kwargs)


class ItemFilter:
    """
    This will be used by FeedExporter to decide if an item should be allowed
    to be exported to a particular feed.

    :param feed_options: feed specific options passed from FeedExporter
    :type feed_options: dict
    """

    feed_options: Optional[Dict[str, Any]]
    item_classes: Tuple[type, ...]

    def __init__(self, feed_options: Optional[Dict[str, Any]]) -> None:
        self.feed_options = feed_options
        if feed_options is not None:
            self.item_classes = tuple(
                load_object(item_class)
                for item_class in feed_options.get("item_classes") or ()
            )
        else:
            self.item_classes = ()

    def accepts(self, item: Any) -> bool:
        """
        Return ``True`` if `item` should be exported or ``False`` otherwise.

        :param item: scraped item which user wants to check if is acceptable
        :type item: :ref:`Scrapy items <topics-items>`
        :return: `True` if accepted, `False` otherwise
        :rtype: bool
        """
        if self.item_classes:
            return isinstance(item, self.item_classes)
        return True  # accept all items by default


class IFeedStorage(Interface):
    """Interface that all Feed Storages must implement"""

    def __init__(uri, *, feed_options=None):
        """Initialize the storage with the parameters given in the URI and the
        feed-specific options (see :setting:`FEEDS`)"""

    def open(spider):
        """Open the storage for the given spider. It must return a file-like
        object that will be used for the exporters"""

    def store(file):
        """Store the given file stream"""


class FeedStorageProtocol(Protocol):
    """Reimplementation of ``IFeedStorage`` that can be used in type hints."""

    def __init__(self, uri: str, *, feed_options: Optional[Dict[str, Any]] = None):
        """Initialize the storage with the parameters given in the URI and the
        feed-specific options (see :setting:`FEEDS`)"""

    def open(self, spider: Spider) -> IO[bytes]:
        """Open the storage for the given spider. It must return a file-like
        object that will be used for the exporters"""

    def store(self, file: IO[bytes]) -> Optional[Deferred[None]]:
        """Store the given file stream"""


@implementer(IFeedStorage)
class BlockingFeedStorage:
    def open(self, spider: Spider) -> IO[bytes]:
        path = spider.crawler.settings["FEED_TEMPDIR"]
        if path and not Path(path).is_dir():
            raise OSError("Not a Directory: " + str(path))

        return NamedTemporaryFile(prefix="feed-", dir=path)

    def store(self, file: IO[bytes]) -> Optional[Deferred[None]]:
        return deferToThread(self._store_in_thread, file)

    def _store_in_thread(self, file: IO[bytes]) -> None:
        raise NotImplementedError


@implementer(IFeedStorage)
class StdoutFeedStorage:
    def __init__(
        self,
        uri: str,
        _stdout: Optional[IO[bytes]] = None,
        *,
        feed_options: Optional[Dict[str, Any]] = None,
    ):
        if not _stdout:
            _stdout = sys.stdout.buffer
        self._stdout: IO[bytes] = _stdout
        if feed_options and feed_options.get("overwrite", False) is True:
            logger.warning(
                "Standard output (stdout) storage does not support "
                "overwriting. To suppress this warning, remove the "
                "overwrite option from your FEEDS setting, or set "
                "it to False."
            )

    def open(self, spider: Spider) -> IO[bytes]:
        return self._stdout

    def store(self, file: IO[bytes]) -> Optional[Deferred[None]]:
        pass


@implementer(IFeedStorage)
class FileFeedStorage:
    def __init__(self, uri: str, *, feed_options: Optional[Dict[str, Any]] = None):
        self.path: str = file_uri_to_path(uri)
        feed_options = feed_options or {}
        self.write_mode: OpenBinaryMode = (
            "wb" if feed_options.get("overwrite", False) else "ab"
        )

    def open(self, spider: Spider) -> IO[bytes]:
        dirname = Path(self.path).parent
        if dirname and not dirname.exists():
            dirname.mkdir(parents=True)
        return Path(self.path).open(self.write_mode)

    def store(self, file: IO[bytes]) -> Optional[Deferred[None]]:
        file.close()
        return None


class S3FeedStorage(BlockingFeedStorage):
    def __init__(
        self,
        uri: str,
        access_key: Optional[str] = None,
        secret_key: Optional[str] = None,
        acl: Optional[str] = None,
        endpoint_url: Optional[str] = None,
        *,
        feed_options: Optional[Dict[str, Any]] = None,
        session_token: Optional[str] = None,
        region_name: Optional[str] = None,
    ):
        if not is_botocore_available():
            raise NotConfigured("missing botocore library")
        u = urlparse(uri)
        assert u.hostname
        self.bucketname: str = u.hostname
        self.access_key: Optional[str] = u.username or access_key
        self.secret_key: Optional[str] = u.password or secret_key
        self.session_token: Optional[str] = session_token
        self.keyname: str = u.path[1:]  # remove first "/"
        self.acl: Optional[str] = acl
        self.endpoint_url: Optional[str] = endpoint_url
        self.region_name: Optional[str] = region_name
        # It can be either botocore.client.BaseClient or mypy_boto3_s3.S3Client,
        # there seems to be no good way to infer it statically.
        self.s3_client: Any

        if IS_BOTO3_AVAILABLE:
            import boto3.session

            boto3_session = boto3.session.Session()

            self.s3_client = boto3_session.client(
                "s3",
                aws_access_key_id=self.access_key,
                aws_secret_access_key=self.secret_key,
                aws_session_token=self.session_token,
                endpoint_url=self.endpoint_url,
                region_name=self.region_name,
            )
        else:
            warnings.warn(
                "`botocore` usage has been deprecated for S3 feed "
                "export, please use `boto3` to avoid problems",
                category=ScrapyDeprecationWarning,
            )

            import botocore.session

            botocore_session = botocore.session.get_session()

            self.s3_client = botocore_session.create_client(
                "s3",
                aws_access_key_id=self.access_key,
                aws_secret_access_key=self.secret_key,
                aws_session_token=self.session_token,
                endpoint_url=self.endpoint_url,
                region_name=self.region_name,
            )

        if feed_options and feed_options.get("overwrite", True) is False:
            logger.warning(
                "S3 does not support appending to files. To "
                "suppress this warning, remove the overwrite "
                "option from your FEEDS setting or set it to True."
            )

    @classmethod
    def from_crawler(
        cls,
        crawler: Crawler,
        uri: str,
        *,
        feed_options: Optional[Dict[str, Any]] = None,
    ) -> Self:
        return build_storage(
            cls,
            uri,
            access_key=crawler.settings["AWS_ACCESS_KEY_ID"],
            secret_key=crawler.settings["AWS_SECRET_ACCESS_KEY"],
            session_token=crawler.settings["AWS_SESSION_TOKEN"],
            acl=crawler.settings["FEED_STORAGE_S3_ACL"] or None,
            endpoint_url=crawler.settings["AWS_ENDPOINT_URL"] or None,
            region_name=crawler.settings["AWS_REGION_NAME"] or None,
            feed_options=feed_options,
        )

    def _store_in_thread(self, file: IO[bytes]) -> None:
        file.seek(0)
        kwargs: Dict[str, Any]
        if IS_BOTO3_AVAILABLE:
            kwargs = {"ExtraArgs": {"ACL": self.acl}} if self.acl else {}
            self.s3_client.upload_fileobj(
                Bucket=self.bucketname, Key=self.keyname, Fileobj=file, **kwargs
            )
        else:
            kwargs = {"ACL": self.acl} if self.acl else {}
            self.s3_client.put_object(
                Bucket=self.bucketname, Key=self.keyname, Body=file, **kwargs
            )
        file.close()


class GCSFeedStorage(BlockingFeedStorage):
    def __init__(self, uri: str, project_id: Optional[str], acl: Optional[str]):
        self.project_id: Optional[str] = project_id
        self.acl: Optional[str] = acl
        u = urlparse(uri)
        assert u.hostname
        self.bucket_name: str = u.hostname
        self.blob_name: str = u.path[1:]  # remove first "/"

    @classmethod
    def from_crawler(cls, crawler: Crawler, uri: str) -> Self:
        return cls(
            uri,
            crawler.settings["GCS_PROJECT_ID"],
            crawler.settings["FEED_STORAGE_GCS_ACL"] or None,
        )

    def _store_in_thread(self, file: IO[bytes]) -> None:
        file.seek(0)
        from google.cloud.storage import Client

        client = Client(project=self.project_id)
        bucket = client.get_bucket(self.bucket_name)
        blob = bucket.blob(self.blob_name)
        blob.upload_from_file(file, predefined_acl=self.acl)


class FTPFeedStorage(BlockingFeedStorage):
    def __init__(
        self,
        uri: str,
        use_active_mode: bool = False,
        *,
        feed_options: Optional[Dict[str, Any]] = None,
    ):
        u = urlparse(uri)
        if not u.hostname:
            raise ValueError(f"Got a storage URI without a hostname: {uri}")
        self.host: str = u.hostname
        self.port: int = int(u.port or "21")
        self.username: str = u.username or ""
        self.password: str = unquote(u.password or "")
        self.path: str = u.path
        self.use_active_mode: bool = use_active_mode
        self.overwrite: bool = not feed_options or feed_options.get("overwrite", True)

    @classmethod
    def from_crawler(
        cls,
        crawler: Crawler,
        uri: str,
        *,
        feed_options: Optional[Dict[str, Any]] = None,
    ) -> Self:
        return build_storage(
            cls,
            uri,
            crawler.settings.getbool("FEED_STORAGE_FTP_ACTIVE"),
            feed_options=feed_options,
        )

    def _store_in_thread(self, file: IO[bytes]) -> None:
        ftp_store_file(
            path=self.path,
            file=file,
            host=self.host,
            port=self.port,
            username=self.username,
            password=self.password,
            use_active_mode=self.use_active_mode,
            overwrite=self.overwrite,
        )


class FeedSlot:
    def __init__(
        self,
        storage: FeedStorageProtocol,
        uri: str,
        format: str,
        store_empty: bool,
        batch_id: int,
        uri_template: str,
        filter: ItemFilter,
        feed_options: Dict[str, Any],
        spider: Spider,
        exporters: Dict[str, Type[BaseItemExporter]],
        settings: BaseSettings,
        crawler: Crawler,
    ):
        self.file: Optional[IO[bytes]] = None
        self.exporter: Optional[BaseItemExporter] = None
        self.storage: FeedStorageProtocol = storage
        # feed params
        self.batch_id: int = batch_id
        self.format: str = format
        self.store_empty: bool = store_empty
        self.uri_template: str = uri_template
        self.uri: str = uri
        self.filter: ItemFilter = filter
        # exporter params
        self.feed_options: Dict[str, Any] = feed_options
        self.spider: Spider = spider
        self.exporters: Dict[str, Type[BaseItemExporter]] = exporters
        self.settings: BaseSettings = settings
        self.crawler: Crawler = crawler
        # flags
        self.itemcount: int = 0
        self._exporting: bool = False
        self._fileloaded: bool = False

    def start_exporting(self) -> None:
        if not self._fileloaded:
            self.file = self.storage.open(self.spider)
            if "postprocessing" in self.feed_options:
                self.file = cast(
                    IO[bytes],
                    PostProcessingManager(
                        self.feed_options["postprocessing"],
                        self.file,
                        self.feed_options,
                    ),
                )
            self.exporter = self._get_exporter(
                file=self.file,
                format=self.feed_options["format"],
                fields_to_export=self.feed_options["fields"],
                encoding=self.feed_options["encoding"],
                indent=self.feed_options["indent"],
                **self.feed_options["item_export_kwargs"],
            )
            self._fileloaded = True

        if not self._exporting:
            assert self.exporter
            self.exporter.start_exporting()
            self._exporting = True

    def _get_instance(
        self, objcls: Type[BaseItemExporter], *args: Any, **kwargs: Any
    ) -> BaseItemExporter:
        return build_from_crawler(objcls, self.crawler, *args, **kwargs)

    def _get_exporter(
        self, file: IO[bytes], format: str, *args: Any, **kwargs: Any
    ) -> BaseItemExporter:
        return self._get_instance(self.exporters[format], file, *args, **kwargs)

    def finish_exporting(self) -> None:
        if self._exporting:
            assert self.exporter
            self.exporter.finish_exporting()
            self._exporting = False


_FeedSlot = create_deprecated_class(
    name="_FeedSlot",
    new_class=FeedSlot,
)


class FeedExporter:
    _pending_deferreds: List[Deferred[None]] = []

    @classmethod
    def from_crawler(cls, crawler: Crawler) -> Self:
        exporter = cls(crawler)
        crawler.signals.connect(exporter.open_spider, signals.spider_opened)
        crawler.signals.connect(exporter.close_spider, signals.spider_closed)
        crawler.signals.connect(exporter.item_scraped, signals.item_scraped)
        return exporter

    def __init__(self, crawler: Crawler):
        self.crawler: Crawler = crawler
        self.settings: Settings = crawler.settings
        self.feeds = {}
        self.slots: List[FeedSlot] = []
        self.filters: Dict[str, ItemFilter] = {}

        if not self.settings["FEEDS"] and not self.settings["FEED_URI"]:
            raise NotConfigured

        # Begin: Backward compatibility for FEED_URI and FEED_FORMAT settings
        if self.settings["FEED_URI"]:
            warnings.warn(
                "The `FEED_URI` and `FEED_FORMAT` settings have been deprecated in favor of "
                "the `FEEDS` setting. Please see the `FEEDS` setting docs for more details",
                category=ScrapyDeprecationWarning,
                stacklevel=2,
            )
            uri = self.settings["FEED_URI"]
            # handle pathlib.Path objects
            uri = str(uri) if not isinstance(uri, Path) else uri.absolute().as_uri()
            feed_options = {"format": self.settings.get("FEED_FORMAT", "jsonlines")}
            self.feeds[uri] = feed_complete_default_values_from_settings(
                feed_options, self.settings
            )
            self.filters[uri] = self._load_filter(feed_options)
        # End: Backward compatibility for FEED_URI and FEED_FORMAT settings

        # 'FEEDS' setting takes precedence over 'FEED_URI'
        for uri, feed_options in self.settings.getdict("FEEDS").items():
            # handle pathlib.Path objects
            uri = str(uri) if not isinstance(uri, Path) else uri.absolute().as_uri()
            self.feeds[uri] = feed_complete_default_values_from_settings(
                feed_options, self.settings
            )
            self.filters[uri] = self._load_filter(feed_options)

        self.storages: Dict[str, Type[FeedStorageProtocol]] = self._load_components(
            "FEED_STORAGES"
        )
        self.exporters: Dict[str, Type[BaseItemExporter]] = self._load_components(
            "FEED_EXPORTERS"
        )
        for uri, feed_options in self.feeds.items():
            if not self._storage_supported(uri, feed_options):
                raise NotConfigured
            if not self._settings_are_valid():
                raise NotConfigured
            if not self._exporter_supported(feed_options["format"]):
                raise NotConfigured

    def open_spider(self, spider: Spider) -> None:
        for uri, feed_options in self.feeds.items():
            uri_params = self._get_uri_params(spider, feed_options["uri_params"])
            self.slots.append(
                self._start_new_batch(
                    batch_id=1,
                    uri=uri % uri_params,
                    feed_options=feed_options,
                    spider=spider,
                    uri_template=uri,
                )
            )

    async def close_spider(self, spider: Spider) -> None:
        for slot in self.slots:
            self._close_slot(slot, spider)

        # Await all deferreds
        if self._pending_deferreds:
            await maybe_deferred_to_future(DeferredList(self._pending_deferreds))

        # Send FEED_EXPORTER_CLOSED signal
        await maybe_deferred_to_future(
            self.crawler.signals.send_catch_log_deferred(signals.feed_exporter_closed)
        )

    def _close_slot(self, slot: FeedSlot, spider: Spider) -> Optional[Deferred[None]]:
        def get_file(slot_: FeedSlot) -> IO[bytes]:
            assert slot_.file
            if isinstance(slot_.file, PostProcessingManager):
                slot_.file.close()
                return slot_.file.file
            return slot_.file

        if slot.itemcount:
            # Normal case
            slot.finish_exporting()
        elif slot.store_empty and slot.batch_id == 1:
            # Need to store the empty file
            slot.start_exporting()
            slot.finish_exporting()
        else:
            # In this case, the file is not stored, so no processing is required.
            return None

        logmsg = f"{slot.format} feed ({slot.itemcount} items) in: {slot.uri}"
        d: Deferred[None] = maybeDeferred(slot.storage.store, get_file(slot))  # type: ignore[arg-type]

        d.addCallback(
            self._handle_store_success, logmsg, spider, type(slot.storage).__name__
        )
        d.addErrback(
            self._handle_store_error, logmsg, spider, type(slot.storage).__name__
        )
        self._pending_deferreds.append(d)
        d.addCallback(
            lambda _: self.crawler.signals.send_catch_log_deferred(
                signals.feed_slot_closed, slot=slot
            )
        )
        d.addBoth(lambda _: self._pending_deferreds.remove(d))

        return d

    def _handle_store_error(
        self, f: Failure, logmsg: str, spider: Spider, slot_type: str
    ) -> None:
        logger.error(
            "Error storing %s",
            logmsg,
            exc_info=failure_to_exc_info(f),
            extra={"spider": spider},
        )
        assert self.crawler.stats
        self.crawler.stats.inc_value(f"feedexport/failed_count/{slot_type}")

    def _handle_store_success(
        self, result: Any, logmsg: str, spider: Spider, slot_type: str
    ) -> None:
        logger.info("Stored %s", logmsg, extra={"spider": spider})
        assert self.crawler.stats
        self.crawler.stats.inc_value(f"feedexport/success_count/{slot_type}")

    def _start_new_batch(
        self,
        batch_id: int,
        uri: str,
        feed_options: Dict[str, Any],
        spider: Spider,
        uri_template: str,
    ) -> FeedSlot:
        """
        Redirect the output data stream to a new file.
        Execute multiple times if FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified
        :param batch_id: sequence number of current batch
        :param uri: uri of the new batch to start
        :param feed_options: dict with parameters of feed
        :param spider: user spider
        :param uri_template: template of uri which contains %(batch_time)s or %(batch_id)d to create new uri
        """
        storage = self._get_storage(uri, feed_options)
        slot = FeedSlot(
            storage=storage,
            uri=uri,
            format=feed_options["format"],
            store_empty=feed_options["store_empty"],
            batch_id=batch_id,
            uri_template=uri_template,
            filter=self.filters[uri_template],
            feed_options=feed_options,
            spider=spider,
            exporters=self.exporters,
            settings=self.settings,
            crawler=self.crawler,
        )
        return slot

    def item_scraped(self, item: Any, spider: Spider) -> None:
        slots = []
        for slot in self.slots:
            if not slot.filter.accepts(item):
                slots.append(
                    slot
                )  # if slot doesn't accept item, continue with next slot
                continue

            slot.start_exporting()
            assert slot.exporter
            slot.exporter.export_item(item)
            slot.itemcount += 1
            # create new slot for each slot with itemcount == FEED_EXPORT_BATCH_ITEM_COUNT and close the old one
            if (
                self.feeds[slot.uri_template]["batch_item_count"]
                and slot.itemcount >= self.feeds[slot.uri_template]["batch_item_count"]
            ):
                uri_params = self._get_uri_params(
                    spider, self.feeds[slot.uri_template]["uri_params"], slot
                )
                self._close_slot(slot, spider)
                slots.append(
                    self._start_new_batch(
                        batch_id=slot.batch_id + 1,
                        uri=slot.uri_template % uri_params,
                        feed_options=self.feeds[slot.uri_template],
                        spider=spider,
                        uri_template=slot.uri_template,
                    )
                )
            else:
                slots.append(slot)
        self.slots = slots

    def _load_components(self, setting_prefix: str) -> Dict[str, Any]:
        conf = without_none_values(
            cast(Dict[str, str], self.settings.getwithbase(setting_prefix))
        )
        d = {}
        for k, v in conf.items():
            try:
                d[k] = load_object(v)
            except NotConfigured:
                pass
        return d

    def _exporter_supported(self, format: str) -> bool:
        if format in self.exporters:
            return True
        logger.error("Unknown feed format: %(format)s", {"format": format})
        return False

    def _settings_are_valid(self) -> bool:
        """
        If FEED_EXPORT_BATCH_ITEM_COUNT setting or FEEDS.batch_item_count is specified uri has to contain
        %(batch_time)s or %(batch_id)d to distinguish different files of partial output
        """
        for uri_template, values in self.feeds.items():
            if values["batch_item_count"] and not re.search(
                r"%\(batch_time\)s|%\(batch_id\)", uri_template
            ):
                logger.error(
                    "%%(batch_time)s or %%(batch_id)d must be in the feed URI (%s) if FEED_EXPORT_BATCH_ITEM_COUNT "
                    "setting or FEEDS.batch_item_count is specified and greater than 0. For more info see: "
                    "https://docs.scrapy.org/en/latest/topics/feed-exports.html#feed-export-batch-item-count",
                    uri_template,
                )
                return False
        return True

    def _storage_supported(self, uri: str, feed_options: Dict[str, Any]) -> bool:
        scheme = urlparse(uri).scheme
        if scheme in self.storages or PureWindowsPath(uri).drive:
            try:
                self._get_storage(uri, feed_options)
                return True
            except NotConfigured as e:
                logger.error(
                    "Disabled feed storage scheme: %(scheme)s. " "Reason: %(reason)s",
                    {"scheme": scheme, "reason": str(e)},
                )
        else:
            logger.error("Unknown feed storage scheme: %(scheme)s", {"scheme": scheme})
        return False

    def _get_storage(
        self, uri: str, feed_options: Dict[str, Any]
    ) -> FeedStorageProtocol:
        """Fork of create_instance specific to feed storage classes

        It supports not passing the *feed_options* parameters to classes that
        do not support it, and issuing a deprecation warning instead.
        """
        feedcls = self.storages.get(urlparse(uri).scheme, self.storages["file"])
        crawler = getattr(self, "crawler", None)

        def build_instance(
            builder: Type[FeedStorageProtocol], *preargs: Any
        ) -> FeedStorageProtocol:
            return build_storage(
                builder, uri, feed_options=feed_options, preargs=preargs
            )

        instance: FeedStorageProtocol
        if crawler and hasattr(feedcls, "from_crawler"):
            instance = build_instance(feedcls.from_crawler, crawler)
            method_name = "from_crawler"
        elif hasattr(feedcls, "from_settings"):
            instance = build_instance(feedcls.from_settings, self.settings)
            method_name = "from_settings"
        else:
            instance = build_instance(feedcls)
            method_name = "__new__"
        if instance is None:
            raise TypeError(f"{feedcls.__qualname__}.{method_name} returned None")
        return instance

    def _get_uri_params(
        self,
        spider: Spider,
        uri_params_function: Union[str, UriParamsCallableT, None],
        slot: Optional[FeedSlot] = None,
    ) -> Dict[str, Any]:
        params = {}
        for k in dir(spider):
            params[k] = getattr(spider, k)
        utc_now = datetime.now(tz=timezone.utc)
        params["time"] = utc_now.replace(microsecond=0).isoformat().replace(":", "-")
        params["batch_time"] = utc_now.isoformat().replace(":", "-")
        params["batch_id"] = slot.batch_id + 1 if slot is not None else 1
        uripar_function: UriParamsCallableT = (
            load_object(uri_params_function)
            if uri_params_function
            else lambda params, _: params
        )
        new_params = uripar_function(params, spider)
        return new_params if new_params is not None else params

    def _load_filter(self, feed_options: Dict[str, Any]) -> ItemFilter:
        # load the item filter if declared else load the default filter class
        item_filter_class: Type[ItemFilter] = load_object(
            feed_options.get("item_filter", ItemFilter)
        )
        return item_filter_class(feed_options)