Upload folder using huggingface_hub

modules/dropbox.py

@@ -10,6 +10,23 @@ from datetime import datetime, timedelta, timezone
 
 load_dotenv()
 
+REFRESH_TOKEN = os.getenv("DROPBOX_REFRESH_TOKEN")
+APP_KEY = os.getenv("DROPBOX_APP_KEY")
+APP_SECRET = os.getenv("DROPBOX_APP_SECRET")
+
+if not REFRESH_TOKEN:
+    raise Exception("DROPBOX_REFRESH_TOKEN missing")
+
+if not APP_KEY:
+    raise Exception("APP_KEY missing")
+
+if not APP_SECRET:
+    raise Exception("APP_SECRET missing")
+
+dbx = dropbox.Dropbox(
+    app_key=APP_KEY, app_secret=APP_SECRET, oauth2_refresh_token=REFRESH_TOKEN
+)
+
 
 def list_dropbox_folder_hierarchy(dbx: dropbox.Dropbox, base_path: str = ""):
     """
@@ -62,22 +79,21 @@ def list_dropbox_folder_hierarchy(dbx: dropbox.Dropbox, base_path: str = ""):
     return hierarchy
 
 
-dbx = dropbox.Dropbox(os.getenv("DROPBOX_ACCESS_TOKEN"))
-
-
 class AudioRequest(BaseModel):
     scripture_name: str
     global_index: int
 
+
 # cache = {(scripture_name, global_index, type): {"url": ..., "expiry": ...}}
 audio_cache: dict[tuple[str, int, str], dict] = {}
 CACHE_TTL = timedelta(hours=3, minutes=30)  # refresh before 4h expiry
 
+
 async def get_audio_urls(req: AudioRequest):
     base_path = f"/{req.scripture_name}/audio"
     files_to_check = {
         "recitation": f"{req.global_index}-recitation.mp3",
-        "santhai": f"{req.global_index}-santhai.mp3"
+        "santhai": f"{req.global_index}-santhai.mp3",
     }
 
     urls = {}
@@ -100,10 +116,7 @@ async def get_audio_urls(req: AudioRequest):
                 temp_link = dbx.files_get_temporary_link(file_path).link
                 urls[key] = temp_link
                 # store in cache with expiry
-                audio_cache[cache_key] = {
-                    "url": temp_link,
-                    "expiry": now + CACHE_TTL
-                }
+                audio_cache[cache_key] = {"url": temp_link, "expiry": now + CACHE_TTL}
         except dropbox.exceptions.ApiError:
             urls[key] = None
 
@@ -112,6 +125,7 @@ async def get_audio_urls(req: AudioRequest):
 
     return urls
 
+
 async def cleanup_audio_url_cache(interval_seconds: int = 600):
     """Periodically remove expired entries from audio_cache."""
     while True:
@@ -127,13 +141,8 @@ async def cleanup_audio_url_cache(interval_seconds: int = 600):
 
 if __name__ == "__main__":
     # Create Dropbox client with your access token
-    ACCESS_TOKEN = os.getenv("DROPBOX_ACCESS_TOKEN")
-    if not ACCESS_TOKEN:
-        raise Exception("No access token available")
-    dbx = dropbox.Dropbox(ACCESS_TOKEN)
-
     # data = list_dropbox_folder_hierarchy(dbx, "")
-    data = asyncio.run(get_audio_urls(
-        AudioRequest(scripture_name="divya_prabandham", global_index=0)
-    ))
+    data = asyncio.run(
+        get_audio_urls(AudioRequest(scripture_name="divya_prabandham", global_index=0))
+    )
     print(json.dumps(data, indent=2))

tests/dropbox_token_generator.py

@@ -0,0 +1,35 @@
+import os
+import requests
+
+from dotenv import load_dotenv
+
+load_dotenv()
+
+AUTH_CODE = os.getenv("DROPBOX_AUTH_CODE")
+APP_KEY = os.getenv("DROPBOX_APP_KEY")
+APP_SECRET = os.getenv("DROPBOX_APP_SECRET")
+
+if not AUTH_CODE:
+    raise Exception("DROPBOX_AUTH_CODE missing")
+
+if not APP_KEY:
+    raise Exception("APP_KEY missing")
+
+if not APP_SECRET:
+    raise Exception("APP_SECRET missing")
+
+
+resp = requests.post(
+  "https://api.dropbox.com/oauth2/token",
+  data={
+    "code": os.getenv("DROPBOX_AUTH_CODE"),
+    "grant_type": "authorization_code",
+    # either include client_id/client_secret in the form:
+    "client_id": APP_KEY,
+    "client_secret": APP_SECRET
+  },
+  timeout=10
+)
+data = resp.json()
+print(data)
+# data contains 'access_token', 'refresh_token', 'expires_in'...