From 3eba8700cd5e8b53d6e2c4919f20684e07c16b1d Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E3=83=9D=E3=82=AD?= Date: Sat, 7 Jun 2025 01:06:55 -0700 Subject: [PATCH] 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 --- .gitignore | 173 +++++++++++++++++++++++++ LICENSE | 11 ++ README.md | 284 +++++++++++++++++++++++++++++++++++++++++ requirements.txt | 1 + setup.py | 58 +++++++++ songbox/__init__.py | 32 +++++ songbox/__main__.py | 290 ++++++++++++++++++++++++++++++++++++++++++ songbox/auth.py | 178 ++++++++++++++++++++++++++ songbox/client.py | 167 ++++++++++++++++++++++++ songbox/exceptions.py | 54 ++++++++ songbox/music.py | 247 +++++++++++++++++++++++++++++++++++ songbox/user.py | 216 +++++++++++++++++++++++++++++++ 12 files changed, 1711 insertions(+) create mode 100644 .gitignore create mode 100644 LICENSE create mode 100644 README.md create mode 100644 requirements.txt create mode 100644 setup.py create mode 100644 songbox/__init__.py create mode 100644 songbox/__main__.py create mode 100644 songbox/auth.py create mode 100644 songbox/client.py create mode 100644 songbox/exceptions.py create mode 100644 songbox/music.py create mode 100644 songbox/user.py diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..0bbadef --- /dev/null +++ b/.gitignore @@ -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 \ No newline at end of file diff --git a/LICENSE b/LICENSE new file mode 100644 index 0000000..7a3094a --- /dev/null +++ b/LICENSE @@ -0,0 +1,11 @@ +DO WHAT THE FUCK YOU WANT TO PUBLIC LICENSE +Version 2, December 2004 + +Copyright (C) 2004 Sam Hocevar + +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. diff --git a/README.md b/README.md new file mode 100644 index 0000000..7b448f1 --- /dev/null +++ b/README.md @@ -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 + +# 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 --mobile +songbox auth verify --otp --verify-id +songbox auth refresh +songbox auth logout +``` + +### User Commands + +```bash +songbox user me +songbox user following +``` + +### Music Commands + +```bash +songbox music search +songbox music album +songbox music song +songbox music artist +songbox music trending [--limit ] +``` + +### 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. \ No newline at end of file diff --git a/requirements.txt b/requirements.txt new file mode 100644 index 0000000..aebd9d2 --- /dev/null +++ b/requirements.txt @@ -0,0 +1 @@ +httpx==0.28.1 \ No newline at end of file diff --git a/setup.py b/setup.py new file mode 100644 index 0000000..35591a2 --- /dev/null +++ b/setup.py @@ -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", + }, +) \ No newline at end of file diff --git a/songbox/__init__.py b/songbox/__init__.py new file mode 100644 index 0000000..8a44032 --- /dev/null +++ b/songbox/__init__.py @@ -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", +] \ No newline at end of file diff --git a/songbox/__main__.py b/songbox/__main__.py new file mode 100644 index 0000000..50e5339 --- /dev/null +++ b/songbox/__main__.py @@ -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 + 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 --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() \ No newline at end of file diff --git a/songbox/auth.py b/songbox/auth.py new file mode 100644 index 0000000..ba3905f --- /dev/null +++ b/songbox/auth.py @@ -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 \ No newline at end of file diff --git a/songbox/client.py b/songbox/client.py new file mode 100644 index 0000000..5851baf --- /dev/null +++ b/songbox/client.py @@ -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() \ No newline at end of file diff --git a/songbox/exceptions.py b/songbox/exceptions.py new file mode 100644 index 0000000..1c8f598 --- /dev/null +++ b/songbox/exceptions.py @@ -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 \ No newline at end of file diff --git a/songbox/music.py b/songbox/music.py new file mode 100644 index 0000000..80cee78 --- /dev/null +++ b/songbox/music.py @@ -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 \ No newline at end of file diff --git a/songbox/user.py b/songbox/user.py new file mode 100644 index 0000000..3d8de0f --- /dev/null +++ b/songbox/user.py @@ -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 \ No newline at end of file