Spaces:
Running
on
CPU Upgrade
Running
on
CPU Upgrade
File size: 2,863 Bytes
338ea9e a683f71 338ea9e a683f71 8d9bc54 a683f71 07505ba 338ea9e 07505ba 8d9bc54 338ea9e 8eb4507 338ea9e 17f2f08 07505ba a683f71 338ea9e a683f71 338ea9e a683f71 338ea9e a683f71 338ea9e |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 |
from fastapi import HTTPException
from modules.audio.model import AudioRequest, AudioType
from modules.dropbox.art import get_audible_art_url
from modules.dropbox.audibles import get_audible_audio_url
from modules.dropbox.audio import get_audio_urls, get_global_indices_with_audio
from config import SanatanConfig
from db import SanatanDatabase
from typing import List
async def svc_get_audio_urls(req: AudioRequest):
config = SanatanConfig().get_scripture_by_name(req.scripture_name)
audio_storage = config.get("audio_storage", "dropbox")
if audio_storage == "dropbox":
urls = await get_audio_urls(req)
else:
db = SanatanDatabase()
data = db.fetch_document_by_index(
collection_name=config["collection_name"], index=req.global_index
)
url = data.get("audio", data.get("audio_url", ""))
## Temporary fix for bhagavat gita audio being moved in the source.
if req.scripture_name == "bhagavat_gita":
url = url.replace(
"https://cdn.vivekavani.com/wp-content/",
"https://vivekavani.com/wp-content/",
)
urls = {"recitation": url}
return urls
async def svc_get_indices_with_audio(
scripture_name: str, audio_type: AudioType
) -> List[int]:
"""
Service function to get all global indices for a scripture
that have audio files of the specified type.
Args:
scripture_name: Name of the scripture.
audio_type: AudioType enum value.
Returns:
List[int]: Sorted list of global indices.
"""
config = SanatanConfig().get_scripture_by_name(scripture_name)
audio_storage = config.get("audio_storage", "dropbox")
if audio_storage == "dropbox":
indices = await get_global_indices_with_audio(scripture_name, audio_type)
else:
# Fallback for database storage: iterate all documents and filter by audio_type
db = SanatanDatabase()
collection_name = config["collection_name"]
total = db.count(collection_name=collection_name)
all_docs = db.fetch_all_matches(collection_name, page_size=total)
indices = []
for doc in all_docs:
audio_field = doc.get("audio", "")
if audio_field.lower().startswith(audio_type.value):
indices.append(doc["_global_index"])
indices.sort()
return indices
async def svc_get_audible_audio_url(path: str):
if not path.startswith("_audibles/audio/"):
raise HTTPException(status_code=400, detail="Invalid audible path")
url = await get_audible_audio_url(path)
return url
async def svc_get_audible_art_url(path: str):
if not path.startswith("_audibles/art/"):
raise HTTPException(status_code=400, detail="Invalid audible path")
url = await get_audible_art_url(path)
return url
|