feat(music): add song download functionality and rename client class

- Implement download_song method in MusicClient with progress tracking
- Rename songboxClient to SongboxClient for consistency
- Update documentation and examples to reflect class name change
- Add download command to CLI interface
This commit is contained in:
2025-06-07 06:16:23 -07:00
parent 7fe48acdaf
commit ed9f33e71f
8 changed files with 181 additions and 32 deletions

View File

@ -1,7 +1,7 @@
DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
Version 2, December 2004
Copyright (C) 2004 Sam Hocevar <sam@hocevar.net>
Copyright (C) 2025 CustomIcon <custom.icon@vk.com>
Everyone is permitted to copy and distribute verbatim or modified copies of this license document, and changing it is allowed as long as the name is changed.

View File

@ -31,10 +31,10 @@ pip install -e .
### Python Library Usage
```python
from songbox import songboxClient
from songbox import SongboxClient
# Create client
client = songboxClient()
client = SongboxClient()
# Authenticate
verify_id = client.auth.login("960", "7777777")
@ -125,7 +125,7 @@ songbox user me --format table
### Client Configuration
```python
client = songboxClient(
client = SongboxClient(
timeout=30.0, # Request timeout in seconds
headers={"Custom-Header": "value"} # Additional headers
)
@ -195,10 +195,10 @@ songbox music trending [--limit <number>]
### Complete Authentication Flow
```python
from songbox import songboxClient
from songbox import SongboxClient
from songbox.exceptions import AuthenticationError
client = songboxClient()
client = SongboxClient()
try:
# Step 1: Initiate login

14
example.py Normal file
View File

@ -0,0 +1,14 @@
from songbox import SongboxClient
client = SongboxClient()
verify_id = client.auth.login("960", "777777")
otp_code = input("Enter OTP: ")
token = client.auth.verify_otp(otp_code, verify_id)
print(f"Your Token is: {token}")
client.set_token(token)
song = client.music.get_song(6200)
file_path = client.music.download_song(song['url_original'])
print(f"\nDownload completed successfully!")
print(f"File saved to: {file_path}")
client.close()

View File

@ -51,8 +51,8 @@ setup(
},
keywords="songbox music api client streaming maldives",
project_urls={
"Bug Reports": "https://github.com/yourusername/songbox/issues",
"Source": "https://github.com/yourusername/songbox",
"Documentation": "https://github.com/yourusername/songbox#readme",
"Bug Reports": "https://git.cubable.date/CustomIcon/songbox/issues",
"Source": "https://git.cubable.date/CustomIcon/songbox",
"Documentation": "https://git.cubable.date/CustomIcon/songbox/src/branch/master/README.md",
},
)

View File

@ -3,7 +3,7 @@
A Python client library for the songbox music streaming API.
"""
from .client import songboxClient
from .client import SongboxClient
from .auth import AuthClient
from .user import UserClient
from .music import MusicClient
@ -20,7 +20,7 @@ __author__ = "songbox Library"
__email__ = "custom.icon@vk.com"
__all__ = [
"songboxClient",
"SongboxClient",
"AuthClient",
"UserClient",
"MusicClient",

View File

@ -4,7 +4,7 @@ import argparse
import json
import sys
from typing import Optional
from .client import songboxClient
from .client import SongboxClient
from .exceptions import songboxError, AuthenticationError
@ -27,6 +27,10 @@ Examples:
# Get album information
songbox music album 808
# Download a song
songbox music download "https://example.com/song.mp3"
songbox music download "https://example.com/song.mp3" --path ./my_music
# Get user's following list
songbox user following
"""
@ -103,6 +107,11 @@ Examples:
trending_parser = music_subparsers.add_parser("trending", help="Get trending music")
trending_parser.add_argument("--limit", type=int, help="Limit number of results")
# Download command
download_parser = music_subparsers.add_parser("download", help="Download a song")
download_parser.add_argument("song_url", help="Song URL to download (e.g., from song['url_original'])")
download_parser.add_argument("--path", help="Download directory path (default: ./downloads)")
return parser
@ -144,7 +153,7 @@ def format_output(data, format_type: str) -> str:
return str(data)
def handle_auth_commands(client: songboxClient, args) -> None:
def handle_auth_commands(client: SongboxClient, args) -> None:
"""Handle authentication commands."""
if args.auth_command == "login":
try:
@ -179,7 +188,7 @@ def handle_auth_commands(client: songboxClient, args) -> None:
print("Logged out successfully!")
def handle_user_commands(client: songboxClient, args) -> None:
def handle_user_commands(client: SongboxClient, args) -> None:
"""Handle user commands."""
if args.user_command == "me":
try:
@ -201,7 +210,7 @@ def handle_user_commands(client: songboxClient, args) -> None:
sys.exit(1)
def handle_music_commands(client: songboxClient, args) -> None:
def handle_music_commands(client: SongboxClient, args) -> None:
"""Handle music commands."""
if args.music_command == "search":
try:
@ -237,6 +246,15 @@ def handle_music_commands(client: songboxClient, args) -> None:
except Exception as e:
print(f"Failed to get artist: {e}", file=sys.stderr)
sys.exit(1)
elif args.music_command == "download":
try:
file_path = client.music.download_song(args.song_url, args.path)
print(f"\nDownload completed successfully!")
print(f"File saved to: {file_path}")
except Exception as e:
print(f"Download failed: {e}", file=sys.stderr)
sys.exit(1)
@ -255,7 +273,7 @@ def main():
# Create client
try:
client = songboxClient(base_url=args.base_url)
client = SongboxClient(base_url=args.base_url)
# Set token if provided
if token:

View File

@ -8,7 +8,7 @@ from .music import MusicClient
from .exceptions import songboxError, APIError
class songboxClient:
class SongboxClient:
"""Main client for interacting with the songbox API.
This client provides access to all songbox API endpoints through
@ -20,7 +20,7 @@ class songboxClient:
headers: Additional headers to include with requests
Example:
>>> client = songboxClient()
>>> client = SongboxClient()
>>> # Authenticate
>>> verify_id = client.auth.login("960", "7777777")
>>> token = client.auth.verify_otp("123456", verify_id)
@ -50,16 +50,12 @@ class songboxClient:
if headers:
default_headers.update(headers)
# Create HTTP client
self._http_client = httpx.Client(
base_url=self.base_url,
timeout=timeout,
headers=default_headers,
follow_redirects=True
)
# Initialize sub-clients
self.auth = AuthClient(self)
self.user = UserClient(self)
self.music = MusicClient(self)
@ -120,7 +116,6 @@ class songboxClient:
if require_auth and not self.is_authenticated:
raise songboxError("Authentication required for this endpoint")
# Prepare request
url = endpoint if endpoint.startswith('http') else f"/api{endpoint}"
try:
@ -133,18 +128,14 @@ class songboxClient:
headers=headers
)
# Check for HTTP errors
response.raise_for_status()
# Parse JSON response
try:
return response.json()
except ValueError:
# Handle empty responses
return {}
except httpx.HTTPStatusError as e:
# Try to get error message from response
try:
error_data = e.response.json()
message = error_data.get('message', f'HTTP {e.response.status_code}')

View File

@ -1,5 +1,9 @@
"""Music client for songbox API"""
import os
import httpx
from pathlib import Path
from urllib.parse import quote
from typing import Dict, Any, List, Optional
from .exceptions import AuthenticationError, NotFoundError, ValidationError
@ -43,11 +47,9 @@ class MusicClient:
require_auth=True
)
# Ensure we return a list
if isinstance(response, list):
return response
elif isinstance(response, dict):
# If response is a dict, try to extract results
return response.get('results', response.get('data', []))
else:
return []
@ -199,7 +201,47 @@ class MusicClient:
require_auth=True
)
# Ensure we return a list
if isinstance(response, list):
return response
elif isinstance(response, dict):
return response.get('data', response.get('items', []))
else:
return []
except Exception as e:
if "401" in str(e) or "unauthorized" in str(e).lower():
raise AuthenticationError("Authentication required or token expired")
raise
def get_trending(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
"""Get trending music.
Args:
limit: Maximum number of items to return
Returns:
List of trending music items
Raises:
AuthenticationError: If not authenticated
Example:
>>> trending = client.music.get_trending(limit=10)
>>> for item in trending:
... print(f"Trending: {item.get('name', 'Unknown')}")
"""
params = {}
if limit is not None:
params['limit'] = limit
try:
response = self._client.request(
method="GET",
endpoint="/v1/trending",
params=params,
require_auth=True
)
if isinstance(response, list):
return response
elif isinstance(response, dict):
@ -233,7 +275,6 @@ class MusicClient:
require_auth=True
)
# Ensure we return a list
if isinstance(response, list):
return response
elif isinstance(response, dict):
@ -244,4 +285,89 @@ class MusicClient:
except Exception as e:
if "401" in str(e) or "unauthorized" in str(e).lower():
raise AuthenticationError("Authentication required or token expired")
raise
raise
def download_song(self, song_url: str, download_path: Optional[str] = None) -> str:
"""Download a song from the given URL.
Args:
song_url: The song URL (typically from song['url_original'])
download_path: Optional download directory path. If not provided, uses './downloads'
Returns:
Path to the downloaded file
Raises:
ValidationError: If song_url is empty
Exception: If download fails
Example:
>>> song = client.music.get_song("6200")
>>> file_path = client.music.download_song(song['url_original'])
>>> print(f"Downloaded to: {file_path}")
"""
if not song_url or not song_url.strip():
raise ValidationError("Song URL cannot be empty")
if download_path is None:
download_path = os.path.join(os.getcwd(), "downloads")
Path(download_path).mkdir(parents=True, exist_ok=True)
webhook_url = f"https://n8n.cubable.date/webhook/songbox?song_url={quote(song_url)}"
try:
with httpx.Client(follow_redirects=True, timeout=60.0) as download_client:
with download_client.stream("GET", webhook_url) as response:
response.raise_for_status()
filename = None
if "content-disposition" in response.headers:
content_disposition = response.headers["content-disposition"]
if "filename=" in content_disposition:
filename = content_disposition.split("filename=")[1].strip('"')
if not filename:
final_url = str(response.url)
if "/" in final_url:
filename = final_url.split("/")[-1]
if "?" in filename:
filename = filename.split("?")[0]
if not filename or "." not in filename:
filename = "song.mp3"
if not any(filename.lower().endswith(ext) for ext in [".mp3", ".m4a", ".wav", ".flac"]):
filename += ".mp3"
file_path = os.path.join(download_path, filename)
total_size = int(response.headers.get("content-length", 0))
downloaded_size = 0
print(f"Downloading: {filename}")
if total_size > 0:
print(f"File size: {total_size / (1024*1024):.2f} MB")
with open(file_path, "wb") as file:
for chunk in response.iter_bytes(chunk_size=8192):
if chunk:
file.write(chunk)
downloaded_size += len(chunk)
if total_size > 0:
progress = (downloaded_size / total_size) * 100
bar_length = 50
filled_length = int(bar_length * downloaded_size // total_size)
bar = "" * filled_length + "-" * (bar_length - filled_length)
print(f"\r[{bar}] {progress:.1f}% ({downloaded_size / (1024*1024):.2f}/{total_size / (1024*1024):.2f} MB)", end="")
else:
print(f"\rDownloaded: {downloaded_size / (1024*1024):.2f} MB", end="")
print(f"\nDownload completed: {file_path}")
return file_path
except httpx.HTTPStatusError as e:
raise Exception(f"Failed to download song: HTTP {e.response.status_code}")
except httpx.RequestError as e:
raise Exception(f"Failed to download song: {str(e)}")
except Exception as e:
raise Exception(f"Download failed: {str(e)}")