-
Notifications
You must be signed in to change notification settings - Fork 0
/
main.py
77 lines (60 loc) · 2.58 KB
/
main.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
import logging
import os
import time
import warnings
import requests
warnings.filterwarnings("ignore", category=UserWarning, module="urllib3")
logger = logging.getLogger(__name__)
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s - %(levelname)s - %(message)s",
datefmt="%Y-%m-%d %H:%M:%S",
)
# Predefined constants
DEFAULT_AUDIO_URL = "https://bytefm--di--nacs-ice-01--02--cdn.cast.addradio.de/bytefm/main/high/stream.mp3"
DEFAULT_TARGET_PATH = "stream.mp3"
def download_audio(
audio_url: str = DEFAULT_AUDIO_URL,
target_path: str = DEFAULT_TARGET_PATH,
retries: int = 3,
) -> None:
"""
Download audio files from the provided URL and store them at the specific location on disk.
:param audio_url: A string containing the URL of the audio file.
:param target_path: A string specifying where the downloaded file should be saved.
:param retries: An integer defining how many attempts are made before giving up when hitting exceptions during download. Default is set to 3.
"""
# Ensure directories exist
directory = os.path.dirname(target_path)
if directory and not os.path.exists(directory):
os.makedirs(directory)
logger.info(f"Starting download from '{audio_url}' to '{target_path}'.")
start_time = time.time()
for attempt in range(retries + 1):
try:
response = requests.get(audio_url, stream=True)
response.raise_for_status()
with open(target_path, "wb") as output_file:
for chunk in response.iter_content(chunk_size=1024):
output_file.write(chunk)
duration = round(time.time() - start_time, 2)
logger.info(f"Download completed in {duration} seconds.")
return
except requests.exceptions.HTTPError as http_error:
message = f"HTTP error {response.status_code} encountered while downloading audio from '{audio_url}'."
logg(message, http_error, attempt)
except (
requests.exceptions.ConnectionError,
requests.exceptions.Timeout,
) as network_exception:
message = (
f"A network error occurred while downloading audio from '{audio_url}'."
)
logg(message, network_exception, attempt)
def logg(message, arg1, attempt):
logger.warning(message)
logger.warning(arg1)
logger.warning(f"Retry #{attempt}: Retrying in {attempt*2} seconds.\n")
time.sleep(attempt * 2)
if __name__ == "__main__":
download_audio()