add source code

.gitignore

@@ -0,0 +1,4 @@
+
+output.mp3
+.idea
+__pycache__

Dockerfile

@@ -0,0 +1,20 @@
+FROM python:3.9
+
+WORKDIR /code
+
+COPY ./requirements.txt /code/requirements.txt
+
+RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+
+RUN useradd -m -u 1000 user
+
+USER user
+
+ENV HOME=/home/user \
+ PATH=/home/user/.local/bin:$PATH
+
+WORKDIR $HOME/app
+
+COPY --chown=user . $HOME/app
+
+CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "7860"]

credentials.json

@@ -0,0 +1,13 @@
+{
+  "type": "service_account",
+  "project_id": "vlsp2023-tts-api",
+  "private_key_id": "411335f1452da99ac34dbaeaaea212dab3ff2400",
+  "private_key": "-----BEGIN PRIVATE KEY-----\nMIIEvwIBADANBgkqhkiG9w0BAQEFAASCBKkwggSlAgEAAoIBAQClPtZ/kYuBkYrP\nzbfJ/ye2DmHm4xt2o0Fvdqnuddri63662YA0GmQ+3RpH/yMg4b9Kgcyl0GXJ6Q6r\nZZzdueg3Edx5B9DkBWgYkPaTLFFe5XXOj1w2vgzc8UBN5+oomzS/WaTCpKmrs/D1\nVi05jt7Uo9mrALUkPD8wr2D9vkXQdM1G4NNxuFWCDoAF0MoVKcS5Le3f6bH4OgpL\nl5BniW4sh/UCLcWrWfXS6tuzK+8z/L6AI1ezNUJHlt58AzLjLi5Q8fOdeXviJT60\nf4w22W4gtY5YHdVCG7b7frM7ig6V9bWyVvHjGWqVkmXjYepOCIWnwQHVqja6NpVg\nR45HForBAgMBAAECggEATeqhTamdNE0iPPXtcWvEl82UUEBKFNjJ4/r6CZy8xz7v\nlL81+ltvZUzwNX6SW9DWWBV4H79yH5CrABp7qvkcC8t6P/91ee8qtFq2SZMeEzbz\nI6DphE581jlTbuiputfkOU3VqInoDzRbq/Mkg/1gCLfxzPYac6mMyjIH892iIbYP\nhlexZiXhZWRLqGFu9J3JHQAjlAtCxjhb/vBXkxzxO2j1MXv7BNrp6t7MAgFvX57k\ngRfIg/rmWohHni/NxQgWWQqTydy2JvxEVUUdPr2hgWF7hHgMk4WFbb7hzy6wH/T3\nMcjd5OHMHZcBF/NgrnxMD9NRuhD1cSbSNETJVTt9WQKBgQDS9I3MGOz5/j+jkWY7\n4CdjbbLCgat9RvvyBAkSonogLhVap1/jfKK4XQz1+aDFFY9YWW4MX/hCxKWOM5Fw\nnvDka/m2yjxha66ismflDdxvpfcjAvAt6WNX6cbiPNxFBkpJXQRH55ug9a9PKjXY\nUtc7ZOM5R/NyxfsqPjNm01faVwKBgQDIh6X0ypwCKr0CF8GP7gYPfVeK1Zh22s21\neIOEyGLB40HWOsi+PQxQHkva8teVaGMPKqppdxjs+d26d3h4Lo1rSYQrF9x+BoGm\nUU+d+3vfp1aLOj3VHMIKHKrQVOF0JSy/VdYFkz6vu6nFy3n4WdLg0XXFIQTS4Z+d\nFAD81nhEpwKBgQCqI9WNa/kNM7MuACH9Xq9F8P7BA4ZFVw/yxLBwmBx5gdF1ORMM\nTcSLf3jpljjFW7suHYq1bl2ztBh2lT7TH03YXQGdHIUQaaIC1HMY+VH1tlyZn1AJ\nJ3gZOpJOe5mIDiex/dRrDfCmJCENb1TYMRAodhkRZOeDhQwqqNoaL5BmpwKBgQCM\nWMQB67v8mETorg++2GxNcwBOHugyZzkKBWqnCEh2QsPVWBcfbkKr4Ehe2Q+hdgm+\nl7HlVoGPeeGBnBQoqQw5Rp7GOlELsyoSaV47x8MO6WNc1kpoWVRFF4NFg+K3Ez2a\nPE0qYb/B5qoP0TVwaA17Y531dgKWRWsc2N9IFiLeiQKBgQDNc2ygLHTkAKmqltcF\nzcccEnUTabPHvFkt6qycmx6oOrF6IXr1/x9ODZnwo/2JD5H/Mbub47M2k3I+BDdW\nwb0ETSEW3rdrtk85yPx3eVDIVtHoEr/Zhr9a9EYK+V8uAku3UM0L0TrnHtgqbM0W\nyGvFQOUt1hYxgoydw/Lx7h+1dA==\n-----END PRIVATE KEY-----\n",
+  "client_email": "hahunavth@vlsp2023-tts-api.iam.gserviceaccount.com",
+  "client_id": "102309854252079057442",
+  "auth_uri": "https://accounts.google.com/o/oauth2/auth",
+  "token_uri": "https://oauth2.googleapis.com/token",
+  "auth_provider_x509_cert_url": "https://www.googleapis.com/oauth2/v1/certs",
+  "client_x509_cert_url": "https://www.googleapis.com/robot/v1/metadata/x509/hahunavth%40vlsp2023-tts-api.iam.gserviceaccount.com",
+  "universe_domain": "googleapis.com"
+}

