Spaces:
Sleeping
Sleeping
File size: 6,414 Bytes
63d1495 |
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 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 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峄梚 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:
print(f"Error deleting files: {e}") |