version 1.0.0

pyflowery/__init__.py

@@ -0,0 +1,12 @@
+from pyflowery.models import FloweryAPIConfig, Language, Result, Voice
+from pyflowery.pyflowery import FloweryAPI
+from pyflowery.version import version
+
+__all__ = [
+    'FloweryAPI',
+    'FloweryAPIConfig',
+    'Result',
+    'Voice',
+    'Language',
+    'version',
+]

pyflowery/models.py

@@ -0,0 +1,62 @@
+from dataclasses import dataclass, field
+from logging import Logger, getLogger
+from typing import Dict, List, Union
+
+from pyflowery.version import version
+
+
+@dataclass
+class Voice:
+    """Voice object returned from the Flowery API
+
+    Attributes:
+        id (str): UUID of the voice
+        name (str): Name of the voice
+        gender (str): Gender of the voice
+        source (str): Source of the voice
+        language (Language): Language object
+    """
+    id: str
+    name: str
+    gender: str
+    source: str
+    language: 'Language'
+
+@dataclass
+class Language:
+    """Language object returned from the Flowery API
+
+    Attributes:
+        name (str): Name of the language
+        code (str): Code of the language
+    """
+    name: str
+    code: str
+
+@dataclass
+class Result:
+    """Result returned from low-level RestAdapter
+
+    Attributes:
+        success (bool): Boolean of whether the request was successful
+        status_code (int): Standard HTTP Status code
+        message (str = ''): Human readable result
+        data (Union[List[Dict], Dict]): Python List of Dictionaries (or maybe just a single Dictionary on error)
+    """
+    success: bool
+    status_code: int
+    message: str = ''
+    data: Union[List[Dict], Dict] = field(default_factory=dict)
+
+
+@dataclass
+class FloweryAPIConfig:
+    """Configuration for the Flowery API
+
+    Attributes:
+        user_agent (str): User-Agent string to use for the HTTP requests
+        logger (Logger): Logger to use for logging messages
+    """
+    user_agent: str = f"PyFlowery/{version}"
+    logger: Logger = getLogger('pyflowery')
+    allow_truncation: bool = False

pyflowery/pyflowery.py

@@ -0,0 +1,62 @@
+from typing import AsyncGenerator
+
+from pyflowery.models import FloweryAPIConfig, Language, Voice
+from pyflowery.rest_adapter import RestAdapter
+
+
+class FloweryAPI:
+    """Main class for interacting with the Flowery API
+
+    Attributes:
+        config (FloweryAPIConfig): Configuration object for the API
+        adapter (RestAdapter): Adapter for making HTTP requests
+    """
+    def __init__(self, config: FloweryAPIConfig = FloweryAPIConfig()):
+        self.config = config
+        self.adapter = RestAdapter(config)
+
+    async def get_voices(self) -> AsyncGenerator[Voice, None]:
+        """Get a list of voices from the Flowery API
+
+        Returns:
+            AsyncGenerator[Voice, None]: A generator of Voices
+        """
+        request = await self.adapter.get('/tts/voices')
+        for voice in request.data['voices']:
+            yield Voice(
+                id=voice['id'],
+                name=voice['name'],
+                gender=voice['gender'],
+                source=voice['source'],
+                language=Language(**voice['language']),
+            )
+
+    async def get_tts(self, text: str, voice: Voice | str | None = None, translate: bool = False, silence: int = 0, audio_format: str = 'mp3', speed: float = 1.0):
+        """Get a TTS audio file from the Flowery API
+
+        Args:
+            text (str): The text to convert to speech
+            voice (Voice | str): The voice to use for the speech
+            translate (bool): Whether to translate the text
+            silence (int): Number of seconds of silence to add to the end of the audio
+            audio_format (str): The audio format to return
+            speed (float): The speed of the speech
+
+        Returns:
+            bytes: The audio file
+        """
+        if len(text) > 2048:
+            if not self.config.allow_truncation:
+                raise ValueError('Text must be less than 2048 characters')
+            self.config.logger.warning('Text is too long, will be truncated to 2048 characters by the API')
+        params = {
+            'text': text,
+            'translate': str(translate).lower(),
+            'silence': silence,
+            'audio_format': audio_format,
+            'speed': speed,
+        }
+        if voice:
+            params['voice'] = voice.id if isinstance(voice, Voice) else voice
+        request = await self.adapter.get('/tts', params, timeout=180)
+        return request.data

pyflowery/rest_adapter.py

@@ -0,0 +1,69 @@
+"""This module contains the RestAdapter class, which is used to make requests to the Flowery API."""""
+from json import JSONDecodeError
+
+import aiohttp
+
+from pyflowery.models import FloweryAPIConfig, Result
+
+
+class RestAdapter:
+    """Constructor for RestAdapter
+
+    Args:
+        config (FloweryAPIConfig): Configuration object for the FloweryAPI class
+
+    Raises:
+        ValueError: Raised when the keyword arguments passed to the class constructor conflict.
+    """
+    def __init__(self, config = FloweryAPIConfig):
+        self._url = "https://api.flowery.pw/v1"
+        self._user_agent = config.user_agent
+        self._logger = config.logger
+
+    async def _do(self, http_method: str, endpoint: str, params: dict = None, timeout: float = 60):
+        """Internal method to make a request to the Flowery API. You shouldn't use this directly.
+
+        Args:
+            http_method (str): The [HTTP method](https://developer.mozilla.org/en-US/docs/Web/HTTP/Methods) to use
+            endpoint (str): The endpoint to make the request to.
+            params (dict): Python dictionary of query parameters to send with the request.
+            timeout (float): Number of seconds to wait for the request to complete.
+
+        Returns:
+            Result: A Result object containing the status code, message, and data from the request.
+        """
+        full_url = self._url + endpoint
+        headers = {
+            'User-Agent': self._user_agent,
+        }
+        sanitized_params = {k: str(v) if isinstance(v, bool) else v for k, v in params.items()} if params else None
+        self._logger.debug(f"Making {http_method} request to {full_url} with params {sanitized_params}")
+
+        async with aiohttp.ClientSession() as session:
+            async with session.request(method=http_method, url=full_url, params=sanitized_params, headers=headers, timeout=timeout) as response:
+                try:
+                    data = await response.json()
+                except (JSONDecodeError, aiohttp.ContentTypeError):
+                    data = await response.read()
+
+                result = Result(
+                    success=response.status in range(200, 299),
+                    status_code=response.status,
+                    message=response.reason,
+                    data=data,
+                )
+                self._logger.debug(f"Received response: {result.status_code} {result.message}")
+                return result
+
+    async def get(self, endpoint: str, params: dict = None, timeout: float = 60) -> Result:
+        """Make a GET request to the Flowery API. You should almost never have to use this directly.
+
+        Args:
+            endpoint (str): The endpoint to make the request to.
+            params (dict): Python dictionary of query parameters to send with the request.
+            timeout (float): Number of seconds to wait for the request to complete.
+
+        Returns:
+            Result: A Result object containing the status code, message, and data from the request.
+        """
+        return await self._do(http_method='GET', endpoint=endpoint, params=params, timeout=timeout)

pyflowery/version.py

@@ -0,0 +1 @@
+version = "1.0.0"