google_sheet.py

@@ -0,0 +1,53 @@
+import gspread
+from oauth2client.service_account import ServiceAccountCredentials
+from typing import Dict
+
+class SheetCRUDRepository:
+    def __init__(self, worksheet):
+        self.worksheet = worksheet
+        self.titles = self.worksheet.row_values(1)  # Assuming titles are in the first row
+        assert len(set(self.titles)) == len(self.titles), f"Failed to init {SheetCRUDRepository.__class__}, titles: {self.titles} contain duplicated values!"
+
+    def create(self, data: Dict):
+        values = [data.get(title, '') for title in self.titles]
+        self.worksheet.append_row(values)
+
+    def read(self, row_index: int) -> Dict:
+        values = self.worksheet.row_values(row_index)
+        return {title: value for title, value in zip(self.titles, values)}
+
+    def update(self, row_index: int, data: Dict):
+        values = [data.get(title, '') for title in self.titles]
+        self.worksheet.update(f"A{row_index}:Z{row_index}", [values])
+
+    def delete(self, row_index: int):
+        self.worksheet.delete_row(row_index)
+
+    def find(self, search_dict):
+        for col_title, value in search_dict.items():
+            if col_title in self.titles:
+                col_index = self.titles.index(col_title) + 1  # Adding 1 to match gspread indexing
+                cell = self.worksheet.find(value, in_column=col_index)
+                if cell is None:
+                    break
+                row_number = cell.row
+                return row_number, self.read(row_number)
+        return None
+
+def create_repositories():
+    scope = [
+        'https://www.googleapis.com/auth/spreadsheets',
+        'https://www.googleapis.com/auth/drive'
+    ]
+    creds = ServiceAccountCredentials.from_json_keyfile_name('credentials.json', scope)
+    client = gspread.authorize(creds)
+    sheet_url = "https://docs.google.com/spreadsheets/d/17OxKF0iP_aJJ0HCgJkwFsH762EUrtcEIYcPmyiiKnaM"
+    sheet = client.open_by_url(sheet_url)
+    worksheet = sheet.get_worksheet(0)
+    account_repository = SheetCRUDRepository(worksheet)
+    return account_repository
+
+
+if __name__ == "__main__":
+    a = create_repositories()
+    print(a)

login.py

