✨ Do not process tracks with missing segments. Refactored some method to pathlib.Path.

This commit is contained in:
Robert Honz
2024-11-13 09:49:33 +01:00
parent c1a102f646
commit 3171b1570b
2 changed files with 52 additions and 50 deletions
+24 -22
View File
@@ -92,7 +92,7 @@ class Download:
self, self,
media: Track | Video, media: Track | Video,
path_file: pathlib.Path, path_file: pathlib.Path,
) -> pathlib.Path: ) -> (bool, pathlib.Path):
media_name: str = name_builder_item(media) media_name: str = name_builder_item(media)
urls: [str] urls: [str]
path_base: pathlib.Path = path_file.parent path_base: pathlib.Path = path_file.parent
@@ -165,21 +165,20 @@ class Download:
# mark the whole thing as corrupt. # mark the whole thing as corrupt.
result_segments = False result_segments = False
tmp_path_file_decrypted: pathlib.Path = path_file
# Only if no error happened while downloading.
if result_segments:
# Bring list into right order, so segments can be easily merged. # Bring list into right order, so segments can be easily merged.
dl_segment_results.sort(key=lambda x: x.id_segment) dl_segment_results.sort(key=lambda x: x.id_segment)
# TODO: Implement error handling on corrupt segments (also in the following method.
reult_merge: bool = self._segments_merge(path_file, dl_segment_results) reult_merge: bool = self._segments_merge(path_file, dl_segment_results)
tmp_path_file_decrypted: pathlib.Path = pathlib.Path("")
if reult_merge: if reult_merge and isinstance(media, Track) and stream_manifest.is_encrypted:
if isinstance(media, Track) and stream_manifest.is_encrypted:
key, nonce = decrypt_security_token(stream_manifest.encryption_key) key, nonce = decrypt_security_token(stream_manifest.encryption_key)
tmp_path_file_decrypted = path_file + "_decrypted" tmp_path_file_decrypted = path_file.with_suffix("_decrypted")
decrypt_file(path_file, tmp_path_file_decrypted, key, nonce) decrypt_file(path_file, tmp_path_file_decrypted, key, nonce)
else:
tmp_path_file_decrypted = path_file
return tmp_path_file_decrypted return result_segments, tmp_path_file_decrypted
def _segments_merge(self, path_file, dl_segment_results) -> bool: def _segments_merge(self, path_file, dl_segment_results) -> bool:
result: bool result: bool
@@ -229,7 +228,7 @@ class Download:
except HTTPError as e: except HTTPError as e:
error = e error = e
self.progress.advance(p_task) self.progress.advance(p_task)
self.fn_logger.exception(e.charachter_written) self.fn_logger.exception(e)
# To send the progress to the GUI, we need to emit the percentage. # To send the progress to the GUI, we need to emit the percentage.
if not progress_to_stdout: if not progress_to_stdout:
@@ -281,12 +280,14 @@ class Download:
# Create file name and path # Create file name and path
file_extension_dummy: str = AudioExtensions.FLAC file_extension_dummy: str = AudioExtensions.FLAC
file_name_relative = format_path_media(file_template, media) file_name_relative = format_path_media(file_template, media)
path_media_dst = os.path.abspath( path_media_dst: pathlib.Path = (
os.path.normpath(os.path.join(os.path.expanduser(self.path_base), file_name_relative)) (pathlib.Path(self.path_base).expanduser() / file_name_relative).resolve().absolute()
) )
# Sanitize final path_file to fit into OS boundaries. # Sanitize final path_file to fit into OS boundaries.
path_media_dst = path_file_sanitize(path_media_dst + file_extension_dummy, adapt=True) path_media_dst = pathlib.Path(
path_file_sanitize(str(path_media_dst.with_suffix(file_extension_dummy)), adapt=True)
)
# Compute if and how downloads need to be skipped. # Compute if and how downloads need to be skipped.
if self.skip_existing: if self.skip_existing:
@@ -336,8 +337,9 @@ class Download:
tmp_path_file.touch() tmp_path_file.touch()
# Download media. # Download media.
tmp_path_file = self._download(media=media, path_file=tmp_path_file) result_download, tmp_path_file = self._download(media=media, path_file=tmp_path_file)
if result_download:
# Convert video from TS to MP4 # Convert video from TS to MP4
if isinstance(media, Video) and self.settings.data.video_convert_mp4: if isinstance(media, Video) and self.settings.data.video_convert_mp4:
# Convert `*.ts` file to `*.mp4` using ffmpeg # Convert `*.ts` file to `*.mp4` using ffmpeg
@@ -356,10 +358,6 @@ class Download:
media, tmp_path_file, is_parent_album media, tmp_path_file, is_parent_album
) )
# Move final file to the configured destination directory.
os.makedirs(os.path.dirname(path_media_dst), exist_ok=True)
shutil.move(tmp_path_file, path_media_dst)
# Move lyrics file # Move lyrics file
if self.settings.data.lyrics_file and not isinstance(media, Video): if self.settings.data.lyrics_file and not isinstance(media, Video):
self._move_lyrics(tmp_path_lyrics, path_media_dst) self._move_lyrics(tmp_path_lyrics, path_media_dst)
@@ -368,6 +366,10 @@ class Download:
if self.settings.data.cover_album_file: if self.settings.data.cover_album_file:
self._move_cover(tmp_path_cover, path_media_dst) self._move_cover(tmp_path_cover, path_media_dst)
# Move final file to the configured destination directory.
os.makedirs(path_media_dst.parent, exist_ok=True)
shutil.move(tmp_path_file, path_media_dst)
if quality_audio: if quality_audio:
# Set quality back to the global user value # Set quality back to the global user value
self.adjust_quality_audio(quality_audio_old) self.adjust_quality_audio(quality_audio_old)
@@ -418,16 +420,16 @@ class Download:
return result return result
def _move_lyrics(self, path_lyrics: pathlib.Path, file_media_dst: str) -> bool: def _move_lyrics(self, path_lyrics: pathlib.Path, file_media_dst: pathlib.Path) -> bool:
# Build tmp lyrics filename # Build tmp lyrics filename
path_file_lyrics: str = os.path.splitext(file_media_dst)[0] + EXTENSION_LYRICS path_file_lyrics: pathlib.Path = file_media_dst.with_suffix(EXTENSION_LYRICS)
result: bool = self._move_file(path_lyrics, path_file_lyrics) result: bool = self._move_file(path_lyrics, path_file_lyrics)
return result return result
def _move_cover(self, path_cover: pathlib.Path, file_media_dst: str) -> bool: def _move_cover(self, path_cover: pathlib.Path, file_media_dst: pathlib.Path) -> bool:
# Build tmp lyrics filename # Build tmp lyrics filename
path_file_cover: pathlib.Path = pathlib.Path(file_media_dst).parent.absolute() / COVER_NAME path_file_cover: pathlib.Path = file_media_dst.parent / COVER_NAME
result: bool = self._move_file(path_cover, path_file_cover) result: bool = self._move_file(path_cover, path_file_cover)
return result return result
+6 -6
View File
@@ -1,9 +1,9 @@
import math import math
import os import os
import pathlib
import posixpath import posixpath
import re import re
import sys import sys
from pathlib import Path, PosixPath
from urllib.parse import unquote, urlsplit from urllib.parse import unquote, urlsplit
from pathvalidate import sanitize_filename, sanitize_filepath from pathvalidate import sanitize_filename, sanitize_filepath
@@ -197,7 +197,7 @@ def get_format_template(
def path_file_sanitize(path_file: str, adapt: bool = False, uniquify: bool = False) -> (bool, str): def path_file_sanitize(path_file: str, adapt: bool = False, uniquify: bool = False) -> (bool, str):
# Split into path and filename # Split into path and filename
pathname, filename = os.path.split(path_file) pathname, filename = os.path.split(path_file)
file_extension: str = Path(path_file).suffix file_extension: str = pathlib.Path(path_file).suffix
# Sanitize path # Sanitize path
try: try:
@@ -207,7 +207,7 @@ def path_file_sanitize(path_file: str, adapt: bool = False, uniquify: bool = Fal
except ValidationError: except ValidationError:
# If adaption of path is allowed in case of an error set path to HOME. # If adaption of path is allowed in case of an error set path to HOME.
if adapt: if adapt:
pathname_sanitized: str = Path.home() pathname_sanitized: str = str(pathlib.Path.home())
else: else:
raise raise
@@ -274,10 +274,10 @@ def file_unique_suffix(path_file: str, seperator: str = "_") -> str:
return unique_suffix return unique_suffix
def check_file_exists(path_file: str, extension_ignore: bool = False) -> bool: def check_file_exists(path_file: pathlib.Path, extension_ignore: bool = False) -> bool:
if extension_ignore: if extension_ignore:
path_file_stem: str = Path(path_file).stem path_file_stem: str = pathlib.Path(path_file).stem
path_parent: PosixPath = Path(path_file).parent path_parent: pathlib.Path = pathlib.Path(path_file).parent
path_files: [str] = [] path_files: [str] = []
for extension in AudioExtensions: for extension in AudioExtensions: