Update function/dropbox.py

function/dropbox.py

@@ -1,155 +1,155 @@
-import dropbox.files
-import os
-import shutil
-import requests, base64
-from fastapi import HTTPException
-from dotenv import load_dotenv
-import os
-load_dotenv()
-DROPBOX_APP_KEY=os.getenv('DROPBOX_APP_KEY')
-DROPBOX_APP_SECRET=os.getenv('DROPBOX_APP_SECRET')
-DROPBOX_REFRESH_TOKEN=os.getenv('DROPBOX_REFRESH_TOKEN')
-
-def refresh_token_dropbox():
-    app_key = DROPBOX_APP_KEY
-    app_secret = DROPBOX_APP_SECRET
-    refresh_token = DROPBOX_REFRESH_TOKEN
-    url = 'https://api.dropbox.com/oauth2/token'
-    auth_string = f"{app_key}:{app_secret}"
-    base64authorization = base64.b64encode(auth_string.encode()).decode('utf-8')
-    headers = {
-        'Authorization': f'Basic {base64authorization}',
-        'Content-Type': 'application/x-www-form-urlencoded'
-    }
-    data = {
-        'refresh_token': refresh_token,
-        'grant_type': 'refresh_token'
-    }
-    response = requests.post(url, headers=headers, data=data)
-    response_json = response.json()
-    access_token = response_json.get('access_token', None)
-    return access_token
-
-def delete_file(id,name_file):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx=dropbox.Dropbox(TOKEN)
-        file_path = f"/{id}/{name_file}"
-        dbx.files_delete_v2(file_path)
-        print(f"Xóa file '{file_path}' thành công.")
-    except dropbox.exceptions.ApiError as e:
-        print(f"Lỗi khi xóa file '{file_path}': {e}")
-
-def list_files(id):
-    file_names = []
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx=dropbox.Dropbox(TOKEN)
-        result = dbx.files_list_folder(f"/{id}")
-        for entry in result.entries:
-            if isinstance(entry, dropbox.files.FileMetadata):
-                file_names.append(os.path.basename(entry.path_display))
-    except dropbox.exceptions.ApiError as e:
-        print(f"Error listing files: {e}") 
-    return file_names
-
-def upload_file_fix(local_path,cloud_path,token):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx=dropbox.Dropbox(TOKEN)
-        with open(local_path, "rb") as f:
-            data = f.read()
-        dbx.files_upload(data, cloud_path)
-        print(f"Uploaded file '{local_path}' to '{cloud_path}'")
-    except dropbox.exceptions.ApiError as e:
-        print(f"Error uploading file '{local_path}': {e}")
-
-def upload_file(local_path, cloud_path):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx=dropbox.Dropbox(TOKEN)
-        with open(local_path, "rb") as f:
-            data = f.read()
-        dbx.files_upload(data, cloud_path)
-        print(f"Uploaded file '{local_path}' to '{cloud_path}'")
-    except dropbox.exceptions.ApiError as e:
-        upload_file_fix()
-
-def clear_local_folder(path):
-    try:
-        for filename in os.listdir(path):
-            file_path = os.path.join(path, filename)
-            if os.path.isfile(file_path) or os.path.islink(file_path):
-                os.unlink(file_path)
-            elif os.path.isdir(file_path):
-                shutil.rmtree(file_path)
-    except Exception as e:
-        print(f"Failed to delete contents of {path}. Reason: {e}")
-
-def download_folder(id):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx = dropbox.Dropbox(TOKEN)
-        local_path = f"./user_file/{id}"
-        os.makedirs(local_path, exist_ok=True)
-        clear_local_folder(local_path)
-        result = dbx.files_list_folder(f"/{id}")
-        for entry in result.entries:
-            if isinstance(entry, dropbox.files.FileMetadata):
-                cloud_file_path = entry.path_display
-                file_name = os.path.basename(cloud_file_path)
-                local_file_path = os.path.join(local_path, file_name)
-                dbx.files_download_to_file(local_file_path, cloud_file_path)
-                print(f"Downloaded file '{file_name}' to '{local_file_path}'")
-    except dropbox.exceptions.ApiError as e:
-        print(f"Error downloading file '{id}': {e}")
-
-def download_file_id(file_name, id):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx = dropbox.Dropbox(TOKEN)
-        local_folder_path = f"./user_file/{id}"
-        os.makedirs(local_folder_path, exist_ok=True)
-        local_file_path = os.path.join(local_folder_path, file_name)
-        with open(local_file_path, "wb") as f:
-            metadata, response = dbx.files_download(f"/{id}/{file_name}")
-            f.write(response.content)
-        print(f"Downloaded file '{file_name}' to '{local_file_path}'")
-    except dropbox.exceptions.ApiError as e:
-        print(f"Error downloading file '{file_name}': {e}")
-        raise HTTPException(status_code=500, detail="Internal Server Error")
-
-def search_and_download_file(start_char, id):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx = dropbox.Dropbox(TOKEN)
-        result = dbx.files_list_folder(f"/{id}")
-        files_starting_with_char = [entry.name for entry in result.entries if entry.name.startswith(start_char)]
-        if len(files_starting_with_char) == 0:
-            print(f"No file found starting with '{start_char}' in folder '{id}'")
-            return
-        file_name = files_starting_with_char[0]
-        local_folder_path = f"./user_file/{id}"
-        os.makedirs(local_folder_path, exist_ok=True)
-        local_file_path = os.path.join(local_folder_path, file_name)
-        with open(local_file_path, "wb") as f:
-            metadata, response = dbx.files_download(f"/{id}/{file_name}")
-            f.write(response.content)
-        print(f"Downloaded file '{file_name}' to '{local_file_path}'")
-    except dropbox.exceptions.ApiError as e:
-        print(f"Error searching or downloading file: {e}")
-        raise HTTPException(status_code=500, detail="Internal Server Error")
-
-def delete_all_files_in_folder(folder_id):
-    try:
-        TOKEN = refresh_token_dropbox()
-        dbx = dropbox.Dropbox(TOKEN)
-        result = dbx.files_list_folder(f"/{folder_id}")
-        for entry in result.entries:
-            if isinstance(entry, dropbox.files.FileMetadata):
-                file_path = entry.path_display
-                dbx.files_delete_v2(file_path)
-                print(f"Deleted file '{file_path}'")
-        print(f"All files in folder '{folder_id}' have been deleted.")
-    except dropbox.exceptions.ApiError as e:
+import dropbox.files
+import os
+import shutil
+import requests, base64
+from fastapi import HTTPException
+from dotenv import load_dotenv
+import os
+load_dotenv()
+DROPBOX_APP_KEY=os.getenv('DROPBOX_APP_KEY')
+DROPBOX_APP_SECRET=os.getenv('DROPBOX_APP_SECRET')
+DROPBOX_REFRESH_TOKEN=os.getenv('DROPBOX_REFRESH_TOKEN')
+
+def refresh_token_dropbox():
+    app_key = DROPBOX_APP_KEY
+    app_secret = DROPBOX_APP_SECRET
+    refresh_token = DROPBOX_REFRESH_TOKEN
+    url = 'https://api.dropbox.com/oauth2/token'
+    auth_string = f"{app_key}:{app_secret}"
+    base64authorization = base64.b64encode(auth_string.encode()).decode('utf-8')
+    headers = {
+        'Authorization': f'Basic {base64authorization}',
+        'Content-Type': 'application/x-www-form-urlencoded'
+    }
+    data = {
+        'refresh_token': refresh_token,
+        'grant_type': 'refresh_token'
+    }
+    response = requests.post(url, headers=headers, data=data)
+    response_json = response.json()
+    access_token = response_json.get('access_token', None)
+    return access_token
+
+def delete_file(id,name_file):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx=dropbox.Dropbox(TOKEN)
+        file_path = f"/{id}/{name_file}"
+        dbx.files_delete_v2(file_path)
+        print(f"Xóa file '{file_path}' thành công.")
+    except dropbox.exceptions.ApiError as e:
+        print(f"Lỗi khi xóa file '{file_path}': {e}")
+
+def list_files(id):
+    file_names = []
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx=dropbox.Dropbox(TOKEN)
+        result = dbx.files_list_folder(f"/{id}")
+        for entry in result.entries:
+            if isinstance(entry, dropbox.files.FileMetadata):
+                file_names.append(os.path.basename(entry.path_display))
+    except dropbox.exceptions.ApiError as e:
+        print(f"Error listing files: {e}") 
+    return file_names
+
+def upload_file_fix(local_path,cloud_path,token):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx=dropbox.Dropbox(TOKEN)
+        with open(local_path, "rb") as f:
+            data = f.read()
+        dbx.files_upload(data, cloud_path)
+        print(f"Uploaded file '{local_path}' to '{cloud_path}'")
+    except dropbox.exceptions.ApiError as e:
+        print(f"Error uploading file '{local_path}': {e}")
+
+def upload_file(local_path, cloud_path):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx=dropbox.Dropbox(TOKEN)
+        with open(local_path, "rb") as f:
+            data = f.read()
+        dbx.files_upload(data, cloud_path)
+        print(f"Uploaded file '{local_path}' to '{cloud_path}'")
+    except dropbox.exceptions.ApiError as e:
+        upload_file_fix()
+
+def clear_local_folder(path):
+    try:
+        for filename in os.listdir(path):
+            file_path = os.path.join(path, filename)
+            if os.path.isfile(file_path) or os.path.islink(file_path):
+                os.unlink(file_path)
+            elif os.path.isdir(file_path):
+                shutil.rmtree(file_path)
+    except Exception as e:
+        print(f"Failed to delete contents of {path}. Reason: {e}")
+
+def download_folder(id):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx = dropbox.Dropbox(TOKEN)
+        local_path = f"./code/user_file/{id}"
+        os.makedirs(local_path, exist_ok=True)
+        clear_local_folder(local_path)
+        result = dbx.files_list_folder(f"/{id}")
+        for entry in result.entries:
+            if isinstance(entry, dropbox.files.FileMetadata):
+                cloud_file_path = entry.path_display
+                file_name = os.path.basename(cloud_file_path)
+                local_file_path = os.path.join(local_path, file_name)
+                dbx.files_download_to_file(local_file_path, cloud_file_path)
+                print(f"Downloaded file '{file_name}' to '{local_file_path}'")
+    except dropbox.exceptions.ApiError as e:
+        print(f"Error downloading file '{id}': {e}")
+
+def download_file_id(file_name, id):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx = dropbox.Dropbox(TOKEN)
+        local_folder_path = f"/code/user_file/{id}"
+        os.makedirs(local_folder_path, exist_ok=True)
+        local_file_path = os.path.join(local_folder_path, file_name)
+        with open(local_file_path, "wb") as f:
+            metadata, response = dbx.files_download(f"/{id}/{file_name}")
+            f.write(response.content)
+        print(f"Downloaded file '{file_name}' to '{local_file_path}'")
+    except dropbox.exceptions.ApiError as e:
+        print(f"Error downloading file '{file_name}': {e}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+
+def search_and_download_file(start_char, id):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx = dropbox.Dropbox(TOKEN)
+        result = dbx.files_list_folder(f"/{id}")
+        files_starting_with_char = [entry.name for entry in result.entries if entry.name.startswith(start_char)]
+        if len(files_starting_with_char) == 0:
+            print(f"No file found starting with '{start_char}' in folder '{id}'")
+            return
+        file_name = files_starting_with_char[0]
+        local_folder_path = f"./code/user_file/{id}"
+        os.makedirs(local_folder_path, exist_ok=True)
+        local_file_path = os.path.join(local_folder_path, file_name)
+        with open(local_file_path, "wb") as f:
+            metadata, response = dbx.files_download(f"/{id}/{file_name}")
+            f.write(response.content)
+        print(f"Downloaded file '{file_name}' to '{local_file_path}'")
+    except dropbox.exceptions.ApiError as e:
+        print(f"Error searching or downloading file: {e}")
+        raise HTTPException(status_code=500, detail="Internal Server Error")
+
+def delete_all_files_in_folder(folder_id):
+    try:
+        TOKEN = refresh_token_dropbox()
+        dbx = dropbox.Dropbox(TOKEN)
+        result = dbx.files_list_folder(f"/{folder_id}")
+        for entry in result.entries:
+            if isinstance(entry, dropbox.files.FileMetadata):
+                file_path = entry.path_display
+                dbx.files_delete_v2(file_path)
+                print(f"Deleted file '{file_path}'")
+        print(f"All files in folder '{folder_id}' have been deleted.")
+    except dropbox.exceptions.ApiError as e:
         print(f"Error deleting files: {e}")