Delete service

service/AuthService.py

@@ -1,292 +0,0 @@
-from datetime import timedelta, datetime
-from request import RequestAuth as req
-from response import ResponseAuth as res
-import requests
-import json, re
-from auth.authentication import signJWT
-from firebase_admin import credentials, auth, exceptions
-import firebase_admin
-from repository import UserLoginRepository, UserRepository, UserInfoRepository, OTPRepository
-import service.OTPService
-from function import support_function as sf
-from dotenv import load_dotenv
-import os
-from response import ResponseUser as res
-load_dotenv()
-CLIENT_ID_GOOGLE = os.getenv('CLIENT_ID')
-API_SIGN_UP_FIREBASE_PATH = os.getenv('API_SIGN_UP_FIREBASE')
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-def get_user1(email):
-    try:
-        user = auth.get_user_by_email(email)
-        return user
-    except exceptions.FirebaseError as e:
-        return None
-
-def check_email(email):
-    if isinstance(email, str) and re.fullmatch(regex, email):
-        return True
-    else:
-        return False
-from pathlib import Path
-try:
- if not firebase_admin._apps:
-     json_path = Path(__file__).resolve().parent / 'app' / 'firebase_certificate.json'
-     cred = credentials.Certificate(str(json_path))
-     fred = firebase_admin.initialize_app(cred)
-except:
-    if not firebase_admin._apps:
-     cred = credentials.Certificate("firebase_certificate.json")
-     fred = firebase_admin.initialize_app(cred)
-def sign_up_with_email_and_password(email, password, username=None, return_secure_token=True):
-    try:
-        rest_api_url = "https://identitytoolkit.googleapis.com/v1/accounts:signUp"
-        payload = {
-            "email": email,
-            "password": password,
-            "returnSecureToken": return_secure_token
-        }
-        if username:
-            payload["displayName"] = username
-        payload = json.dumps(payload)
-        r = requests.post(rest_api_url, params={"key": API_SIGN_UP_FIREBASE_PATH}, data=payload)
-        try:
-            if r.status_code == 200:
-                response_data = r.json()
-                email = response_data.get('email')
-                display_name = response_data.get('displayName')
-                return email, display_name
-        except Exception as e:
-            pass
-    except Exception as e:
-        pass
-
-def sign_in_with_email_and_password(email=None, password=None, return_secure_token=True):
-    rest_api_url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword"
-    try:
-        payload = {
-            "returnSecureToken": return_secure_token
-        }
-        if email:
-            payload["email"] = email
-        if password:
-            payload["password"] = password
-        payload = json.dumps(payload)
-        r = requests.post(rest_api_url, params={"key": API_SIGN_UP_FIREBASE_PATH}, data=payload)
-        r.raise_for_status()
-        data = r.json()
-        if 'idToken' in data:
-            return data['email']
-        else:
-            return False
-    except requests.exceptions.RequestException as e:
-        print(f"Error signing in: {e}")
-        return False
-
-
-def login(request: req.RequestLoginEmail):
- try:
-    email = request.email
-    password = request.password
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    if password is None:
-        return res.ReponseError(
-            status=400,
-            data=res.Message(message="Password is empty")
-        )
-    user_check = get_user1(email)
-    if user_check is None:
-        return res.ReponseError(
-            status=404,
-            data=res.Message(message="Email not exits")
-        )
-    user = sign_in_with_email_and_password(email, password)
-
-    if user:
-        check = signJWT(user)
-        if check is False:
-            return res.ReponseError(
-                status=500,
-                data=res.Message(message="Invalid authorization code.")
-            )
-        else:
-            access_token = check["access_token"]
-            refresh_token = check["refresh_token"]
-            expires_in = check["expires_in"]
-            session_id = check["session_id"]
-            return res.ResponseLoginEmail(
-                status=200,
-                data=res.DataLogin(access_token=access_token, refresh_token=refresh_token, expires_in=expires_in,
-                                         session_id=session_id)
-            )
-    else:
-        return res.ReponseError(
-            status=400,
-            data=res.Message(message="Passwords do not match")
-        )
- except:
-    return res.ReponseError(
-        status=500,
-        data=res.Message(message="Server Error")
-    )
-
-def verify_token_google(token):
-    from google.oauth2 import id_token
-    from google.auth.transport import requests
-    try:
-        CLIENT_ID = CLIENT_ID_GOOGLE
-        id_info = id_token.verify_oauth2_token(token, requests.Request(), CLIENT_ID)
-        check = id_info['email_verified']
-        if check is True:
-            return True
-    except ValueError as e:
-        return False
-
-def login_google(request: req.RequestLoginGoogle):
-  try:
-    email = request.email
-    token_google = request.token_google
-    check_google = verify_token_google(token_google)
-    if check_google is False:
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="Login google failed")
-        )
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    user = get_user1(email)
-    if user:
-     check = signJWT(email)
-     if check == False:
-         return res.ReponseError(
-            status=500,
-            data =res.Message(message="Invalid authorization code.")
-        )
-     else:
-         access_token = check["access_token"]
-         refresh_token = check["refresh_token"]
-         expires_in = check["expires_in"]
-         session_id = check["session_id"]
-         return res.ResponseLoginGoogle(
-           status= 200,
-           data = res.DataLogin(access_token=access_token,refresh_token=refresh_token,expires_in=expires_in,session_id=session_id)
-    )
-    else:
-          return res.ReponseError(
-            status= 404,
-            data = res.Message(message="Email not exist")
-        )
-  except:
-      return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def sign_up(request: req.RequestRegister):
- try:
-    email = request.email
-    password = request.password
-    confirm_password = request.confirm_password
-    username = request.username
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    if password is None or password == "":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="password is empty")
-        )
-    if confirm_password is None or confirm_password == "":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="confirm password is empty")
-        )
-    if confirm_password != password:
-        return res.ReponseError(status=400,
-                                      data =res.Message(message="password and confirm_password do not match"))
-    user_signup = get_user1(email)
-    if user_signup is not None:
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="Email exist")
-        )
-
-    user_info, display_name = sign_up_with_email_and_password(email, password, username)
-    if user_info:
-       return res.ResponseSignUp(
-            status=200,
-            data =res.DataSignUp(email=user_info)
-        )
-    else:
-        return res.ReponseError(
-            status=500,
-            data =res.Message(message="Internal Server Error")
-        )
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-import pytz, datetime
-import auth.authentication as auth123
-from response import ResponseDefault as res1
-def refresh_token(request: req.RequestRefreshTokenLogin):
-     try:
-         token = request.refresh_token
-         if token is None:
-             return res.ReponseError(
-                 status=400,
-                 data=res.Message(message="token is empty")
-             )
-         user_token = UserRepository.getUserIdByRefreshToken(token)
-         if user_token is None:
-             return res.ReponseError(
-                 status=404,
-                 data=res.Message(message="Not found refresh token")
-             )
-         email = UserRepository.getEmailUserById(user_token)
-         if email:
-             rf_token = token
-         result = auth123.get_refresh_token(token, email)
-         token_new = result["access_token"]
-         rf_token_new = result["refresh_token"]
-         seconds1 = result["expires_in"]
-         session_id = result["session_id"]
-         vn_timezone = pytz.timezone('Asia/Ho_Chi_Minh')
-         current_time = datetime.datetime.utcnow().replace(tzinfo=pytz.utc).astimezone(vn_timezone) + timedelta(
-             seconds=seconds1)
-         formatted_time = current_time.strftime('%Y-%m-%d %H:%M:%S ')
-         if rf_token == rf_token_new:
-             UserRepository.updateAccessToken(user_token, token_new, formatted_time)
-         else:
-             UserRepository.UpdateAccessTokenRefreshTokenById(user_token, token_new, rf_token_new, formatted_time)
-         user = token_new
-         if user == False:
-             return res.ReponseError(
-                 status=400,
-                 data=res.Message(message="Refresh token error")
-             )
-         return res.ResponseRefreshTokenLogin(
-             status=200,
-             data=res.DataRefreshToken(token_new=user, session_id=session_id)
-         )
-     except:
-         return res.ReponseError(
-             status=500,
-             data=res.Message(message="Server Error")
-         )
-def check_token_is_valid(token):
-   try:
-    check = UserRepository.getEmailUserByAccessToken(token)
-    if check is None:
-        return False
-    return True
-   except:
-       return res.ReponseError(
-           status=500,
-           data=res.Message(message="Server Error")
-       )

service/ChatService.py

@@ -1,141 +0,0 @@
-from typing import Union
-from datetime import timedelta
-from request import RequestChat as req
-from response import ResponseChat as res
-from repository import ChatHistoryRepository, DetailChatRepository, UserRepository
-import function.chatbot as sf
-from typing import Dict
-import json, re
-from pydantic import BaseModel
-from function import support_function
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def check_email(email):
-    if(re.fullmatch(regex, email)):
-        return True
-    else:
-        return False
-
-class Document(BaseModel):
-    page_content: str
-    metadata: Dict[str, str]
-    type: str
-
-def query2_upgrade_old(request: req.RequestQuery2UpgradeOld):
-  try:
-    user_id = request.user_id
-    question = request.question
-    text_all = request.text_all
-    chat_name = request.chat_name
-    email = support_function.check_email_service(str(user_id))
-    if isinstance(email, res.ReponseError):
-        return email
-    if question is None:
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="question is empty")
-        )
-    if chat_name is None:
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="chat_name is empty")
-        )
-    text_all_dicts = json.loads(text_all)
-    text_all1 = [Document(**doc) for doc in text_all_dicts]
-    test, list1, list2 = sf.handle_query_upgrade_keyword_old(question, text_all1, email)
-    text1 = "<Data_Relevant>".join(list1)
-    text2 = "<Source_File>".join(list2)
-    id = 0
-    if test:
-       chat_id = ChatHistoryRepository.getIdChatHistoryByUserIdAndNameChat(user_id,chat_name)
-       if chat_id:
-           id = DetailChatRepository.addDetailChat(chat_id,question, test, text1, text2)
-       if chat_id is None:
-           ChatHistoryRepository.addChatHistory(user_id,chat_name)
-           chat_id_new = ChatHistoryRepository.getIdChatHistoryByUserIdAndNameChat(user_id,chat_name)
-           id = DetailChatRepository.addDetailChat(chat_id_new, question, test, text1, text2)
-       return res.ResponseQuery2UpgradeOld(
-               status= 200,
-               data = res.DataAnswer1(id = id,
-                                      answer=test,
-                                      data_relevant=list1,
-                                      sources=list2))
-    if test is None or test == "":
-         return res.ReponseError(
-            status=500,
-            data =res.Message(message="No answer")
-        )
-  except:
-       return res.ReponseError(
-            status=500,
-            data =res.Message(message="Server Error")
-        )
-
-def extract_file(request: req.RequestExtractFile):
- try:
-   user_id = request.user_id
-   email = support_function.check_email_service(str(user_id))
-   if isinstance(email, res.ReponseError):
-       return email
-   text_all1 = sf.extract_data2(email)
-   if text_all1 is False:
-       return res.ResponseExtractFile(
-               status= 200,
-               data = res.DataExtractFile(text_all="No data response"))
-   return res.ResponseExtractFile(
-               status= 200,
-               data = res.DataExtractFile(text_all=text_all1))
- except:
-     return res.ReponseError(
-            status=500,
-            data =res.Message(message="Server Error")
-        )
-
-def generate_question(request: req.RequestGenerateQuestion):
- try:
-   user_id = request.user_id
-   email = support_function.check_email_service(str(user_id))
-   if isinstance(email, res.ReponseError):
-       return email
-   text_all1 = sf.generate_question(email)
-   if text_all1 is False:
-       return res.ResponseGenerateQuestion(
-               status= 200,
-               data = res.GenerateQuestion(question=False))
-     
-   return res.ResponseGenerateQuestion(
-               status= 200,
-               data = res.GenerateQuestion(question=text_all1))
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def delete_chat(request: req.RequestDeleteChat):
-  try:
-    user_id = request.user_id
-    email = support_function.check_email_service(str(user_id))
-    if isinstance(email, res.ReponseError):
-        return email
-    chat_name = request.chat_name
-    if chat_name is None:
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="chat_name is empty")
-        )
-    DetailChatRepository.delete_chat_detail(chat_name)
-    check = ChatHistoryRepository.deleteChatHistory(user_id,chat_name)
-    if check is False:
-       return res.ResponseDeleteChat(
-               status = 500,
-               data = res.Message(message="Delete conversation chat failed"))
-    else:
-       return res.ResponseDeleteChat(
-               status = 200,
-               data = res.Message(message="Delete conversation chat success"))
-  except:
-     return res.ReponseError(
-            status= 500,
-            data = res.Message(message="Server Error")
-        )