@@ -0,0 +1,47 @@
+from datetime import datetime, timedelta
+from jose import JWTError, jwt, ExpiredSignatureError
+# from passlib.context import CryptContext
+
+
+class AuthService:
+    def __init__(self, account_repo, secret_key="123"):
+        assert account_repo is not None
+        assert secret_key is not None
+
+        self.account_repo = account_repo
+        self.secret_key = secret_key
+        self.encode_alg = "HS256"
+        # self.pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
+
+    async def authenticate_user(self, email, password):
+        key, user = self.account_repo.find({"email": email})
+        if not user:
+            return False
+
+        assert 'password' in user.keys()
+        if user['password'] == password:
+            return True
+        # if not self.pwd_context.verify(password, user["hashed_password"]):
+        #     return False
+        return user
+
+    async def create_token(self, email):
+        expire = datetime.utcnow() + timedelta(minutes=30)
+        encoded_jwt = jwt.encode(
+            {"sub": email, "exp": expire},
+            self.secret_key,
+            algorithm=self.encode_alg
+        )
+        return encoded_jwt
+
+    async def validate_token(self, encoded_token):
+        try:
+            decoded_token = jwt.decode(encoded_token, self.secret_key, algorithms=[self.encode_alg])
+            key, user = self.account_repo.find({"email": decoded_token['sub']})
+            return user['email']
+        except ExpiredSignatureError:
+            return False # Expired
+        except JWTError:
+            return False
+        except Exception as e:
+            raise e

main.py

@@ -0,0 +1,103 @@
+from typing import Annotated
+
+from fastapi.encoders import jsonable_encoder
+from fastapi.exceptions import RequestValidationError
+from starlette.middleware.cors import CORSMiddleware
+from fastapi import FastAPI, Header, UploadFile, Depends, HTTPException, status
+import base64
+from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
+from starlette.responses import JSONResponse
+import soundfile as sf
+from collections import defaultdict
+
+from model import SynthesisRequest, SynthesisResponse, TransferRequest, TransferResponse, LoginRequest, LoginResponse, BaseResponse
+from google_sheet import create_repositories
+from login import AuthService
+from tts import TTSService
+
+account_repo = create_repositories()
+auth_service = AuthService(account_repo=account_repo)
+tts_service = TTSService()
+
+
+app = FastAPI()
+
+auth = HTTPBearer()
+
+
+@app.exception_handler(HTTPException)
+async def http_exception_handler(request, exc: HTTPException):
+    return JSONResponse(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        content=jsonable_encoder(BaseResponse(status=0, message=exc.detail))
+    )
+
+@app.exception_handler(RequestValidationError)
+def validation_exception_handler(request, exc: RequestValidationError) -> JSONResponse:
+    reformatted_message = defaultdict(list)
+    for pydantic_error in exc.errors():
+        loc, msg = pydantic_error["loc"], pydantic_error["msg"]
+        filtered_loc = loc[1:] if loc[0] in ("body", "query", "path") else loc
+        field_string = ".".join(filtered_loc)
+        reformatted_message[field_string].append(msg)
+
+    return JSONResponse(
+        status_code=status.HTTP_400_BAD_REQUEST,
+        content=jsonable_encoder(BaseResponse(status=0, message="Invalid request", result=reformatted_message))
+    )
+    # return JSONResponse(content=jsonable_encoder(BaseResponse(status=0, message="RequestValidationError", result=str(exc))))
+
+
+async def get_current_user(access_token: Annotated[str, Header(convert_underscores=False)] = None):
+    if access_token is None:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Token missing")
+
+    username = await auth_service.validate_token(access_token)
+    if not username:
+        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid Token")
+    return username
+
+
+@app.post("/login", response_model=BaseResponse)
+async def login(request: LoginRequest):
+    email = request.email
+    password = request.password
+    user = await auth_service.authenticate_user(email, password)
+    if not user:
+        raise HTTPException(status_code=400, detail="Incorrect username or password")
+    else:
+        encoded_jwt = await auth_service.create_token(email)
+        return BaseResponse(result={"access_token": encoded_jwt})
+
+
+@app.post("/test-auth", response_model=BaseResponse)
+def test_auth(username: str = Depends(get_current_user)):
+    return BaseResponse(result={"email": username})
+
+
+@app.post("/tts/sub-task-1", response_model=BaseResponse)
+def synthesis(request: SynthesisRequest): # todo: , username: str = Depends(get_current_user)):
+    audio_data = tts_service.synthesis(request.input_text)
+    return BaseResponse(result=SynthesisResponse(data=audio_data))
+
+@app.post("/tts/sub-task-2", response_model=BaseResponse)
+async def transfer(input_text: str, ref_audio: UploadFile): # request: TransferRequest # todo: , username: str = Depends(get_current_user)):
+    if ref_audio.content_type != "audio/mpeg":
+        raise HTTPException(status_code=400, detail="Only audio files allowed")
+
+    # ref_audio_contents = await request.ref_audio.read()
+    # Convert the audio file to a NumPy array
+    with sf.SoundFile(ref_audio.file, 'rb') as f:
+        audio_np_array = f.read(dtype='float32')
+
+        audio_out = tts_service.transfer(input_text, audio_np_array)
+        audio_data = base64.b64encode(audio_out)
+        return BaseResponse(result=TransferResponse(data=audio_data))
+
+app.add_middleware(
+    CORSMiddleware,
+    allow_origins=["*"],
+    allow_credentials=True,
+    allow_methods=["*"],
+    allow_headers=["*"],
+)

