music-kraken-core/music_kraken/objects/target.py

117 lines
3.5 KiB
Python
Raw Permalink Normal View History

2023-12-29 20:16:09 +00:00
from __future__ import annotations
2023-02-25 21:16:32 +00:00
from pathlib import Path
2023-12-29 20:16:09 +00:00
from typing import List, Tuple, TextIO, Union
2023-06-22 10:44:46 +00:00
import logging
2024-04-10 16:18:52 +00:00
import random
2023-03-31 07:47:03 +00:00
import requests
2023-04-04 15:20:27 +00:00
from tqdm import tqdm
2023-02-25 21:16:32 +00:00
2023-12-19 21:11:46 +00:00
from .parents import OuterProxy
2024-04-10 16:18:52 +00:00
from ..utils.shared import HIGHEST_ID
2023-09-10 15:27:07 +00:00
from ..utils.config import main_settings, logging_settings
2023-09-13 16:55:04 +00:00
from ..utils.string_processing import fit_to_file_system
2023-02-25 21:16:32 +00:00
2023-06-22 10:44:46 +00:00
LOGGER = logging.getLogger("target")
2023-12-19 21:11:46 +00:00
class Target(OuterProxy):
2023-02-25 21:16:32 +00:00
"""
create somehow like that
```python
# I know path is pointless, and I will change that (don't worry about backwards compatibility there)
Target(file="song.mp3", path="~/Music/genre/artist/album")
```
"""
2023-12-29 20:16:09 +00:00
file_path: Path
2023-12-20 08:55:09 +00:00
_default_factories = {
}
2024-04-10 16:18:52 +00:00
@classmethod
def temp(cls, name: str = str(random.randint(0, HIGHEST_ID))) -> P:
return cls(main_settings["temp_directory"] / name)
2023-12-29 20:16:09 +00:00
# This is automatically generated
def __init__(self, file_path: Union[Path, str], relative_to_music_dir: bool = False, **kwargs) -> None:
if not isinstance(file_path, Path):
file_path = Path(file_path)
if relative_to_music_dir:
file_path = Path(main_settings["music_directory"], file_path)
super().__init__(file_path=fit_to_file_system(file_path), **kwargs)
2023-02-25 21:16:32 +00:00
self.is_relative_to_music_dir: bool = relative_to_music_dir
2023-03-30 12:39:28 +00:00
def __repr__(self) -> str:
return str(self.file_path)
@property
def indexing_values(self) -> List[Tuple[str, object]]:
return [('filepath', self.file_path)]
2023-04-04 15:54:05 +00:00
@property
def exists(self) -> bool:
2023-03-30 14:10:48 +00:00
return self.file_path.is_file()
2023-04-05 09:54:02 +00:00
@property
def size(self) -> int:
"""
2023-12-29 20:16:09 +00:00
returns the size the downloaded audio takes up in bytes
returns 0 if the file doesn't exist
2023-04-05 09:54:02 +00:00
"""
if not self.exists:
return 0
return self.file_path.stat().st_size
2023-04-04 15:54:05 +00:00
def create_path(self):
2024-01-15 10:40:48 +00:00
self.file_path.parent.mkdir(parents=True, exist_ok=True)
2023-04-04 15:54:05 +00:00
2023-12-29 20:16:09 +00:00
def copy_content(self, copy_to: Target):
2023-03-30 13:28:23 +00:00
if not self.exists:
2023-06-22 10:44:46 +00:00
LOGGER.warning(f"No file exists at: {self.file_path}")
2023-03-30 13:28:23 +00:00
return
2023-04-04 15:54:05 +00:00
2023-03-30 13:28:23 +00:00
with open(self.file_path, "rb") as read_from:
copy_to.create_path()
2023-03-31 07:54:31 +00:00
with open(copy_to.file_path, "wb") as write_to:
2023-03-30 13:28:23 +00:00
write_to.write(read_from.read())
2023-03-31 07:47:03 +00:00
def stream_into(self, r: requests.Response, desc: str = None) -> bool:
if r is None:
return False
2023-04-04 15:54:05 +00:00
2023-03-31 07:47:03 +00:00
self.create_path()
2023-04-04 15:54:05 +00:00
2023-03-31 07:47:03 +00:00
total_size = int(r.headers.get('content-length'))
2023-04-04 15:54:05 +00:00
with open(self.file_path, 'wb') as f:
try:
2023-04-04 15:20:27 +00:00
"""
https://en.wikipedia.org/wiki/Kilobyte
> The internationally recommended unit symbol for the kilobyte is kB.
"""
with tqdm(total=total_size, unit='B', unit_scale=True, unit_divisor=1024, desc=desc) as t:
2023-09-10 15:27:07 +00:00
for chunk in r.iter_content(chunk_size=main_settings["chunk_size"]):
2023-04-04 15:54:05 +00:00
size = f.write(chunk)
t.update(size)
2023-04-04 18:58:22 +00:00
return True
2023-04-04 15:54:05 +00:00
except requests.exceptions.Timeout:
2023-09-10 15:27:07 +00:00
logging_settings["download_logger"].error("Stream timed out.")
return False
2023-06-15 16:22:00 +00:00
def open(self, file_mode: str, **kwargs) -> TextIO:
return self.file_path.open(file_mode, **kwargs)
2023-05-25 09:21:39 +00:00
def delete(self):
self.file_path.unlink(missing_ok=True)
2024-04-10 16:18:52 +00:00
def read_bytes(self) -> bytes:
return self.file_path.read_bytes()