neda working

Author: mahdikiani

app/apps/neda/routes.py

@@ -3,6 +3,7 @@
 import fastapi
 from fastapi import BackgroundTasks
 from fastapi_mongo_base.routes import AbstractTaskRouter
+from fastapi_mongo_base.core import exceptions
 from usso.fastapi import jwt_access_security
 
 from .models import VoiceConvert
@@ -28,13 +29,51 @@ def __init__(self):
     def config_routes(self, **kwargs):
         super().config_routes(update_route=False)
 
+    async def retrieve_item(
+        self,
+        request: fastapi.Request,
+        uid: uuid.UUID,
+    ):
+        user = await self.get_user(request)
+        is_admin = "admin" in user.data.get("scopes", [])
+        if is_admin:
+            item = await self.get_item(uid, ignore_user_id=True)
+        else:
+            item = await self.get_item(uid, user_id=user.uid)
+        return item
+
     async def create_item(
         self,
         request: fastapi.Request,
         data: VoiceConvertTaskCreateSchema,
         background_tasks: BackgroundTasks,
+        # user_id: uuid.UUID | None = fastapi.Body(
+        #     default=None,
+        #     embed=True,
+        #     description="Request for another User ID. It is possible only if the request user is admin. If not provided, the request user will be used.",
+        # ),
     ):
-        return await super().create_item(request, data.model_dump(), background_tasks)
+        user = await self.get_user(request)
+        is_admin = "admin" in user.data.get("scopes", [])
+        # import logging
+        # logging.info(f"user_id: {user_id} is_admin: {is_admin} user.uid: {user.uid}")
+
+        # if user_id and not is_admin and user_id != user.uid:
+        #     raise exceptions.BaseHTTPException(
+        #         status_code=403,
+        #         error="Forbidden",
+        #         message={
+        #             "en": "You are not allowed to request another user's ID.",
+        #             "fa": "شما مجوز دسترسی به آیدی کاربر دیگری را ندارید.",
+        #         },
+        #     )
+
+        user_id = user.uid
+        item = await self.model.create_item({**data.model_dump(), "user_id": user_id})
+
+        if item.task_status == "init" or not self.draftable:
+            background_tasks.add_task(item.start_processing)
+        return item
 
     async def webhook(
         self,
@@ -48,3 +87,12 @@ async def webhook(
 
 
 router = VoiceConvertRouter().router
+
+
+@router.post("/pitch")
+async def get_pitch(url: str = fastapi.Body(..., embed=True)):
+    from utils import voice
+    from .services import get_voice
+
+    pitch_data = voice.get_voice_pitch_parselmouth(await get_voice(url))
+    return pitch_data

app/apps/neda/schemas.py

@@ -50,7 +50,7 @@ def progress(self):
 
 class VoiceConvertTaskCreateSchema(BaseModel):
     url: str
-    pitch_difference: float = 0.0
+    pitch_difference: float | None = None
     target_voice: str 
 
     meta_data: dict | None = None

app/apps/neda/services.py

@@ -15,7 +15,7 @@
 )
 
 
-@cached(ttl=60)
+@cached(ttl=60 * 10)
 async def get_voice(url: str) -> BytesIO:
     async with httpx.AsyncClient() as client:
         response = await client.get(url)
@@ -24,11 +24,14 @@ async def get_voice(url: str) -> BytesIO:
 
 
 async def register_cost(voice_task: VoiceConvert):
-    duration = voice_task.meta_data.get("duration", 0)
-    price = Settings.minutes_price * duration
-    usage = await finance.meter_cost(voice_task.user_id, amount=price)
-    if usage:
-        return usage
+    try:
+        duration = voice_task.meta_data.get("duration", 0)
+        price = Settings.minutes_price * duration
+        usage = await finance.meter_cost(voice_task.user_id, amount=price)
+        if usage:
+            return usage
+    except Exception as e:
+        logging.error(f"Error registering cost. {voice_task.user_id} {voice_task.uid} {e}")
 
     logging.error(f"Insufficient balance. {voice_task.user_id} {voice_task.id}")
     await voice_task.fail("Insufficient balance.")
@@ -44,7 +47,6 @@ async def convert_voice(voice_task: VoiceConvert, **kwargs):
     usage = await register_cost(voice_task)
 
     if usage is None:
-        await voice_task.fail("Insufficient balance.")
         return
 
     model = await VoiceModel.get_by_slug(voice_task.target_voice)
@@ -59,13 +61,14 @@ async def convert_voice(voice_task: VoiceConvert, **kwargs):
         await voice_task.fail("Model not found.")
         return
 
-    voice_task._status = VoiceConvertStatus.pitch_conversion
-    await voice_task.save()
+    if voice_task.pitch_difference is None:
+        voice_task._status = VoiceConvertStatus.pitch_conversion
+        await voice_task.save()
 
-    pitch_data = voice.get_voice_pitch_parselmouth(await get_voice(voice_task.url))
-    voice_task.pitch_difference = voice.calculate_pitch_shift_log(
-        pitch_data["robust_average"], model.base_pitch
-    )
+        pitch_data = voice.get_voice_pitch_parselmouth(await get_voice(voice_task.url))
+        voice_task.pitch_difference = voice.calculate_pitch_shift_log(
+            pitch_data["robust_average"], model.base_pitch
+        )
 
     run_id = voice.create_rvc_conversion_runpod(
         voice_task.url,

app/utils/finance.py

@@ -9,7 +9,7 @@
 from ufaas import AsyncUFaaS, exceptions
 from ufaas.apps.saas.schemas import UsageCreateSchema, UsageSchema
 
-resource_variant = getattr(Settings, "UFAAS_RESOURCE_VARIANT", "subtitle")
+resource_variant = getattr(Settings, "UFAAS_RESOURCE_VARIANT", "neda")
 
 
 @asynccontextmanager
@@ -26,6 +26,7 @@ async def get_ufaas_client() -> AsyncGenerator[AsyncUFaaS, None]:
         pass
 
 
+@basic.retry_execution(attempts=2, delay=0.1)
 async def meter_cost(
     user_id: uuid.UUID, amount: float, meta_data: dict = None
 ) -> UsageSchema: