69 lines
1.8 KiB
Python
69 lines
1.8 KiB
Python
"""
|
|
Direct link downloads
|
|
"""
|
|
import aiohttp
|
|
import aiofiles
|
|
from pathlib import Path
|
|
from typing import Optional
|
|
import logging
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def download_file(url: str, output_path: str, chunk_size: int = 8192) -> bool:
|
|
"""
|
|
Download file from direct link
|
|
|
|
Args:
|
|
url: File URL
|
|
output_path: Path to save
|
|
chunk_size: Chunk size for download
|
|
|
|
Returns:
|
|
True if successful, False otherwise
|
|
"""
|
|
try:
|
|
# Create directory if it doesn't exist
|
|
Path(output_path).parent.mkdir(parents=True, exist_ok=True)
|
|
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.get(url) as response:
|
|
if response.status != 200:
|
|
logger.error(f"Download error: HTTP {response.status}")
|
|
return False
|
|
|
|
async with aiofiles.open(output_path, 'wb') as f:
|
|
async for chunk in response.content.iter_chunked(chunk_size):
|
|
await f.write(chunk)
|
|
|
|
logger.info(f"File downloaded: {output_path}")
|
|
return True
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error downloading file: {e}")
|
|
return False
|
|
|
|
|
|
async def get_file_size(url: str) -> Optional[int]:
|
|
"""
|
|
Get file size from URL
|
|
|
|
Args:
|
|
url: File URL
|
|
|
|
Returns:
|
|
File size in bytes or None
|
|
"""
|
|
try:
|
|
async with aiohttp.ClientSession() as session:
|
|
async with session.head(url) as response:
|
|
if response.status == 200:
|
|
content_length = response.headers.get('Content-Length')
|
|
if content_length:
|
|
return int(content_length)
|
|
except Exception as e:
|
|
logger.error(f"Error getting file size: {e}")
|
|
|
|
return None
|
|
|