Upload folder using huggingface_hub

.gitignore

@@ -5,6 +5,10 @@
 .venv/
 logs/
 results/
+qdrant_storage/
+assets/audio/
+assets/video/
+
 .github/
 .trunk/
 .idea/

README.md

@@ -8,4 +8,24 @@ app_port: 7860
 short_description: Chat with Characters
 ---
 
-# **Chattr**: App part of the Chatacter Backend
+## **Chattr**: App part of the Chatacter Backend
+
+### Environment Variables
+
+The configuration of the server is done using environment variables:
+
+| Name                       | Description                      | Required | Default Value                              |
+|:---------------------------|:---------------------------------|:--------:|:-------------------------------------------|
+| `MODEL__URL`               | OpenAI-compatible endpoint       |    ✘     | `https://api.groq.com/openai/v1`           |
+| `MODEL__NAME`              | Model name to use for chat       |    ✘     | `llama3-70b-8192`                          |
+| `MODEL__API_KEY`           | API key for model access         |    ✔     | `None`                                     |
+| `MODEL__TEMPERATURE`       | Model temperature (0.0-1.0)      |    ✘     | `0.0`                                      |
+| `SHORT_TERM_MEMORY__URL`   | Redis URL for memory store       |    ✘     | `redis://localhost:6379`                   |
+| `VECTOR_DATABASE__NAME`    | Vector database collection name  |    ✘     | `chattr`                                   |
+| `VOICE_GENERATOR_MCP__URL` | MCP service for audio generation |    ✘     | `http://localhost:8001/gradio_api/mcp/sse` |
+| `VIDEO_GENERATOR_MCP__URL` | MCP service for video generation |    ✘     | `http://localhost:8002/gradio_api/mcp/sse` |
+| `DIRECTORY__ASSETS`        | Base assets directory            |    ✘     | `./assets`                                 |
+| `DIRECTORY__LOG`           | Log files directory              |    ✘     | `./logs`                                   |
+| `DIRECTORY__IMAGE`         | Image assets directory           |    ✘     | `./assets/image`                           |
+| `DIRECTORY__AUDIO`         | Audio assets directory           |    ✘     | `./assets/audio`                           |
+| `DIRECTORY__VIDEO`         | Video assets directory           |    ✘     | `./assets/video`                           |

pyproject.toml

@@ -8,13 +8,13 @@ authors = [
 ]
 requires-python = ">=3.12"
 dependencies = [
-  "gradio>=5.36.2",
-  "langchain>=0.3.26",
+  "gradio>=5.39.0",
+  "langchain>=0.3.27",
   "langchain-mcp-adapters>=0.1.9",
-  "langchain-openai>=0.3.27",
-  "langgraph>=0.5.2",
-  "loguru>=0.7.3",
-  "python-dotenv>=1.1.1",
+  "langchain-openai>=0.3.28",
+  "langgraph>=0.6.3",
+  "langgraph-checkpoint-redis>=0.0.8",
+  "m3u8>=6.0.0",
 ]
 
 [project.scripts]
@@ -25,4 +25,4 @@ requires = ["uv_build"]
 build-backend = "uv_build"
 
 [dependency-groups]
-dev = ["ruff>=0.12.2", "ty>=0.0.1a12"]
+dev = ["ruff>=0.12.7", "ty>=0.0.1a16"]

src/chattr/__init__.py

