from os import PathLike from pathlib import PurePosixPath from typing import IO, DefaultDict, Optional, Set, Union from urllib.parse import urlparse import repub.utils from repub.exporters import RssExporter from scrapy.pipelines.images import FilesPipeline as BaseFilesPipeline from scrapy.pipelines.images import ImagesPipeline as BaseImagesPipeline class ImagePipeline(BaseImagesPipeline): def file_path(self, request, response=None, info=None, *, item=None): return repub.utils.local_image_path(request.url) def thumb_path(self, request, thumb_id, response=None, info=None, *, item=None): raise NotImplementedError() class FilePipeline(BaseFilesPipeline): def file_path(self, request, response=None, info=None, *, item=None): return repub.utils.local_file_path(request.url) class AudioPipeline(BaseFilesPipeline): def __init__(self, store_uri: Union[str, PathLike], **kwargs): self.FILES_URLS_FIELD = "audio_urls" self.FILES_RESULT_FIELD = "audios" store_uri = kwargs["settings"]["AUDIO_STORE"] super().__init__(store_uri, **kwargs) def file_path(self, request, response=None, info=None, *, item=None): return repub.utils.local_audio_path(request.url) class VideoPipeline(BaseFilesPipeline): def __init__(self, store_uri: Union[str, PathLike], **kwargs): self.FILES_URLS_FIELD = "video_urls" self.FILES_RESULT_FIELD = "videos" store_uri = kwargs["settings"]["VIDEO_STORE"] super().__init__(store_uri, **kwargs) def file_path(self, request, response=None, info=None, *, item=None): return repub.utils.local_video_path(request.url)