PlanOpticon

Add Google Drive and Dropbox cloud source integrations - BaseSource ABC with SourceFile pydantic model - GoogleDriveSource: service account + OAuth2 installed app flow, auto-detects auth method, stores tokens in ~/.planopticon/ - DropboxSource: OAuth2 PKCE flow with refresh tokens, direct access token fallback - CLI: batch --source gdrive/dropbox with --folder-id/--folder-path - CLI: planopticon auth google|dropbox for interactive auth - 20 tests for cloud sources (204 total passing)

leo 2026-02-14 22:36 trunk
Commit a6b6869d6fbc8545f3c62629a15b1ddeed39c77157cfe61c6cc9cce64685812e
Parent f0106a3cb04f75e…
6 files changed +162 +72 -5 +5 +66 +196 +169
+ tests/test_cloud_sources.py ~ video_processor/cli/commands.py + video_processor/sources/__init__.py + video_processor/sources/base.py + video_processor/sources/dropbox_source.py + video_processor/sources/google_drive.py
A tests/test_cloud_sources.py
+162
--- a/tests/test_cloud_sources.py
+++ b/tests/test_cloud_sources.py
@@ -0,0 +1,162 @@
1
+"""Tests for cloud source integrations."""
2
+
3
+import json
4
+from pathlib import Path
5
+from unittest.mock import MagicMock, patch
6
+
7
+import pytest
8
+
9
+from video_processor.sources.base import BaseSource, SourceFile
10
+
11
+
12
+class TestSourceFile:
13
+ def test_basic(self):
14
+ f = SourceFile(name="video.mp4", id="abc123")
15
+ assert f.name == "video.mp4"
16
+ assert f.id == "abc123"
17
+
18
+ def test_full(self):
19
+ f = SourceFile(
20
+ name="meeting.mp4",
21
+ id="xyz",
22
+ size_bytes=1024000,
23
+ mime_type="video/mp4",
24
+ modified_at="2025-01-01T00:00:00Z",
25
+ path="/recordings/meeting.mp4",
26
+ )
27
+ assert f.size_bytes == 1024000
28
+ assert f.path == "/recordings/meeting.mp4"
29
+
30
+ def test_round_trip(self):
31
+ f = SourceFile(name="test.mp4", id="1")
32
+ data = f.model_dump_json()
33
+ restored = SourceFile.model_validate_json(data)
34
+ assert restored.name == f.name
35
+
36
+
37
+class TestBaseSource:
38
+ def test_download_all(self, tmp_path):
39
+ class FakeSource(BaseSource):
40
+ def authenticate(self):
41
+ return True
42
+
43
+ def list_videos(self, **kwargs):
44
+ return []
45
+
46
+ def download(self, file, destination):
47
+ destination.write_text("fake video data")
48
+ return destination
49
+
50
+ source = FakeSource()
51
+ files = [
52
+ SourceFile(name="a.mp4", id="1"),
53
+ SourceFile(name="b.mp4", id="2"),
54
+ ]
55
+ paths = source.download_all(files, tmp_path / "downloads")
56
+ assert len(paths) == 2
57
+ assert (tmp_path / "downloads" / "a.mp4").exists()
58
+ assert (tmp_path / "downloads" / "b.mp4").exists()
59
+
60
+ def test_download_all_handles_errors(self, tmp_path):
61
+ class FailingSource(BaseSource):
62
+ def authenticate(self):
63
+ return True
64
+
65
+ def list_videos(self, **kwargs):
66
+ return []
67
+
68
+ def download(self, file, destination):
69
+ raise RuntimeError("Download failed")
70
+
71
+ source = FailingSource()
72
+ files = [SourceFile(name="fail.mp4", id="1")]
73
+ paths = source.download_all(files, tmp_path / "downloads")
74
+ assert len(paths) == 0
75
+
76
+
77
+class TestGoogleDriveSource:
78
+ def test_init_defaults(self):
79
+ from video_processor.sources.google_drive import GoogleDriveSource
80
+
81
+ source = GoogleDriveSource()
82
+ assert source.service is None
83
+ assert source.use_service_account is None
84
+
85
+ def test_init_with_credentials(self):
86
+ from video_processor.sources.google_drive import GoogleDriveSource
87
+
88
+ source = GoogleDriveSource(
89
+ credentials_path="/path/to/creds.json",
90
+ use_service_account=True,
91
+ )
92
+ assert source.credentials_path == "/path/to/creds.json"
93
+ assert source.use_service_account is True
94
+
95
+ def test_is_service_account_true(self, tmp_path):
96
+ from video_processor.sources.google_drive import GoogleDriveSource
97
+
98
+ creds_file = tmp_path / "sa.json"
99
+ creds_file.write_text(json.dumps({"type": "service_account"}))
100
+ source = GoogleDriveSource(credentials_path=str(creds_file))
101
+ assert source._is_service_account() is True
102
+
103
+ def test_is_service_account_false(self, tmp_path):
104
+ from video_processor.sources.google_drive import GoogleDriveSource
105
+
106
+ creds_file = tmp_path / "oauth.json"
107
+ creds_file.write_text(json.dumps({"installed": {}}))
108
+ source = GoogleDriveSource(credentials_path=str(creds_file))
109
+ assert source._is_service_account() is False
110
+
111
+ @patch.dict("os.environ", {}, clear=True)
112
+ def test_is_service_account_no_path(self):
113
+ from video_processor.sources.google_drive import GoogleDriveSource
114
+
115
+ source = GoogleDriveSource(credentials_path=None)
116
+ source.credentials_path = None # Override any env var fallback
117
+ assert source._is_service_account() is False
118
+
119
+ def test_list_videos_not_authenticated(self):
120
+ from video_processor.sources.google_drive import GoogleDriveSource
121
+
122
+ source = GoogleDriveSource()
123
+ with pytest.raises(RuntimeError, match="Not authenticated"):
124
+ source.list_videos(folder_id="abc")
125
+
126
+ def test_download_not_authenticated(self):
127
+ from video_processor.sources.google_drive import GoogleDriveSource
128
+
129
+ source = GoogleDriveSource()
130
+ f = SourceFile(name="test.mp4", id="1")
131
+ with pytest.raises(RuntimeError, match="Not authenticated"):
132
+ source.download(f, Path("/tmp/test.mp4"))
133
+
134
+ @patch("video_processor.sources.google_drive.GoogleDriveSource._auth_service_account")
135
+ def test_authenticate_import_error(self, mock_auth):
136
+ from video_processor.sources.google_drive import GoogleDriveSource
137
+
138
+ source = GoogleDriveSourceth patch.dict(
139
+ "sys.modules", {"google.oauth2": None, "google.oau):
140
+ # The import will fail inside authenticate
141
+ result = source.authenticate()
142
+ assert result is False
143
+
144
+
145
+class TestDropboxSource:
146
+ def test_init_defaults(self):
147
+ from video_processor.sources.dropbox_source import DropboxSource
148
+
149
+ source = DropboxSource()
150
+ assert source.dbx is None
151
+
152
+ def test_init_with_token(self):
153
+ from video_processor.sources.dropbox_source import DropboxSource
154
+
155
+ source = DropboxSource(access_token="test_token")
156
+ assert source.access_token == "test_token"
157
+
158
+ def test_init_with_app_key(self):
159
+ from video_processor.sources.dropbox_source import DropboxSource
160
+
161
+ source = DropboxSource(app_key="key", app_secret="secret")
162
+
--- a/tests/test_cloud_sources.py
+++ b/tests/test_cloud_sources.py
@@ -0,0 +1,162 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/tests/test_cloud_sources.py
+++ b/tests/test_cloud_sources.py
@@ -0,0 +1,162 @@
1 """Tests for cloud source integrations."""
2
3 import json
4 from pathlib import Path
5 from unittest.mock import MagicMock, patch
6
7 import pytest
8
9 from video_processor.sources.base import BaseSource, SourceFile
10
11
12 class TestSourceFile:
13 def test_basic(self):
14 f = SourceFile(name="video.mp4", id="abc123")
15 assert f.name == "video.mp4"
16 assert f.id == "abc123"
17
18 def test_full(self):
19 f = SourceFile(
20 name="meeting.mp4",
21 id="xyz",
22 size_bytes=1024000,
23 mime_type="video/mp4",
24 modified_at="2025-01-01T00:00:00Z",
25 path="/recordings/meeting.mp4",
26 )
27 assert f.size_bytes == 1024000
28 assert f.path == "/recordings/meeting.mp4"
29
30 def test_round_trip(self):
31 f = SourceFile(name="test.mp4", id="1")
32 data = f.model_dump_json()
33 restored = SourceFile.model_validate_json(data)
34 assert restored.name == f.name
35
36
37 class TestBaseSource:
38 def test_download_all(self, tmp_path):
39 class FakeSource(BaseSource):
40 def authenticate(self):
41 return True
42
43 def list_videos(self, **kwargs):
44 return []
45
46 def download(self, file, destination):
47 destination.write_text("fake video data")
48 return destination
49
50 source = FakeSource()
51 files = [
52 SourceFile(name="a.mp4", id="1"),
53 SourceFile(name="b.mp4", id="2"),
54 ]
55 paths = source.download_all(files, tmp_path / "downloads")
56 assert len(paths) == 2
57 assert (tmp_path / "downloads" / "a.mp4").exists()
58 assert (tmp_path / "downloads" / "b.mp4").exists()
59
60 def test_download_all_handles_errors(self, tmp_path):
61 class FailingSource(BaseSource):
62 def authenticate(self):
63 return True
64
65 def list_videos(self, **kwargs):
66 return []
67
68 def download(self, file, destination):
69 raise RuntimeError("Download failed")
70
71 source = FailingSource()
72 files = [SourceFile(name="fail.mp4", id="1")]
73 paths = source.download_all(files, tmp_path / "downloads")
74 assert len(paths) == 0
75
76
77 class TestGoogleDriveSource:
78 def test_init_defaults(self):
79 from video_processor.sources.google_drive import GoogleDriveSource
80
81 source = GoogleDriveSource()
82 assert source.service is None
83 assert source.use_service_account is None
84
85 def test_init_with_credentials(self):
86 from video_processor.sources.google_drive import GoogleDriveSource
87
88 source = GoogleDriveSource(
89 credentials_path="/path/to/creds.json",
90 use_service_account=True,
91 )
92 assert source.credentials_path == "/path/to/creds.json"
93 assert source.use_service_account is True
94
95 def test_is_service_account_true(self, tmp_path):
96 from video_processor.sources.google_drive import GoogleDriveSource
97
98 creds_file = tmp_path / "sa.json"
99 creds_file.write_text(json.dumps({"type": "service_account"}))
100 source = GoogleDriveSource(credentials_path=str(creds_file))
101 assert source._is_service_account() is True
102
103 def test_is_service_account_false(self, tmp_path):
104 from video_processor.sources.google_drive import GoogleDriveSource
105
106 creds_file = tmp_path / "oauth.json"
107 creds_file.write_text(json.dumps({"installed": {}}))
108 source = GoogleDriveSource(credentials_path=str(creds_file))
109 assert source._is_service_account() is False
110
111 @patch.dict("os.environ", {}, clear=True)
112 def test_is_service_account_no_path(self):
113 from video_processor.sources.google_drive import GoogleDriveSource
114
115 source = GoogleDriveSource(credentials_path=None)
116 source.credentials_path = None # Override any env var fallback
117 assert source._is_service_account() is False
118
119 def test_list_videos_not_authenticated(self):
120 from video_processor.sources.google_drive import GoogleDriveSource
121
122 source = GoogleDriveSource()
123 with pytest.raises(RuntimeError, match="Not authenticated"):
124 source.list_videos(folder_id="abc")
125
126 def test_download_not_authenticated(self):
127 from video_processor.sources.google_drive import GoogleDriveSource
128
129 source = GoogleDriveSource()
130 f = SourceFile(name="test.mp4", id="1")
131 with pytest.raises(RuntimeError, match="Not authenticated"):
132 source.download(f, Path("/tmp/test.mp4"))
133
134 @patch("video_processor.sources.google_drive.GoogleDriveSource._auth_service_account")
135 def test_authenticate_import_error(self, mock_auth):
136 from video_processor.sources.google_drive import GoogleDriveSource
137
138 source = GoogleDriveSourceth patch.dict(
139 "sys.modules", {"google.oauth2": None, "google.oau):
140 # The import will fail inside authenticate
141 result = source.authenticate()
142 assert result is False
143
144
145 class TestDropboxSource:
146 def test_init_defaults(self):
147 from video_processor.sources.dropbox_source import DropboxSource
148
149 source = DropboxSource()
150 assert source.dbx is None
151
152 def test_init_with_token(self):
153 from video_processor.sources.dropbox_source import DropboxSource
154
155 source = DropboxSource(access_token="test_token")
156 assert source.access_token == "test_token"
157
158 def test_init_with_app_key(self):
159 from video_processor.sources.dropbox_source import DropboxSource
160
161 source = DropboxSource(app_key="key", app_secret="secret")
162
M video_processor/cli/commands.py
+72 -5
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -118,11 +118,11 @@
118118
 traceback.print_exc()
119119
 sys.exit(1)
120120
121121
122122
 @cli.command()
123
-@click.option("--input-dir", "-i", required=True, type=click.Path(exists=True), help="Directory of videos")
123
+@click.option("--input-dir", "-i", type=click.Path(), default=None, help="Local directory of videos")
124124
 @click.option("--output", "-o", required=True, type=click.Path(), help="Output directory")
125125
 @click.option(
126126
 "--depth",
127127
 type=click.Choice(["basic", "standard", "comprehensive"]),
128128
 default="standard",
@@ -142,12 +142,20 @@
142142
 default="auto",
143143
 help="API provider",
144144
 )
145145
 @click.option("--vision-model", type=str, default=None, help="Override model for vision tasks")
146146
 @click.option("--chat-model", type=str, default=None, help="Override model for LLM/chat tasks")
147
+@click.option(
148
+ "--source",
149
+ type=click.Choice(["local", "gdrive", "dropbox"]),
150
+ default="local",
151
+ help="Video source (local directory, Google Drive, or Dropbox)",
152
+)
153
+@click.option("--folder-id", type=str, default=None, help="Google Drive folder ID")
154
+@click.option("--folder-path", type=str, default=None, help="Cloud folder path")
147155
 @click.pass_context
148
-def batch(ctx, input_dir, output, depth, pattern, title, provider, vision_model, chat_model):
156
+def batch(ctx, input_dir, output, depth, pattern, title, provider, vision_model, chat_model, source, folder_id, folder_path):
149157
 """Process a folder of videos in batch."""
150158
 from video_processor.integrators.knowledge_graph import KnowledgeGraph
151159
 from video_processor.integrators.plan_generator import PlanGenerator
152160
 from video_processor.models import BatchManifest, BatchVideoEntry
153161
 from video_processor.output_structure import (
@@ -156,16 +164,49 @@
156164
 write_batch_manifest,
157165
 )
158166
 from video_processor.pipeline import process_single_video
159167
 from video_processor.providers.manager import ProviderManager
160168
161
- input_dir = Path(input_dir)
162169
 prov = None if provider == "auto" else provider
163170
 pm = ProviderManager(vision_model=vision_model, chat_model=chat_model, provider=prov)
164
-
165
- # Find videos
166171
 patterns = [p.strip() for p in pattern.split(",")]
172
+
173
+ # Handle cloud sources
174
+ if source != "local":
175
+ download_dir = Path(output) / "_downloads"
176
+ download_dir.mkdir(parents=True, exist_ok=True)
177
+
178
+ if source == "gdrive":
179
+ from video_processor.sources.google_drive import GoogleDriveSource
180
+
181
+ cloud = GoogleDriveSource()
182
+ if not cloud.authenticate():
183
+ logging.error("Google Drive authentication failed")
184
+ sys.exit(1)
185
+ cloud_files = cloud.list_videos(folder_id=folder_id, folder_path=folder_path, patterns=patterns)
186
+ local_paths = cloud.download_all(cloud_files, download_dir)
187
+ elif source == "dropbox":
188
+ from video_processor.sources.dropbox_source import DropboxSource
189
+
190
+ cloud = DropboxSource()
191
+ if not cloud.authenticate():
192
+ logging.error("Dropbox authentication failed")
193
+ sys.exit(1)
194
+ cloud_files = cloud.list_videos(folder_path=folder_path, patterns=patterns)
195
+ local_paths = cloud.download_all(cloud_files, download_dir)
196
+ else:
197
+ logging.error(f"Unknown source: {source}")
198
+ sys.exit(1)
199
+
200
+ input_dir = download_dir
201
+ else:
202
+ if not input_dir:
203
+ logging.error("--input-dir is required for local source")
204
+ sys.exit(1)
205
+ input_dir = Path(input_dir)
206
+
207
+ # Find videos
167208
 videos = []
168209
 for pat in patterns:
169210
 videos.extend(sorted(input_dir.glob(pat)))
170211
 videos = sorted(set(videos))
171212
@@ -318,13 +359,39 @@
318359
 import traceback
319360
320361
 traceback.print_exc()
321362
 sys.exit(1)
322363
364
+
365
+@cli.command()
366
+@click.argument("service", type=click.Choice(["google", "dropbox"]))
367
+@click.pass_context
368
+def auth(ctx, service):
369
+ """Authenticate with a cloud service (google or dropbox)."""
370
+ if service == "google":
371
+ from video_processor.sources.google_drive import GoogleDriveSource
372
+
373
+ source = GoogleDriveSource(use_service_account=False)
374
+ if source.authenticate():
375
+ click.echo("Google Drive authentication successful.")
376
+ else:
377
+ click.echo("Google Drive authentication failed.", err=True)
378
+ sys.exit(1)
379
+
380
+ elif service == "dropbox":
381
+ from video_processor.sources.dropbox_source import DropboxSource
382
+
383
+ source = DropboxSource()
384
+ if source.authenticate():
385
+ click.echo("Dropbox authentication successful.")
386
+ else:
387
+ click.echo("Dropbox authentication failed.", err=True)
388
+ sys.exit(1)
389
+
323390
324391
 def main():
325392
 """Entry point for command-line usage."""
326393
 cli(obj={})
327394
328395
329396
 if __name__ == "__main__":
330397
 main()
331398
332399
ADDED video_processor/sources/__init__.py
333400
ADDED video_processor/sources/base.py
334401
ADDED video_processor/sources/dropbox_source.py
335402
ADDED video_processor/sources/google_drive.py
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -118,11 +118,11 @@
118 traceback.print_exc()
119 sys.exit(1)
120
121
122 @cli.command()
123 @click.option("--input-dir", "-i", required=True, type=click.Path(exists=True), help="Directory of videos")
124 @click.option("--output", "-o", required=True, type=click.Path(), help="Output directory")
125 @click.option(
126 "--depth",
127 type=click.Choice(["basic", "standard", "comprehensive"]),
128 default="standard",
@@ -142,12 +142,20 @@
142 default="auto",
143 help="API provider",
144 )
145 @click.option("--vision-model", type=str, default=None, help="Override model for vision tasks")
146 @click.option("--chat-model", type=str, default=None, help="Override model for LLM/chat tasks")
 
 
 
 
 
 
 
 
147 @click.pass_context
148 def batch(ctx, input_dir, output, depth, pattern, title, provider, vision_model, chat_model):
149 """Process a folder of videos in batch."""
150 from video_processor.integrators.knowledge_graph import KnowledgeGraph
151 from video_processor.integrators.plan_generator import PlanGenerator
152 from video_processor.models import BatchManifest, BatchVideoEntry
153 from video_processor.output_structure import (
@@ -156,16 +164,49 @@
156 write_batch_manifest,
157 )
158 from video_processor.pipeline import process_single_video
159 from video_processor.providers.manager import ProviderManager
160
161 input_dir = Path(input_dir)
162 prov = None if provider == "auto" else provider
163 pm = ProviderManager(vision_model=vision_model, chat_model=chat_model, provider=prov)
164
165 # Find videos
166 patterns = [p.strip() for p in pattern.split(",")]
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
167 videos = []
168 for pat in patterns:
169 videos.extend(sorted(input_dir.glob(pat)))
170 videos = sorted(set(videos))
171
@@ -318,13 +359,39 @@
318 import traceback
319
320 traceback.print_exc()
321 sys.exit(1)
322
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
323
324 def main():
325 """Entry point for command-line usage."""
326 cli(obj={})
327
328
329 if __name__ == "__main__":
330 main()
331
332 DDED video_processor/sources/__init__.py
333 DDED video_processor/sources/base.py
334 DDED video_processor/sources/dropbox_source.py
335 DDED video_processor/sources/google_drive.py
--- video_processor/cli/commands.py
+++ video_processor/cli/commands.py
@@ -118,11 +118,11 @@
118 traceback.print_exc()
119 sys.exit(1)
120
121
122 @cli.command()
123 @click.option("--input-dir", "-i", type=click.Path(), default=None, help="Local directory of videos")
124 @click.option("--output", "-o", required=True, type=click.Path(), help="Output directory")
125 @click.option(
126 "--depth",
127 type=click.Choice(["basic", "standard", "comprehensive"]),
128 default="standard",
@@ -142,12 +142,20 @@
142 default="auto",
143 help="API provider",
144 )
145 @click.option("--vision-model", type=str, default=None, help="Override model for vision tasks")
146 @click.option("--chat-model", type=str, default=None, help="Override model for LLM/chat tasks")
147 @click.option(
148 "--source",
149 type=click.Choice(["local", "gdrive", "dropbox"]),
150 default="local",
151 help="Video source (local directory, Google Drive, or Dropbox)",
152 )
153 @click.option("--folder-id", type=str, default=None, help="Google Drive folder ID")
154 @click.option("--folder-path", type=str, default=None, help="Cloud folder path")
155 @click.pass_context
156 def batch(ctx, input_dir, output, depth, pattern, title, provider, vision_model, chat_model, source, folder_id, folder_path):
157 """Process a folder of videos in batch."""
158 from video_processor.integrators.knowledge_graph import KnowledgeGraph
159 from video_processor.integrators.plan_generator import PlanGenerator
160 from video_processor.models import BatchManifest, BatchVideoEntry
161 from video_processor.output_structure import (
@@ -156,16 +164,49 @@
164 write_batch_manifest,
165 )
166 from video_processor.pipeline import process_single_video
167 from video_processor.providers.manager import ProviderManager
168
 
169 prov = None if provider == "auto" else provider
170 pm = ProviderManager(vision_model=vision_model, chat_model=chat_model, provider=prov)
 
 
171 patterns = [p.strip() for p in pattern.split(",")]
172
173 # Handle cloud sources
174 if source != "local":
175 download_dir = Path(output) / "_downloads"
176 download_dir.mkdir(parents=True, exist_ok=True)
177
178 if source == "gdrive":
179 from video_processor.sources.google_drive import GoogleDriveSource
180
181 cloud = GoogleDriveSource()
182 if not cloud.authenticate():
183 logging.error("Google Drive authentication failed")
184 sys.exit(1)
185 cloud_files = cloud.list_videos(folder_id=folder_id, folder_path=folder_path, patterns=patterns)
186 local_paths = cloud.download_all(cloud_files, download_dir)
187 elif source == "dropbox":
188 from video_processor.sources.dropbox_source import DropboxSource
189
190 cloud = DropboxSource()
191 if not cloud.authenticate():
192 logging.error("Dropbox authentication failed")
193 sys.exit(1)
194 cloud_files = cloud.list_videos(folder_path=folder_path, patterns=patterns)
195 local_paths = cloud.download_all(cloud_files, download_dir)
196 else:
197 logging.error(f"Unknown source: {source}")
198 sys.exit(1)
199
200 input_dir = download_dir
201 else:
202 if not input_dir:
203 logging.error("--input-dir is required for local source")
204 sys.exit(1)
205 input_dir = Path(input_dir)
206
207 # Find videos
208 videos = []
209 for pat in patterns:
210 videos.extend(sorted(input_dir.glob(pat)))
211 videos = sorted(set(videos))
212
@@ -318,13 +359,39 @@
359 import traceback
360
361 traceback.print_exc()
362 sys.exit(1)
363
364
365 @cli.command()
366 @click.argument("service", type=click.Choice(["google", "dropbox"]))
367 @click.pass_context
368 def auth(ctx, service):
369 """Authenticate with a cloud service (google or dropbox)."""
370 if service == "google":
371 from video_processor.sources.google_drive import GoogleDriveSource
372
373 source = GoogleDriveSource(use_service_account=False)
374 if source.authenticate():
375 click.echo("Google Drive authentication successful.")
376 else:
377 click.echo("Google Drive authentication failed.", err=True)
378 sys.exit(1)
379
380 elif service == "dropbox":
381 from video_processor.sources.dropbox_source import DropboxSource
382
383 source = DropboxSource()
384 if source.authenticate():
385 click.echo("Dropbox authentication successful.")
386 else:
387 click.echo("Dropbox authentication failed.", err=True)
388 sys.exit(1)
389
390
391 def main():
392 """Entry point for command-line usage."""
393 cli(obj={})
394
395
396 if __name__ == "__main__":
397 main()
398
399 DDED video_processor/sources/__init__.py
400 DDED video_processor/sources/base.py
401 DDED video_processor/sources/dropbox_source.py
402 DDED video_processor/sources/google_drive.py
A video_processor/sources/__init__.py
+5
--- a/video_processor/sources/__init__.py
+++ b/video_processor/sources/__init__.py
@@ -0,0 +1,5 @@
1
+"""Cloud source integrations for fetching videos from remote storage."""
2
+
3
+from video_processor.sources.base import BaseSource, SourceFile
4
+
5
+__all__ = ["
--- a/video_processor/sources/__init__.py
+++ b/video_processor/sources/__init__.py
@@ -0,0 +1,5 @@
 
 
 
 
 
--- a/video_processor/sources/__init__.py
+++ b/video_processor/sources/__init__.py
@@ -0,0 +1,5 @@
1 """Cloud source integrations for fetching videos from remote storage."""
2
3 from video_processor.sources.base import BaseSource, SourceFile
4
5 __all__ = ["
A video_processor/sources/base.py
+66
--- a/video_processor/sources/base.py
+++ b/video_processor/sources/base.py
@@ -0,0 +1,66 @@
1
+"""Base interface for cloud source integrations."""
2
+
3
+import logging
4
+from abc import ABC, abstractmethod
5
+from pathlib import Path
6
+from typing import List, Optional
7
+
8
+from pydantic import BaseModel, Field
9
+
10
+logger = logging.getLogger(__name__)
11
+
12
+
13
+class SourceFile(BaseModel):
14
+ """A file availble in a cloud source."""
15
+
16
+ name: str = Field(description="File name")
17
+ id: str = Field(description="Provider-specific file identifier")
18
+ size_bytes: Optional[int] = Field(default=None, description="File size in bytes")
19
+ mime_type: Optional[str] = Field(default=None, description="MIME type")
20
+ modified_at: Optional[str] = Field(default=None, description="Last modified timestamp")
21
+ path: Optional[str] = Field(default=None, description="Path within the source folder")
22
+
23
+
24
+class BaseSource(ABC):
25
+ """Abstract base class for cloud source integrations."""
26
+
27
+ @abstractmethod
28
+ def authenticate(self) -> bool:
29
+ """Authenticate with the cloud provider. Returns True on success."""
30
+ ...
31
+
32
+ @abstractmethod
33
+ def list_videos(
34
+ self,
35
+ folder_id: Optional[str] = None,
36
+ folder_path: Optional[str] = None,
37
+ patterns: Optional[List[str]] = None,
38
+ ) -> List[SourceFile]:
39
+ """List video files in a folder."""
40
+ ...
41
+
42
+ @abstractmethod
43
+ def download(
44
+ self,
45
+ file: SourceFile,
46
+ destination: Path,
47
+ ) -> Path:
48
+ """Download a file to a local path. Returns the local path."""
49
+ ...
50
+
51
+ def download_all(
52
+ self,
53
+ files: List[SourceFile],
54
+ destination_dir: Path,
55
+ ) -> List[Path]:
56
+ """Download mu."""
57
+ destination_dir"""
58
+ destination_dir.mkdir(parents=True, exist_ok=True)
59
+ paths = []
60
+ fdest = destination_dir / f.nam= destination_dir / relative
61
+ try:
62
+ local_path = self.download(f, dest)
63
+ paths.append(local_path)
64
+ f.name}")
65
+ except Exception as e:
66
+ logger.error(f"Failed to download {f.namFailed to download {relative}:
--- a/video_processor/sources/base.py
+++ b/video_processor/sources/base.py
@@ -0,0 +1,66 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/sources/base.py
+++ b/video_processor/sources/base.py
@@ -0,0 +1,66 @@
1 """Base interface for cloud source integrations."""
2
3 import logging
4 from abc import ABC, abstractmethod
5 from pathlib import Path
6 from typing import List, Optional
7
8 from pydantic import BaseModel, Field
9
10 logger = logging.getLogger(__name__)
11
12
13 class SourceFile(BaseModel):
14 """A file availble in a cloud source."""
15
16 name: str = Field(description="File name")
17 id: str = Field(description="Provider-specific file identifier")
18 size_bytes: Optional[int] = Field(default=None, description="File size in bytes")
19 mime_type: Optional[str] = Field(default=None, description="MIME type")
20 modified_at: Optional[str] = Field(default=None, description="Last modified timestamp")
21 path: Optional[str] = Field(default=None, description="Path within the source folder")
22
23
24 class BaseSource(ABC):
25 """Abstract base class for cloud source integrations."""
26
27 @abstractmethod
28 def authenticate(self) -> bool:
29 """Authenticate with the cloud provider. Returns True on success."""
30 ...
31
32 @abstractmethod
33 def list_videos(
34 self,
35 folder_id: Optional[str] = None,
36 folder_path: Optional[str] = None,
37 patterns: Optional[List[str]] = None,
38 ) -> List[SourceFile]:
39 """List video files in a folder."""
40 ...
41
42 @abstractmethod
43 def download(
44 self,
45 file: SourceFile,
46 destination: Path,
47 ) -> Path:
48 """Download a file to a local path. Returns the local path."""
49 ...
50
51 def download_all(
52 self,
53 files: List[SourceFile],
54 destination_dir: Path,
55 ) -> List[Path]:
56 """Download mu."""
57 destination_dir"""
58 destination_dir.mkdir(parents=True, exist_ok=True)
59 paths = []
60 fdest = destination_dir / f.nam= destination_dir / relative
61 try:
62 local_path = self.download(f, dest)
63 paths.append(local_path)
64 f.name}")
65 except Exception as e:
66 logger.error(f"Failed to download {f.namFailed to download {relative}:
A video_processor/sources/dropbox_source.py
+196
--- a/video_processor/sources/dropbox_source.py
+++ b/video_processor/sources/dropbox_source.py
@@ -0,0 +1,196 @@
1
+"""Dropbox source integration with OAuth support."""
2
+
3
+import json
4
+import logging
5
+import os
6
+import webbrowser
7
+from pathlib import Path
8
+from typing import List, Optional
9
+
10
+from video_processor.sources.base import BaseSource, SourceFile
11
+
12
+logger = logging.getLogger(__name__)
13
+
14
+# Video extensions we look for
15
+VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv"}
16
+
17
+_TOKEN_PATH = Path.home() / ".planopticon" / "dropbox_token.json"
18
+
19
+
20
+class DropboxSource(BaseSource):
21
+ """
22
+ Dropbox source with OAuth2 support.
23
+
24
+ Auth methods:
25
+ - Access token: Set DROPBOX_ACCESS_TOKEN env var for simple usage
26
+ - OAuth2: Interactive browser-based flow with refresh tokens
27
+ """
28
+
29
+ def __init__(
30
+ self,
31
+ access_token: Optional[str] = None,
32
+ app_key: Optional[str] = None,
33
+ app_secret: Optional[str] = None,
34
+ token_path: Optional[Path] = None,
35
+ ):
36
+ """
37
+ Initialize Dropbox source.
38
+
39
+ Parameters
40
+ ----------
41
+ access_token : str, optional
42
+ Direct access token. Falls back to DROPBOX_ACCESS_TOKEN env var.
43
+ app_key : str, optional
44
+ Dropbox app key for OAuth. Falls back to DROPBOX_APP_KEY env var.
45
+ app_secret : str, optional
46
+ Dropbox app secret for OAuth. Falls back to DROPBOX_APP_SECRET env var.
47
+ token_path : Path, optional
48
+ Where to store/load OAuth tokens.
49
+ """
50
+ self.access_token = access_token or os.environ.get("DROPBOX_ACCESS_TOKEN")
51
+ self.app_key = app_key or os.environ.get("DROPBOX_APP_KEY")
52
+ self.app_secret = app_secret or os.environ.get("DROPBOX_APP_SECRET")
53
+ self.token_path = token_path or _TOKEN_PATH
54
+ self.dbx = None
55
+
56
+ def authenticate(self) -> bool:
57
+ """Authenticate with Dropbox API."""
58
+ try:
59
+ import dropbox
60
+ except ImportErrorpath}")
61
+ :
62
+ logger.error("Dropbox SDK not installed. Run: pip in
63
+""Dropbox so with OAuth support."pbox]")
64
+ return False
65
+
66
+ # Try direct access token first
67
+ if self.access_token:
68
+ return self._auth_token(dropbox)
69
+
70
+ # Try saved OAuth token
71
+ if self.token_path.exists():
72
+ if self._auth_saved_token(dropbox):
73
+ return True
74
+
75
+ # Run OAuth flow
76
+ return self._auth_oauth(dropbox)
77
+
78
+ def _auth_token(self, dropbox) -> bool:
79
+ """Authenticate with a direct access token."""
80
+ try:
81
+ self.dbx = dropbox.Dropbox(self.access_token)
82
+ self.dbx.users_get_current_account()
83
+ logger.info("Authenticated with Dropbox via access token")
84
+ return True
85
+ except Exception as e:
86
+ logger.error(f"Dropbox access token auth failed: {e}")
87
+ return False
88
+
89
+ def _auth_saved_token(self, dropbox) -> bool:
90
+ """Authenticate using a saved OAuth refresh token."""
91
+ try:
92
+ data = json.loads(self.token_path.read_text())
93
+ refresh_token = data.get("refresh_token")
94
+ app_key = data.get("app_key") or self.app_key
95
+ app_secret = data.get("app_secret") or self.app_secret
96
+
97
+ if not refresh_token or not app_key:
98
+ return False
99
+
100
+ self.dbx = dropbox.Dropbox(
101
+ oauth2_refresh_token=refresh_token,
102
+ app_key=app_key,
103
+ app_secret=app_secret,
104
+ )
105
+ self.dbx.users_get_current_account()
106
+ logger.info("Authenticated with Dropbox via saved token")
107
+ return True
108
+ except Exception:
109
+ return False
110
+
111
+ def _auth_oauth(self, dropbox) -> bool:
112
+ """Run OAuth2 PKCE flow."""
113
+path}")
114
+ " str, optional
115
+ # Try direct acc
116
+""Dropbox so with OAuth support." return self._auth_token(dropbox)
117
+
118
+ # Try saved OAuth token
119
+ if self.token_path.exists():
120
+ if self._auth_saved_token(dropbox):
121
+ return True
122
+
123
+ # Run OAuth flow
124
+ return self._auth_oauth(dropbox)
125
+
126
+ def _auth_token(self, dropbox) -> bool:
127
+ """Authenticate with a direct access token."""
128
+ try:
129
+ self.dbx = dropbox.Dropbox(self.access_token)
130
+ self.dbx.users_get_current_account()
131
+ logger.info("Authenticated with Dropbox via access token")
132
+ return True
133
+ except Exception as e:
134
+ logger.error(f"Dropbox access token auth failed: {e}")
135
+ return False
136
+
137
+ def _auth_saved_token(self, dropbox) -> bool:
138
+ """Authenticate upath}")
139
+ logger.inentry.name.endswith(p.replace(path}")
140
+ logge):path}")
141
+ logger.incontinue
142
+
143
+ ""Dropbox source integration with OAuth support."""
144
+
145
+import json
146
+import logging
147
+import os
148
+import webbrowser
149
+from pathlib import Path
150
+from typing import List, Optional
151
+
152
+from video_processor.sources.base import BaseSource, SourceFile
153
+
154
+logger = logging.getLogger(__name__)
155
+
156
+# Video extensions we look for
157
+VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv"}
158
+
159
+_TOKEN_PATH = Path.home() / ".planopticon" / "dropbox_token.json"
160
+
161
+
162
+class DropboxSource(BaseSource):
163
+ """
164
+ Dropbox source with OAuth2 support.
165
+
166
+ Auth methods:
167
+ - Access token: Set DROPBOX_ACCESS_TOKEN env var for simple usage
168
+ - OAuth2: Interactive browser-based flow with refresh tokens
169
+ """
170
+
171
+ def __init__(
172
+ self,
173
+ access_token: Optional[str] = None,
174
+ app_key: Optional[str] = None,
175
+ app_secret: Optional[str] = None,
176
+ token_path: Optional[Path] = None,
177
+ ):
178
+ """
179
+ Initialize Dropbox source.
180
+
181
+ Parameters
182
+ ----------
183
+ access_token : str, optional
184
+ Direct access token. Falls back to DROPBOX_ACCESS_TOKEN env var.
185
+ app_key : str, optional
186
+ Dropbox app key for OAuth. Falls back to DROPBOX_APP_KEY env var.
187
+ app_secret : str, optional
188
+ Dropbox app secret for OAuth. Falls back to DROPBOX_APP_SECRET env var.
189
+ token_path : Path, optional
190
+ ropbox source integration with OAuth support."""
191
+
192
+import json
193
+import logging
194
+import os
195
+import webbrowser
196
+fro
--- a/video_processor/sources/dropbox_source.py
+++ b/video_processor/sources/dropbox_source.py
@@ -0,0 +1,196 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/sources/dropbox_source.py
+++ b/video_processor/sources/dropbox_source.py
@@ -0,0 +1,196 @@
1 """Dropbox source integration with OAuth support."""
2
3 import json
4 import logging
5 import os
6 import webbrowser
7 from pathlib import Path
8 from typing import List, Optional
9
10 from video_processor.sources.base import BaseSource, SourceFile
11
12 logger = logging.getLogger(__name__)
13
14 # Video extensions we look for
15 VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv"}
16
17 _TOKEN_PATH = Path.home() / ".planopticon" / "dropbox_token.json"
18
19
20 class DropboxSource(BaseSource):
21 """
22 Dropbox source with OAuth2 support.
23
24 Auth methods:
25 - Access token: Set DROPBOX_ACCESS_TOKEN env var for simple usage
26 - OAuth2: Interactive browser-based flow with refresh tokens
27 """
28
29 def __init__(
30 self,
31 access_token: Optional[str] = None,
32 app_key: Optional[str] = None,
33 app_secret: Optional[str] = None,
34 token_path: Optional[Path] = None,
35 ):
36 """
37 Initialize Dropbox source.
38
39 Parameters
40 ----------
41 access_token : str, optional
42 Direct access token. Falls back to DROPBOX_ACCESS_TOKEN env var.
43 app_key : str, optional
44 Dropbox app key for OAuth. Falls back to DROPBOX_APP_KEY env var.
45 app_secret : str, optional
46 Dropbox app secret for OAuth. Falls back to DROPBOX_APP_SECRET env var.
47 token_path : Path, optional
48 Where to store/load OAuth tokens.
49 """
50 self.access_token = access_token or os.environ.get("DROPBOX_ACCESS_TOKEN")
51 self.app_key = app_key or os.environ.get("DROPBOX_APP_KEY")
52 self.app_secret = app_secret or os.environ.get("DROPBOX_APP_SECRET")
53 self.token_path = token_path or _TOKEN_PATH
54 self.dbx = None
55
56 def authenticate(self) -> bool:
57 """Authenticate with Dropbox API."""
58 try:
59 import dropbox
60 except ImportErrorpath}")
61 :
62 logger.error("Dropbox SDK not installed. Run: pip in
63 ""Dropbox so with OAuth support."pbox]")
64 return False
65
66 # Try direct access token first
67 if self.access_token:
68 return self._auth_token(dropbox)
69
70 # Try saved OAuth token
71 if self.token_path.exists():
72 if self._auth_saved_token(dropbox):
73 return True
74
75 # Run OAuth flow
76 return self._auth_oauth(dropbox)
77
78 def _auth_token(self, dropbox) -> bool:
79 """Authenticate with a direct access token."""
80 try:
81 self.dbx = dropbox.Dropbox(self.access_token)
82 self.dbx.users_get_current_account()
83 logger.info("Authenticated with Dropbox via access token")
84 return True
85 except Exception as e:
86 logger.error(f"Dropbox access token auth failed: {e}")
87 return False
88
89 def _auth_saved_token(self, dropbox) -> bool:
90 """Authenticate using a saved OAuth refresh token."""
91 try:
92 data = json.loads(self.token_path.read_text())
93 refresh_token = data.get("refresh_token")
94 app_key = data.get("app_key") or self.app_key
95 app_secret = data.get("app_secret") or self.app_secret
96
97 if not refresh_token or not app_key:
98 return False
99
100 self.dbx = dropbox.Dropbox(
101 oauth2_refresh_token=refresh_token,
102 app_key=app_key,
103 app_secret=app_secret,
104 )
105 self.dbx.users_get_current_account()
106 logger.info("Authenticated with Dropbox via saved token")
107 return True
108 except Exception:
109 return False
110
111 def _auth_oauth(self, dropbox) -> bool:
112 """Run OAuth2 PKCE flow."""
113 path}")
114 " str, optional
115 # Try direct acc
116 ""Dropbox so with OAuth support." return self._auth_token(dropbox)
117
118 # Try saved OAuth token
119 if self.token_path.exists():
120 if self._auth_saved_token(dropbox):
121 return True
122
123 # Run OAuth flow
124 return self._auth_oauth(dropbox)
125
126 def _auth_token(self, dropbox) -> bool:
127 """Authenticate with a direct access token."""
128 try:
129 self.dbx = dropbox.Dropbox(self.access_token)
130 self.dbx.users_get_current_account()
131 logger.info("Authenticated with Dropbox via access token")
132 return True
133 except Exception as e:
134 logger.error(f"Dropbox access token auth failed: {e}")
135 return False
136
137 def _auth_saved_token(self, dropbox) -> bool:
138 """Authenticate upath}")
139 logger.inentry.name.endswith(p.replace(path}")
140 logge):path}")
141 logger.incontinue
142
143 ""Dropbox source integration with OAuth support."""
144
145 import json
146 import logging
147 import os
148 import webbrowser
149 from pathlib import Path
150 from typing import List, Optional
151
152 from video_processor.sources.base import BaseSource, SourceFile
153
154 logger = logging.getLogger(__name__)
155
156 # Video extensions we look for
157 VIDEO_EXTENSIONS = {".mp4", ".mkv", ".avi", ".mov", ".webm", ".wmv"}
158
159 _TOKEN_PATH = Path.home() / ".planopticon" / "dropbox_token.json"
160
161
162 class DropboxSource(BaseSource):
163 """
164 Dropbox source with OAuth2 support.
165
166 Auth methods:
167 - Access token: Set DROPBOX_ACCESS_TOKEN env var for simple usage
168 - OAuth2: Interactive browser-based flow with refresh tokens
169 """
170
171 def __init__(
172 self,
173 access_token: Optional[str] = None,
174 app_key: Optional[str] = None,
175 app_secret: Optional[str] = None,
176 token_path: Optional[Path] = None,
177 ):
178 """
179 Initialize Dropbox source.
180
181 Parameters
182 ----------
183 access_token : str, optional
184 Direct access token. Falls back to DROPBOX_ACCESS_TOKEN env var.
185 app_key : str, optional
186 Dropbox app key for OAuth. Falls back to DROPBOX_APP_KEY env var.
187 app_secret : str, optional
188 Dropbox app secret for OAuth. Falls back to DROPBOX_APP_SECRET env var.
189 token_path : Path, optional
190 ropbox source integration with OAuth support."""
191
192 import json
193 import logging
194 import os
195 import webbrowser
196 fro
A video_processor/sources/google_drive.py
+169
--- a/video_processor/sources/google_drive.py
+++ b/video_processor/sources/google_drive.py
@@ -0,0 +1,169 @@
1
+"""Google Drive source integration with service account and OAuth support."""
2
+
3
+import json
4
+import logging
5
+import os
6
+from pathlib import Path
7
+from typing import List, Optional
8
+
9
+from video_processor.sources.base import BaseSource, SourceFile
10
+
11
+logger = logging.getLogger(__name__)
12
+
13
+# Video MIME types we support
14
+VIDEO_MIME_TYPES = {
15
+ "video/mp4",
16
+ "video/x-matroska",
17
+ "video/avi",
18
+ "video/quicktime",
19
+ "video/webm",
20
+ "video/x-msvideo",
21
+ "video/x-ms-wmv",
22
+}
23
+
24
+# Default OAuth scopes
25
+SCOPES = ["https://www.googleapis.com/auth/drive.readonly"]
26
+
27
+# OAuth client config for installed app flow
28
+_DEFAULT_CLIENT_CONFIG = {
29
+ "installed": {
30
+ "client_id": os.environ.get("GOOGLE_OAUTH_CLIENT_ID", ""),
31
+ "client_secret": os.environ.get("GOOGLE_OAUTH_CLIENT_SECRET", ""),
32
+ "auth_uri": "https://accounts.google.com/o/oauth2/auth",
33
+ "token_uri": "https://oauth2.googleapis.com/token",
34
+ "redirect_uris": ["http://localhost"],
35
+ }
36
+}
37
+
38
+_TOKEN_PATH = Path.home() / ".planopticon" / "google_drive_token.json"
39
+
40
+
41
+class GoogleDriveSource(BaseSource):
42
+ """
43
+ Google Drive source with dual auth support.
44
+
45
+ Auth methods:
46
+ - Service account: Set GOOGLE_APPLICATION_CREDENTIALS env var
47
+ - OAuth2: Interactive browser-based flow for user accounts
48
+ """
49
+
50
+ def __init__(
51
+ self,
52
+ credentials_path: Optional[str] = None,
53
+ use_service_account: Optional[bool] = None,
54
+ token_path: Optional[Path] = None,
55
+ ):
56
+ """
57
+ Initialize Google Drive source.
58
+
59
+ Parameters
60
+ ----------
61
+ credentials_path : str, optional
62
+ Path to service account JSON or OAuth client secrets.
63
+ Falls back to GOOGLE_APPLICATION_CREDENTIALS env var.
64
+ use_service_account : bool, optional
65
+ If True, force service account auth. If False, force OAuth.
66
+ If None, auto-detect from credentials file.
67
+ token_path : Path, optional
68
+ Where to store/load OAuth tokens. Defaults to ~/.planopticon/google_drive_token.json
69
+ """
70
+ self.credentials_path = credentials_path or os.environ.get(
71
+ if no"
72
+ Falls back to GOO"
73
+h
74
+from typing impo"""Google Drive source integration with service account and OAuth support."""
75
+
76
+import json
77
+import logging
78
+import os
79
+from pathlib import Path
80
+from typing import List, Optional
81
+
82
+from video_processor.sources.base import BaseSource, SourceFile
83
+
84
+logger = logging.getLogger(__name__)
85
+
86
+# Video MIME types we support
87
+VIDEO_MIME_
88
+ id=f["id"],
89
+ ET", ""),
90
+ "auth_uri": "https://accounts.google.com/o/oauth2/auth",
91
+ e:
92
+ re "token_uri": "https://oauth2.googleapis.com/token",
93
+ "red
94
+ prefix else name,
95
+ return False
96
+
97
+ home() / ".planopticon" / "google_drive_token.json"
98
+
99
+
100
+class GoogleDriveSource(BaseSource):
101
+ """
102
+ Google Drive source with dual auth support.
103
+
104
+ Auth methods:
105
+ - Service account: Set GOOGLE_APPLICATION_CREDENTIALS env var
106
+ - OAuth2: Interactive browser-based flow for user accounts
107
+ """
108
+
109
+ def __init__(
110
+ self,
111
+ credentials_path: Optional[str] = None,
112
+ use_service_account: Optional[bool] = None,
113
+ token_path: Optional[Path] = None,
114
+ ):
115
+ """
116
+ Initialize Google Drive source.
117
+
118
+ Parameters
119
+ ----------
120
+ credentials_path : str, optional
121
+ Path to service account JSON or OAuth client secrets.
122
+ Falls back to GOOGLE_APPLICATION_CREDENTIALS env var.
123
+ use_service_account : bool, optional
124
+ If True, force service account auth. If False, force OAuth.
125
+ If None, auto-detect from credentials file.
126
+ token_path : Path, optional
127
+ Where to store/load OAuth tokens. Defaults to ~/.planopticon/google_drive_token.json
128
+ """
129
+ self.credentials_path = credentials_path or os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
130
+ self.use_service_account = use_service_account
131
+ self.token_path = token_path or _TOKEN_PATH
132
+ self.service = None
133
+ self._creds = None
134
+
135
+ def authenticate(self) -> bool:
136
+ """Authenticate with Google Drive API."""
137
+ try:
138
+ from google.oauth2 import service_account as sa_module # noqa: F401
139
+ from googleapiclient.discovery import build
140
+ except ImportError:
141
+ logger.error("Google API client not installed. Run: pip install planopticon[gdrive]")
142
+ return False
143
+
144
+ # Determine auth method
145
+ if self.use_service_account is True or (
146
+ e:
147
+ re:
148
+ logger.error("OAuth libraries not installed. Run: pip i
149
+ prefix else name,
150
+ return False
151
+
152
+ creds = None
153
+
154
+ # Load existing token
155
+ if self.token_path.exists():
156
+ try:
157
+ c= Credentialse:
158
+ re re)
159
+ ptional[bool] = None,
160
+ # Build query# Filter for video MIME types
161
+ files = []
162
+ self.c# Apply patif patternst json
163
+import logging
164
+"""Google Driv iftyping import List, Optional
165
+
166
+"""Google Drive source integration with service account and OAuth supporfiles.append(SourceFile(name=f["name"]) e:
167
+ re name.endswith(p.replace(e:
168
+ ree_token,
169
+
--- a/video_processor/sources/google_drive.py
+++ b/video_processor/sources/google_drive.py
@@ -0,0 +1,169 @@
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
--- a/video_processor/sources/google_drive.py
+++ b/video_processor/sources/google_drive.py
@@ -0,0 +1,169 @@
1 """Google Drive source integration with service account and OAuth support."""
2
3 import json
4 import logging
5 import os
6 from pathlib import Path
7 from typing import List, Optional
8
9 from video_processor.sources.base import BaseSource, SourceFile
10
11 logger = logging.getLogger(__name__)
12
13 # Video MIME types we support
14 VIDEO_MIME_TYPES = {
15 "video/mp4",
16 "video/x-matroska",
17 "video/avi",
18 "video/quicktime",
19 "video/webm",
20 "video/x-msvideo",
21 "video/x-ms-wmv",
22 }
23
24 # Default OAuth scopes
25 SCOPES = ["https://www.googleapis.com/auth/drive.readonly"]
26
27 # OAuth client config for installed app flow
28 _DEFAULT_CLIENT_CONFIG = {
29 "installed": {
30 "client_id": os.environ.get("GOOGLE_OAUTH_CLIENT_ID", ""),
31 "client_secret": os.environ.get("GOOGLE_OAUTH_CLIENT_SECRET", ""),
32 "auth_uri": "https://accounts.google.com/o/oauth2/auth",
33 "token_uri": "https://oauth2.googleapis.com/token",
34 "redirect_uris": ["http://localhost"],
35 }
36 }
37
38 _TOKEN_PATH = Path.home() / ".planopticon" / "google_drive_token.json"
39
40
41 class GoogleDriveSource(BaseSource):
42 """
43 Google Drive source with dual auth support.
44
45 Auth methods:
46 - Service account: Set GOOGLE_APPLICATION_CREDENTIALS env var
47 - OAuth2: Interactive browser-based flow for user accounts
48 """
49
50 def __init__(
51 self,
52 credentials_path: Optional[str] = None,
53 use_service_account: Optional[bool] = None,
54 token_path: Optional[Path] = None,
55 ):
56 """
57 Initialize Google Drive source.
58
59 Parameters
60 ----------
61 credentials_path : str, optional
62 Path to service account JSON or OAuth client secrets.
63 Falls back to GOOGLE_APPLICATION_CREDENTIALS env var.
64 use_service_account : bool, optional
65 If True, force service account auth. If False, force OAuth.
66 If None, auto-detect from credentials file.
67 token_path : Path, optional
68 Where to store/load OAuth tokens. Defaults to ~/.planopticon/google_drive_token.json
69 """
70 self.credentials_path = credentials_path or os.environ.get(
71 if no"
72 Falls back to GOO"
73 h
74 from typing impo"""Google Drive source integration with service account and OAuth support."""
75
76 import json
77 import logging
78 import os
79 from pathlib import Path
80 from typing import List, Optional
81
82 from video_processor.sources.base import BaseSource, SourceFile
83
84 logger = logging.getLogger(__name__)
85
86 # Video MIME types we support
87 VIDEO_MIME_
88 id=f["id"],
89 ET", ""),
90 "auth_uri": "https://accounts.google.com/o/oauth2/auth",
91 e:
92 re "token_uri": "https://oauth2.googleapis.com/token",
93 "red
94 prefix else name,
95 return False
96
97 home() / ".planopticon" / "google_drive_token.json"
98
99
100 class GoogleDriveSource(BaseSource):
101 """
102 Google Drive source with dual auth support.
103
104 Auth methods:
105 - Service account: Set GOOGLE_APPLICATION_CREDENTIALS env var
106 - OAuth2: Interactive browser-based flow for user accounts
107 """
108
109 def __init__(
110 self,
111 credentials_path: Optional[str] = None,
112 use_service_account: Optional[bool] = None,
113 token_path: Optional[Path] = None,
114 ):
115 """
116 Initialize Google Drive source.
117
118 Parameters
119 ----------
120 credentials_path : str, optional
121 Path to service account JSON or OAuth client secrets.
122 Falls back to GOOGLE_APPLICATION_CREDENTIALS env var.
123 use_service_account : bool, optional
124 If True, force service account auth. If False, force OAuth.
125 If None, auto-detect from credentials file.
126 token_path : Path, optional
127 Where to store/load OAuth tokens. Defaults to ~/.planopticon/google_drive_token.json
128 """
129 self.credentials_path = credentials_path or os.environ.get("GOOGLE_APPLICATION_CREDENTIALS")
130 self.use_service_account = use_service_account
131 self.token_path = token_path or _TOKEN_PATH
132 self.service = None
133 self._creds = None
134
135 def authenticate(self) -> bool:
136 """Authenticate with Google Drive API."""
137 try:
138 from google.oauth2 import service_account as sa_module # noqa: F401
139 from googleapiclient.discovery import build
140 except ImportError:
141 logger.error("Google API client not installed. Run: pip install planopticon[gdrive]")
142 return False
143
144 # Determine auth method
145 if self.use_service_account is True or (
146 e:
147 re:
148 logger.error("OAuth libraries not installed. Run: pip i
149 prefix else name,
150 return False
151
152 creds = None
153
154 # Load existing token
155 if self.token_path.exists():
156 try:
157 c= Credentialse:
158 re re)
159 ptional[bool] = None,
160 # Build query# Filter for video MIME types
161 files = []
162 self.c# Apply patif patternst json
163 import logging
164 """Google Driv iftyping import List, Optional
165
166 """Google Drive source integration with service account and OAuth supporfiles.append(SourceFile(name=f["name"]) e:
167 re name.endswith(p.replace(e:
168 ree_token,
169

Keyboard Shortcuts

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