@@ -1,82 +0,0 @@
-from datetime import datetime
-from os import getenv
-from pathlib import Path
-
-from dotenv import load_dotenv
-from loguru import logger
-from requests import get
-
-load_dotenv()
-
-SERVER_URL: str = getenv(key="SERVER_URL", default="127.0.0.1")
-SERVER_PORT: int = int(getenv(key="SERVER_PORT", default="7860"))
-CURRENT_DATE: str = datetime.now().strftime(format="%Y-%m-%d_%H-%M-%S")
-MCP_VOICE_GENERATOR: str = getenv(
-    key="MCP_VOICE_GENERATOR", default="http://localhost:8001/"
-)
-MCP_VIDEO_GENERATOR: str = getenv(
-    key="MCP_VIDEO_GENERATOR", default="http://localhost:8002/"
-)
-VECTOR_DATABASE_NAME: str = getenv(
-    key="VECTOR_DATABASE_NAME", default="chattr"
-)
-DOCKER_MODEL_RUNNER_URL: str = getenv(
-    key="DOCKER_MODEL_RUNNER_URL", default="http://127.0.0.1:12434/engines/v1"
-)
-DOCKER_MODEL_RUNNER_MODEL_NAME: str = getenv(
-    key="DOCKER_MODEL_RUNNER_MODEL_NAME",
-    default="ai/qwen3:0.6B-Q4_0",
-)
-GROQ_URL: str = getenv(
-    key="MODEL_URL", default="https://api.groq.com/openai/v1"
-)
-GROQ_MODEL_NAME: str = getenv(key="GROQ_MODEL_NAME", default="llama3-70b-8192")
-
-BASE_DIR: Path = Path.cwd()
-ASSETS_DIR: Path = BASE_DIR / "assets"
-LOG_DIR: Path = BASE_DIR / "logs"
-IMAGE_DIR: Path = ASSETS_DIR / "image"
-AUDIO_DIR: Path = ASSETS_DIR / "audio"
-VIDEO_DIR: Path = ASSETS_DIR / "video"
-
-LOG_FILE_PATH: Path = LOG_DIR / f"{CURRENT_DATE}.log"
-AUDIO_FILE_PATH: Path = AUDIO_DIR / f"{CURRENT_DATE}.wav"
-VIDEO_FILE_PATH: Path = VIDEO_DIR / f"{CURRENT_DATE}.mp4"
-
-ASSETS_DIR.mkdir(exist_ok=True)
-IMAGE_DIR.mkdir(exist_ok=True)
-AUDIO_DIR.mkdir(exist_ok=True)
-VIDEO_DIR.mkdir(exist_ok=True)
-LOG_DIR.mkdir(exist_ok=True)
-
-MODEL_URL: str = (
-    DOCKER_MODEL_RUNNER_URL
-    if get(DOCKER_MODEL_RUNNER_URL, timeout=10).status_code == 200
-    else GROQ_URL
-)
-MODEL_NAME: str = (
-    DOCKER_MODEL_RUNNER_MODEL_NAME
-    if MODEL_URL == DOCKER_MODEL_RUNNER_URL
-    else GROQ_MODEL_NAME
-)
-MODEL_API_KEY: str = (
-    "not-needed"
-    if MODEL_URL == DOCKER_MODEL_RUNNER_URL
-    else getenv("GROQ_API_KEY")
-)
-MODEL_TEMPERATURE: float = float(getenv(key="MODEL_TEMPERATURE", default=0.0))
-
-logger.add(
-    sink=LOG_FILE_PATH,
-    format="{time:YYYY-MM-DD at HH:mm:ss} | {level} | {message}",
-    colorize=True,
-)
-logger.info(f"Current date: {CURRENT_DATE}")
-logger.info(f"Base directory: {BASE_DIR}")
-logger.info(f"Assets directory: {ASSETS_DIR}")
-logger.info(f"Log directory: {LOG_DIR}")
-logger.info(f"Audio file path: {AUDIO_FILE_PATH}")
-logger.info(f"Log file path: {LOG_FILE_PATH}")
-logger.info(f"Model URL is going to be used is {MODEL_URL}")
-logger.info(f"Model name is going to be used is {MODEL_NAME}")
-logger.info(f"Model temperature is going to be used is {MODEL_TEMPERATURE}")

src/chattr/__main__.py

@@ -1,17 +1,13 @@
 from gradio import Blocks
 
-from chattr import SERVER_PORT, SERVER_URL
 from chattr.gui import app_block
 
 
 def main() -> None:
-    """
-    Initializes and launches the Gradio-based Chattr application server with API access, monitoring, and PWA support enabled.
-    """
+    """Initializes and launches the Gradio-based Chattr application server with API access, monitoring, and PWA support enabled."""
     app: Blocks = app_block()
     app.queue(api_open=True).launch(
-        server_name=SERVER_URL,
-        server_port=SERVER_PORT,
+        server_port=7860,
         debug=True,
         show_api=True,
         enable_monitoring=True,

src/chattr/graph/builder.py

@@ -0,0 +1,255 @@
+"""This module contains the Graph class, which represents the main orchestration graph for the Chattr application."""
+
+from json import dumps
+from logging import getLogger
+from pathlib import Path
+from typing import AsyncGenerator, Self
+
+from gradio import ChatMessage
+from gradio.components.chatbot import MetadataDict
+from langchain_core.messages import HumanMessage
+from langchain_core.runnables import Runnable, RunnableConfig
+from langchain_core.tools import BaseTool
+from langchain_mcp_adapters.client import MultiServerMCPClient
+from langchain_mcp_adapters.sessions import (
+    SSEConnection,
+    StdioConnection,
+    StreamableHttpConnection,
+    WebsocketConnection,
+)
+from langchain_openai import ChatOpenAI
+from langgraph.checkpoint.redis.aio import AsyncRedisSaver
+from langgraph.graph import START, StateGraph
+from langgraph.graph.message import MessagesState
+from langgraph.graph.state import CompiledStateGraph
+from langgraph.prebuilt import ToolNode, tools_condition
+from langgraph.store.redis.aio import AsyncRedisStore
+
+from chattr.settings import Settings
+from chattr.utils import convert_audio_to_wav, download_file, is_url
+
+logger = getLogger(__name__)
+
+
+class Graph:
+    """
+    Represents the main orchestration graph for the Chattr application.
+    This class manages the setup and execution of the conversational agent, tools, and state graph.
+    """
+
+    settings: Settings
+
+    def __init__(
+        self,
+        store: AsyncRedisStore,
+        saver: AsyncRedisSaver,
+        tools: list[BaseTool],
+    ):
+        self._long_term_memory: AsyncRedisStore = store
+        self._short_term_memory: AsyncRedisSaver = saver
+        self._tools: list[BaseTool] = tools
+        self._llm: ChatOpenAI = self._initialize_llm()
+        self._model: Runnable = self._llm.bind_tools(self._tools)
+        self._graph: CompiledStateGraph = self._build_state_graph()
+
+    @classmethod
+    async def create(cls, settings: Settings) -> Self:
+        """Async factory method to create a Graph instance."""
+        cls.settings: Settings = settings
+        store, saver = await cls._setup_memory()
+        tools: list[BaseTool] = await cls._setup_tools(
+            MultiServerMCPClient(cls._create_mcp_config())
+        )
+        return cls(store, saver, tools)
+
+    def _build_state_graph(self) -> CompiledStateGraph:
+        """
+        Construct and compile the state graph for the Chattr application.
+        This method defines the nodes and edges for the conversational agent and tool interactions.
+
+        Returns:
+            CompiledStateGraph: The compiled state graph is ready for execution.
+        """
+
+        async def _call_model(state: MessagesState) -> MessagesState:
+            response = await self._model.ainvoke(
+                [self.settings.model.system_message] + state["messages"]
+            )
+            return MessagesState(messages=[response])
+
+        graph_builder: StateGraph = StateGraph(MessagesState)
+        graph_builder.add_node("agent", _call_model)
+        graph_builder.add_node("tools", ToolNode(self._tools))
+        graph_builder.add_edge(START, "agent")
+        graph_builder.add_conditional_edges("agent", tools_condition)
+        graph_builder.add_edge("tools", "agent")
+        return graph_builder.compile(
+            debug=True,
+            checkpointer=self._short_term_memory,
+            store=self._long_term_memory,
+        )
+
+    @classmethod
+    def _create_mcp_config(
+        cls,
+    ) -> dict[
+        str,
+        StdioConnection
+        | SSEConnection
+        | StreamableHttpConnection
+        | WebsocketConnection,
+    ]:
+        """
+        Create the configuration dictionary for MCP (Multi-Component Protocol) servers.
+        This method sets up the connection details for each MCP server used by the application.
+
+        Returns:
+            dict: A dictionary mapping server names to their connection configurations.
+        """
+
+        return {
+            "vector_database": StdioConnection(
+                command="uvx",
+                args=["mcp-server-qdrant"],
+                env={
+                    "QDRANT_URL": str(cls.settings.vector_database.url),
+                    "COLLECTION_NAME": cls.settings.vector_database.name,
+                },
+                transport="stdio",
+            ),
+            "time": StdioConnection(
+                command="uvx",
+                args=["mcp-server-time"],
+                transport="stdio",
+            ),
+            cls.settings.voice_generator_mcp.name: SSEConnection(
+                url=str(cls.settings.voice_generator_mcp.url),
+                transport=cls.settings.voice_generator_mcp.transport,
+            ),
+        }
+
+    def _initialize_llm(self) -> ChatOpenAI:
+        """
+        Initialize the ChatOpenAI language model using the provided settings.
+        This method creates and returns a ChatOpenAI instance configured with the model's URL, name, API key, and temperature.
+
+        Returns:
+            ChatOpenAI: The initialized ChatOpenAI language model instance.
+
+        Raises:
+            Exception: If the model initialization fails.
+        """
+        try:
+            return ChatOpenAI(
+                base_url=str(self.settings.model.url),
+                model=self.settings.model.name,
+                api_key=self.settings.model.api_key,
+                temperature=self.settings.model.temperature,
+            )
+        except Exception as e:
+            logger.error(f"Failed to initialize ChatOpenAI model: {e}")
+            raise
+
+    @classmethod
+    async def _setup_memory(cls) -> tuple[AsyncRedisStore, AsyncRedisSaver]:
+        """
+        Initialize and set up the Redis store and checkpointer for state persistence.
+
+        Returns:
+            tuple[AsyncRedisStore, AsyncRedisSaver]: Configured Redis store and saver instances.
+        """
+        store_ctx = AsyncRedisStore.from_conn_string(str(cls.settings.memory.url))
+        checkpointer_ctx = AsyncRedisSaver.from_conn_string(
+            str(cls.settings.memory.url)
+        )
+        store = await store_ctx.__aenter__()
+        checkpointer = await checkpointer_ctx.__aenter__()
+        await store.setup()
+        await checkpointer.asetup()
+        return store, checkpointer
+
+    @staticmethod
+    async def _setup_tools(_mcp_client: MultiServerMCPClient) -> list[BaseTool]:
+        """
+        Retrieve a list of tools from the provided MCP client.
+
+        Args:
+            _mcp_client: The MultiServerMCPClient instance used to fetch available tools.
+
+        Returns:
+            list[BaseTool]: A list of BaseTool objects retrieved from the MCP client.
+        """
+        return await _mcp_client.get_tools()
+
+    def draw_graph(self) -> None:
+        """Render the compiled state graph as a Mermaid PNG image and save it."""
+        self._graph.get_graph().draw_mermaid_png(
+            output_file_path=self.settings.directory.assets / "graph.png"
+        )
+
+    async def generate_response(
+        self, message: str, history: list[ChatMessage]
+    ) -> AsyncGenerator[tuple[str, list[ChatMessage], Path | None]]:
+        """
+        Generate a response to a user message and update the conversation history.
+        This asynchronous method streams responses from the state graph and yields updated history and audio file paths as needed.
+
+        Args:
+            message: The user's input message as a string.
+            history: The conversation history as a list of ChatMessage objects.
+
+        Returns:
+            AsyncGenerator[tuple[str, list[ChatMessage], Path]]: Yields a tuple containing an empty string, the updated history, and a Path to an audio file if generated.
+        """
+        async for response in self._graph.astream(
+            MessagesState(messages=[HumanMessage(content=message)]),
+            RunnableConfig(configurable={"thread_id": "1", "user_id": "1"}),
+            stream_mode="updates",
+        ):
+            if response.keys() == {"agent"}:
+                last_agent_message = response["agent"]["messages"][-1]
+                if last_agent_message.tool_calls:
+                    history.append(
+                        ChatMessage(
+                            role="assistant",
+                            content=dumps(
+                                last_agent_message.tool_calls[0]["args"], indent=4
+                            ),
+                            metadata=MetadataDict(
+                                title=last_agent_message.tool_calls[0]["name"],
+                                id=last_agent_message.tool_calls[0]["id"],
+                            ),
+                        )
+                    )
+                else:
+                    history.append(
+                        ChatMessage(
+                            role="assistant", content=last_agent_message.content
+                        )
+                    )
+            else:
+                last_tool_message = response["tools"]["messages"][-1]
+                history.append(
+                    ChatMessage(
+                        role="assistant",
+                        content=last_tool_message.content,
+                        metadata=MetadataDict(
+                            title=last_tool_message.name,
+                            id=last_tool_message.id,
+                        ),
+                    )
+                )
+                if is_url(last_tool_message.content):
+                    logger.info(f"Downloading audio from {last_tool_message.content}")
+                    file_path: Path = (
+                        self.settings.directory.audio / last_tool_message.id
+                    )
+                    download_file(
+                        last_tool_message.content, file_path.with_suffix(".aac")
+                    )
+                    logger.info(f"Audio downloaded to {file_path.with_suffix('.aac')}")
+                    convert_audio_to_wav(
+                        file_path.with_suffix(".aac"), file_path.with_suffix(".wav")
+                    )
+                    yield "", history, file_path.with_suffix(".wav")
+            yield "", history, None

src/chattr/graph/runner.py

@@ -0,0 +1,6 @@
+from asyncio import run
+
+from chattr.graph.builder import Graph
+from chattr.settings import Settings
+
+graph: Graph = run(Graph.create(Settings()))

src/chattr/gui.py

@@ -1,61 +1,41 @@
-import gradio
-from gradio import Blocks, Button, Chatbot, ChatMessage, Row, Textbox
+"""This module contains the Gradio-based GUI for the Chattr app."""
 
+from gradio import (
+    Audio,
+    Blocks,
+    Button,
+    Chatbot,
+    ClearButton,
+    Column,
+    PlayableVideo,
+    Row,
+    Textbox,
+)
 
-def generate_response(
-    history: list[ChatMessage], thread_id: str
-) -> list[ChatMessage]:
-    """
-    Appends an assistant message about a quarterly sales plot to the chat history for the specified thread ID.
-
-    If the thread ID is 0, raises a Gradio error prompting for a valid thread ID.
-
-    Returns:
-        The updated chat history including the new assistant message.
-    """
-    if thread_id == 0:
-        raise gradio.Error("Please enter a thread ID.")
-    history.append(
-        ChatMessage(
-            role="assistant",
-            content=f"Here is the plot of quarterly sales for {thread_id}.",
-            metadata={"title": "🛠️ Used tool Weather API"},
-        )
-    )
-    return history
+from chattr.graph.runner import graph
 
 
 def app_block() -> Blocks:
-    """
-    Constructs and returns the main Gradio chat application interface with a thread ID input, chatbot display, and control buttons.
+    """Creates and returns the main Gradio Blocks interface for the Chattr app.
+
+    This function sets up the user interface, including video, audio, chatbot, and input controls.
 
     Returns:
-        Blocks: The complete Gradio Blocks interface for the chat application.
+        Blocks: The constructed Gradio Blocks interface for the chat application.
     """
-
-    history = [
-        ChatMessage(role="assistant", content="How can I help you?"),
-        ChatMessage(
-            role="user", content="Can you make me a plot of quarterly sales?"
-        ),
-        ChatMessage(
-            role="assistant",
-            content="I am happy to provide you that report and plot.",
-        ),
-    ]
-    with Blocks() as app:
-        with Row():
-            thread_id: Textbox = Textbox(
-                label="Thread ID", info="Enter Thread ID"
-            )
-
-        chatbot: Chatbot = Chatbot(history, type="messages")
-
+    with Blocks() as chat:
         with Row():
-            generate_btn: Button = Button(value="Generate", variant="primary")
-            stop_btn: Button = Button(value="Stop", variant="stop")
-        _event = generate_btn.click(
-            generate_response, [chatbot, thread_id], chatbot
-        )
-        stop_btn.click(cancels=[_event])
-    return app
+            with Column():
+                video = PlayableVideo()
+                audio = Audio(sources="upload", type="filepath", format="wav")
+            with Column():
+                chatbot = Chatbot(
+                    type="messages", show_copy_button=True, show_share_button=True
+                )
+                msg = Textbox()
+                with Row():
+                    button = Button("Send", variant="primary")
+                    ClearButton([msg, chatbot, video], variant="stop")
+        button.click(graph.generate_response, [msg, chatbot], [msg, chatbot, audio])
+        msg.submit(graph.generate_response, [msg, chatbot], [msg, chatbot, audio])
+    return chat

src/chattr/settings.py

@@ -0,0 +1,139 @@
+"""This module contains the settings for the Chattr app."""
+
+from logging import getLogger
+from pathlib import Path
+from typing import List, Literal, Self
+
+from dotenv import load_dotenv
+from langchain_core.messages import SystemMessage
+from pydantic import (
+    BaseModel,
+    DirectoryPath,
+    Field,
+    HttpUrl,
+    RedisDsn,
+    SecretStr,
+    StrictStr,
+    model_validator,
+)
+from pydantic_settings import BaseSettings, SettingsConfigDict
+
+logger = getLogger(__name__)
+
+load_dotenv()
+
+
+class ModelSettings(BaseModel):
+    url: HttpUrl = Field(default=None)
+    name: StrictStr = Field(default=None)
+    api_key: SecretStr = Field(default=None)
+    temperature: float = Field(default=0.0, ge=0.0, le=1.0)
+    system_message: SystemMessage = SystemMessage(
+        content="You are a helpful assistant that can answer questions about the time and generate audio files from text."
+    )
+
+    @model_validator(mode="after")
+    def check_api_key_exist(self) -> Self:
+        """
+        Ensure that an API key and model name are provided if a model URL is set.
+        This method validates the presence of required credentials for the model provider.
+
+        Returns:
+            Self: The validated ModelSettings instance.
+
+        Raises:
+            ValueError: If the API key or model name is missing when a model URL is provided.
+        """
+        if self.url:
+            if not self.api_key or not self.api_key.get_secret_value():
+                raise ValueError(
+                    "You need to provide API Key for the Model provider via `MODEL__API_KEY`"
+                )
+            if not self.name:
+                raise ValueError("You need to provide Model name via `MODEL__NAME`")
+        return self
+
+
+class MemorySettings(BaseModel):
+    url: RedisDsn = Field(default=RedisDsn(url="redis://localhost:6379"))
+
+
+class VectorDatabaseSettings(BaseModel):
+    name: StrictStr = Field(default="chattr")
+    url: HttpUrl = Field(default=HttpUrl(url="http://localhost:6333"))
+
+
+class MCPSettings(BaseModel):
+    name: StrictStr = Field(default=None)
+    url: HttpUrl = Field(default=None)
+    command: StrictStr = Field(default=None)
+    args: List[StrictStr] = Field(default=[])
+    transport: Literal["sse", "stdio", "streamable_http", "websocket"] = Field(
+        default=None
+    )
+
+
+class DirectorySettings(BaseModel):
+    base: DirectoryPath = Field(default_factory=lambda: Path.cwd())
+    assets: DirectoryPath = Field(default_factory=lambda: Path.cwd() / "assets")
+    log: DirectoryPath = Field(default_factory=lambda: Path.cwd() / "logs")
+    image: DirectoryPath = Field(
+        default_factory=lambda: Path.cwd() / "assets" / "image"
+    )
+    audio: DirectoryPath = Field(
+        default_factory=lambda: Path.cwd() / "assets" / "audio"
+    )
+    video: DirectoryPath = Field(
+        default_factory=lambda: Path.cwd() / "assets" / "video"
+    )
+
+    @model_validator(mode="after")
+    def create_missing_dirs(self) -> Self:
+        """
+        Ensure that all specified directories exist, creating them if necessary.
+        This method checks and creates any missing directories defined in the DirectorySettings.
+
+        Returns:
+            Self: The validated DirectorySettings instance.
+        """
+        for directory in [
+            self.base,
+            self.assets,
+            self.log,
+            self.image,
+            self.audio,
+            self.video,
+        ]:
+            directory.mkdir(exist_ok=True)
+            logger.info(f"Created directory: {directory}")
+        return self
+
+
+class Settings(BaseSettings):
+    """Configuration for the Chattr app."""
+
+    model_config = SettingsConfigDict(
+        env_nested_delimiter="__",
+        env_parse_none_str="None",
+        env_file=".env",
+        extra="ignore",
+    )
+
+    model: ModelSettings = ModelSettings()
+    memory: MemorySettings = MemorySettings()
+    vector_database: VectorDatabaseSettings = VectorDatabaseSettings()
+    voice_generator_mcp: MCPSettings = MCPSettings(
+        url="http://localhost:8080/gradio_api/mcp/sse",
+        transport="sse",
+        name="voice_generator",
+    )
+    video_generator_mcp: MCPSettings = MCPSettings(
+        url="http://localhost:8002/gradio_api/mcp/sse",
+        transport="sse",
+        name="video_generator",
+    )
+    directory: DirectorySettings = DirectorySettings()
+
+
+if __name__ == "__main__":
+    print(Settings().model_dump())

src/chattr/utils.py

@@ -0,0 +1,77 @@
+"""This module contains utility functions for the Chattr app."""
+
+from logging import getLogger
+from pathlib import Path
+from typing import Optional
+
+from m3u8 import M3U8, load
+from pydantic import HttpUrl, ValidationError
+from pydub import AudioSegment
+from requests import Session
+
+logger = getLogger(__name__)
+
+
+def is_url(value: Optional[str]) -> bool:
+    """
+    Check if a string is a valid URL.
+
+    Args:
+        value: The string to check. Can be None.
+
+    Returns:
+        bool: True if the string is a valid URL, False otherwise.
+    """
+    if value is None:
+        return False
+
+    try:
+        HttpUrl(value)
+        return True
+    except ValidationError:
+        return False
+
+
+def download_file(url: HttpUrl, path: Path) -> None:
+    """
+    Download a file from a URL and save it to a local path.
+
+    Args:
+        url: The URL to download the file from.
+        path: The local file path where the downloaded file will be saved.
+
+    Returns:
+        None
+
+    Raises:
+        requests.RequestException: If the HTTP request fails.
+        IOError: If file writing fails.
+    """
+    if str(url).endswith(".m3u8"):
+        _playlist: M3U8 = load(url)
+        url: str = str(url).replace("playlist.m3u8", _playlist.segments[0].uri)
+    print(url)
+    session = Session()
+    response = session.get(url, stream=True, timeout=30)
+    response.raise_for_status()
+    with open(path, "wb") as f:
+        for chunk in response.iter_content(chunk_size=8192):
+            if chunk:
+                f.write(chunk)
+
+
+def convert_audio_to_wav(input_path: Path, output_path: Path) -> None:
+    """
+    Convert an audio file from aac to WAV format.
+
+    Args:
+        input_path: The path to the input aac file.
+        output_path: The path to the output WAV file.
+
+    Returns:
+        None
+    """
+    logger.info(f"Converting {input_path} to WAV format")
+    audio = AudioSegment.from_file(input_path, "aac")
+    audio.export(output_path, "wav")
+    logger.info(f"Converted {input_path} to {output_path}")