feat: add initial songbox API client library
This commit introduces the initial version of the songbox API client library, including: - Core client implementation with authentication and HTTP handling - Modular sub-clients for auth, user, and music operations - Comprehensive exception hierarchy for error handling - CLI interface with support for all API operations - Setup configuration with dependencies and package metadata - Documentation including README with usage examples
This commit is contained in:
173
.gitignore
vendored
Normal file
173
.gitignore
vendored
Normal file
@ -0,0 +1,173 @@
|
||||
# ---> Python
|
||||
# Byte-compiled / optimized / DLL files
|
||||
__pycache__/
|
||||
*.py[cod]
|
||||
*$py.class
|
||||
|
||||
# C extensions
|
||||
*.so
|
||||
|
||||
# Distribution / packaging
|
||||
.Python
|
||||
build/
|
||||
develop-eggs/
|
||||
dist/
|
||||
downloads/
|
||||
eggs/
|
||||
.eggs/
|
||||
lib/
|
||||
lib64/
|
||||
parts/
|
||||
sdist/
|
||||
var/
|
||||
wheels/
|
||||
share/python-wheels/
|
||||
*.egg-info/
|
||||
.installed.cfg
|
||||
*.egg
|
||||
MANIFEST
|
||||
|
||||
# PyInstaller
|
||||
# Usually these files are written by a python script from a template
|
||||
# before PyInstaller builds the exe, so as to inject date/other infos into it.
|
||||
*.manifest
|
||||
*.spec
|
||||
|
||||
# Installer logs
|
||||
pip-log.txt
|
||||
pip-delete-this-directory.txt
|
||||
|
||||
# Unit test / coverage reports
|
||||
htmlcov/
|
||||
.tox/
|
||||
.nox/
|
||||
.coverage
|
||||
.coverage.*
|
||||
.cache
|
||||
nosetests.xml
|
||||
coverage.xml
|
||||
*.cover
|
||||
*.py,cover
|
||||
.hypothesis/
|
||||
.pytest_cache/
|
||||
cover/
|
||||
|
||||
# Translations
|
||||
*.mo
|
||||
*.pot
|
||||
|
||||
# Django stuff:
|
||||
*.log
|
||||
local_settings.py
|
||||
db.sqlite3
|
||||
db.sqlite3-journal
|
||||
|
||||
# Flask stuff:
|
||||
instance/
|
||||
.webassets-cache
|
||||
|
||||
# Scrapy stuff:
|
||||
.scrapy
|
||||
|
||||
# Sphinx documentation
|
||||
docs/_build/
|
||||
|
||||
# PyBuilder
|
||||
.pybuilder/
|
||||
target/
|
||||
|
||||
# Jupyter Notebook
|
||||
.ipynb_checkpoints
|
||||
|
||||
# IPython
|
||||
profile_default/
|
||||
ipython_config.py
|
||||
|
||||
# pyenv
|
||||
# For a library or package, you might want to ignore these files since the code is
|
||||
# intended to run in multiple environments; otherwise, check them in:
|
||||
# .python-version
|
||||
|
||||
# pipenv
|
||||
# According to pypa/pipenv#598, it is recommended to include Pipfile.lock in version control.
|
||||
# However, in case of collaboration, if having platform-specific dependencies or dependencies
|
||||
# having no cross-platform support, pipenv may install dependencies that don't work, or not
|
||||
# install all needed dependencies.
|
||||
#Pipfile.lock
|
||||
|
||||
# UV
|
||||
# Similar to Pipfile.lock, it is generally recommended to include uv.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
#uv.lock
|
||||
|
||||
# poetry
|
||||
# Similar to Pipfile.lock, it is generally recommended to include poetry.lock in version control.
|
||||
# This is especially recommended for binary packages to ensure reproducibility, and is more
|
||||
# commonly ignored for libraries.
|
||||
# https://python-poetry.org/docs/basic-usage/#commit-your-poetrylock-file-to-version-control
|
||||
#poetry.lock
|
||||
|
||||
# pdm
|
||||
# Similar to Pipfile.lock, it is generally recommended to include pdm.lock in version control.
|
||||
#pdm.lock
|
||||
# pdm stores project-wide configurations in .pdm.toml, but it is recommended to not include it
|
||||
# in version control.
|
||||
# https://pdm.fming.dev/latest/usage/project/#working-with-version-control
|
||||
.pdm.toml
|
||||
.pdm-python
|
||||
.pdm-build/
|
||||
|
||||
# PEP 582; used by e.g. github.com/David-OConnor/pyflow and github.com/pdm-project/pdm
|
||||
__pypackages__/
|
||||
|
||||
# Celery stuff
|
||||
celerybeat-schedule
|
||||
celerybeat.pid
|
||||
|
||||
# SageMath parsed files
|
||||
*.sage.py
|
||||
|
||||
# Environments
|
||||
.env
|
||||
.venv
|
||||
env/
|
||||
venv/
|
||||
ENV/
|
||||
env.bak/
|
||||
venv.bak/
|
||||
|
||||
# Spyder project settings
|
||||
.spyderproject
|
||||
.spyproject
|
||||
|
||||
# Rope project settings
|
||||
.ropeproject
|
||||
|
||||
# mkdocs documentation
|
||||
/site
|
||||
|
||||
# mypy
|
||||
.mypy_cache/
|
||||
.dmypy.json
|
||||
dmypy.json
|
||||
|
||||
# Pyre type checker
|
||||
.pyre/
|
||||
|
||||
# pytype static type analyzer
|
||||
.pytype/
|
||||
|
||||
# Cython debug symbols
|
||||
cython_debug/
|
||||
|
||||
# PyCharm
|
||||
# JetBrains specific template is maintained in a separate JetBrains.gitignore that can
|
||||
# be found at https://github.com/github/gitignore/blob/main/Global/JetBrains.gitignore
|
||||
# and can be added to the global gitignore or merged into this file. For a more nuclear
|
||||
# option (not recommended) you can uncomment the following to ignore the entire idea folder.
|
||||
#.idea/
|
||||
|
||||
venv/*
|
||||
Documents.md
|
||||
test.py
|
||||
11
LICENSE
Normal file
11
LICENSE
Normal file
@ -0,0 +1,11 @@
|
||||
DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
|
||||
Version 2, December 2004
|
||||
|
||||
Copyright (C) 2004 Sam Hocevar <sam@hocevar.net>
|
||||
|
||||
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.
|
||||
|
||||
DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE
|
||||
TERMS AND CONDITIONS FOR COPYING, DISTRIBUTION AND MODIFICATION
|
||||
|
||||
0. You just DO WHAT THE FUCK YOU WANT TO.
|
||||
284
README.md
Normal file
284
README.md
Normal file
@ -0,0 +1,284 @@
|
||||
# API Client Library
|
||||
|
||||
A comprehensive Python client library for a music streaming API. This library provides both programmatic access and command-line interface (CLI) for interacting with songbox's music platform.
|
||||
|
||||
## Features
|
||||
|
||||
- 🔐 **Authentication**: Login with mobile number and OTP verification
|
||||
- 👤 **User Management**: Get user profile, following lists, and user operations
|
||||
- 🎵 **Music Operations**: Search songs, get albums, artists, and trending music
|
||||
- 🖥️ **CLI Support**: Full command-line interface for all operations
|
||||
- 🔧 **Modular Design**: Separate clients for different API categories
|
||||
- 📦 **Easy Installation**: Simple pip install with all dependencies
|
||||
- 🛡️ **Error Handling**: Comprehensive exception handling with meaningful errors
|
||||
|
||||
## Installation
|
||||
|
||||
```bash
|
||||
pip install songbox
|
||||
```
|
||||
|
||||
Or install from source:
|
||||
|
||||
```bash
|
||||
git clone https://github.com/yourusername/songbox.git
|
||||
cd songbox
|
||||
pip install -e .
|
||||
```
|
||||
|
||||
## Quick Start
|
||||
|
||||
### Python Library Usage
|
||||
|
||||
```python
|
||||
from songbox import songboxClient
|
||||
|
||||
# Create client
|
||||
client = songboxClient()
|
||||
|
||||
# Authenticate
|
||||
verify_id = client.auth.login("960", "7777777")
|
||||
opt_code = input("Enter OTP: ")
|
||||
token = client.auth.verify_otp(otp_code, verify_id)
|
||||
|
||||
# Get user info
|
||||
user_info = client.user.get_me()
|
||||
print(f"Welcome, {user_info['username']}!")
|
||||
|
||||
# Search for music
|
||||
results = client.music.search("huttaa")
|
||||
for category in results:
|
||||
print(f"Category: {category['title']}")
|
||||
for item in category['items']:
|
||||
print(f" - {item['heading']} by {item['sub_heading']}")
|
||||
|
||||
# Get album details
|
||||
album = client.music.get_album("808")
|
||||
print(f"Album: {album['name']} by {album['artist_name']}")
|
||||
print(f"Songs: {len(album['songs'])}")
|
||||
|
||||
# Close client when done
|
||||
client.close()
|
||||
```
|
||||
|
||||
### CLI Usage
|
||||
|
||||
```bash
|
||||
# Login
|
||||
songbox auth login --country-code 960 --mobile 7777777
|
||||
songbox auth verify --otp 123456 --verify-id <verify_id>
|
||||
|
||||
# Set token for subsequent commands
|
||||
export songbox_TOKEN="your_jwt_token_here"
|
||||
|
||||
# Get user info
|
||||
songbox user me
|
||||
|
||||
# Search for music
|
||||
songbox music search "huttaa"
|
||||
|
||||
# Get album information
|
||||
songbox music album 808
|
||||
|
||||
# Get trending music
|
||||
songbox music trending --limit 10
|
||||
|
||||
# Different output formats
|
||||
songbox music search "huttaa" --format json
|
||||
songbox user me --format table
|
||||
```
|
||||
|
||||
## API Reference
|
||||
|
||||
### Authentication (`client.auth`)
|
||||
|
||||
- `login(country_code, mobile_number)` - Initiate login with mobile number
|
||||
- `verify_otp(otp, verify_id)` - Verify OTP and get authentication token
|
||||
- `refresh_token()` - Refresh the current authentication token
|
||||
- `logout()` - Clear authentication token
|
||||
- `get_token_info()` - Get information about current token
|
||||
|
||||
### User Management (`client.user`)
|
||||
|
||||
- `get_me()` - Get current user's profile information
|
||||
- `get_following()` - Get list of followed users/artists
|
||||
- `get_user_profile(user_id)` - Get profile of specific user
|
||||
- `follow_user(user_id)` - Follow a user or artist
|
||||
- `unfollow_user(user_id)` - Unfollow a user or artist
|
||||
- `update_profile(**kwargs)` - Update current user's profile
|
||||
|
||||
### Music Operations (`client.music`)
|
||||
|
||||
- `search(query)` - Search for songs, artists, and albums
|
||||
- `get_album(album_id)` - Get detailed album information
|
||||
- `get_song(song_id)` - Get detailed song information
|
||||
- `get_artist(artist_id)` - Get detailed artist information
|
||||
- `get_recommendations(limit=None)` - Get personalized recommendations
|
||||
- `get_playlists()` - Get user's playlists
|
||||
|
||||
## Configuration
|
||||
|
||||
### Environment Variables
|
||||
|
||||
- `songbox_TOKEN` - Authentication token for CLI usage
|
||||
|
||||
### Client Configuration
|
||||
|
||||
```python
|
||||
client = songboxClient(
|
||||
timeout=30.0, # Request timeout in seconds
|
||||
headers={"Custom-Header": "value"} # Additional headers
|
||||
)
|
||||
```
|
||||
|
||||
## Error Handling
|
||||
|
||||
The library provides specific exceptions for different error types:
|
||||
|
||||
```python
|
||||
from songbox.exceptions import (
|
||||
songboxError,
|
||||
AuthenticationError,
|
||||
APIError,
|
||||
NotFoundError,
|
||||
ValidationError
|
||||
)
|
||||
|
||||
try:
|
||||
user_info = client.user.get_me()
|
||||
except AuthenticationError:
|
||||
print("Please login first")
|
||||
except NotFoundError:
|
||||
print("User not found")
|
||||
except APIError as e:
|
||||
print(f"API Error: {e.message} (Status: {e.status_code})")
|
||||
except songboxError as e:
|
||||
print(f"General error: {e}")
|
||||
```
|
||||
|
||||
## CLI Commands
|
||||
|
||||
### Authentication Commands
|
||||
|
||||
```bash
|
||||
songbox auth login --country-code <code> --mobile <number>
|
||||
songbox auth verify --otp <otp> --verify-id <verify_id>
|
||||
songbox auth refresh
|
||||
songbox auth logout
|
||||
```
|
||||
|
||||
### User Commands
|
||||
|
||||
```bash
|
||||
songbox user me
|
||||
songbox user following
|
||||
```
|
||||
|
||||
### Music Commands
|
||||
|
||||
```bash
|
||||
songbox music search <query>
|
||||
songbox music album <album_id>
|
||||
songbox music song <song_id>
|
||||
songbox music artist <artist_id>
|
||||
songbox music trending [--limit <number>]
|
||||
```
|
||||
|
||||
### Output Formats
|
||||
|
||||
- `--format simple` (default) - Human-readable format
|
||||
- `--format json` - JSON output
|
||||
- `--format table` - Tabular format
|
||||
|
||||
## Examples
|
||||
|
||||
### Complete Authentication Flow
|
||||
|
||||
```python
|
||||
from songbox import songboxClient
|
||||
from songbox.exceptions import AuthenticationError
|
||||
|
||||
client = songboxClient()
|
||||
|
||||
try:
|
||||
# Step 1: Initiate login
|
||||
verify_id = client.auth.login("960", "7777777")
|
||||
print(f"OTP sent! Verification ID: {verify_id}")
|
||||
|
||||
# Step 2: Get OTP from user
|
||||
otp = input("Enter OTP: ")
|
||||
|
||||
# Step 3: Verify OTP and get token
|
||||
token = client.auth.verify_otp(otp, verify_id)
|
||||
print(f"Login successful! Token: {token}")
|
||||
|
||||
# Step 4: Use the API
|
||||
user_info = client.user.get_me()
|
||||
print(f"Welcome, {user_info['username']}!")
|
||||
|
||||
except AuthenticationError as e:
|
||||
print(f"Authentication failed: {e}")
|
||||
finally:
|
||||
client.close()
|
||||
```
|
||||
|
||||
### Music Discovery
|
||||
|
||||
```python
|
||||
# Search and explore music
|
||||
results = client.music.search("maldivian music")
|
||||
|
||||
for category in results:
|
||||
print(f"\n{category['title']}:")
|
||||
for item in category['items'][:3]: # Show top 3
|
||||
print(f" {item['heading']} - {item['sub_heading']}")
|
||||
|
||||
# Get detailed info if it's a song
|
||||
if category['title'] == 'Songs':
|
||||
try:
|
||||
song_details = client.music.get_song(item['destination_id'])
|
||||
print(f" Duration: {song_details.get('duration', 'Unknown')}s")
|
||||
except:
|
||||
pass
|
||||
|
||||
```
|
||||
|
||||
## Development
|
||||
|
||||
### Setting up for Development
|
||||
|
||||
```bash
|
||||
git clone https://github.com/yourusername/songbox.git
|
||||
cd songbox
|
||||
pip install -e ".[dev]"
|
||||
```
|
||||
|
||||
### Running Tests
|
||||
|
||||
```bash
|
||||
pytest
|
||||
```
|
||||
|
||||
### Code Formatting
|
||||
|
||||
```bash
|
||||
black songbox/
|
||||
flake8 songbox/
|
||||
mypy songbox/
|
||||
```
|
||||
|
||||
## Contributing
|
||||
|
||||
1. Fork the repository
|
||||
2. Create a feature branch (`git checkout -b feature/amazing-feature`)
|
||||
3. Commit your changes (`git commit -m 'Add amazing feature'`)
|
||||
4. Push to the branch (`git push origin feature/amazing-feature`)
|
||||
5. Open a Pull Request
|
||||
|
||||
## License
|
||||
|
||||
This project is licensed under the MIT License - see the [LICENSE](LICENSE) file for details.
|
||||
|
||||
---
|
||||
|
||||
**Note**: This is an unofficial client library. songbox is a trademark of its respective owners.
|
||||
1
requirements.txt
Normal file
1
requirements.txt
Normal file
@ -0,0 +1 @@
|
||||
httpx==0.28.1
|
||||
58
setup.py
Normal file
58
setup.py
Normal file
@ -0,0 +1,58 @@
|
||||
"""Setup script for songbox API Client Library"""
|
||||
|
||||
from setuptools import setup, find_packages
|
||||
|
||||
with open("README.md", "r", encoding="utf-8") as fh:
|
||||
long_description = fh.read()
|
||||
|
||||
with open("requirements.txt", "r", encoding="utf-8") as fh:
|
||||
requirements = [line.strip() for line in fh if line.strip() and not line.startswith("#")]
|
||||
|
||||
setup(
|
||||
name="songbox",
|
||||
version="1.0.0",
|
||||
author="CustomIcon",
|
||||
author_email="custom.icon@vk.com",
|
||||
description="A Python client library for a music streaming API",
|
||||
long_description=long_description,
|
||||
long_description_content_type="text/markdown",
|
||||
url="https://git.cubable.date/CustomIcon/songbox",
|
||||
packages=find_packages(),
|
||||
classifiers=[
|
||||
"Development Status :: 4 - Beta",
|
||||
"Intended Audience :: Developers",
|
||||
"License :: OSI Approved :: MIT License",
|
||||
"Operating System :: OS Independent",
|
||||
"Programming Language :: Python :: 3",
|
||||
"Programming Language :: Python :: 3.7",
|
||||
"Programming Language :: Python :: 3.8",
|
||||
"Programming Language :: Python :: 3.9",
|
||||
"Programming Language :: Python :: 3.10",
|
||||
"Programming Language :: Python :: 3.11",
|
||||
"Topic :: Software Development :: Libraries :: Python Modules",
|
||||
"Topic :: Multimedia :: Sound/Audio",
|
||||
"Topic :: Internet :: WWW/HTTP :: Dynamic Content",
|
||||
],
|
||||
python_requires=">=3.7",
|
||||
install_requires=requirements,
|
||||
extras_require={
|
||||
"dev": [
|
||||
"pytest>=6.0",
|
||||
"pytest-asyncio",
|
||||
"black",
|
||||
"flake8",
|
||||
"mypy",
|
||||
],
|
||||
},
|
||||
entry_points={
|
||||
"console_scripts": [
|
||||
"songbox=songbox.__main__:main",
|
||||
],
|
||||
},
|
||||
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",
|
||||
},
|
||||
)
|
||||
32
songbox/__init__.py
Normal file
32
songbox/__init__.py
Normal file
@ -0,0 +1,32 @@
|
||||
"""songbox API Client Library
|
||||
|
||||
A Python client library for the songbox music streaming API.
|
||||
"""
|
||||
|
||||
from .client import songboxClient
|
||||
from .auth import AuthClient
|
||||
from .user import UserClient
|
||||
from .music import MusicClient
|
||||
from .exceptions import (
|
||||
songboxError,
|
||||
AuthenticationError,
|
||||
APIError,
|
||||
RateLimitError,
|
||||
NotFoundError
|
||||
)
|
||||
|
||||
__version__ = "1.0.0"
|
||||
__author__ = "songbox Library"
|
||||
__email__ = "custom.icon@vk.com"
|
||||
|
||||
__all__ = [
|
||||
"songboxClient",
|
||||
"AuthClient",
|
||||
"UserClient",
|
||||
"MusicClient",
|
||||
"songboxError",
|
||||
"AuthenticationError",
|
||||
"APIError",
|
||||
"RateLimitError",
|
||||
"NotFoundError",
|
||||
]
|
||||
290
songbox/__main__.py
Normal file
290
songbox/__main__.py
Normal file
@ -0,0 +1,290 @@
|
||||
"""Command-line interface for songbox API Client"""
|
||||
|
||||
import argparse
|
||||
import json
|
||||
import sys
|
||||
from typing import Optional
|
||||
from .client import songboxClient
|
||||
from .exceptions import songboxError, AuthenticationError
|
||||
|
||||
|
||||
def create_parser() -> argparse.ArgumentParser:
|
||||
"""Create the argument parser for the CLI."""
|
||||
parser = argparse.ArgumentParser(
|
||||
prog="songbox",
|
||||
description="songbox API Client - Access songbox music streaming API from command line",
|
||||
formatter_class=argparse.RawDescriptionHelpFormatter,
|
||||
epilog="""
|
||||
Examples:
|
||||
# Login and get user info
|
||||
songbox auth login --country-code 960 --mobile 7777777
|
||||
songbox auth verify --otp 123456 --verify-id <verify_id>
|
||||
songbox user me
|
||||
|
||||
# Search for music
|
||||
songbox music search "huttaa"
|
||||
|
||||
# Get album information
|
||||
songbox music album 808
|
||||
|
||||
# Get user's following list
|
||||
songbox user following
|
||||
"""
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--token",
|
||||
help="Authentication token (can also be set via songbox_TOKEN env var)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--base-url",
|
||||
default="https://lavafoshi.online",
|
||||
help="Base URL for the API (default: https://lavafoshi.online)"
|
||||
)
|
||||
|
||||
parser.add_argument(
|
||||
"--format",
|
||||
choices=["json", "table", "simple"],
|
||||
default="simple",
|
||||
help="Output format (default: simple)"
|
||||
)
|
||||
|
||||
subparsers = parser.add_subparsers(dest="command", help="Available commands")
|
||||
|
||||
# Auth commands
|
||||
auth_parser = subparsers.add_parser("auth", help="Authentication commands")
|
||||
auth_subparsers = auth_parser.add_subparsers(dest="auth_command")
|
||||
|
||||
# Login command
|
||||
login_parser = auth_subparsers.add_parser("login", help="Initiate login with mobile number")
|
||||
login_parser.add_argument("--country-code", required=True, help="Country code (e.g., 960)")
|
||||
login_parser.add_argument("--mobile", required=True, help="Mobile number")
|
||||
|
||||
# Verify OTP command
|
||||
verify_parser = auth_subparsers.add_parser("verify", help="Verify OTP and get token")
|
||||
verify_parser.add_argument("--otp", required=True, help="OTP code")
|
||||
verify_parser.add_argument("--verify-id", required=True, help="Verification ID from login")
|
||||
|
||||
# Refresh token command
|
||||
auth_subparsers.add_parser("refresh", help="Refresh authentication token")
|
||||
|
||||
# Logout command
|
||||
auth_subparsers.add_parser("logout", help="Logout (clear token)")
|
||||
|
||||
# User commands
|
||||
user_parser = subparsers.add_parser("user", help="User commands")
|
||||
user_subparsers = user_parser.add_subparsers(dest="user_command")
|
||||
|
||||
user_subparsers.add_parser("me", help="Get current user information")
|
||||
user_subparsers.add_parser("following", help="Get following list")
|
||||
|
||||
# Music commands
|
||||
music_parser = subparsers.add_parser("music", help="Music commands")
|
||||
music_subparsers = music_parser.add_subparsers(dest="music_command")
|
||||
|
||||
# Search command
|
||||
search_parser = music_subparsers.add_parser("search", help="Search for music")
|
||||
search_parser.add_argument("query", help="Search query")
|
||||
|
||||
# Album command
|
||||
album_parser = music_subparsers.add_parser("album", help="Get album information")
|
||||
album_parser.add_argument("album_id", help="Album ID")
|
||||
|
||||
# Song command
|
||||
song_parser = music_subparsers.add_parser("song", help="Get song information")
|
||||
song_parser.add_argument("song_id", help="Song ID")
|
||||
|
||||
# Artist command
|
||||
artist_parser = music_subparsers.add_parser("artist", help="Get artist information")
|
||||
artist_parser.add_argument("artist_id", help="Artist ID")
|
||||
|
||||
# Trending command
|
||||
trending_parser = music_subparsers.add_parser("trending", help="Get trending music")
|
||||
trending_parser.add_argument("--limit", type=int, help="Limit number of results")
|
||||
|
||||
return parser
|
||||
|
||||
|
||||
def format_output(data, format_type: str) -> str:
|
||||
"""Format output based on the specified format."""
|
||||
if format_type == "json":
|
||||
return json.dumps(data, indent=2, ensure_ascii=False)
|
||||
elif format_type == "table":
|
||||
# Simple table format for lists
|
||||
if isinstance(data, list) and data:
|
||||
if isinstance(data[0], dict):
|
||||
# Print headers
|
||||
headers = list(data[0].keys())
|
||||
result = "\t".join(headers) + "\n"
|
||||
# Print rows
|
||||
for item in data:
|
||||
row = [str(item.get(h, "")) for h in headers]
|
||||
result += "\t".join(row) + "\n"
|
||||
return result
|
||||
return str(data)
|
||||
else:
|
||||
if isinstance(data, dict):
|
||||
result = ""
|
||||
for key, value in data.items():
|
||||
result += f"{key}: {value}\n"
|
||||
return result
|
||||
elif isinstance(data, list):
|
||||
result = ""
|
||||
for i, item in enumerate(data):
|
||||
if isinstance(item, dict):
|
||||
result += f"Item {i + 1}:\n"
|
||||
for key, value in item.items():
|
||||
result += f" {key}: {value}\n"
|
||||
result += "\n"
|
||||
else:
|
||||
result += f"{i + 1}. {item}\n"
|
||||
return result
|
||||
else:
|
||||
return str(data)
|
||||
|
||||
|
||||
def handle_auth_commands(client: songboxClient, args) -> None:
|
||||
"""Handle authentication commands."""
|
||||
if args.auth_command == "login":
|
||||
try:
|
||||
verify_id = client.auth.login(args.country_code, args.mobile)
|
||||
print(f"OTP sent! Verification ID: {verify_id}")
|
||||
print(f"Use this command to verify: songbox auth verify --otp <OTP> --verify-id {verify_id}")
|
||||
except Exception as e:
|
||||
print(f"Login failed: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.auth_command == "verify":
|
||||
try:
|
||||
token = client.auth.verify_otp(args.otp, args.verify_id)
|
||||
print(f"Authentication successful!")
|
||||
print(f"Token: {token}")
|
||||
print(f"Save this token and use it with --token flag or songbox_TOKEN env var")
|
||||
except Exception as e:
|
||||
print(f"Verification failed: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.auth_command == "refresh":
|
||||
try:
|
||||
new_token = client.auth.refresh_token()
|
||||
print(f"Token refreshed successfully!")
|
||||
print(f"New token: {new_token}")
|
||||
except Exception as e:
|
||||
print(f"Token refresh failed: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.auth_command == "logout":
|
||||
client.auth.logout()
|
||||
print("Logged out successfully!")
|
||||
|
||||
|
||||
def handle_user_commands(client: songboxClient, args) -> None:
|
||||
"""Handle user commands."""
|
||||
if args.user_command == "me":
|
||||
try:
|
||||
user_info = client.user.get_me()
|
||||
print(format_output(user_info, args.format))
|
||||
except Exception as e:
|
||||
print(f"Failed to get user info: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.user_command == "following":
|
||||
try:
|
||||
following = client.user.get_following()
|
||||
if not following:
|
||||
print("Not following anyone yet.")
|
||||
else:
|
||||
print(format_output(following, args.format))
|
||||
except Exception as e:
|
||||
print(f"Failed to get following list: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
def handle_music_commands(client: songboxClient, args) -> None:
|
||||
"""Handle music commands."""
|
||||
if args.music_command == "search":
|
||||
try:
|
||||
results = client.music.search(args.query)
|
||||
if not results:
|
||||
print(f"No results found for '{args.query}'")
|
||||
else:
|
||||
print(format_output(results, args.format))
|
||||
except Exception as e:
|
||||
print(f"Search failed: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.music_command == "album":
|
||||
try:
|
||||
album = client.music.get_album(args.album_id)
|
||||
print(format_output(album, args.format))
|
||||
except Exception as e:
|
||||
print(f"Failed to get album: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.music_command == "song":
|
||||
try:
|
||||
song = client.music.get_song(args.song_id)
|
||||
print(format_output(song, args.format))
|
||||
except Exception as e:
|
||||
print(f"Failed to get song: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
elif args.music_command == "artist":
|
||||
try:
|
||||
artist = client.music.get_artist(args.artist_id)
|
||||
print(format_output(artist, args.format))
|
||||
except Exception as e:
|
||||
print(f"Failed to get artist: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
|
||||
|
||||
|
||||
def main():
|
||||
"""Main CLI entry point."""
|
||||
parser = create_parser()
|
||||
args = parser.parse_args()
|
||||
|
||||
if not args.command:
|
||||
parser.print_help()
|
||||
sys.exit(1)
|
||||
|
||||
# Get token from args or environment
|
||||
import os
|
||||
token = args.token or os.getenv("songbox_TOKEN")
|
||||
|
||||
# Create client
|
||||
try:
|
||||
client = songboxClient(base_url=args.base_url)
|
||||
|
||||
# Set token if provided
|
||||
if token:
|
||||
client.set_token(token)
|
||||
|
||||
# Handle commands
|
||||
if args.command == "auth":
|
||||
handle_auth_commands(client, args)
|
||||
elif args.command == "user":
|
||||
if not client.is_authenticated:
|
||||
print("Authentication required. Please login first or provide a token.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
handle_user_commands(client, args)
|
||||
elif args.command == "music":
|
||||
if not client.is_authenticated:
|
||||
print("Authentication required. Please login first or provide a token.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
handle_music_commands(client, args)
|
||||
|
||||
except KeyboardInterrupt:
|
||||
print("\nOperation cancelled.", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
except Exception as e:
|
||||
print(f"Error: {e}", file=sys.stderr)
|
||||
sys.exit(1)
|
||||
finally:
|
||||
if 'client' in locals():
|
||||
client.close()
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
178
songbox/auth.py
Normal file
178
songbox/auth.py
Normal file
@ -0,0 +1,178 @@
|
||||
"""Authentication client for songbox API"""
|
||||
|
||||
from typing import Dict, Any, Optional
|
||||
from .exceptions import AuthenticationError, ValidationError
|
||||
|
||||
|
||||
class AuthClient:
|
||||
"""Client for handling authentication with the songbox API.
|
||||
|
||||
This client handles login, OTP verification, and token refresh operations.
|
||||
"""
|
||||
|
||||
def __init__(self, client):
|
||||
self._client = client
|
||||
|
||||
def login(self, country_code: str, mobile_number: str) -> str:
|
||||
"""Initiate login process by sending OTP to mobile number.
|
||||
|
||||
Args:
|
||||
country_code: Country code (e.g., "960" for Maldives)
|
||||
mobile_number: Mobile number without country code
|
||||
|
||||
Returns:
|
||||
verify_id: ID needed for OTP verification
|
||||
|
||||
Raises:
|
||||
ValidationError: If input parameters are invalid
|
||||
AuthenticationError: If login initiation fails
|
||||
|
||||
Example:
|
||||
>>> auth = AuthClient(client)
|
||||
>>> verify_id = auth.login("960", "7777777")
|
||||
"""
|
||||
if not country_code or not mobile_number:
|
||||
raise ValidationError("Country code and mobile number are required")
|
||||
|
||||
data = {
|
||||
"country_code": country_code,
|
||||
"mobile_number": mobile_number
|
||||
}
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="POST",
|
||||
endpoint="/v2/auth/init",
|
||||
data=data,
|
||||
require_auth=False
|
||||
)
|
||||
|
||||
verify_id = response.get("verify_id")
|
||||
if not verify_id:
|
||||
raise AuthenticationError("Failed to get verification ID")
|
||||
|
||||
return verify_id
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, (AuthenticationError, ValidationError)):
|
||||
raise
|
||||
raise AuthenticationError(f"Login failed: {str(e)}")
|
||||
|
||||
def verify_otp(self, otp: str, verify_id: str) -> str:
|
||||
"""Verify OTP and get authentication token.
|
||||
|
||||
Args:
|
||||
otp: One-time password received via SMS
|
||||
verify_id: Verification ID from login step
|
||||
|
||||
Returns:
|
||||
token: JWT authentication token
|
||||
|
||||
Raises:
|
||||
ValidationError: If input parameters are invalid
|
||||
AuthenticationError: If OTP verification fails
|
||||
|
||||
Example:
|
||||
>>> token = auth.verify_otp("123456", verify_id)
|
||||
>>> client.set_token(token)
|
||||
"""
|
||||
if not otp or not verify_id:
|
||||
raise ValidationError("OTP and verify_id are required")
|
||||
|
||||
data = {
|
||||
"otp": otp,
|
||||
"verify_id": verify_id
|
||||
}
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="POST",
|
||||
endpoint="/v2/sign_in?auth_method=otp",
|
||||
data=data,
|
||||
require_auth=False
|
||||
)
|
||||
|
||||
token = response.get("token")
|
||||
if not token:
|
||||
raise AuthenticationError("Failed to get authentication token")
|
||||
|
||||
# Automatically set the token in the client
|
||||
self._client.set_token(token)
|
||||
|
||||
return token
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, (AuthenticationError, ValidationError)):
|
||||
raise
|
||||
raise AuthenticationError(f"OTP verification failed: {str(e)}")
|
||||
|
||||
def refresh_token(self) -> str:
|
||||
"""Refresh the current authentication token.
|
||||
|
||||
Returns:
|
||||
token: New JWT authentication token
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If token refresh fails
|
||||
|
||||
Example:
|
||||
>>> new_token = auth.refresh_token()
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/refresh_token",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
token = response.get("token")
|
||||
if not token:
|
||||
raise AuthenticationError("Failed to refresh token")
|
||||
|
||||
# Update the token in the client
|
||||
self._client.set_token(token)
|
||||
|
||||
return token
|
||||
|
||||
except Exception as e:
|
||||
if isinstance(e, AuthenticationError):
|
||||
raise
|
||||
raise AuthenticationError(f"Token refresh failed: {str(e)}")
|
||||
|
||||
def logout(self) -> None:
|
||||
"""Logout by clearing the authentication token.
|
||||
|
||||
Example:
|
||||
>>> auth.logout()
|
||||
"""
|
||||
self._client.clear_token()
|
||||
|
||||
def get_token_info(self) -> Optional[Dict[str, Any]]:
|
||||
"""Get information about the current token.
|
||||
|
||||
Returns:
|
||||
Token information including expiry date, or None if not authenticated
|
||||
|
||||
Example:
|
||||
>>> info = auth.get_token_info()
|
||||
>>> if info:
|
||||
... print(f"Token expires: {info.get('expire')}")
|
||||
"""
|
||||
if not self._client.is_authenticated:
|
||||
return None
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/refresh_token",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return {
|
||||
"token": response.get("token"),
|
||||
"expire": response.get("expire"),
|
||||
"code": response.get("code")
|
||||
}
|
||||
|
||||
except Exception:
|
||||
return None
|
||||
167
songbox/client.py
Normal file
167
songbox/client.py
Normal file
@ -0,0 +1,167 @@
|
||||
"""Main songbox API Client"""
|
||||
|
||||
import httpx
|
||||
from typing import Optional, Dict, Any
|
||||
from .auth import AuthClient
|
||||
from .user import UserClient
|
||||
from .music import MusicClient
|
||||
from .exceptions import songboxError, APIError
|
||||
|
||||
|
||||
class songboxClient:
|
||||
"""Main client for interacting with the songbox API.
|
||||
|
||||
This client provides access to all songbox API endpoints through
|
||||
modular sub-clients for authentication, user management, and music.
|
||||
|
||||
Args:
|
||||
base_url: The base URL for the songbox API
|
||||
timeout: Request timeout in seconds (default: 30)
|
||||
headers: Additional headers to include with requests
|
||||
|
||||
Example:
|
||||
>>> client = songboxClient()
|
||||
>>> # Authenticate
|
||||
>>> verify_id = client.auth.login("960", "7777777")
|
||||
>>> token = client.auth.verify_otp("123456", verify_id)
|
||||
>>> client.set_token(token)
|
||||
>>> # Get user info
|
||||
>>> user_info = client.user.get_me()
|
||||
>>> print(user_info)
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
base_url: str = "https://lavafoshi.online",
|
||||
timeout: float = 30.0,
|
||||
headers: Optional[Dict[str, str]] = None
|
||||
):
|
||||
self.base_url = base_url.rstrip('/')
|
||||
self._token: Optional[str] = None
|
||||
|
||||
# Default headers
|
||||
default_headers = {
|
||||
"User-Agent": "songbox/2.1.4 (songbox; Python) httpx",
|
||||
"Accept": "*/*",
|
||||
"Accept-Language": "en-US;q=1.0",
|
||||
"Accept-Encoding": "gzip;q=1.0, compress;q=0.5",
|
||||
"Connection": "keep-alive"
|
||||
}
|
||||
|
||||
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)
|
||||
|
||||
def set_token(self, token: str) -> None:
|
||||
"""Set the authentication token for API requests.
|
||||
|
||||
Args:
|
||||
token: JWT token obtained from authentication
|
||||
"""
|
||||
self._token = token
|
||||
self._http_client.headers["Authorization"] = f"Bearer {token}"
|
||||
|
||||
def clear_token(self) -> None:
|
||||
"""Clear the authentication token."""
|
||||
self._token = None
|
||||
if "Authorization" in self._http_client.headers:
|
||||
del self._http_client.headers["Authorization"]
|
||||
|
||||
@property
|
||||
def token(self) -> Optional[str]:
|
||||
"""Get the current authentication token."""
|
||||
return self._token
|
||||
|
||||
@property
|
||||
def is_authenticated(self) -> bool:
|
||||
"""Check if the client is authenticated."""
|
||||
return self._token is not None
|
||||
|
||||
def request(
|
||||
self,
|
||||
method: str,
|
||||
endpoint: str,
|
||||
params: Optional[Dict[str, Any]] = None,
|
||||
data: Optional[Dict[str, Any]] = None,
|
||||
json: Optional[Dict[str, Any]] = None,
|
||||
headers: Optional[Dict[str, str]] = None,
|
||||
require_auth: bool = True
|
||||
) -> Dict[str, Any]:
|
||||
"""Make an HTTP request to the songbox API.
|
||||
|
||||
Args:
|
||||
method: HTTP method (GET, POST, etc.)
|
||||
endpoint: API endpoint (without base URL)
|
||||
params: Query parameters
|
||||
data: Form data for POST requests
|
||||
json: JSON data for POST requests
|
||||
headers: Additional headers
|
||||
require_auth: Whether authentication is required
|
||||
|
||||
Returns:
|
||||
JSON response data
|
||||
|
||||
Raises:
|
||||
songboxError: If authentication is required but not provided
|
||||
APIError: If the API returns an error response
|
||||
"""
|
||||
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:
|
||||
response = self._http_client.request(
|
||||
method=method,
|
||||
url=url,
|
||||
params=params,
|
||||
data=data,
|
||||
json=json,
|
||||
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}')
|
||||
except ValueError:
|
||||
message = f'HTTP {e.response.status_code}'
|
||||
|
||||
raise APIError(message, status_code=e.response.status_code)
|
||||
|
||||
except httpx.RequestError as e:
|
||||
raise songboxError(f"Request failed: {str(e)}")
|
||||
|
||||
def close(self) -> None:
|
||||
"""Close the HTTP client."""
|
||||
self._http_client.close()
|
||||
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||
self.close()
|
||||
54
songbox/exceptions.py
Normal file
54
songbox/exceptions.py
Normal file
@ -0,0 +1,54 @@
|
||||
"""songbox API Client Exceptions"""
|
||||
|
||||
from typing import Optional
|
||||
|
||||
|
||||
class songboxError(Exception):
|
||||
"""Base exception for all songbox API errors."""
|
||||
pass
|
||||
|
||||
|
||||
class APIError(songboxError):
|
||||
"""Exception raised when the API returns an error response.
|
||||
|
||||
Args:
|
||||
message: Error message
|
||||
status_code: HTTP status code
|
||||
response_data: Raw response data from the API
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
message: str,
|
||||
status_code: Optional[int] = None,
|
||||
response_data: Optional[dict] = None
|
||||
):
|
||||
super().__init__(message)
|
||||
self.message = message
|
||||
self.status_code = status_code
|
||||
self.response_data = response_data or {}
|
||||
|
||||
def __str__(self) -> str:
|
||||
if self.status_code:
|
||||
return f"API Error {self.status_code}: {self.message}"
|
||||
return f"API Error: {self.message}"
|
||||
|
||||
|
||||
class AuthenticationError(songboxError):
|
||||
"""Exception raised for authentication-related errors."""
|
||||
pass
|
||||
|
||||
|
||||
class RateLimitError(APIError):
|
||||
"""Exception raised when rate limit is exceeded."""
|
||||
pass
|
||||
|
||||
|
||||
class NotFoundError(APIError):
|
||||
"""Exception raised when a resource is not found (404)."""
|
||||
pass
|
||||
|
||||
|
||||
class ValidationError(songboxError):
|
||||
"""Exception raised for input validation errors."""
|
||||
pass
|
||||
247
songbox/music.py
Normal file
247
songbox/music.py
Normal file
@ -0,0 +1,247 @@
|
||||
"""Music client for songbox API"""
|
||||
|
||||
from typing import Dict, Any, List, Optional
|
||||
from .exceptions import AuthenticationError, NotFoundError, ValidationError
|
||||
|
||||
|
||||
class MusicClient:
|
||||
"""Client for handling music-related operations with the songbox API.
|
||||
|
||||
This client handles search, albums, songs, and other music operations.
|
||||
"""
|
||||
|
||||
def __init__(self, client):
|
||||
self._client = client
|
||||
|
||||
def search(self, query: str) -> List[Dict[str, Any]]:
|
||||
"""Search for songs, artists, and albums.
|
||||
|
||||
Args:
|
||||
query: Search query string
|
||||
|
||||
Returns:
|
||||
List of search results categorized by type (Songs, Artists, Albums)
|
||||
|
||||
Raises:
|
||||
ValidationError: If query is empty
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> results = client.music.search("huttaa")
|
||||
>>> for category in results:
|
||||
... print(f"Category: {category['title']}")
|
||||
... for item in category['items']:
|
||||
... print(f" - {item['heading']}")
|
||||
"""
|
||||
if not query or not query.strip():
|
||||
raise ValidationError("Search query cannot be empty")
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint=f"/v1/search/{query.strip()}",
|
||||
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 []
|
||||
|
||||
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_album(self, album_id: str) -> Dict[str, Any]:
|
||||
"""Get detailed information about an album.
|
||||
|
||||
Args:
|
||||
album_id: ID of the album
|
||||
|
||||
Returns:
|
||||
Album information including songs, artist, artwork, etc.
|
||||
|
||||
Raises:
|
||||
ValidationError: If album_id is empty
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If album not found
|
||||
|
||||
Example:
|
||||
>>> album = client.music.get_album("808")
|
||||
>>> print(f"Album: {album['name']} by {album['artist_name']}")
|
||||
>>> print(f"Songs: {len(album['songs'])}")
|
||||
"""
|
||||
if not album_id or not str(album_id).strip():
|
||||
raise ValidationError("Album ID cannot be empty")
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint=f"/v1/albums/{album_id}",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"Album with ID {album_id} not found")
|
||||
raise
|
||||
|
||||
def get_song(self, song_id: str) -> Dict[str, Any]:
|
||||
"""Get detailed information about a song.
|
||||
|
||||
Args:
|
||||
song_id: ID of the song
|
||||
|
||||
Returns:
|
||||
Song information including artist, album, duration, URLs, etc.
|
||||
|
||||
Raises:
|
||||
ValidationError: If song_id is empty
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If song not found
|
||||
|
||||
Example:
|
||||
>>> song = client.music.get_song("6200")
|
||||
>>> print(f"Song: {song['name']} by {song['artist_name']}")
|
||||
>>> print(f"Duration: {song['duration']} seconds")
|
||||
"""
|
||||
if not song_id or not str(song_id).strip():
|
||||
raise ValidationError("Song ID cannot be empty")
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint=f"/v1/songs/{song_id}",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"Song with ID {song_id} not found")
|
||||
raise
|
||||
|
||||
def get_artist(self, artist_id: str) -> Dict[str, Any]:
|
||||
"""Get detailed information about an artist.
|
||||
|
||||
Args:
|
||||
artist_id: ID of the artist
|
||||
|
||||
Returns:
|
||||
Artist information including albums, songs, etc.
|
||||
|
||||
Raises:
|
||||
ValidationError: If artist_id is empty
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If artist not found
|
||||
|
||||
Example:
|
||||
>>> artist = client.music.get_artist("124")
|
||||
>>> print(f"Artist: {artist['name']}")
|
||||
"""
|
||||
if not artist_id or not str(artist_id).strip():
|
||||
raise ValidationError("Artist ID cannot be empty")
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint=f"/v1/artists/{artist_id}",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"Artist with ID {artist_id} not found")
|
||||
raise
|
||||
|
||||
def get_recommendations(self, limit: Optional[int] = None) -> List[Dict[str, Any]]:
|
||||
"""Get personalized music recommendations.
|
||||
|
||||
Args:
|
||||
limit: Maximum number of items to return
|
||||
|
||||
Returns:
|
||||
List of recommended music items
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> recommendations = client.music.get_recommendations(limit=5)
|
||||
>>> for item in recommendations:
|
||||
... print(f"Recommended: {item.get('name', 'Unknown')}")
|
||||
"""
|
||||
params = {}
|
||||
if limit is not None:
|
||||
params['limit'] = limit
|
||||
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/recommendations",
|
||||
params=params,
|
||||
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_playlists(self) -> List[Dict[str, Any]]:
|
||||
"""Get user's playlists.
|
||||
|
||||
Returns:
|
||||
List of user's playlists
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> playlists = client.music.get_playlists()
|
||||
>>> for playlist in playlists:
|
||||
... print(f"Playlist: {playlist.get('name', 'Unknown')}")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/playlists",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
# Ensure we return a list
|
||||
if isinstance(response, list):
|
||||
return response
|
||||
elif isinstance(response, dict):
|
||||
return response.get('data', response.get('playlists', []))
|
||||
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
|
||||
216
songbox/user.py
Normal file
216
songbox/user.py
Normal file
@ -0,0 +1,216 @@
|
||||
"""User client for songbox API"""
|
||||
|
||||
from typing import Dict, Any, List, Optional
|
||||
from .exceptions import AuthenticationError, NotFoundError
|
||||
|
||||
|
||||
class UserClient:
|
||||
"""Client for handling user-related operations with the songbox API.
|
||||
|
||||
This client handles user profile information, following, and other user operations.
|
||||
"""
|
||||
|
||||
def __init__(self, client):
|
||||
self._client = client
|
||||
|
||||
def get_me(self) -> Dict[str, Any]:
|
||||
"""Get current user's profile information.
|
||||
|
||||
Returns:
|
||||
User profile data including id, username, email, subscription info
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> user_info = client.user.get_me()
|
||||
>>> print(f"Username: {user_info['username']}")
|
||||
>>> print(f"Subscription: {user_info['subscription']}")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/me",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
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_following(self) -> List[Dict[str, Any]]:
|
||||
"""Get list of users/artists that the current user is following.
|
||||
|
||||
Returns:
|
||||
List of followed users/artists
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> following = client.user.get_following()
|
||||
>>> for user in following:
|
||||
... print(f"Following: {user.get('name', 'Unknown')}")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint="/v1/me/following",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
# Handle empty response (API returns {} when no following)
|
||||
if isinstance(response, dict) and not response:
|
||||
return []
|
||||
|
||||
# If response is a list, return it
|
||||
if isinstance(response, list):
|
||||
return response
|
||||
|
||||
# If response has a 'data' or 'following' key, return that
|
||||
if isinstance(response, dict):
|
||||
return response.get('data', response.get('following', []))
|
||||
|
||||
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_user_profile(self, user_id: str) -> Dict[str, Any]:
|
||||
"""Get profile information for a specific user.
|
||||
|
||||
Args:
|
||||
user_id: ID of the user to get profile for
|
||||
|
||||
Returns:
|
||||
User profile data
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If user not found
|
||||
|
||||
Example:
|
||||
>>> profile = client.user.get_user_profile("12345")
|
||||
>>> print(f"User: {profile['username']}")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="GET",
|
||||
endpoint=f"/v1/users/{user_id}",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"User with ID {user_id} not found")
|
||||
raise
|
||||
|
||||
def follow_user(self, user_id: str) -> bool:
|
||||
"""Follow a user or artist.
|
||||
|
||||
Args:
|
||||
user_id: ID of the user/artist to follow
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If user not found
|
||||
|
||||
Example:
|
||||
>>> success = client.user.follow_user("12345")
|
||||
>>> if success:
|
||||
... print("Successfully followed user")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="POST",
|
||||
endpoint=f"/v1/users/{user_id}/follow",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
# Check if the response indicates success
|
||||
return response.get('success', True)
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"User with ID {user_id} not found")
|
||||
raise
|
||||
|
||||
def unfollow_user(self, user_id: str) -> bool:
|
||||
"""Unfollow a user or artist.
|
||||
|
||||
Args:
|
||||
user_id: ID of the user/artist to unfollow
|
||||
|
||||
Returns:
|
||||
True if successful
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
NotFoundError: If user not found
|
||||
|
||||
Example:
|
||||
>>> success = client.user.unfollow_user("12345")
|
||||
>>> if success:
|
||||
... print("Successfully unfollowed user")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="DELETE",
|
||||
endpoint=f"/v1/users/{user_id}/follow",
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
# Check if the response indicates success
|
||||
return response.get('success', True)
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
elif "404" in str(e) or "not found" in str(e).lower():
|
||||
raise NotFoundError(f"User with ID {user_id} not found")
|
||||
raise
|
||||
|
||||
def update_profile(self, **kwargs) -> Dict[str, Any]:
|
||||
"""Update current user's profile information.
|
||||
|
||||
Args:
|
||||
**kwargs: Profile fields to update (e.g., username, email)
|
||||
|
||||
Returns:
|
||||
Updated profile data
|
||||
|
||||
Raises:
|
||||
AuthenticationError: If not authenticated
|
||||
|
||||
Example:
|
||||
>>> updated = client.user.update_profile(username="newname")
|
||||
>>> print(f"Updated username: {updated['username']}")
|
||||
"""
|
||||
try:
|
||||
response = self._client.request(
|
||||
method="PUT",
|
||||
endpoint="/v1/me",
|
||||
json=kwargs,
|
||||
require_auth=True
|
||||
)
|
||||
|
||||
return response
|
||||
|
||||
except Exception as e:
|
||||
if "401" in str(e) or "unauthorized" in str(e).lower():
|
||||
raise AuthenticationError("Authentication required or token expired")
|
||||
raise
|
||||
Reference in New Issue
Block a user