service/DefaultService.py

@@ -1,205 +0,0 @@
-from typing import Union
-from datetime import timedelta
-from request import RequestDefault as req
-from response import ResponseDefault as res
-from response import ResponseUser as res_login
-from firebase_admin import credentials, auth, exceptions
-import firebase_admin
-import base64
-import auth.authentication as auth123
-import re
-from repository import UserRepository, UserInfoRepository
-from pathlib import Path
-import cloudinary
-import cloudinary.uploader
-from fastapi import FastAPI, File, UploadFile
-from function import support_function as sf
-from dotenv import load_dotenv
-from service import AuthService as authservice
-import os
-load_dotenv()
-CLOUDINARY_CLOUD_NAME=os.getenv("CLOUDINARY_CLOUD_NAME")
-CLOUDINARY_API_KEY=os.getenv("CLOUDINARY_API_KEY")
-CLOUDINARY_API_SECRET=os.getenv("CLOUDINARY_API_SECRET")
-
-cloudinary.config(
-     cloud_name=CLOUDINARY_CLOUD_NAME,
-     api_key=CLOUDINARY_API_KEY,
-     api_secret=CLOUDINARY_API_SECRET
-)
-import os
-
-try:
- if not firebase_admin._apps:
-    cred = credentials.Certificate("../certificate/firebase_certificate.json")
-    fred = firebase_admin.initialize_app(cred)
-except:
-   try:
-    if not firebase_admin._apps:
-     cred = credentials.Certificate("firebase_certificate.json")
-     fred = firebase_admin.initialize_app(cred)
-   except:
-    if not firebase_admin._apps:
-     json_path = Path(__file__).resolve().parent / 'app' / 'firebase_certificate.json'
-     cred = credentials.Certificate(str(json_path))
-     fred = firebase_admin.initialize_app(cred)
-
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def check_email(email):
-    if(re.fullmatch(regex, email)):
-        return True
-    else:
-        return False
-
-def get_user(email):
-    try:
-        user = auth.get_user_by_email(email)
-        return user
-    except exceptions.FirebaseError as e:
-        return None
-        
-def create_user(email):
-    user = auth.create_user(email=email)
-    return user
-
-def get_user1(email):
-    try:
-        user = auth.get_user_by_email(email)
-        return user
-    except exceptions.FirebaseError as e:
-        return None
-
-def create_firebase_user(request: req.RequestCreateFireBaseUserGoogle):
-    try:
-     email = request.email
-     token_google = request.token_google
-     check_email_fc = sf.check_email_empty_invalid(email)
-     if check_email_fc is not True:
-        return check_email_fc
-     user = get_user(email)
-     if token_google is None or token_google == "":
-         return res.ReponseError(
-             status=400,
-             data=res.Message(message="token google not empty")
-         )
-     check_google = authservice.verify_token_google(token_google)
-     if check_google == False:
-         return res.ReponseError(
-             status=400,
-             data=res.Message(message="Create user failed")
-         )
-     if user: 
-        email1 = user.email
-        display_name = user.display_name
-        uid = user.uid
-        photo_url = user.photo_url
-        return res.ResponseCreateFireBaseUser(
-           status=200,
-           data = res.DataCreateFireBaseUser(localId=uid,
-                                             email=email1,
-                                             displayName=display_name,
-                                             photoUrl=photo_url)
-    )
-     else:
-        return res.ReponseError(
-            status=500,
-            data =res.Message(message="Error")
-        ) 
-    except:
-        return res.ReponseError(
-            status=500,
-            data =res.Message(message="Server Error")
-        )
-
-def info_user(request: req.RequestInfoUser):
-    try:
-        user_id = request.user_id
-        email = sf.check_email_service(user_id)
-        if isinstance(email, res.ReponseError):
-            return email
-        user = get_user(email)
-        if user is None:
-            return res.ReponseError(
-                status=404,
-                data=res.Message(message="User not found")
-            )
-        uid = user.uid if user.uid else ""
-        email = user.email if user.email else ""
-        display_name = user.display_name if user.display_name else "N/A"
-        photo_url = user.photo_url if user.photo_url else "N/A"
-        return res.ResponseInfoUser(
-            status=200,
-            data=res.DataInfoUser(
-                uid=uid,
-                email=email,
-                display_name=display_name,
-                photo_url=photo_url
-            )
-        )
-    except Exception as e:
-        return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error: " + str(e))
-        )
-
-def check_email_token(token):
-  try:
-    decoded_token = auth123.decodeJWT(token)
-    sub_value = decoded_token.get("sub")
-    name_user = base64.b85decode(sub_value.encode('ascii')).decode('ascii')
-    return name_user
-  except:
-    return False
-
-def is_me(request: req.RequestIsMe):
- try:
-   token = request.token
-   if token is None or token == "":
-          return res.ReponseError(
-            status=400,
-            data =res.Message(message="token is empty")
-        )
-   test = check_email_token(token)
-   if test is not False:
-      user_id = UserRepository.getUserByEmail(test).id
-      return res.ResponseIsMe(
-           status=200,
-           data = res.DataIsMe(user_id = user_id))
- except:
-    return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-ALLOWED_EXTENSIONS = {'png', 'jpg','jpeg'}
-def allowed_file(filename):
-    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
-
-def upload_image_service(request: req.RequestUpLoadImage):
-  try:
-        user_id = request.user_id
-        file = request.files
-        email = sf.check_email_service(user_id)
-        if isinstance(email, res.ReponseError):
-            return email
-        if not allowed_file(file.filename):
-            return res.ReponseError(
-                status=415,
-                data=res.Message(message=f"File type not allow")
-            )
-        temp_file_path = f"temp_image_{email}.png"
-        contents = file.file.read()
-        with open(temp_file_path, "wb") as temp_file:
-            temp_file.write(contents)
-        upload_result = cloudinary.uploader.upload(temp_file_path, public_id=email)
-        os.remove(temp_file_path)
-        return res.ResponseUploadImage(
-          status=200,
-          data=res.DataUploadImage(url=upload_result["secure_url"])
-    )
-  except:
-      return res.ReponseError(
-          status=500,
-          data=res.Message(message="Server Error")
-      )

service/FileService.py

@@ -1,151 +0,0 @@
-from request import RequestFile as req
-from response import ResponseFile as res
-from response import ResponseDefault as res1
-import function.dropbox as sf_dropbox
-import os
-import shutil
-import re
-from repository import UserRepository
-from function import  support_function as sf
-ALLOWED_EXTENSIONS = {'csv', 'txt', 'doc', 'docx', 'pdf', 'xlsx', 'pptx', 'json','md'}
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def check_email(email):
-    if(re.fullmatch(regex, email)):
-        return True
-    else:
-        return False
-
-def allowed_file(filename):
-    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
-
-def listNameFiles(request: req.RequestGetNameFile ):
-  try:
-    user_id = request.user_id
-    email = sf.check_email_service(user_id)
-    if isinstance(email,res1.ReponseError):
-        return email
-    list_files = sf_dropbox.list_files(email)
-    return res.ResponseGetNameFile(
-       status= 200,
-       data = res.DataGetNameFile(files=list_files)
-    )
-  except:
-      return res.ReponseError(
-            status=500,
-            data =res.Message(message="Server Error")
-        )
-
-def deleteFile(request: req.RequestDeleteFile):
-  try:
-   user_id = request.user_id
-   name_file = request.name_file
-   email = sf.check_email_service(user_id)
-   if isinstance(email, res1.ReponseError):
-        return email
-   if name_file is None or name_file == "":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="Name file is empty")
-        )
-   sf_dropbox.delete_file(email,name_file)
-   return res.ResponseDeleteFile(
-            status=200,
-            data =res.Message(message=f"delete {name_file} success")
-        )
-  except:
-       return res.ReponseError(
-            status=500,
-            data =res.Message(message=f"delete {name_file} error")
-        )
-
-def download_folder(request:req.RequestDownLoadFolder):
-   try:
-       user_id = request.user_id
-       email = sf.check_email_service(user_id)
-       if isinstance(email, res1.ReponseError):
-           return email
-       sf_dropbox.download_folder(email)
-       return res.ResponseDownloadFolder(
-            status=200,
-            data =res.Message(message=f"Downloaded folder {email} success")
-        )
-   except: 
-      return res.ReponseError(
-            status=500,
-            data =res.Message(message=f"Server error")
-        )
-
-def download_file(request:req.RequestDownLoadFile):
-   try:
-       user_id = request.user_id
-       name_file = request.name_file
-       email = sf.check_email_service(user_id)
-       if isinstance(email, res1.ReponseError):
-           return email
-       if name_file is None or name_file == "":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="name_file is empty")
-        )
-       sf_dropbox.search_and_download_file(name_file,email)
-       return res.ResponseDownloadFile(
-            status=200,
-            data =res.Message(message=f"Downloaded file '{name_file}' by email: '{email}' success")
-        )
-   except: 
-      return res.ReponseError(
-            status=500,
-            data =res.Message(message=f"Server error")
-        )
-
-ALLOWED_EXTENSIONS = {'csv', 'txt', 'doc', 'docx', 'pdf', 'xlsx', 'pptx', 'json','md'}
-
-def allowed_file1(filename):
-    return '.' in filename and filename.rsplit('.', 1)[1].lower() in ALLOWED_EXTENSIONS
-def upload_files(request: req.RequestUploadFile):
- try:
-   user_id = request.user_id
-   files = request.files
-   email = sf.check_email_service(user_id)
-   if isinstance(email, res1.ReponseError):
-       return email
-   for file in files:
-            if not allowed_file(file.filename):
-                return res.ReponseError(
-            status=415,
-            data =res.Message(message=f"File type not allow")
-        )
-            temp_dir = f"/code/temp/{email}"
-            os.makedirs(temp_dir, exist_ok=True)
-            file_path = os.path.join(temp_dir, file.filename)
-            with open(file_path, "wb") as buffer:
-                shutil.copyfileobj(file.file, buffer)
-            cloud_path = f"/{email}/{file.filename}"
-            sf_dropbox.upload_file(file_path, cloud_path)
-   return res.ResponseUploadedFile(
-            status=200,
-            data =res.Message(message=f"Load file success")
-        )
- except:
-    return res.ReponseError(
-            status=500,
-            data =res.Message(message=f"Load file error")
-        )
-
-def deleteAllFile(request: req.RequestDeleteAllFile):
- try:
-    user_id = request.user_id
-    email = sf.check_email_service(user_id)
-    if isinstance(email, res.ReponseError):
-        return email
-    sf_dropbox.delete_all_files_in_folder(email)
-    return res.ResponseDeleteAllFile(
-            status=200,
-            data=res.Message(message=f"Delete all file success")
-        )
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message=f"Delete all file error")
-        )

service/MySQLService.py

@@ -1,201 +0,0 @@
-import json
-
-from request import RequestMySQL as req
-from response import ResponseMySQL as res
-from response import ResponseDefault as res1
-from repository import ChatHistoryRepository
-from repository import DetailChatRepository,UserRepository
-from fastapi.responses import  JSONResponse
-from function import  support_function as sf
-import re
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def check_email(email):
-    if(re.fullmatch(regex, email)):
-        return True
-    else:
-        return False
-    
-def edit_chat(request: req.RequestEditNameChat):
-  try:
-    user_id = request.user_id
-    name_old = request.name_old
-    name_new = request.name_new
-    email = sf.check_email_service(user_id)
-    if isinstance(email, res1.ReponseError):
-        return email
-    if name_old is None or name_old =="":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="name_old is empty")
-        )
-    if name_new is None or name_new == "":
-        return res.ReponseError(
-            status=400,
-            data =res.Message(message="name_new is empty")
-        )
-    chat_exist = ChatHistoryRepository.getIdChatHistoryByUserIdAndNameChatNew(user_id, name_new)
-    if chat_exist:
-        return res.ReponseError(
-            status=400,
-            data=res.Message(message="name_new duplicate")
-        )
-    id_chat = ChatHistoryRepository.getIdChatHistoryByUserIdAndNameChat(user_id, name_old)
-    if id_chat:
-       ChatHistoryRepository.updateNameChatHistory(user_id,name_old,name_new)
-       check = True
-    else:
-       check = False
-    if check is True:
-        return  res.ResponseEditChat(
-           status= 200,
-           data= res.Message(message=check)
-        )
-    else:
-       return res.ReponseError(
-            status=500,
-            data =res.Message(message="Update chat error")
-        )     
-  except:
-     return res.ReponseError(
-            status=500,
-            data =res.Message(message="Server Error")
-        )
-
-def delete_chat(request: req.RequestDeleteChat):
-    try:
-        user_id = request.user_id
-        chat_name = request.chat_name
-        email = sf.check_email_service(user_id)
-        if isinstance(email, res1.ReponseError):
-            return email
-        if chat_name is None or chat_name =="":
-            return res.ReponseError(
-                status=400,
-                data =res.Message(message="chat_name is empty")
-            )
-        DetailChatRepository.delete_chat_detail(chat_name)
-        chat_exist = ChatHistoryRepository.getIdChatHistoryByUserIdAndNameChat(user_id, chat_name)
-        if chat_exist is None:
-            return res.ReponseError(
-                status=404,
-                data=res.Message(message="chat_name not exist")
-            )
-        check = ChatHistoryRepository.deleteChatHistory(user_id,chat_name)
-        if check is True:
-            return  res.ResponseDeleteChat(
-               status= 200,
-               data= res.Message(message="Delete conversation chat success")
-            )
-        else:
-            return res.ReponseError(
-                status=500,
-                data =res.Message(message="Delete conversation chat error")
-            )
-    except Exception as e:
-        return res.ResponseDeleteChat(
-            status=500,
-            data=res.Message(message=str(e))
-        )
-
-def delete_chat_detail_by_id(request: req.RequestDeleteDetailChat):
-    try:
-        user_id = request.user_id
-        chat_detail_id = request.id_chat_detail
-        email = sf.check_email_service(user_id)
-        if isinstance(email, res1.ReponseError):
-            return email
-        if chat_detail_id is None or chat_detail_id == " ":
-            return res.ReponseError(
-                status=400,
-                data=res.Message(message="id chat_detail is empty")
-            )
-        check = DetailChatRepository.delete_chat_detail_by_id((chat_detail_id))
-        if check is True:
-            return res.ResponseDeleteChatDetailById(
-               status= 200,
-               data= res.CheckModel(check=check)
-            )
-        else:
-            return res.ResponseDeleteChatDetailById(
-                status=200,
-                data=res.CheckModel(check=check)
-            )
-    except Exception as e:
-        return res.ResponseDeleteChat(
-            status=500,
-            data=res.Message(message=str(e))
-        )
-
-def render_chat_history(request: req.RequestRenderChatHistory):
-    try:
-        user_id = request.user_id
-        email = sf.check_email_service(user_id)
-        if isinstance(email, res1.ReponseError):
-            return email
-        chat_detail = ChatHistoryRepository.getChatHistoryById(user_id)
-        for item in chat_detail:
-            print(item)
-        chat1 = [res.ListUserChat(id=item.id, email=item.email, chat_name=item.name_chat) for item in chat_detail]
-        return res.ResponseRenderChatHistory(
-            status=200,
-            data=res.UserInfoListResponse(chat=chat1)
-        )
-    except Exception as e:
-        return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def get_detail_chat_by_chat_id(request: req.RequestGetChatDetails):
-      id = request.id
-      if id is None or id == "":
-          return res.ReponseError(
-              status=400,
-              data=res.Message(message="Id is empty")
-          )
-      chat_detail1 = DetailChatRepository.getDetailChatByChatId(id)
-      if chat_detail1:
-          return res.ResponseChatDetailById(
-              status=200,
-              data=res.ChatDetailById(
-                  id = chat_detail1.id,
-                  data_relevant = chat_detail1.data_relevant,
-                  source_file = chat_detail1.source_file
-              )
-          )
-      else:
-          return res.ReponseError(
-              status=404,
-              data=res.Message(message="Chat not exist")
-          )
-
-def load_chat_history(request: req.RequestLoadChatHistory):
-  try:
-    chat_id = request.chat_id
-    user_id = request.user_id
-    email = sf.check_email_service(str(user_id))
-    if isinstance(email, res1.ReponseError):
-        return email
-    if chat_id is None or chat_id == "":
-        return res.ReponseError(
-            status = 400,
-            data = res.Message(message="chat_id is empty")
-        )
-    check_exist_chatid_width_user_id = ChatHistoryRepository.getChatHistoryByChatIdAndUserId(chat_id,user_id)
-    if check_exist_chatid_width_user_id is None:
-        return res.ReponseError(
-            status=404,
-            data=res.Message(message="Not found chat width chat_id and user_id")
-        )
-    result = DetailChatRepository.getListDetailChatByChatId(chat_id)
-    chat1 = [res.ChatDetail(id=item.id, chat_id = item.chat_id, question=item.YouMessage,answer=item.AiMessage,data_relevant = item.data_relevant,source_file=item.source_file) for item in result]
-    return res.ResponseLoadChatHistory(
-         status = 200,
-         data = res.ListChatDeTail(detail_chat=chat1)
-      )
-  except:
-      return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )

service/OTPService.py

@@ -1,168 +0,0 @@
-from datetime import timedelta,datetime
-from request import RequestOTP as req
-from response import ResponseOTP as res
-import re
-from firebase_admin import auth, exceptions
-from repository import OTPRepository
-from datetime import datetime, timedelta
-import datetime, string,random
-from function import  support_function as sf
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def get_user1(email):
-    try:
-        user = auth.get_user_by_email(email)
-        return user
-    except exceptions.FirebaseError as e:
-        return None
-
-def check_email(email):
-    if isinstance(email, str) and re.fullmatch(regex, email):
-        return True
-    else:
-        return False
-    
-def generate_otp(length=6):
-    characters = string.ascii_uppercase + string.digits
-    otp = ''.join(random.choice(characters) for _ in range(length))
-    return otp
-
-def createOTP(request: req.RequestCreateOTP):
-  try:
-    email = request.email
-    otp = generate_otp()
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    OTPRepository.addOTP(email,otp)
-    return res.ResponseCreateOTP(
-            status=200,
-            data = res.CheckModel(check=True),
-            otp = otp
-        )
-  except:
-      return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def verifyOTP(request: req.RequestVerifyOTP):
- try:
-    email = request.email
-    otp = request.otp
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    if otp is None:
-        return res.ReponseError(
-            status=400,
-            data=res.Message(message="otp is empty")
-        ) 
-    user_otp = OTPRepository.getOtpByEmail(email)
-    if user_otp:
-        otp_db = user_otp.otp
-        otp_created_at = user_otp.created_at
-        if otp == otp_db:
-            current_time = datetime.datetime.now()
-            otp_expiry_time = otp_created_at + timedelta(minutes=15)
-            if current_time <= otp_expiry_time:
-                OTPRepository.deleteOTP(email, otp)
-                return res.ResponseVerifyOTPSignUp(
-                    status=200,
-                    data=res.Message(message="OTP is valid")
-                )
-            else:
-                return res.ReponseError(
-                    status=400,
-                    data=res.Message(message="OTP has expired")
-                )
-        else:
-            return res.ReponseError(
-                status=400,
-                data=res.Message(message="Invalid OTP")
-            )
-    else:
-        return res.ReponseError(
-            status=404,
-            data=res.Message(message="No OTP found for this email")
-        )
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def createOTPReset(email):
- try:
-    otp = generate_otp()
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    OTPRepository.addOTP(email,otp)
-    return res.ResponseCreateOTP(
-            status=200,
-            data = res.CheckModel(check=True),
-            otp = otp
-        )
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )
-
-def generate_random_password(length=8):
-    characters = string.ascii_letters + string.digits
-    password = ''.join(random.choice(characters) for i in range(length))
-    return password
-
-def verifyOTPReset(request: req.RequestVerifyOTP):
- try:
-    email = request.email
-    otp = request.otp
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    if otp is None:
-        return res.ReponseError(
-            status=400,
-            data=res.Message(message="OTP is empty")
-        ) 
-    user_otp = OTPRepository.getOtpByEmail(email)
-    if user_otp:
-        otp_db = user_otp.otp
-        otp_created_at = user_otp.created_at
-        if otp == otp_db:
-            current_time = datetime.datetime.now()  # Lấy thời gian hiện tại với múi giờ hệ thống (múi giờ +7)
-            otp_expiry_time = otp_created_at + timedelta(minutes=15)
-            new_password = generate_random_password()
-            if current_time <= otp_expiry_time:
-                OTPRepository.deleteOTP(email, otp)
-                user_email = auth.get_user_by_email(email)
-                auth.update_user(
-                    user_email.uid,
-                    password=new_password)
-                return res.ResponseVerifyOTP(
-                    status=200,
-                    data=res.Message(message="New Password send to Email"),
-                    newpassword=new_password
-                )
-            else:
-                return res.ReponseError(
-                    status=400,
-                    data=res.Message(message="OTP has expired")
-                )
-        else:
-            return res.ReponseError(
-                status=400,
-                data=res.Message(message="Invalid OTP")
-            )
-    else:
-        return res.ReponseError(
-            status=404,
-            data=res.Message(message="No OTP found for this email")
-        )
- except:
-     return res.ReponseError(
-            status=500,
-            data=res.Message(message="Server Error")
-        )

