feat: Refactor and improved modularity and clarity

- Separated functionality into distinct functions for better readability and maintainability:
- Enhanced logging for better tracking of execution flow.
- Improved documentation with clear function docstrings.
This commit is contained in:
Hetari 2024-09-25 10:47:55 +03:00
parent 82e50c64f0
commit 88aa00a723

View file

@ -5,12 +5,43 @@ import time
import aiohttp import aiohttp
import asyncio import asyncio
from urllib.parse import urlparse from urllib.parse import urlparse
url = 'https://storage.googleapis.com/panels-api/data/20240916/media-1a-i-p~s' import logging
async def delay(ms):
# Constants
URL = 'https://storage.googleapis.com/panels-api/data/20240916/media-1a-i-p~s'
DOWNLOAD_DIR = os.path.join(os.getcwd(), 'downloads')
DELAY_MS = 250
# Configure logging
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')
async def delay(ms: int) -> None:
"""Delay the execution for the specified milliseconds."""
await asyncio.sleep(ms / 1000) await asyncio.sleep(ms / 1000)
async def download_image(session, image_url, file_path):
def create_download_dir(directory: str) -> None:
"""Create the download directory if it doesn't exist."""
if not os.path.exists(directory):
try:
os.makedirs(directory)
logging.info(f"📁 Created directory: {directory}")
except OSError as e:
logging.error(f"Failed to create directory {directory}: {e}")
raise
def get_file_extension(url: str) -> str:
"""Extract the file extension from the URL."""
parsed_url = urlparse(url)
ext = os.path.splitext(parsed_url.path)[-1]
return ext or '.jpg' # Default to .jpg if no extension found
async def download_image(session: aiohttp.ClientSession, image_url: str, file_path: str) -> None:
"""Download an image from the provided URL and save it to the specified file path."""
try: try:
async with session.get(image_url) as response: async with session.get(image_url) as response:
if response.status != 200: if response.status != 200:
@ -18,44 +49,48 @@ async def download_image(session, image_url, file_path):
content = await response.read() content = await response.read()
with open(file_path, 'wb') as f: with open(file_path, 'wb') as f:
f.write(content) f.write(content)
logging.info(f"🖼️ Image saved to {file_path}")
except Exception as e: except Exception as e:
print(f"Error downloading image: {str(e)}") logging.error(f"Error downloading image from {image_url}: {e}")
async def main() -> None:
"""Main function that fetches JSON data and downloads images based on URLs."""
create_download_dir(DOWNLOAD_DIR)
async def main():
try: try:
async with aiohttp.ClientSession() as session: async with aiohttp.ClientSession() as session:
async with session.get(url) as response: async with session.get(URL) as response:
if response.status != 200: if response.status != 200:
raise Exception(f"⛔ Failed to fetch JSON file: {response.status}") raise Exception(f"⛔ Failed to fetch JSON file: {response.status}")
json_data = await response.json() json_data = await response.json()
data = json_data.get('data') data = json_data.get('data')
if not data: if not data:
raise Exception('⛔ JSON does not have a "data" property at its root.') raise Exception('⛔ JSON does not have a "data" property at its root.')
download_dir = os.path.join(os.getcwd(), 'downloads')
if not os.path.exists(download_dir):
os.makedirs(download_dir)
print(f"📁 Created directory: {download_dir}")
file_index = 1 file_index = 1
for key, subproperty in data.items(): for key, subproperty in data.items():
if subproperty and subproperty.get('dhd'): if subproperty and subproperty.get('dhd'):
image_url = subproperty['dhd'] image_url = subproperty['dhd']
print(f"🔍 Found image URL!") logging.info("🔍 Found image URL")
parsed_url = urlparse(image_url)
ext = os.path.splitext(parsed_url.path)[-1] or '.jpg' file_extension = get_file_extension(image_url)
filename = f"{file_index}{ext}" filename = f"{file_index}{file_extension}"
file_path = os.path.join(download_dir, filename) file_path = os.path.join(DOWNLOAD_DIR, filename)
await download_image(session, image_url, file_path) await download_image(session, image_url, file_path)
print(f"🖼️ Saved image to {file_path}")
file_index += 1 file_index += 1
await delay(250) await delay(DELAY_MS)
# for debug purposes
# else:
# logging.warning(f"⚠️ No image URL found for key: {key}")
except Exception as e: except Exception as e:
print(f"Error: {str(e)}") logging.error(f"Error: {e}")
def ascii_art(): def ascii_art():
print(""" print("""
@ -67,8 +102,8 @@ def ascii_art():
| $$\\ $ | $$| $$\\ $$ | $$ \\ $$ /$$ \\ $$| $$ | $$ | $$\\ $ | $$| $$\\ $$ | $$ \\ $$ /$$ \\ $$| $$ | $$
| $$ \\/ | $$| $$ \\ $$| $$$$$$$/| $$$$$$/| $$$$$$$/ | $$ \\/ | $$| $$ \\ $$| $$$$$$$/| $$$$$$/| $$$$$$$/
|__/ |__/|__/ \\__/|_______/ \\______/ |_______/""") |__/ |__/|__/ \\__/|_______/ \\______/ |_______/""")
print("") print("\n🤑 Starting downloads from your favorite sellout grifter's wallpaper app...")
print("🤑 Starting downloads from your favorite sellout grifter's wallpaper app...")
if __name__ == "__main__": if __name__ == "__main__":
ascii_art() ascii_art()