model.py

@@ -0,0 +1,36 @@
+from fastapi import UploadFile
+from pydantic import BaseModel
+
+
+class SynthesisRequest(BaseModel):
+    input_text: str
+    emotion: str
+
+
+class SynthesisResponse(BaseModel):
+    data: str
+
+
+class TransferRequest(BaseModel):
+    input_text: str
+    ref_audio: UploadFile
+
+
+
+class TransferResponse(BaseModel):
+    data: str
+
+
+class LoginRequest(BaseModel):
+    email: str
+    password: str
+
+
+class LoginResponse(BaseModel):
+    access_token: str
+
+
+class BaseResponse(BaseModel):
+    status: int = 1
+    message: str = ""
+    result: object = None

requirements.txt

@@ -0,0 +1,10 @@
+fastapi
+uvicorn
+python-multipart
+passlib
+python-jose
+motor
+pyjwt
+gspread
+google-api-python-client
+oauth2client

tts.py

@@ -0,0 +1,78 @@
+import base64
+from abc import ABC, abstractmethod
+
+from gtts import gTTS
+from io import BytesIO
+import numpy as np
+
+
+class ExpressiveModel(ABC):
+    @abstractmethod
+    def load(self):
+        pass
+
+    @abstractmethod
+    def synthesize(self, text: str, emotion: str):
+        """
+        Synthesis audio with emotion
+        :param text: (str)
+        :param emotion: (str) neutral | happy | ...
+        :return: np.array
+        """
+        pass
+
+
+class StyleTransferModel(ABC):
+    @abstractmethod
+    def load(self):
+        pass
+
+    @abstractmethod
+    def synthesize(self, text: str, ref_audio):
+        """
+        Synthesis audio with reference audio
+        :param text: (str)
+        :param ref_audio: (np.array)
+        :return: np.array
+        """
+        pass
+
+
+class TTSService:
+    """
+    Get input text (str), emotion label (str) or reference audio (np.array)
+    Synthesis audio (np.array)
+    Convert audio to base64
+    """
+    @staticmethod
+    def synthesis(text: str) -> str:
+        tts = gTTS(text)
+        # Using in-memory handling
+        audio_data = BytesIO()
+        tts.write_to_fp(audio_data)
+        encoded_audio = base64.b64encode(audio_data.getvalue()).decode('utf-8')
+        return encoded_audio
+        #
+        # tts.save("output.mp3")
+        # with open("output.mp3", "rb") as audio_file:
+        #     audio_data = audio_file.read()
+        #     encoded_audio = base64.b64encode(audio_data).decode('utf-8')
+        #     return encoded_audio
+
+    @staticmethod
+    def transfer(input_text: str, ref_audio: np.array) -> str:
+        # Process reference audio
+        # ..
+        # np.array to audio
+        # tts_output_np_array = np.array([0, 1, 0, 1])
+        # tts_output_bytes = tts_output_np_array.tobytes()
+        # audio_data = base64.b64encode(tts_output_bytes).decode('utf-8')
+        # return audio_data
+        #
+        # example
+        tts_text = input_text
+        tts = gTTS(tts_text)
+        audio_data = BytesIO()
+        tts.write_to_fp(audio_data)
+        encoded_audio = base64.b64encode(audio_data.getvalue()).decode('utf-8')
+        return encoded_audio