PlanOpticon

planopticon / video_processor / sources / dropbox_source.py
Source Blame History 227 lines
a6b6869… leo 1 """Dropbox source integration with OAuth support."""
a6b6869… leo 2
a6b6869… leo 3 import json
a6b6869… leo 4 import logging
a6b6869… leo 5 import os
a6b6869… leo 6 import webbrowser
a6b6869… leo 7 from pathlib import Path
a6b6869… leo 8 from typing import List, Optional
a6b6869… leo 9
a6b6869… leo 10 from video_processor.sources.base import BaseSource, SourceFile
a6b6869… leo 11
a6b6869… leo 12 logger = logging.getLogger(__name__)
a6b6869… leo 13
a6b6869… leo 14 # Video extensions we look for
a6b6869… leo 15 VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv"}
a6b6869… leo 16
a6b6869… leo 17 _TOKEN_PATH = Path.home() / ".planopticon" / "dropbox_token.json"
a6b6869… leo 18
a6b6869… leo 19
a6b6869… leo 20 class DropboxSource(BaseSource):
a6b6869… leo 21 """
a6b6869… leo 22 Dropbox source with OAuth2 support.
a6b6869… leo 23
a6b6869… leo 24 Auth methods:
a6b6869… leo 25 - Access token: Set DROPBOX_ACCESS_TOKEN env var for simple usage
a6b6869… leo 26 - OAuth2: Interactive browser-based flow with refresh tokens
a6b6869… leo 27 """
a6b6869… leo 28
a6b6869… leo 29 def __init__(
a6b6869… leo 30 self,
a6b6869… leo 31 access_token: Optional[str] = None,
a6b6869… leo 32 app_key: Optional[str] = None,
a6b6869… leo 33 app_secret: Optional[str] = None,
a6b6869… leo 34 token_path: Optional[Path] = None,
a6b6869… leo 35 ):
a6b6869… leo 36 """
a6b6869… leo 37 Initialize Dropbox source.
a6b6869… leo 38
a6b6869… leo 39 Parameters
a6b6869… leo 40 ----------
a6b6869… leo 41 access_token : str, optional
a6b6869… leo 42 Direct access token. Falls back to DROPBOX_ACCESS_TOKEN env var.
a6b6869… leo 43 app_key : str, optional
a6b6869… leo 44 Dropbox app key for OAuth. Falls back to DROPBOX_APP_KEY env var.
a6b6869… leo 45 app_secret : str, optional
a6b6869… leo 46 Dropbox app secret for OAuth. Falls back to DROPBOX_APP_SECRET env var.
a6b6869… leo 47 token_path : Path, optional
a6b6869… leo 48 Where to store/load OAuth tokens.
a6b6869… leo 49 """
a6b6869… leo 50 self.access_token = access_token or os.environ.get("DROPBOX_ACCESS_TOKEN")
a6b6869… leo 51 self.app_key = app_key or os.environ.get("DROPBOX_APP_KEY")
a6b6869… leo 52 self.app_secret = app_secret or os.environ.get("DROPBOX_APP_SECRET")
a6b6869… leo 53 self.token_path = token_path or _TOKEN_PATH
a6b6869… leo 54 self.dbx = None
a6b6869… leo 55
a6b6869… leo 56 def authenticate(self) -> bool:
a6b6869… leo 57 """Authenticate with Dropbox API."""
a6b6869… leo 58 try:
a6b6869… leo 59 import dropbox
a6b6869… leo 60 except ImportError:
829e24a… leo 61 logger.error("Dropbox SDK not installed. Run: pip install planopticon[dropbox]")
a6b6869… leo 62 return False
a6b6869… leo 63
a6b6869… leo 64 # Try direct access token first
a6b6869… leo 65 if self.access_token:
a6b6869… leo 66 return self._auth_token(dropbox)
a6b6869… leo 67
a6b6869… leo 68 # Try saved OAuth token
a6b6869… leo 69 if self.token_path.exists():
a6b6869… leo 70 if self._auth_saved_token(dropbox):
a6b6869… leo 71 return True
a6b6869… leo 72
a6b6869… leo 73 # Run OAuth flow
a6b6869… leo 74 return self._auth_oauth(dropbox)
a6b6869… leo 75
a6b6869… leo 76 def _auth_token(self, dropbox) -> bool:
a6b6869… leo 77 """Authenticate with a direct access token."""
a6b6869… leo 78 try:
a6b6869… leo 79 self.dbx = dropbox.Dropbox(self.access_token)
a6b6869… leo 80 self.dbx.users_get_current_account()
a6b6869… leo 81 logger.info("Authenticated with Dropbox via access token")
a6b6869… leo 82 return True
a6b6869… leo 83 except Exception as e:
a6b6869… leo 84 logger.error(f"Dropbox access token auth failed: {e}")
a6b6869… leo 85 return False
a6b6869… leo 86
a6b6869… leo 87 def _auth_saved_token(self, dropbox) -> bool:
a6b6869… leo 88 """Authenticate using a saved OAuth refresh token."""
a6b6869… leo 89 try:
a6b6869… leo 90 data = json.loads(self.token_path.read_text())
a6b6869… leo 91 refresh_token = data.get("refresh_token")
a6b6869… leo 92 app_key = data.get("app_key") or self.app_key
a6b6869… leo 93 app_secret = data.get("app_secret") or self.app_secret
a6b6869… leo 94
a6b6869… leo 95 if not refresh_token or not app_key:
a6b6869… leo 96 return False
a6b6869… leo 97
a6b6869… leo 98 self.dbx = dropbox.Dropbox(
a6b6869… leo 99 oauth2_refresh_token=refresh_token,
a6b6869… leo 100 app_key=app_key,
a6b6869… leo 101 app_secret=app_secret,
a6b6869… leo 102 )
a6b6869… leo 103 self.dbx.users_get_current_account()
a6b6869… leo 104 logger.info("Authenticated with Dropbox via saved token")
a6b6869… leo 105 return True
a6b6869… leo 106 except Exception:
a6b6869… leo 107 return False
a6b6869… leo 108
a6b6869… leo 109 def _auth_oauth(self, dropbox) -> bool:
a6b6869… leo 110 """Run OAuth2 PKCE flow."""
a6b6869… leo 111 if not self.app_key:
829e24a… leo 112 logger.error("Dropbox app key not configured. Set DROPBOX_APP_KEY env var.")
a6b6869… leo 113 return False
a6b6869… leo 114
a6b6869… leo 115 try:
a6b6869… leo 116 flow = dropbox.DropboxOAuth2FlowNoRedirect(
a6b6869… leo 117 consumer_key=self.app_key,
a6b6869… leo 118 consumer_secret=self.app_secret,
a6b6869… leo 119 token_access_type="offline",
a6b6869… leo 120 use_pkce=True,
a6b6869… leo 121 )
a6b6869… leo 122
a6b6869… leo 123 authorize_url = flow.start()
a6b6869… leo 124 print(f"\nOpen this URL to authorize PlanOpticon:\n{authorize_url}\n")
a6b6869… leo 125
a6b6869… leo 126 try:
a6b6869… leo 127 webbrowser.open(authorize_url)
a6b6869… leo 128 except Exception:
a6b6869… leo 129 pass
a6b6869… leo 130
a6b6869… leo 131 auth_code = input("Enter the authorization code: ").strip()
a6b6869… leo 132 result = flow.finish(auth_code)
a6b6869… leo 133
a6b6869… leo 134 self.dbx = dropbox.Dropbox(
a6b6869… leo 135 oauth2_refresh_token=result.refresh_token,
a6b6869… leo 136 app_key=self.app_key,
a6b6869… leo 137 app_secret=self.app_secret,
a6b6869… leo 138 )
a6b6869… leo 139
a6b6869… leo 140 # Save token
a6b6869… leo 141 self.token_path.parent.mkdir(parents=True, exist_ok=True)
a6b6869… leo 142 self.token_path.write_text(
a6b6869… leo 143 json.dumps(
a6b6869… leo 144 {
a6b6869… leo 145 "refresh_token": result.refresh_token,
a6b6869… leo 146 "app_key": self.app_key,
a6b6869… leo 147 "app_secret": self.app_secret or "",
a6b6869… leo 148 }
a6b6869… leo 149 )
a6b6869… leo 150 )
a6b6869… leo 151 logger.info(f"OAuth token saved to {self.token_path}")
a6b6869… leo 152 logger.info("Authenticated with Dropbox via OAuth")
a6b6869… leo 153 return True
a6b6869… leo 154 except Exception as e:
a6b6869… leo 155 logger.error(f"Dropbox OAuth failed: {e}")
a6b6869… leo 156 return False
a6b6869… leo 157
a6b6869… leo 158 def list_videos(
a6b6869… leo 159 self,
a6b6869… leo 160 folder_id: Optional[str] = None,
a6b6869… leo 161 folder_path: Optional[str] = None,
a6b6869… leo 162 patterns: Optional[List[str]] = None,
a6b6869… leo 163 ) -> List[SourceFile]:
a6b6869… leo 164 """List video files in a Dropbox folder."""
a6b6869… leo 165 if not self.dbx:
a6b6869… leo 166 raise RuntimeError("Not authenticated. Call authenticate() first.")
a6b6869… leo 167
a6b6869… leo 168 path = folder_path or ""
a6b6869… leo 169 if path and not path.startswith("/"):
a6b6869… leo 170 path = f"/{path}"
a6b6869… leo 171
a6b6869… leo 172 files = []
a6b6869… leo 173 try:
a6b6869… leo 174 result = self.dbx.files_list_folder(path, recursive=False)
a6b6869… leo 175
a6b6869… leo 176 while True:
a6b6869… leo 177 for entry in result.entries:
a6b6869… leo 178 import dropbox as dbx_module
a6b6869… leo 179
a6b6869… leo 180 if not isinstance(entry, dbx_module.files.FileMetadata):
a6b6869… leo 181 continue
a6b6869… leo 182
a6b6869… leo 183 ext = Path(entry.name).suffix.lower()
a6b6869… leo 184 if ext not in VIDEO_EXTENSIONS:
a6b6869… leo 185 continue
a6b6869… leo 186
a6b6869… leo 187 if patterns:
829e24a… leo 188 if not any(entry.name.endswith(p.replace("*", "")) for p in patterns):
a6b6869… leo 189 continue
a6b6869… leo 190
a6b6869… leo 191 files.append(
a6b6869… leo 192 SourceFile(
a6b6869… leo 193 name=entry.name,
a6b6869… leo 194 id=entry.id,
a6b6869… leo 195 size_bytes=entry.size,
a6b6869… leo 196 mime_type=None,
a6b6869… leo 197 modified_at=entry.server_modified.isoformat()
a6b6869… leo 198 if entry.server_modified
a6b6869… leo 199 else None,
a6b6869… leo 200 path=entry.path_display,
a6b6869… leo 201 )
a6b6869… leo 202 )
a6b6869… leo 203
a6b6869… leo 204 if not result.has_more:
a6b6869… leo 205 break
a6b6869… leo 206 result = self.dbx.files_list_folder_continue(result.cursor)
a6b6869… leo 207
a6b6869… leo 208 except Exception as e:
a6b6869… leo 209 logger.error(f"Failed to list Dropbox folder: {e}")
a6b6869… leo 210 raise
a6b6869… leo 211
a6b6869… leo 212 logger.info(f"Found {len(files)} videos in Dropbox")
a6b6869… leo 213 return files
a6b6869… leo 214
a6b6869… leo 215 def download(self, file: SourceFile, destination: Path) -> Path:
a6b6869… leo 216 """Download a file from Dropbox."""
a6b6869… leo 217 if not self.dbx:
a6b6869… leo 218 raise RuntimeError("Not authenticated. Call authenticate() first.")
a6b6869… leo 219
a6b6869… leo 220 destination = Path(destination)
a6b6869… leo 221 destination.parent.mkdir(parents=True, exist_ok=True)
a6b6869… leo 222
a6b6869… leo 223 path = file.path or f"/{file.name}"
a6b6869… leo 224 self.dbx.files_download_to_file(str(destination), path)
a6b6869… leo 225
a6b6869… leo 226 logger.info(f"Downloaded {file.name} to {destination}")
a6b6869… leo 227 return destination

Keyboard Shortcuts

Open search /
Next entry (timeline) j
Previous entry (timeline) k
Open focused entry Enter
Show this help ?
Toggle theme Top nav button