Merge pull request #13 from jmbannon/test
Github hook for linting, 10/10 pylint score
This commit is contained in:
commit
cae3c458dd
20 changed files with 516 additions and 55 deletions
|
|
@ -18,6 +18,6 @@ jobs:
|
|||
run: |
|
||||
python -m pip install --upgrade pip
|
||||
pip install .[lint]
|
||||
- name: Analysing the code with pylint
|
||||
- name: Run linters
|
||||
run: |
|
||||
./tools/lint
|
||||
./tools/linter check
|
||||
|
|
@ -1,5 +0,0 @@
|
|||
[MASTER]
|
||||
disable=
|
||||
C0114, # missing-module-docstring
|
||||
R0903, # too-few-public-methods
|
||||
R0801, # similar lines
|
||||
|
|
@ -4,4 +4,34 @@ line_length = 100
|
|||
force_single_line = true
|
||||
|
||||
[tool.black]
|
||||
line_length = 100
|
||||
line_length = 100
|
||||
|
||||
[tool.pylint.MASTER]
|
||||
disable = [
|
||||
"C0115", # Missing class docstring
|
||||
"C0114", # missing-module-docstring
|
||||
"R0903", # too-few-public-methods
|
||||
"R0801", # similar lines
|
||||
"R0913", # Too many arguments
|
||||
"W0511", # TODO
|
||||
]
|
||||
|
||||
load-plugins = "pylint.extensions.docparams"
|
||||
|
||||
[tool.pydocstyle]
|
||||
inherit = false
|
||||
match = "[^test_].*\\.py"
|
||||
ignore = [
|
||||
"D100", # docstring in public module
|
||||
"D101", # Missing docstring in public class (covered by pylint)
|
||||
"D104", # docstring in public package
|
||||
"D107", # docstring in init
|
||||
"D200", # One-line should fit on one line
|
||||
"D203", # 1 blank line before class docstring
|
||||
"D205", # 1 blank line between summary and description
|
||||
"D212", # Multi-line should start at first line
|
||||
"D400", # Should end with a period
|
||||
"D401", # Return vs Returns
|
||||
"D413", # Missing blank line after last section
|
||||
"D415", # Should end with a period
|
||||
]
|
||||
|
|
|
|||
|
|
@ -38,6 +38,7 @@ lint =
|
|||
black==22.3.0
|
||||
isort==5.10.1
|
||||
pylint==2.13.5
|
||||
pydocstyle[toml]==6.1.1
|
||||
docs =
|
||||
sphinx==4.5.0
|
||||
sphinx-rtd-theme==1.0.0
|
||||
|
|
|
|||
|
|
@ -31,11 +31,31 @@ class ConfigFile(StrictDictValidator):
|
|||
self.presets = self._validate_key("presets", LiteralDictValidator)
|
||||
|
||||
@classmethod
|
||||
def from_dict(cls, config_dict) -> "ConfigFile":
|
||||
def from_dict(cls, config_dict: dict) -> "ConfigFile":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
config_dict:
|
||||
The config in dictionary format
|
||||
|
||||
Returns
|
||||
-------
|
||||
Config file validator
|
||||
"""
|
||||
return ConfigFile(name="", value=config_dict)
|
||||
|
||||
@classmethod
|
||||
def from_file_path(cls, config_path) -> "ConfigFile":
|
||||
def from_file_path(cls, config_path: str) -> "ConfigFile":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
config_path:
|
||||
Path to the config yaml
|
||||
|
||||
Returns
|
||||
-------
|
||||
Config file validator
|
||||
"""
|
||||
# TODO: Create separate yaml file loader class
|
||||
with open(config_path, "r", encoding="utf-8") as file:
|
||||
config_dict = yaml.safe_load(file)
|
||||
|
|
|
|||
|
|
@ -50,6 +50,16 @@ class DownloadStrategyValidator(StrictDictValidator):
|
|||
).value
|
||||
|
||||
def get(self, downloader_source: str) -> Type[Downloader]:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
downloader_source:
|
||||
Name of the download source
|
||||
|
||||
Returns
|
||||
-------
|
||||
The downloader class
|
||||
"""
|
||||
try:
|
||||
return DownloadStrategyMapping.get(
|
||||
source=downloader_source, download_strategy=self.download_strategy_name
|
||||
|
|
|
|||
|
|
@ -79,6 +79,18 @@ class DownloadStrategyMapping:
|
|||
|
||||
@classmethod
|
||||
def get(cls, source: str, download_strategy: str) -> Type[Downloader]:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
source:
|
||||
The source (i.e. 'youtube', 'soundcloud') of the downloader
|
||||
download_strategy:
|
||||
The download strategy name
|
||||
|
||||
Returns
|
||||
-------
|
||||
The downloader class
|
||||
"""
|
||||
cls._validate_is_download_strategy(source, download_strategy)
|
||||
return cls._MAPPING[source][download_strategy]
|
||||
|
||||
|
|
@ -106,6 +118,21 @@ class PluginMapping:
|
|||
|
||||
@classmethod
|
||||
def get(cls, plugin: str) -> Type[Plugin]:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
plugin
|
||||
Name of the plugin
|
||||
|
||||
Returns
|
||||
-------
|
||||
The plugin class
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
Raised if the plugin does not exist
|
||||
"""
|
||||
if plugin not in cls.plugins():
|
||||
raise ValueError(
|
||||
f"Tried to use plugin '{plugin}' that does not exist. Available plugins: "
|
||||
|
|
|
|||
|
|
@ -58,6 +58,11 @@ class SubscriptionValidator(StrictDictValidator):
|
|||
)
|
||||
|
||||
def to_subscription(self) -> Subscription:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The subscription after the config and preset have been validated
|
||||
"""
|
||||
return Subscription(
|
||||
name=self._name,
|
||||
config_options=self.config.config_options,
|
||||
|
|
@ -66,14 +71,40 @@ class SubscriptionValidator(StrictDictValidator):
|
|||
|
||||
@classmethod
|
||||
def from_dict(
|
||||
cls, config: ConfigFile, subscription_name, subscription_dict: Dict
|
||||
cls, config: ConfigFile, subscription_name: str, subscription_dict: Dict
|
||||
) -> "SubscriptionValidator":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
config:
|
||||
Validated instance of the config
|
||||
subscription_name:
|
||||
Name of the subscription
|
||||
subscription_dict:
|
||||
The subscription config in dict format
|
||||
|
||||
Returns
|
||||
-------
|
||||
The Subscription validator
|
||||
"""
|
||||
return SubscriptionValidator(config=config, name=subscription_name, value=subscription_dict)
|
||||
|
||||
@classmethod
|
||||
def from_file_path(
|
||||
cls, config: ConfigFile, subscription_path: str
|
||||
) -> List["SubscriptionValidator"]:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
config:
|
||||
Validated instance of the config
|
||||
subscription_path:
|
||||
File path to the subscription yaml file
|
||||
|
||||
Returns
|
||||
-------
|
||||
List of subscription validators, for each one in the subscription yaml
|
||||
"""
|
||||
# TODO: Create separate yaml file loader class
|
||||
with open(subscription_path, "r", encoding="utf-8") as file:
|
||||
subscription_dict = yaml.safe_load(file)
|
||||
|
|
|
|||
|
|
@ -77,6 +77,18 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
|
|||
ytdl_options: Optional[Dict] = None,
|
||||
download_archive_file_name: Optional[str] = None,
|
||||
):
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
working_directory
|
||||
Path to the working directory
|
||||
download_options
|
||||
Options validator for this downloader
|
||||
ytdl_options
|
||||
YTDL options validator
|
||||
download_archive_file_name
|
||||
Optional. Name of the download archive file that should reside in the working directory
|
||||
"""
|
||||
self.working_directory = working_directory
|
||||
self.download_options = download_options
|
||||
self.ytdl_options = Downloader._configure_ytdl_options(
|
||||
|
|
@ -120,6 +132,8 @@ class Downloader(Generic[DownloaderOptionsT, DownloaderEntryT], ABC):
|
|||
|
||||
Parameters
|
||||
----------
|
||||
output_directory
|
||||
overrides:
|
||||
Subscription overrides
|
||||
output_directory:
|
||||
Output directory to potentially store extra files downloaded
|
||||
"""
|
||||
|
|
|
|||
|
|
@ -197,14 +197,27 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
|
|||
ytdl_options_overrides=ytdl_options_overrides,
|
||||
)
|
||||
|
||||
@classmethod
|
||||
def __download_thumbnail(
|
||||
self,
|
||||
channel_dict: dict,
|
||||
cls,
|
||||
entry_dict: dict,
|
||||
thumbnail_id: str,
|
||||
output_thumbnail_path: str,
|
||||
):
|
||||
"""
|
||||
Downloads a specific thumbnail from a YTDL entry's thumbnail list
|
||||
|
||||
Parameters
|
||||
----------
|
||||
entry_dict:
|
||||
YTDL entry dict
|
||||
thumbnail_id:
|
||||
Id of the thumbnail defined in the YTDL thumnail
|
||||
output_thumbnail_path:
|
||||
Path to store the thumbnail after downloading
|
||||
"""
|
||||
thumbnail_url = None
|
||||
for thumbnail in channel_dict.get("thumbnails", []):
|
||||
for thumbnail in entry_dict.get("thumbnails", []):
|
||||
if thumbnail["id"] == thumbnail_id:
|
||||
thumbnail_url = thumbnail["url"]
|
||||
break
|
||||
|
|
@ -230,13 +243,13 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
|
|||
Output directory path
|
||||
"""
|
||||
channel_json_file_path = Path(self.working_directory) / f"{self.channel_id}.info.json"
|
||||
with open(channel_json_file_path) as channel_json:
|
||||
with open(channel_json_file_path, "r", encoding="utf-8") as channel_json:
|
||||
channel_entry = json.load(channel_json)
|
||||
|
||||
if self.download_options.channel_avatar_path:
|
||||
thumbnail_name = overrides.apply_formatter(self.download_options.channel_avatar_path)
|
||||
self.__download_thumbnail(
|
||||
channel_dict=channel_entry,
|
||||
entry_dict=channel_entry,
|
||||
thumbnail_id="avatar_uncropped",
|
||||
output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
|
||||
)
|
||||
|
|
@ -244,7 +257,7 @@ class YoutubeChannelDownloader(YoutubeDownloader[YoutubeChannelDownloaderOptions
|
|||
if self.download_options.channel_banner_path:
|
||||
thumbnail_name = overrides.apply_formatter(self.download_options.channel_banner_path)
|
||||
self.__download_thumbnail(
|
||||
channel_dict=channel_entry,
|
||||
entry_dict=channel_entry,
|
||||
thumbnail_id="banner_uncropped",
|
||||
output_thumbnail_path=str(Path(output_directory) / thumbnail_name),
|
||||
)
|
||||
|
|
|
|||
|
|
@ -46,6 +46,16 @@ class BaseEntry(ABC):
|
|||
return self._kwargs[key]
|
||||
|
||||
def working_directory(self) -> str:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The working directory
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
The working directory was never defined in the init
|
||||
"""
|
||||
if self._working_directory is None:
|
||||
raise ValueError(
|
||||
"Entry working directory is not set when trying to access its download file path"
|
||||
|
|
|
|||
|
|
@ -64,6 +64,22 @@ class SoundcloudAlbumTrack(SoundcloudTrack):
|
|||
soundcloud_track: SoundcloudTrack,
|
||||
playlist_metadata: PlaylistMetadata,
|
||||
) -> "SoundcloudAlbumTrack":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
album:
|
||||
Album name
|
||||
album_year:
|
||||
Album year
|
||||
soundcloud_track:
|
||||
Track to convert to an album track
|
||||
playlist_metadata:
|
||||
Metadata for playlist ordering
|
||||
|
||||
Returns
|
||||
-------
|
||||
SoundcloudTrack converted to a SoundcloudAlbumTrack
|
||||
"""
|
||||
return SoundcloudAlbumTrack(
|
||||
entry_dict=soundcloud_track._kwargs, # pylint: disable=protected-access
|
||||
working_directory=soundcloud_track.working_directory(),
|
||||
|
|
@ -91,7 +107,9 @@ class SoundcloudAlbum(Entry):
|
|||
|
||||
def album_tracks(self, skip_premiere_tracks: bool = True) -> List[SoundcloudAlbumTrack]:
|
||||
"""
|
||||
Returns all tracks in the album represented as album-tracks. They will share the
|
||||
Returns
|
||||
-------
|
||||
All tracks in the album represented as album-tracks. They will share the
|
||||
same album name, have ordered track numbers, and a shared album year.
|
||||
"""
|
||||
tracks = [
|
||||
|
|
@ -118,17 +136,26 @@ class SoundcloudAlbum(Entry):
|
|||
|
||||
@property
|
||||
def album_year(self) -> int:
|
||||
"""Returns the album's year, computed by the max upload year amongst all album tracks"""
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The album's year, computed by the max upload year amongst all album tracks
|
||||
"""
|
||||
return max(track.upload_year for track in self._single_tracks)
|
||||
|
||||
@property
|
||||
def track_count(self) -> int:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
Number of tracks in the album (technically a playlist)
|
||||
"""
|
||||
return self.kwargs("playlist_count")
|
||||
|
||||
@property
|
||||
def downloaded_track_count(self) -> int:
|
||||
return len(self.kwargs("entries"))
|
||||
|
||||
def contains(self, track: SoundcloudTrack) -> bool:
|
||||
"""Returns whether the album contains a track"""
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
True if this album contains this track. False otherwise.
|
||||
"""
|
||||
return any(track.uid == t.uid for t in self._single_tracks)
|
||||
|
|
|
|||
|
|
@ -30,7 +30,8 @@ class NfoTagsPlugin(Plugin[NfoTagsOptions]):
|
|||
|
||||
Parameters
|
||||
----------
|
||||
entry: Entry to create an NFO file for
|
||||
entry:
|
||||
Entry to create an NFO file for
|
||||
"""
|
||||
nfo = {}
|
||||
|
||||
|
|
|
|||
|
|
@ -61,7 +61,8 @@ class Plugin(Generic[PluginOptionsT], ABC):
|
|||
|
||||
Parameters
|
||||
----------
|
||||
entry: Entry to post process
|
||||
entry:
|
||||
Entry to post process
|
||||
"""
|
||||
|
||||
def post_process_subscription(self):
|
||||
|
|
|
|||
|
|
@ -60,48 +60,101 @@ class Subscription:
|
|||
|
||||
@property
|
||||
def downloader_class(self) -> Type[Downloader]:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
This subscription's downloader class
|
||||
"""
|
||||
return self.__preset_options.downloader
|
||||
|
||||
@property
|
||||
def downloader_options(self) -> DownloaderValidator:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The download options for this subscription's downloader
|
||||
"""
|
||||
return self.__preset_options.downloader_options
|
||||
|
||||
@property
|
||||
def plugins(self) -> List[Tuple[Type[Plugin], PluginOptions]]:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
List of tuples containing (plugin class, plugin options)
|
||||
"""
|
||||
return self.__preset_options.plugins
|
||||
|
||||
@property
|
||||
def ytdl_options(self) -> YTDLOptions:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
YTDL options for this subscription
|
||||
"""
|
||||
return self.__preset_options.ytdl_options
|
||||
|
||||
@property
|
||||
def output_options(self) -> OutputOptions:
|
||||
"""Returns the output options defined for this subscription"""
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The output options defined for this subscription
|
||||
"""
|
||||
return self.__preset_options.output_options
|
||||
|
||||
@property
|
||||
def overrides(self) -> Overrides:
|
||||
"""Returns the overrides defined for this subscription"""
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The overrides defined for this subscription
|
||||
"""
|
||||
return self.__preset_options.overrides
|
||||
|
||||
@property
|
||||
def working_directory(self) -> str:
|
||||
"""Returns the directory that the downloader saves files to"""
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
The directory that the downloader saves files to
|
||||
"""
|
||||
return str(Path(self.__config_options.working_directory.value) / Path(self.name))
|
||||
|
||||
@property
|
||||
def output_directory(self) -> str:
|
||||
"""
|
||||
:return: The formatted output directory
|
||||
Returns
|
||||
-------
|
||||
The formatted output directory
|
||||
"""
|
||||
return self.overrides.apply_formatter(formatter=self.output_options.output_directory)
|
||||
|
||||
def _copy_file_to_output_directory(self, source_file_path: str, output_file_name: str):
|
||||
"""
|
||||
Helper function to move a file from the working directory to the output directory.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
source_file_path:
|
||||
Path to the source file
|
||||
output_file_name:
|
||||
Desired output file name with the output directory as its relative directory
|
||||
"""
|
||||
destination_file_path = Path(self.output_directory) / Path(output_file_name)
|
||||
os.makedirs(os.path.dirname(destination_file_path), exist_ok=True)
|
||||
copyfile(source_file_path, destination_file_path)
|
||||
|
||||
def _copy_entry_files_to_output_directory(self, entry: Entry):
|
||||
"""
|
||||
Helper function to move the media file and optionally thumbnail file to the output directory
|
||||
for a single entry.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
entry:
|
||||
The entry with files to move
|
||||
"""
|
||||
# Move the file after all direct file modifications are complete
|
||||
output_file_name = self.overrides.apply_formatter(
|
||||
formatter=self.output_options.file_name, entry=entry
|
||||
|
|
@ -125,6 +178,10 @@ class Subscription:
|
|||
|
||||
@contextlib.contextmanager
|
||||
def _prepare_working_directory(self):
|
||||
"""
|
||||
Context manager to create all directories to the working directory. Deletes the entire
|
||||
working directory when cleaning up.
|
||||
"""
|
||||
os.makedirs(self.working_directory, exist_ok=True)
|
||||
|
||||
try:
|
||||
|
|
@ -134,6 +191,9 @@ class Subscription:
|
|||
|
||||
@contextlib.contextmanager
|
||||
def _maintain_archive_file(self):
|
||||
"""
|
||||
Context manager to initialize the enhanced download archive
|
||||
"""
|
||||
if self.output_options.maintain_download_archive:
|
||||
self._enhanced_download_archive.prepare_download_archive()
|
||||
|
||||
|
|
@ -147,9 +207,14 @@ class Subscription:
|
|||
date_range=self.output_options.delete_stale_files.get_date_range()
|
||||
)
|
||||
|
||||
self._enhanced_download_archive.save_download_archive()
|
||||
self._enhanced_download_archive.save_download_mappings()
|
||||
|
||||
def _initialize_plugins(self) -> List[Plugin]:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
List of plugins defined in the subscription, initialized and ready to use.
|
||||
"""
|
||||
plugins: List[Plugin] = []
|
||||
for plugin_type, plugin_options in self.plugins:
|
||||
plugin = plugin_type(
|
||||
|
|
|
|||
|
|
@ -15,6 +15,11 @@ class DateRangeValidator(StrictDictValidator):
|
|||
self.after = self._validate_key_if_present("after", StringDatetimeValidator)
|
||||
|
||||
def get_date_range(self) -> Optional[DateRange]:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
Date range if the 'before' or 'after' is defined. None otherwise.
|
||||
"""
|
||||
if self.before or self.after:
|
||||
return DateRange(
|
||||
start=self.after.datetime_str if self.after else None,
|
||||
|
|
|
|||
|
|
@ -31,7 +31,7 @@ class StringFormatterValidator(Validator):
|
|||
List of format variables in the format string
|
||||
|
||||
Raises
|
||||
-------
|
||||
------
|
||||
ValidationException
|
||||
If the format string contains invalid variable formatting
|
||||
"""
|
||||
|
|
@ -100,6 +100,18 @@ class StringFormatterValidator(Validator):
|
|||
|
||||
@final
|
||||
def apply_formatter(self, variable_dict: Dict[str, str]) -> str:
|
||||
"""
|
||||
Calls `format` on the format string using the variable_dict as input kwargs
|
||||
|
||||
Parameters
|
||||
----------
|
||||
variable_dict
|
||||
kwargs to pass to the format string
|
||||
|
||||
Returns
|
||||
-------
|
||||
Format string formatted
|
||||
"""
|
||||
# Keep formatting the format string until no format_variables are present
|
||||
formatter = self
|
||||
recursion_depth = 0
|
||||
|
|
|
|||
|
|
@ -34,6 +34,16 @@ class DownloadMapping:
|
|||
|
||||
@classmethod
|
||||
def from_dict(cls, mapping_dict: dict) -> "DownloadMapping":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
mapping_dict
|
||||
Download mapping in dict format
|
||||
|
||||
Returns
|
||||
-------
|
||||
Instantiated DownloadMapping class
|
||||
"""
|
||||
return DownloadMapping(
|
||||
upload_date=mapping_dict["upload_date"],
|
||||
extractor=mapping_dict["extractor"],
|
||||
|
|
@ -42,6 +52,16 @@ class DownloadMapping:
|
|||
|
||||
@classmethod
|
||||
def from_entry(cls, entry: Entry) -> "DownloadMapping":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
entry
|
||||
Entry to create a download mapping for
|
||||
|
||||
Returns
|
||||
-------
|
||||
DownloadMapping for the entry
|
||||
"""
|
||||
return DownloadMapping(
|
||||
upload_date=entry.upload_date_standardized,
|
||||
extractor=entry.extractor,
|
||||
|
|
@ -55,30 +75,70 @@ class DownloadArchive:
|
|||
possible in case of future changes.
|
||||
"""
|
||||
|
||||
def __init__(self):
|
||||
self._download_archive_lines: List[str] = []
|
||||
|
||||
@classmethod
|
||||
def from_lines(cls, lines: List[str]) -> "DownloadArchive":
|
||||
download_archive = DownloadArchive()
|
||||
download_archive._download_archive_lines = lines
|
||||
return download_archive
|
||||
def __init__(self, download_archive_lines: List[str]):
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
download_archive_lines
|
||||
Lines found in a YTDL download archive file, i.e. youtube id-32342343423
|
||||
"""
|
||||
self._download_archive_lines = download_archive_lines
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, file_path: str) -> "DownloadArchive":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
file_path
|
||||
Path to a download archive file
|
||||
|
||||
Returns
|
||||
-------
|
||||
Instantiated DownloadArchive class
|
||||
"""
|
||||
with open(file_path, "r", encoding="utf8") as file:
|
||||
return cls.from_lines(lines=file.readlines())
|
||||
return cls(download_archive_lines=file.readlines())
|
||||
|
||||
def to_file(self, file_path: str) -> "DownloadArchive":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
file_path
|
||||
File path to store this download archive to
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
with open(file_path, "w", encoding="utf8") as file:
|
||||
for line in self._download_archive_lines:
|
||||
file.write(f"{line}\n")
|
||||
return self
|
||||
|
||||
def contains(self, entry_id: str) -> bool:
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
entry_id
|
||||
Id of the entry
|
||||
|
||||
Returns
|
||||
-------
|
||||
True if the entry id is within this download archive. False otherwise.
|
||||
"""
|
||||
return any(entry_id in line for line in self._download_archive_lines)
|
||||
|
||||
def remove_entry(self, entry_id: str) -> "DownloadArchive":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
entry_id
|
||||
Entry ID to remove if it exists in this download archive
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
self._download_archive_lines = [
|
||||
line for line in self._download_archive_lines if entry_id not in line
|
||||
]
|
||||
|
|
@ -89,10 +149,23 @@ class DownloadMappings:
|
|||
_strptime_format = "%Y-%m-%d"
|
||||
|
||||
def __init__(self):
|
||||
"""
|
||||
Initializes an empty mapping
|
||||
"""
|
||||
self._entry_mappings: Dict[str, DownloadMapping] = {}
|
||||
|
||||
@classmethod
|
||||
def from_file(cls, json_file_path: str) -> "DownloadMappings":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
json_file_path
|
||||
Path to a json file that contains download mappings
|
||||
|
||||
Returns
|
||||
-------
|
||||
Instantiated DownloadMappings class
|
||||
"""
|
||||
with open(json_file_path, "r", encoding="utf8") as json_file:
|
||||
entry_mappings_json = json.load(json_file)
|
||||
|
||||
|
|
@ -107,13 +180,37 @@ class DownloadMappings:
|
|||
|
||||
@property
|
||||
def entry_ids(self) -> List[str]:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
List of entry ids in the mapping
|
||||
"""
|
||||
return list(self._entry_mappings.keys())
|
||||
|
||||
@property
|
||||
def is_empty(self) -> bool:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
True if there are no entry mappings. False otherwise.
|
||||
"""
|
||||
return len(self._entry_mappings) == 0
|
||||
|
||||
def add_entry(self, entry: Entry, entry_file_path: str) -> "DownloadMappings":
|
||||
"""
|
||||
Adds a file path for the entry. An entry can map to multiple file paths.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
entry
|
||||
Entry that this file belongs to
|
||||
entry_file_path
|
||||
Relative path to the file that belongs to the entry
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
if entry.uid not in self.entry_ids:
|
||||
self._entry_mappings[entry.uid] = DownloadMapping.from_entry(entry=entry)
|
||||
|
||||
|
|
@ -121,14 +218,30 @@ class DownloadMappings:
|
|||
return self
|
||||
|
||||
def remove_entry(self, entry_id: str) -> "DownloadMappings":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
entry_id
|
||||
Id of the entry to remove
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
if entry_id in self.entry_ids:
|
||||
del self._entry_mappings[entry_id]
|
||||
return self
|
||||
|
||||
def get_entries_out_of_range(self, date_range: DateRange) -> Dict[str, DownloadMapping]:
|
||||
"""
|
||||
:param date_range: range of dates that entries' upload dates must be within
|
||||
:return: dict of entry_id: mapping if the upload date is not in the date range
|
||||
Parameters
|
||||
----------
|
||||
date_range
|
||||
Range of dates that entries' upload dates must be within
|
||||
|
||||
Returns
|
||||
-------
|
||||
Dict of entry_id: mapping if the upload date is not in the date range
|
||||
"""
|
||||
out_of_range_entry_mappings = copy.deepcopy(self._entry_mappings)
|
||||
for uid in list(out_of_range_entry_mappings.keys()):
|
||||
|
|
@ -143,7 +256,16 @@ class DownloadMappings:
|
|||
return out_of_range_entry_mappings
|
||||
|
||||
def to_file(self, output_json_file: str) -> "DownloadMappings":
|
||||
"""
|
||||
Parameters
|
||||
----------
|
||||
output_json_file
|
||||
Output json file path to write the download mappings to
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
# Create json string first to ensure it is valid before writing anything to file
|
||||
json_str = json.dumps(
|
||||
obj={
|
||||
|
|
@ -163,11 +285,17 @@ class DownloadMappings:
|
|||
return self
|
||||
|
||||
def to_download_archive(self) -> DownloadArchive:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
A DownloadArchive created from the DownloadMappings' ids and extractors. YTDL will use this
|
||||
to avoid redownloading entries already downloaded.
|
||||
"""
|
||||
lines: List[str] = []
|
||||
for entry_id, metadata in self._entry_mappings.items():
|
||||
lines.append(f"{metadata.extractor} {entry_id}")
|
||||
|
||||
return DownloadArchive.from_lines(lines)
|
||||
return DownloadArchive(download_archive_lines=lines)
|
||||
|
||||
|
||||
class EnhancedDownloadArchive:
|
||||
|
|
@ -211,39 +339,64 @@ class EnhancedDownloadArchive:
|
|||
@property
|
||||
def archive_file_name(self) -> str:
|
||||
"""
|
||||
:return: The download archive's file name (no path)
|
||||
Returns
|
||||
-------
|
||||
The download archive's file name (no path)
|
||||
"""
|
||||
return f".ytdl-subscribe-{self.subscription_name}-download-archive.txt"
|
||||
|
||||
@property
|
||||
def _mapping_file_name(self) -> str:
|
||||
"""
|
||||
:return: The download mapping's file name (no path)
|
||||
Returns
|
||||
-------
|
||||
The download mapping's file name (no path)
|
||||
"""
|
||||
return f".ytdl-subscribe-{self.subscription_name}-download-mapping.json"
|
||||
|
||||
@property
|
||||
def _mapping_output_file_path(self):
|
||||
"""
|
||||
:return: The download mapping's file path in the output directory.
|
||||
Returns
|
||||
-------
|
||||
The download mapping's file path in the output directory.
|
||||
"""
|
||||
return str(Path(self.output_directory) / self._mapping_file_name)
|
||||
|
||||
@property
|
||||
def _archive_working_file_path(self) -> str:
|
||||
"""
|
||||
:return: The download archive's file path in the working directory.
|
||||
Returns
|
||||
-------
|
||||
The download archive's file path in the working directory.
|
||||
"""
|
||||
return str(Path(self.working_directory) / self.archive_file_name)
|
||||
|
||||
@property
|
||||
def mapping(self) -> DownloadMappings:
|
||||
"""
|
||||
Returns
|
||||
-------
|
||||
Loaded DownloadMappings
|
||||
|
||||
Raises
|
||||
------
|
||||
ValueError
|
||||
If the download mappings was not loaded
|
||||
"""
|
||||
if self._download_mapping is None:
|
||||
raise ValueError("Tried to use download mapping before it was loaded")
|
||||
return self._download_mapping
|
||||
|
||||
def _load(self) -> "EnhancedDownloadArchive":
|
||||
"""
|
||||
Tries to load download mappings if they are present in the output directory.
|
||||
If they are not, initialize an empty mapping.
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
# If a mapping file exists in the output directory, load it up.
|
||||
if os.path.isfile(self._mapping_output_file_path):
|
||||
self._download_mapping = DownloadMappings.from_file(
|
||||
|
|
@ -257,6 +410,14 @@ class EnhancedDownloadArchive:
|
|||
return self
|
||||
|
||||
def _copy_to_working_directory(self) -> "EnhancedDownloadArchive":
|
||||
"""
|
||||
If the mapping is not empty, create a download archive from it and save it into the
|
||||
working directory. This will tell YTDL to not redownload already downloaded entries.
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
# If the download mapping is empty, do nothing since the ytdl downloader will create a new
|
||||
# download archive file
|
||||
if self.mapping.is_empty:
|
||||
|
|
@ -268,11 +429,32 @@ class EnhancedDownloadArchive:
|
|||
return self
|
||||
|
||||
def prepare_download_archive(self) -> "EnhancedDownloadArchive":
|
||||
"""
|
||||
Helper function to load mappings and create a YTDL download archive file in the
|
||||
working directory if mappings exist.
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
self._load()
|
||||
self._copy_to_working_directory()
|
||||
return self
|
||||
|
||||
def remove_stale_files(self, date_range: DateRange) -> "EnhancedDownloadArchive":
|
||||
"""
|
||||
Checks all entries within the mappings. If any entries' upload dates are not within the
|
||||
provided date range, delete them.
|
||||
|
||||
Parameters
|
||||
----------
|
||||
date_range
|
||||
Date range the upload date must be in to not get deleted
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
stale_mappings: Dict[str, DownloadMapping] = self.mapping.get_entries_out_of_range(
|
||||
date_range=date_range
|
||||
)
|
||||
|
|
@ -289,7 +471,15 @@ class EnhancedDownloadArchive:
|
|||
|
||||
return self
|
||||
|
||||
def save_download_archive(self) -> "EnhancedDownloadArchive":
|
||||
def save_download_mappings(self) -> "EnhancedDownloadArchive":
|
||||
"""
|
||||
Saves the updated download mappings to the output directory
|
||||
|
||||
Returns
|
||||
-------
|
||||
self
|
||||
"""
|
||||
# TODO: Make sure this logic is actually right...
|
||||
# Load the download archive from the working directory, which should contain any past
|
||||
# and new entries downloaded in this session
|
||||
self._download_archive = DownloadArchive.from_file(self._archive_working_file_path)
|
||||
|
|
|
|||
|
|
@ -1,4 +1,4 @@
|
|||
# Uncomment to build api docs
|
||||
# sphinx-apidoc -f -o docs/source ytdl_subscribe
|
||||
# sphinx-apidoc -f -o docs/source ytdl_subscribea
|
||||
sphinx-build -a -b html docs docs/_html
|
||||
|
||||
|
|
|
|||
17
tools/linter
17
tools/linter
|
|
@ -1,4 +1,13 @@
|
|||
# Run within root directory. Runs isort, black, then pylint
|
||||
isort .
|
||||
black .
|
||||
pylint src/
|
||||
# Run within root directory
|
||||
|
||||
if [ $1 = "check" ]; then
|
||||
isort . --check-only --diff \
|
||||
&& black . --check \
|
||||
&& pylint src/ \
|
||||
&& pydocstyle src/*
|
||||
else
|
||||
isort .
|
||||
black .
|
||||
pylint src/
|
||||
pydocstyle src/*
|
||||
fi
|
||||
Loading…
Reference in a new issue