service/UserService.py

@@ -1,322 +0,0 @@
-from datetime import timedelta, datetime
-from request import RequestUser as req_login
-from response import ResponseUser as res_login
-import requests
-import json, re
-from auth.authentication import signJWT
-from firebase_admin import credentials, auth, exceptions
-import firebase_admin
-from repository import UserLoginRepository, UserRepository, UserInfoRepository, OTPRepository
-import service.OTPService
-from function import support_function as sf
-from dotenv import load_dotenv
-import os
-from response import ResponseUser as res
-from response import ResponseDefault as res1
-load_dotenv()
-CLIENT_ID_GOOGLE = os.getenv('CLIENT_ID')
-API_SIGN_UP_FIREBASE_PATH = os.getenv('API_SIGN_UP_FIREBASE')
-regex = r'\b[A-Za-z0-9._%+-]+@[A-Za-z0-9.-]+\.[A-Z|a-z]{2,7}\b'
-
-def get_user1(email):
-    try:
-        user = auth.get_user_by_email(email)
-        return user
-    except exceptions.FirebaseError as e:
-        return None
-
-def check_email(email):
-    if isinstance(email, str) and re.fullmatch(regex, email):
-        return True
-    else:
-        return False
-from pathlib import Path
-try:
- if not firebase_admin._apps:
-     json_path = Path(__file__).resolve().parent / 'app' / 'firebase_certificate.json'
-     cred = credentials.Certificate(str(json_path))
-     fred = firebase_admin.initialize_app(cred)
-except:
-    if not firebase_admin._apps:
-     cred = credentials.Certificate("firebase_certificate.json")
-     fred = firebase_admin.initialize_app(cred)
-
-def sign_up_with_email_and_password(email, password, username=None, return_secure_token=True):
-    try:
-        rest_api_url = "https://identitytoolkit.googleapis.com/v1/accounts:signUp"
-        payload = {
-            "email": email,
-            "password": password,
-            "returnSecureToken": return_secure_token
-        }
-        if username:
-            payload["displayName"] = username 
-        payload = json.dumps(payload)
-        r = requests.post(rest_api_url, params={"key": API_SIGN_UP_FIREBASE_PATH}, data=payload)
-        try:
-            return r.json()['email']
-        except Exception as e:
-            pass
-    except Exception as e:
-        pass
-
-def sign_in_with_email_and_password(email=None, password=None, return_secure_token=True):
-    rest_api_url = "https://identitytoolkit.googleapis.com/v1/accounts:signInWithPassword"
-    try:
-        payload = {
-            "returnSecureToken": return_secure_token
-        }
-        if email:
-            payload["email"] = email
-        if password:
-            payload["password"] = password
-        payload = json.dumps(payload)
-        r = requests.post(rest_api_url, params={"key": API_SIGN_UP_FIREBASE_PATH}, data=payload)
-        r.raise_for_status()
-        data = r.json()
-        if 'idToken' in data:
-            return data['email']
-        else:
-            return False
-    except requests.exceptions.RequestException as e:
-        print(f"Error signing in: {e}")
-        return False
-
-def update_info_user(uid, email=None, user_name=None, photo_url=None):
-    user_data = {}
-    if email is not None:
-        user_data['email'] = email
-    if user_name is not None:
-        user_data['display_name'] = user_name
-    if photo_url is not None and photo_url != 'N/A':
-        user_data['photo_url'] = photo_url
-    if user_data:
-        auth.update_user(uid, **user_data)
-
-def update_user_info(request: req_login.RequestUpdateUserInfo):
- try:
-   email = request.email
-   user_id = request.user_id
-   email_check = sf.check_email_service(user_id)
-   if isinstance(email_check, res1.ReponseError):
-       return email_check
-   check_email_fc = sf.check_email_empty_invalid(email)
-   if check_email_fc is not True:
-       return check_email_fc
-   user = get_user1(email)
-   if user: 
-        user_info = UserInfoRepository.getUserInfo(user_id)
-        if user_info:
-          UserInfoRepository.updateUserInfo(
-            request.user_id,
-            request.uid,
-            request.email,
-            request.display_name,
-            request.photo_url)
-        else:
-            UserInfoRepository.addUserInfo(
-                request.uid,
-                request.email,
-                request.display_name,
-                request.photo_url
-            )
-        update_info_user(request.uid,
-            request.email,
-            request.display_name,
-            request.photo_url   
-              )
-        return res_login.ResponseUpdateUserInfo(status=200,
-                data = res_login.Message(message=f"User info updated successfully"))
-   else:
-          return res_login.ReponseError(
-            status = 404,
-            data = res_login.Message(message="Not found user")
-        )
- except:
-     return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error")
-        )
- 
-def check_info_google(request: req_login.RequestCheckInfoGoogle):
- try:
-   user_id = request.user_id
-   check = UserRepository.getEmailUserByIdFix(user_id)
-   if check is None:
-       return res_login.ReponseError(
-            status = 404,
-            data = res_login.Message(message="user_id not exist")
-        )
-   email = sf.check_email_service(str(user_id))
-   if isinstance(email, res.ReponseError):
-       return email
-   user_info = UserInfoRepository.getUserInfo(user_id)
-   if user_info is not None:
-       email_check = True
-   else:
-       email_check = False
-   if email_check is not None:
-    return res_login.ResponseCheckInfoGoogle(status= 200,data = res_login.CheckModel(check=email_check))
- except:
-     return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error")
-        )
-   
-def check_info_google_email(request: req_login.RequestCheckInfoGoogleEmail):
- try:
-   email = request.email
-   check_email_fc = sf.check_email_empty_invalid(email)
-   if check_email_fc is not True:
-       return check_email_fc
-   user_info = UserInfoRepository.getUserInfoByEmail(email)
-   if user_info is not None:
-       email_check = True
-   else:
-       email_check = False
-   if email_check is not None:
-    return res_login.ResponseCheckInfoGoogle(status= 200,data = res_login.CheckModel(check=email_check))
- except:
-     return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error")
-        )
-
-def check_state_login(request: req_login.RequestCheckStateLogin):
- try:
-   user_id = request.user_id
-   session_id_now = request.session_id_now
-   email = sf.check_email_service(user_id)
-   if isinstance(email, res1.ReponseError):
-       return email
-   elif session_id_now is None or session_id_now=="":
-       return res_login.ReponseError(
-            status= 400,
-            data =res_login.Message(message="session_id is empty")
-        )
-   user = get_user1(email)
-   if user:
-       check1 = False
-       session_id = UserLoginRepository.getUserSessionIdByUserEmail(user_id)
-       print(f"session_id: {session_id}")
-       if session_id != session_id_now:
-         check1 = False
-       else:
-         check1 = True
-       return res_login.ResponseCheckInfoGoogle(status= 200,data = res_login.CheckModel(check = check1))
-   else:
-       return res_login.ReponseError(
-            status=404,
-            data =res_login.Message(message="Not found user")
-        )
- except:
-  return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error")
-        )
-
-import string, random
-def generate_otp(length=6):
-    characters = string.ascii_uppercase + string.digits
-    otp = ''.join(random.choice(characters) for _ in range(length))
-    return otp
-
-def createOTPReset(email):
-    otp = generate_otp()
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    OTPRepository.addOTP(email,otp)
-    return otp
-
-def reset_password(request: req_login.RequestResetPassword):
- try:
-    email = request.email
-    check_email_fc = sf.check_email_empty_invalid(email)
-    if check_email_fc is not True:
-        return check_email_fc
-    try:
-       user = get_user1(email)
-       if user is not None:
-           otp = createOTPReset(email)
-           return res_login.ResponseCreateOTP(
-               status= 200,
-               data= res_login.CheckModel(check = True),
-               otp = otp
-           )
-       else:
-           return res_login.ReponseError(
-               status= 404,
-               data =res_login.Message(message="Email not exist")
-           )
-    except auth.UserNotFoundError as e:
-           return res_login.ReponseError(
-            status=500,
-            data =res_login.Message(message=str(e))
-        )
- except:
-     return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error")
-        )
-
-def change_password(request: req_login.RequestChangePassword):
- try:
-    user_id = request.user_id
-    email = sf.check_email_service(user_id)
-    new_password = request.new_password
-    current_password= request.current_password
-    confirm_new_password = request.confirm_new_password
-    if isinstance(email, res1.ReponseError):
-        return email
-    if new_password is None:
-        return res_login.ReponseError(
-            status=400,
-            data =res_login.Message(message="new_password is empty")
-        )
-    if current_password is None or confirm_new_password == "":
-        return res_login.ReponseError(
-            status=400,
-            data =res_login.Message(message="current_password is empty")
-        )
-    if confirm_new_password is None or confirm_new_password == "":
-        return res_login.ReponseError(
-            status=400,
-            data =res_login.Message(message="confirm_new_password is empty")
-        )
-    if current_password == new_password:
-        return res_login.ReponseError(
-            status=400,
-            data=res_login.Message(message="The new_password and the current_password must be different")
-        )
-    if confirm_new_password != new_password:
-        return res_login.ReponseError(
-            status=400,
-            data=res_login.Message(message="The new_password and the confirm_new_password must be similar")
-        )
-    user = sign_in_with_email_and_password(email, current_password)
-    try:
-     if user:
-      user_email = auth.get_user_by_email(email)
-      auth.update_user(
-        user_email.uid,
-        password=new_password
-    ) 
-      return res_login.ResponseChangePassword(
-               status= 200,
-               data = res_login.Message(message=f"Update password success"))
-     else:
-        return res_login.ReponseError(
-            status=400,
-            data =res_login.Message(message="Current password not valid")
-        )
-    except :
-      return res_login.ReponseError(
-            status=500,
-            data =res_login.Message(message="Server Error")
-        )
- except:
-     return res_login.ReponseError(
-            status=500,
-            data=res_login.Message(message="Server Error!!")
-        )