pitts should work

App/TTS/Schemas.py

@@ -30,6 +30,11 @@ class DescriptTranscript(BaseModel):
     file_extenstion: str = ".wav"
 
 
+class PiTTSRequest(BaseModel):
+    text: str
+    voice: Optional[str]
+
+
 class DescriptRequest(BaseModel):
     text: str
     speaker: Optional[str] = Field(default="Lawrance")

App/TTS/TTSRoutes.py

@@ -9,13 +9,19 @@ from .Schemas import (
     DescriptStatusRequest,
     DescriptSfxRequest,
     DescriptTranscript,
+    PiTTSRequest,
 )
 from .utils.Podcastle import PodcastleAPI
 from .utils.HeyGen import HeygenAPI
+from .utils.Pi import PiAIClient
 from .utils.Descript import DescriptTTS
 import os
 import asyncio
 
+from fastapi import FastAPI, Request, HTTPException
+from fastapi.responses import StreamingResponse, FileResponse
+import os
+
 tts_router = APIRouter(tags=["TTS"])
 data = {"username": os.environ.get("USERNAME"), "password": os.environ.get("PASSWORD")}
 tts = PodcastleAPI(**data)
@@ -27,6 +33,7 @@ data = {
 
 descript_tts = DescriptTTS()
 heyGentts = HeygenAPI(**data)
+pi = PiAIClient()
 
 
 @tts_router.post("/generate_tts")
@@ -86,3 +93,44 @@ async def auto_refresh():
 @tts_router.post("/status")
 async def search_id(req: StatusRequest):
     return await tts.check_status(req)
+
+
+@tts_router.post("/pi_tts")
+async def pi_tts(req: PiTTSRequest):
+    return await pi.say(text=req.text, voice=req.voice)
+
+
+@tts_router.get("/audio/{audio_name}")
+async def serve_audio(request: Request, audio_name: str):
+    audio_directory = "/tmp/Audio"
+    audio_path = os.path.join(audio_directory, audio_name)
+    if not os.path.isfile(audio_path):
+        raise HTTPException(status_code=404, detail="Audio not found")
+
+    range_header = request.headers.get("Range", None)
+    audio_size = os.path.getsize(audio_path)
+
+    if range_header:
+        start, end = range_header.strip().split("=")[1].split("-")
+        start = int(start)
+        end = audio_size if end == "" else int(end)
+
+        headers = {
+            "Content-Range": f"bytes {start}-{end}/{audio_size}",
+            "Accept-Ranges": "bytes",
+            # Optionally, you might want to force download by uncommenting the next line:
+            # "Content-Disposition": f"attachment; filename={audio_name}",
+        }
+
+        content = read_file_range(audio_path, start, end)
+        return StreamingResponse(content, media_type="audio/mpeg", headers=headers)
+
+    return FileResponse(audio_path, media_type="audio/mpeg")
+
+
+def read_file_range(path, start, end):
+    """Helper function to read specific range of bytes from a file."""
+    with open(path, "rb") as file:
+        file.seek(start)
+        # Be sure to handle the case where `end` is not the last byte
+        return file.read(end - start + 1)

App/TTS/utils/Pi.py

@@ -0,0 +1,189 @@
+import aiohttp
+import asyncio
+import enum
+import requests
+import os
+from functools import cache
+import tempfile
+import uuid
+
+
+class VoiceType(enum.Enum):
+    voice1 = "voice1"
+    voice2 = "voice2"
+    voice3 = "voice3"
+    voice4 = "voice4"
+    voice5 = "voice5"
+    voice5_update = "voice5-update"
+    voice6 = "voice6"
+    voice7 = "voice7"
+    voice8 = "voice8"
+    voice9 = "voice9"
+    voice10 = "voice10"
+    voice11 = "voice11"
+    voice12 = "voice12"
+    qdpi = "qdpi"
+
+
+class PiAIClient:
+    def __init__(self):
+        self.dir = "/tmp/Audio"
+        self.base_url = "https://pi.ai/api/chat"
+        self.referer = "https://pi.ai/talk"
+        self.origin = "https://pi.ai"
+        self.user_agent = (
+            "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/113.0"
+        )
+        self.cookie = None
+        self.headers = {
+            "User-Agent": self.user_agent,
+            "Accept": "text/event-stream",
+            "Referer": self.referer,
+            "X-Api-Version": "3",
+            "Content-Type": "application/json",
+            "Origin": self.origin,
+            "Connection": "keep-alive",
+            "Sec-Fetch-Dest": "empty",
+            "Sec-Fetch-Mode": "no-cors",
+            "Sec-Fetch-Site": "same-origin",
+            "DNT": "1",
+            "Sec-GPC": "1",
+            "TE": "trailers",
+            "Pragma": "no-cache",
+            "Cache-Control": "no-cache",
+        }
+
+    async def get_cookie(self) -> str:
+        headers = self.headers.copy()
+        async with aiohttp.ClientSession() as session:
+            async with session.post(
+                f"{self.base_url}/start", headers=headers, json={}
+            ) as response:
+                self.cookie = response.headers["Set-Cookie"]
+                return self.cookie
+
+    async def make_request(
+        self, endpoint: str, headers: dict, json: dict = None, method: str = "POST"
+    ):
+        async with aiohttp.ClientSession() as session:
+            if method == "POST":
+                async with session.post(
+                    endpoint, headers=headers, json=json
+                ) as response:
+                    return await response.text()
+            elif method == "GET":
+                async with session.get(endpoint, headers=headers) as response:
+                    return response
+
+    async def get_response(self, input_text) -> tuple[list[str], list[str]]:
+        if self.cookie is None:
+            self.cookie = await self.get_cookie()
+
+        headers = self.headers.copy()
+        headers["Cookie"] = self.cookie
+
+        data = {"text": input_text}
+        response_text = await self.make_request(self.base_url, headers, json=data)
+
+        response_lines = response_text.split("\n")
+        response_texts = []
+        response_sids = []
+        print(response_lines)
+        for line in response_lines:
+            if line.startswith('data: {"text":"'):
+                start = len('data: {"text":')
+                end = line.rindex("}")
+                text_dict = line[start + 1 : end - 1].strip()
+                response_texts.append(text_dict)
+            elif line.startswith('data: {"sid":'):
+                start = len('data: {"sid":')
+                end = line.rindex("}")
+                sid_dict = line[start : end - 1].strip()
+                sid_dict = sid_dict.split(",")[0][1:-1]
+                response_sids.append(sid_dict)
+
+        return response_texts, response_sids
+
+    async def speak_response(
+        self, message_sid: str, voice: VoiceType = VoiceType.voice4
+    ) -> None:
+        if self.cookie is None:
+            self.cookie = await self.get_cookie()
+
+        headers = self.headers.copy()
+        headers.update(
+            {
+                "Host": "pi.ai",
+                "Accept": "audio/webm,audio/ogg,audio/wav,audio/*;q=0.9,application/ogg;q=0.7,video/*;q=0.6,*/*;q=0.5",
+                "Accept-Language": "en-US,en;q=0.9",
+                "Range": "bytes=0-",
+                "Sec-Fetch-Dest": "audio",
+                "Sec-Fetch-Mode": "no-cors",
+                "Sec-Fetch-Site": "same-origin",
+                "Sec-CH-UA": '"Chromium";v="124", "Google Chrome";v="124", "Not-A.Brand";v="99"',
+                "Sec-CH-UA-Mobile": "?0",
+                "Sec-CH-UA-Platform": '"Windows"',
+            }
+        )
+
+        headers = {
+            "User-Agent": "Mozilla/5.0 (X11; Linux x86_64; rv:109.0) Gecko/20100101 Firefox/113.0",
+            "Accept": "audio/webm,audio/ogg,audio/wav,audio/*;q=0.9,application/ogg;q=0.7,video/*;q=0.6,*/*;q=0.5",
+            "Accept-Language": "en-US,en;q=0.5",
+            "Range": "bytes=0-",
+            "Connection": "keep-alive",
+            "Referer": "https://pi.ai/talk",
+            # "Cookie": cookie,
+            "Sec-Fetch-Dest": "audio",
+            "Sec-Fetch-Mode": "no-cors",
+            "Sec-Fetch-Site": "same-origin",
+            "DNT": "1",
+            "Sec-GPC": "1",
+            "Accept-Encoding": "identity",
+            "TE": "trailers",
+        }
+        headers["Cookie"] = self.cookie
+        print(headers)
+        endpoint = f"{self.base_url}/voice?mode=eager&voice={voice.value}&messageSid={message_sid}"
+        async with aiohttp.ClientSession() as session:
+            async with session.get(endpoint, headers=headers) as response:
+                print(response.status)
+                file_name = str(uuid.uuid4()) + ".mp3"
+                file_path = os.path.join(self.dir, file_name)
+                os.makedirs(file_path, exist_ok=True)
+                if response.status == 200:
+                    with open(file_path, "wb") as file:
+                        async for chunk in response.content.iter_chunked(128):
+                            file.write(chunk)
+                    return {
+                        "url": f"https://yakova-embedding.hf.space/audio/{file_name}"
+                    }
+                    # Run command vlc to play the audio file
+                    # os.system("vlc speak.wav --intf dummy --play-and-exit")
+                else:
+                    temp = await response.text()
+                    print(temp)
+                    return "Error: Unable to retrieve audio."
+
+    async def say(self, text, voice=VoiceType.qdpi):
+        _, response_sids = await self.get_response(text)
+
+        if response_sids:
+            return await self.speak_response(response_sids[0], voice=voice)
+
+
+# async def main():
+#     client = PiAIClient()
+#     response_texts, response_sids = await client.get_response(
+#         "Write a ryme to introduce yourself."
+#     )
+#     print(response_texts, response_sids)
+#     import time
+
+#     if response_sids:
+#         return await client.speak_response(response_sids[1])
+
+
+# # Run the main function
+# if __name__ == "__main__":
+#     asyncio.run(main())

App/TTS/utils/Picsart.py

@@ -1,33 +0,0 @@
-import asyncio
-from playwright.async_api import async_playwright
-
-async def extract_bootstrap_data():
-    async with async_playwright() as p:
-        browser = await p.chromium.launch()
-        page = await browser.new_page()
-
-        # Navigate to the Pixabay Sound Effects page
-        await page.goto('https://pixabay.com/sound-effects/search/door%20creaking/')
-
-        # Wait for the content to load (you can adjust the timeout as needed)
-        # await page.wait_for_selector('.js-media-list-wrapper')
-        # await page.wait_for_selector('.js-media-item')
-
-        # Get the content of the 5th script tag
-        # script_content = await page.evaluate('''() => {
-        #     const scripty=document.querySelectorAll('script')[0];
-        #     return scripty.content
-        # }''')
-        # print(script_content)
-        # await page.evaluate(f'''{script_content}(''')
-       
-        page_content = await page.content()
-        # Print the content of the 5th script tag
-        # print(page_content)
-
-        # Close the browser
-        await browser.close()
-
-# Run the extraction function
-if __name__ == '__main__':
-    asyncio.run(extract_bootstrap_data())