-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathcli.py
More file actions
2225 lines (1845 loc) · 94.4 KB
/
cli.py
File metadata and controls
2225 lines (1845 loc) · 94.4 KB
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
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
642
643
644
645
646
647
648
649
650
651
652
653
654
655
656
657
658
659
660
661
662
663
664
665
666
667
668
669
670
671
672
673
674
675
676
677
678
679
680
681
682
683
684
685
686
687
688
689
690
691
692
693
694
695
696
697
698
699
700
701
702
703
704
705
706
707
708
709
710
711
712
713
714
715
716
717
718
719
720
721
722
723
724
725
726
727
728
729
730
731
732
733
734
735
736
737
738
739
740
741
742
743
744
745
746
747
748
749
750
751
752
753
754
755
756
757
758
759
760
761
762
763
764
765
766
767
768
769
770
771
772
773
774
775
776
777
778
779
780
781
782
783
784
785
786
787
788
789
790
791
792
793
794
795
796
797
798
799
800
801
802
803
804
805
806
807
808
809
810
811
812
813
814
815
816
817
818
819
820
821
822
823
824
825
826
827
828
829
830
831
832
833
834
835
836
837
838
839
840
841
842
843
844
845
846
847
848
849
850
851
852
853
854
855
856
857
858
859
860
861
862
863
864
865
866
867
868
869
870
871
872
873
874
875
876
877
878
879
880
881
882
883
884
885
886
887
888
889
890
891
892
893
894
895
896
897
898
899
900
901
902
903
904
905
906
907
908
909
910
911
912
913
914
915
916
917
918
919
920
921
922
923
924
925
926
927
928
929
930
931
932
933
934
935
936
937
938
939
940
941
942
943
944
945
946
947
948
949
950
951
952
953
954
955
956
957
958
959
960
961
962
963
964
965
966
967
968
969
970
971
972
973
974
975
976
977
978
979
980
981
982
983
984
985
986
987
988
989
990
991
992
993
994
995
996
997
998
999
1000
#!/usr/bin/env python3
"""
Internxt CLI - Python implementation with Path Support and Delete Operations
Enhanced with path-based operations and comprehensive delete/trash functionality
"""
import click
import sys
import os
import json
import base64
import hashlib
from pathlib import Path
from datetime import datetime, timezone
from typing import Optional, Dict, Any, List, Tuple
import glob
import time
# Try to import required packages
try:
import requests
import mnemonic
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from cryptography.hazmat.primitives.kdf.pbkdf2 import PBKDF2HMAC
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.backends import default_backend
from tqdm import tqdm
except ImportError as e:
print(f"❌ Missing required dependency: {e}")
print("📦 Install with: pip install cryptography mnemonic tqdm requests click")
sys.exit(1)
# Add current directory to path for imports
current_dir = os.path.dirname(os.path.abspath(__file__))
if current_dir not in sys.path:
sys.path.insert(0, current_dir)
# Import our services
try:
from config.config import config_service
from services.crypto import crypto_service
from services.auth import auth_service
from utils.api import api_client
from services.drive import drive_service
from services.webdav_server import webdav_server
except ImportError as e:
print(f"❌ Failed to import services: {e}")
print("📦 Make sure all service files are in place with fixed implementations")
# Check for WebDAV specific dependencies (NEW LOGIC)
try:
import wsgidav
except ImportError:
print("📦 Missing core WebDAV dependency. Install with:")
print(" pip install WsgiDAV")
sys.exit(1)
try:
# Check for at least one server, matching webdav_server.py
try:
import waitress
except ImportError:
import cheroot
except ImportError:
print("📦 No suitable WSGI server found. Install one of:")
print(" pip install waitress")
print(" pip install cheroot")
sys.exit(1)
sys.exit(1) # Exit from the *original* service import error
def format_size(size_bytes: int) -> str:
"""Format bytes to human readable size"""
if not size_bytes:
return "0 B"
for unit in ['B', 'KB', 'MB', 'GB', 'TB']:
if size_bytes < 1024.0:
return f"{size_bytes:.1f} {unit}"
size_bytes /= 1024.0
return f"{size_bytes:.1f} PB"
def format_date(date_string: str) -> str:
"""Format ISO date string to readable format"""
try:
dt = datetime.fromisoformat(date_string.replace('Z', '+00:00'))
return dt.strftime('%d %B, %Y at %H:%M')
except Exception:
return date_string
@click.group()
@click.version_option(version='1.0.0')
def cli():
"""Internxt Python CLI with Path Support and Delete Operations"""
pass
# ========== AUTHENTICATION COMMANDS ==========
@cli.command()
@click.option('--email', '-e', help='Your Internxt email')
@click.option('--password', '-p', help='Your password')
@click.option('--tfa', '--2fa', help='Two-factor authentication code (6 digits)')
@click.option('--non-interactive', is_flag=True, help='Run in non-interactive mode')
@click.option('--debug', is_flag=True, help='Enable debug output')
def login(email: Optional[str], password: Optional[str], tfa: Optional[str], non_interactive: bool, debug: bool):
"""Login to your Internxt account"""
try:
if debug:
print("🔍 Debug mode enabled")
print(f"🔍 API Endpoints:")
print(f" Drive API: {config_service.get('DRIVE_NEW_API_URL')}")
print(f" Network API: {config_service.get('NETWORK_URL')}")
# Get email
if not email:
if non_interactive:
click.echo("❌ Email is required in non-interactive mode", err=True)
sys.exit(1)
email = click.prompt('What is your email?', type=str)
# Validate email
if '@' not in email or '.' not in email:
click.echo("❌ Invalid email format", err=True)
sys.exit(1)
# Get password
if not password:
if non_interactive:
click.echo("❌ Password is required in non-interactive mode", err=True)
sys.exit(1)
password = click.prompt('What is your password?', hide_input=True, type=str)
if not password.strip():
click.echo("❌ Password cannot be empty", err=True)
sys.exit(1)
# Check 2FA
click.echo("🔍 Checking 2FA requirements...")
try:
is_2fa_needed = auth_service.is_2fa_needed(email)
if debug:
print(f"🔍 2FA needed: {is_2fa_needed}")
except Exception as e:
click.echo(f"⚠️ Could not check 2FA status: {e}")
is_2fa_needed = False
if is_2fa_needed and not tfa:
if non_interactive:
click.echo("❌ 2FA code is required in non-interactive mode", err=True)
sys.exit(1)
tfa = click.prompt('What is your two-factor token?', type=str)
if tfa and (not tfa.isdigit() or len(tfa) != 6):
click.echo("❌ Invalid 2FA code format (must be 6 digits)", err=True)
sys.exit(1)
# Login
click.echo("🔐 Logging in...")
credentials = auth_service.login(email, password, tfa)
user_email = credentials['user']['email']
user_uuid = credentials['user']['uuid']
root_folder_id = credentials['user'].get('rootFolderId', '')
click.echo(f"✅ Successfully logged in as: {user_email}")
if debug:
print(f"🔍 User UUID: {user_uuid}")
print(f"🔍 Root Folder ID: {root_folder_id}")
except Exception as e:
error_msg = str(e)
if "Login failed:" in error_msg:
error_msg = error_msg.replace("Login failed: ", "")
click.echo(f"❌ Login failed: {error_msg}", err=True)
if debug:
import traceback
print("🔍 Full error traceback:")
traceback.print_exc()
sys.exit(1)
@cli.command()
def whoami():
"""Check current login status"""
try:
user_info = auth_service.whoami()
if user_info:
click.echo(f"📧 Logged in as: {user_info['email']}")
click.echo(f"🆔 User ID: {user_info['uuid']}")
click.echo(f"📁 Root Folder ID: {user_info['rootFolderId']}")
else:
click.echo("❌ Not logged in")
click.echo("💡 Use 'python cli.py login' to log in")
except Exception as e:
click.echo(f"❌ Error: {e}", err=True)
@cli.command()
def logout():
"""Logout and clear credentials"""
try:
auth_service.logout()
click.echo("✅ Successfully logged out")
except Exception as e:
click.echo(f"❌ Error during logout: {e}", err=True)
# ========== BASIC FILE OPERATIONS ==========
@cli.command()
@click.option('--folder-id', help='Folder ID to list (defaults to root)')
@click.option('--detailed', '-d', is_flag=True, help='Show detailed information')
def list(folder_id, detailed):
"""List files and folders (UUID-based - legacy)"""
try:
credentials = auth_service.get_auth_details()
if not folder_id:
folder_id = credentials['user'].get('rootFolderId', '')
if not folder_id:
click.echo("❌ No root folder ID found. Please try logging in again.", err=True)
return
click.echo(f"📂 Listing contents of folder: {folder_id}")
contents = drive_service.get_folder_content(folder_id)
folders = contents.get('folders', [])
files = contents.get('files', [])
if not folders and not files:
click.echo("📭 Folder is empty")
return
if folders:
click.echo(f"\n📁 Folders ({len(folders)}):")
for folder in folders:
name = folder.get('plainName', 'Unknown')
# Prioritize preserved timestamps
display_time_iso = folder.get('modificationTime') or \
folder.get('creationTime') or \
folder.get('updatedAt') or \
folder.get('createdAt', '')
if detailed and display_time_iso:
click.echo(f" 📁 {name} (created {format_date(display_time_iso)})")
else:
click.echo(f" 📁 {name}")
if files:
click.echo(f"\n📄 Files ({len(files)}):")
for file in files:
name = file.get('plainName', 'Unknown')
file_type = file.get('type', '')
if file_type:
name = f"{name}.{file_type}"
try:
size = int(file.get('size', 0))
except (ValueError, TypeError):
size = 0
# Prioritize preserved timestamps
display_time_iso = file.get('modificationTime') or \
file.get('creationTime') or \
file.get('updatedAt') or \
file.get('createdAt', '')
if detailed:
size_str = format_size(size)
if display_time_iso:
click.echo(f" 📄 {name} ({size_str}, {format_date(display_time_iso)})")
else:
click.echo(f" 📄 {name} ({size_str})")
else:
click.echo(f" 📄 {name} ({format_size(size)})")
except Exception as e:
click.echo(f"❌ Error listing folder: {e}", err=True)
sys.exit(1)
@cli.command()
@click.argument('path')
@click.option('--parent-folder-id', help='Parent folder ID (defaults to root). If used, path must be a single name.')
def mkdir(path: str, parent_folder_id: Optional[str]):
"""Create a new folder (supports paths like folder/subfolder)"""
try:
# Ensure we are logged in
credentials = auth_service.get_auth_details()
# CASE 1: User specified a specific parent UUID (Legacy/Strict mode)
if parent_folder_id:
if '/' in path or '\\' in path:
click.echo("❌ Error: When providing --parent-folder-id, the name cannot contain slashes.", err=True)
click.echo("💡 Tip: To create nested paths like 'A/B', do not use --parent-folder-id.", err=True)
return
click.echo(f"📁 Creating single folder '{path}' in parent {parent_folder_id}...")
folder = drive_service.create_folder(path, parent_folder_id)
# CASE 2: No parent ID specified (Smart Path mode)
else:
# Use the recursive function to handle "daten/phil" automatically
click.echo(f"📁 Creating folder path: {path}")
folder = drive_service.create_folder_recursive(path)
# Output success results
folder_uuid = folder.get('uuid', folder.get('id', ''))
folder_name = folder.get('plainName', folder.get('name', path))
click.echo(f"✅ Folder created successfully!")
click.echo(f"📁 Name: {folder_name}")
click.echo(f"🆔 UUID: {folder_uuid}")
except Exception as e:
error_msg = str(e)
click.echo(f"❌ Error creating folder: {error_msg}", err=True)
# cli.py
@cli.command('mv')
@click.argument('source_path')
@click.argument('target_path')
@click.option('--verbose', '-v', is_flag=True, help='Enable high verbosity tracing')
def move_path(source_path: str, target_path: str, verbose: bool):
"""
Move a file or folder to a new destination path.
Example: python cli.py mv /Documents/file.txt /Archive
"""
try:
# Ensure we have a fresh, hydrated session
auth_service.get_auth_details()
click.echo(f"🚚 Moving '{source_path}'...")
drive_service.move_by_path(source_path, target_path)
click.echo(f"✅ Successfully moved to {target_path}")
except Exception as e:
click.echo(f"❌ Move operation failed: {e}", err=True)
if verbose:
import traceback
traceback.print_exc()
sys.exit(1)
@cli.command()
@click.argument('sources', nargs=-1, type=str)
@click.option('--target', '-t', 'target_path', default='/', help='Destination path on Internxt Drive (default: /)')
@click.option('--recursive', '-r', is_flag=True, help='Upload directories recursively')
@click.option('--on-conflict', type=click.Choice(['overwrite', 'skip'], case_sensitive=False), default='skip', help='Action if target exists (overwrite/skip)')
@click.option('--preserve-timestamps', '-p', is_flag=True, help='Preserve file creation and modification times')
@click.option('--include', multiple=True, help='Include only files matching pattern (e.g., --include "*.png" --include "*.jpg")')
@click.option('--exclude', multiple=True, help='Exclude files matching pattern (e.g., --exclude "*.tmp" --exclude ".DS_Store")')
@click.option('--verbose', '-v', is_flag=True, help='Show verbose output')
def upload(sources: Tuple[str], target_path: str, recursive: bool, on_conflict: str,
preserve_timestamps: bool, include: Tuple[str], exclude: Tuple[str], verbose: bool):
"""
Encrypts and uploads local files/folders to a remote path.
SOURCES... can be one or more local file or directory paths. Wildcards (*) are supported.
TARGET_PATH is the destination path on your Internxt Drive (e.g., "/Documents/Backup").
Trailing slash behavior (like rsync):
source/ → uploads contents to target (no new folder)
source → creates 'source' folder in target
Use --preserve-timestamps to maintain original file dates (experimental).
Use --include/--exclude to filter files by pattern (supports wildcards).
Examples:
Upload only images:
python cli.py upload photos/ -t /Backup -r --include "*.png" --include "*.jpg"
Upload all except temp files:
python cli.py upload project/ -t /Code -r --exclude "*.tmp" --exclude ".DS_Store"
Preserve timestamps:
python cli.py upload docs/ -t /Documents -r --preserve-timestamps
"""
if not sources:
click.echo("❌ No source files or directories specified.", err=True)
sys.exit(1)
# Convert include/exclude tuples to lists
include_patterns = list(include) if include else []
exclude_patterns = list(exclude) if exclude else []
if verbose or include_patterns or exclude_patterns:
if include_patterns:
click.echo(f"🔍 Include filters: {', '.join(include_patterns)}")
if exclude_patterns:
click.echo(f"🚫 Exclude filters: {', '.join(exclude_patterns)}")
try:
click.echo("🔄 Refreshing authentication token...")
auth_service.refresh_tokens() # This will now work
credentials = auth_service.get_auth_details() # This gets the *new* token
click.echo(f"🎯 Preparing upload to remote path: {target_path}")
# --- Resolve or Create Target Folder ---
target_folder_uuid = None
target_folder_path_str = "/"
try:
target_folder_info = drive_service.resolve_path(target_path)
if target_folder_info['type'] != 'folder':
click.echo(f"❌ Target path '{target_path}' exists but is not a folder.", err=True)
sys.exit(1)
target_folder_uuid = target_folder_info['uuid']
target_folder_path_str = target_folder_info['path']
click.echo(f"✅ Target folder exists: '{target_folder_path_str}' (UUID: {target_folder_uuid[:8]}...)")
except FileNotFoundError:
click.echo(f"⏳ Target path '{target_path}' not found. Attempting to create...")
try:
created_folder = drive_service.create_folder_recursive(target_path)
target_folder_uuid = created_folder['uuid']
target_folder_info = drive_service.resolve_path(target_path)
target_folder_path_str = target_folder_info['path']
click.echo(f"✅ Created target folder '{target_folder_path_str}' (UUID: {target_folder_uuid[:8]}...)")
except Exception as create_err:
click.echo(f"❌ Failed to create target folder '{target_path}': {create_err}", err=True)
sys.exit(1)
except Exception as resolve_err:
click.echo(f"❌ Error resolving target path '{target_path}': {resolve_err}", err=True)
sys.exit(1)
if not target_folder_uuid:
click.echo("❌ Could not determine target folder UUID. Aborting.", err=True)
sys.exit(1)
# --- Process Sources ---
items_to_process = []
click.echo("🔍 Expanding source paths...")
for source_arg in sources:
has_trailing_slash = source_arg.rstrip().endswith('/') or source_arg.rstrip().endswith(os.sep)
source_path = Path(source_arg)
if not source_path.exists():
click.echo(f"⚠️ Source not found: {source_arg}", err=True)
continue
source_path_resolved = source_path.resolve()
source_path_str = str(source_path_resolved)
matches = glob.glob(source_path_str, recursive=recursive)
if not matches:
click.echo(f"⚠️ Source not found or matched nothing: {source_arg}", err=True)
continue
base_dir_str = source_path_str
if "*" in source_arg or "?" in source_arg or "[" in source_arg:
path_parts = Path(source_arg).parts
non_wildcard_parts = []
for part in path_parts:
if "*" in part or "?" in part or "[" in part:
break
non_wildcard_parts.append(part)
if non_wildcard_parts:
base_dir_str = str(Path(*non_wildcard_parts))
if Path(source_arg).is_dir() and not ("*" in source_arg or "?" in source_arg or "[" in source_arg):
base_dir_str = str(Path(source_arg))
else:
base_dir_str = os.getcwd()
if Path(source_arg).is_absolute():
base_dir_path = Path(base_dir_str)
if not base_dir_path.is_absolute():
base_dir_path = Path.cwd() / base_dir_path
base_dir_str = str(base_dir_path.resolve())
base_source_dir = Path(base_dir_str)
if base_source_dir.is_file():
base_source_dir = base_source_dir.parent
for match_str in matches:
match_path = Path(match_str).resolve()
current_base = base_source_dir if not match_path.is_dir() else match_path.parent
copy_contents_only = has_trailing_slash if match_path.is_dir() else False
items_to_process.append((match_path, current_base, copy_contents_only))
if not items_to_process:
click.echo("❌ No valid source files or directories found after expansion.", err=True)
sys.exit(1)
click.echo(f"📦 Found {len(items_to_process)} items/directories to process.")
# --- Upload Loop ---
success_count = 0
skipped_count = 0
error_count = 0
filtered_count = 0
processed_dirs = set()
for local_path, base_source_dir, copy_contents_only in items_to_process:
try:
if verbose:
click.echo("-" * 40)
click.echo(f"Processing: {local_path}")
if local_path.is_file():
# Apply include/exclude filters
if not drive_service.should_include_file(local_path, include_patterns, exclude_patterns):
if verbose:
click.echo(f" -> 🚫 Filtered out: {local_path.name}")
filtered_count += 1
continue
# Get timestamps if preservation requested
creation_time = None
modification_time = None
if preserve_timestamps:
try:
stat_info = local_path.stat()
mtime = datetime.fromtimestamp(stat_info.st_mtime, tz=timezone.utc)
modification_time = mtime.isoformat()
try:
ctime = datetime.fromtimestamp(stat_info.st_birthtime, tz=timezone.utc)
creation_time = ctime.isoformat()
except AttributeError:
ctime = datetime.fromtimestamp(stat_info.st_ctime, tz=timezone.utc)
creation_time = ctime.isoformat()
if verbose:
click.echo(f" 🕐 Local file timestamps:")
click.echo(f" Creation: {creation_time}")
click.echo(f" Modification: {modification_time}")
except Exception as e:
if verbose:
click.echo(f" ⚠️ Could not read timestamps: {e}")
# Upload
upload_result = drive_service.upload_single_item_with_conflict_handling(
local_path,
target_folder_path_str,
target_folder_uuid,
on_conflict,
remote_filename=local_path.name,
creation_time=creation_time,
modification_time=modification_time
)
if upload_result == "uploaded": success_count += 1
elif upload_result == "skipped": skipped_count += 1
else: error_count += 1
elif local_path.is_dir():
if local_path in processed_dirs:
if verbose:
click.echo(f" -> Skipping already processed directory: {local_path}")
continue
click.echo(f"📂 Processing directory recursively: {local_path}")
processed_dirs.add(local_path)
# Create the root upload dir first, with timestamps
if copy_contents_only:
click.echo(f" ✨ Copying contents directly to target (trailing slash detected)")
dir_remote_base_path = Path(target_folder_path_str)
else:
click.echo(f" 📁 Ensuring folder '{local_path.name}' exists in target...")
dir_remote_base_path = Path(target_folder_path_str) / local_path.name
# Get timestamps for THIS directory
creation_time, modification_time = None, None
if preserve_timestamps:
try:
stat_info = local_path.stat()
mtime = datetime.fromtimestamp(stat_info.st_mtime, tz=timezone.utc)
modification_time = mtime.isoformat()
try:
ctime = datetime.fromtimestamp(stat_info.st_birthtime, tz=timezone.utc)
creation_time = ctime.isoformat()
except AttributeError:
ctime = datetime.fromtimestamp(stat_info.st_ctime, tz=timezone.utc)
creation_time = ctime.isoformat()
if verbose:
click.echo(f" 🕐 Applying root dir timestamps: Mod={modification_time}")
except Exception as e:
if verbose:
click.echo(f" ⚠️ Could not read root dir timestamps: {e}")
# Create the root folder WITH timestamps
try:
# We must use str() for the path and normalize separators
drive_service.create_folder_recursive(
str(dir_remote_base_path).replace(os.sep, '/'),
creation_time=creation_time,
modification_time=modification_time
)
except Exception as create_err:
click.echo(f" ❌ Error creating root folder {local_path.name}: {create_err}", err=True)
error_count += 1
continue # Skip this whole directory
click.echo(f" -> Pass 1/2: Creating subdirectory structure...")
# --- FIRST PASS: CREATE SUB-DIRECTORIES ---
dir_list = []
for item in local_path.rglob('*'):
if item.is_dir():
dir_list.append(item)
dir_list.sort(key=lambda x: len(x.parts))
for item in dir_list:
if not drive_service.should_include_file(item, include_patterns, exclude_patterns):
if verbose: click.echo(f" -> 🚫 Filtered dir: {item.name}")
filtered_count += 1
continue
relative_path = item.relative_to(local_path)
item_target_path = dir_remote_base_path / relative_path
item_target_path_str = str(item_target_path).replace(os.sep, '/')
if not item_target_path_str.startswith('/'):
item_target_path_str = '/' + item_target_path_str
# Get directory timestamps
creation_time, modification_time = None, None
if preserve_timestamps:
try:
stat_info = item.stat()
mtime = datetime.fromtimestamp(stat_info.st_mtime, tz=timezone.utc)
modification_time = mtime.isoformat()
try:
ctime = datetime.fromtimestamp(stat_info.st_birthtime, tz=timezone.utc)
creation_time = ctime.isoformat()
except AttributeError:
ctime = datetime.fromtimestamp(stat_info.st_ctime, tz=timezone.utc)
creation_time = ctime.isoformat()
except Exception:
pass # Failed to get timestamps
try:
if verbose:
click.echo(f" -> 📁 Ensuring dir: {item_target_path_str}")
drive_service.create_folder_recursive(
item_target_path_str,
creation_time=creation_time,
modification_time=modification_time
)
except Exception as create_err:
click.echo(f" ❌ Error creating dir {item_target_path_str}: {create_err}", err=True)
error_count += 1
click.echo(f" -> Pass 2/2: Uploading files...")
# --- SECOND PASS: UPLOAD FILES ---
for item in local_path.rglob('*'):
if item.is_file():
# Apply include/exclude filters
if not drive_service.should_include_file(item, include_patterns, exclude_patterns):
if verbose:
click.echo(f" -> 🚫 Filtered: {item.name}")
filtered_count += 1
continue
relative_path = item.relative_to(local_path)
item_target_parent_path = dir_remote_base_path / relative_path.parent
item_target_parent_path_str = str(item_target_parent_path).replace(os.sep, '/')
if not item_target_parent_path_str.startswith('/'):
item_target_parent_path_str = '/' + item_target_parent_path_str
if verbose:
click.echo(f" -> Found file: {item.name} (relative: {relative_path})")
click.echo(f" Target parent path: {item_target_parent_path_str}")
parent_folder_uuid = None
try:
# This call will now be very fast as dirs exist
# And we call it *without* timestamps, as the dir is already made
parent_folder = drive_service.create_folder_recursive(item_target_parent_path_str)
parent_folder_uuid = parent_folder['uuid']
if verbose:
click.echo(f" Ensured parent folder exists (UUID: {parent_folder_uuid[:8]}...)")
except Exception as create_err:
click.echo(f" ❌ Error ensuring parent folder {item_target_parent_path_str}: {create_err}", err=True)
error_count += 1
continue
# Get timestamps if requested
creation_time = None
modification_time = None
if preserve_timestamps:
try:
stat_info = item.stat()
mtime = datetime.fromtimestamp(stat_info.st_mtime, tz=timezone.utc)
modification_time = mtime.isoformat()
try:
ctime = datetime.fromtimestamp(stat_info.st_birthtime, tz=timezone.utc)
creation_time = ctime.isoformat()
except AttributeError:
ctime = datetime.fromtimestamp(stat_info.st_ctime, tz=timezone.utc)
creation_time = ctime.isoformat()
except Exception as e:
# This will tell us why timestamp reading is failing
click.echo(f" -> ⚠️ Could not read timestamps for {item.name}: {e}", err=True)
pass
upload_result = drive_service.upload_single_item_with_conflict_handling(
item,
item_target_parent_path_str,
parent_folder_uuid,
on_conflict,
remote_filename=item.name,
creation_time=creation_time,
modification_time=modification_time
)
if upload_result == "uploaded": success_count += 1
elif upload_result == "skipped": skipped_count += 1
else: error_count += 1
else:
click.echo(f"⚠️ Skipping unknown item type: {local_path}", err=True)
skipped_count += 1
except Exception as e:
click.echo(f"❌ Error processing {local_path}: {e}", err=True)
error_count += 1
continue
# --- Summary ---
click.echo("=" * 40)
click.echo("📊 Upload Summary:")
click.echo(f" ✅ Uploaded: {success_count}")
click.echo(f" ⏭️ Skipped: {skipped_count}")
if filtered_count > 0:
click.echo(f" 🚫 Filtered: {filtered_count}")
click.echo(f" ❌ Errors: {error_count}")
click.echo("=" * 40)
except Exception as e:
click.echo(f"❌ Upload failed: {e}", err=True)
import traceback
traceback.print_exc()
sys.exit(1)
@cli.command()
@click.argument('file_uuid')
@click.option('--destination', '-d', type=click.Path(file_okay=True, writable=True, resolve_path=True), default='.', help='Where to save the file')
@click.option('--preserve-timestamps', '-p', is_flag=True, help='Preserve file modification times')
@click.option('--on-conflict', type=click.Choice(['overwrite', 'skip'], case_sensitive=False), default='overwrite', help='Action if local file exists')
@click.option('--verbose', '-v', is_flag=True, help='Show verbose output')
def download(file_uuid: str, destination: str, preserve_timestamps: bool, on_conflict: str, verbose: bool):
"""Downloads and decrypts a file from your Internxt Drive (by UUID)"""
try:
from pathlib import Path
if verbose:
click.echo(f"📥 Downloading file with UUID: {file_uuid}")
click.echo(f"📁 Destination: {destination}")
if preserve_timestamps:
click.echo(f"🕐 Timestamp preservation: enabled")
# Check if destination file already exists
dest_path = Path(destination)
if dest_path.is_file() and on_conflict == 'skip':
click.echo(f"⏭️ File exists, skipping: {dest_path}")
sys.exit(0)
# Download the file
downloaded_path = drive_service.download_file(
file_uuid,
destination,
preserve_timestamps=preserve_timestamps
)
if not verbose:
click.echo(f"✅ File downloaded successfully to: {downloaded_path}")
except Exception as e:
click.echo(f"❌ Error downloading file: {e}", err=True)
if verbose:
import traceback
traceback.print_exc()
sys.exit(1)
# ========== PATH-BASED OPERATIONS ==========
@cli.command('list-path')
@click.argument('path', default='/')
@click.option('--detailed', '-d', is_flag=True, help='Show detailed information')
@click.option('--all', '-a', is_flag=True, help='Show all attributes (verbose)')
def list_path(path: str, detailed: bool, all: bool):
"""List folder contents with paths (much more user-friendly!)"""
try:
auth_service.get_auth_details()
content = drive_service.list_folder_with_paths(path)
click.echo(f"\n📁 Listing folder: {path}")
click.echo()
click.echo(f"📁 Contents of: {content['current_path']}")
click.echo("=" * 80)
# Show folders first
if content['folders']:
click.echo("📂 Folders:")
click.echo("-" * 60)
for folder in content['folders']:
if all:
# Show ALL attributes
click.echo(f" 📁 {folder['display_name']}")
click.echo(f" UUID: {folder['uuid']}")
click.echo(f" Path: {folder['path']}")
click.echo(f" Plain Name: {folder.get('plainName', 'N/A')}")
click.echo(f" Parent ID: {folder.get('parentId', 'N/A')}")
click.echo(f" User ID: {folder.get('userId', 'N/A')}")
# Timestamps for FOLDERS
created_at = folder.get('createdAt', 'N/A')
updated_at = folder.get('updatedAt', 'N/A')
creation_time = folder.get('creationTime', 'N/A')
modification_time = folder.get('modificationTime', 'N/A')
# Use the correct logic for display
display_creation = creation_time if creation_time != 'N/A' else created_at
display_modification = modification_time if modification_time != 'N/A' else updated_at
if display_creation != 'N/A':
click.echo(f" Creation Time: {format_date(display_creation)} ({display_creation})")
else:
click.echo(f" Creation Time: N/A")
if display_modification != 'N/A':
click.echo(f" Modification Time: {format_date(display_modification)} ({display_modification})")
else:
click.echo(f" Modification Time: N/A")
# Other attributes
click.echo(f" Deleted: {folder.get('deleted', False)}")
# Other attributes
click.echo(f" Deleted: {folder.get('deleted', False)}")
click.echo(f" Removed: {folder.get('removed', False)}")
click.echo()
elif detailed:
modified = folder.get('modified', '')[:10] if folder.get('modified') else ''
click.echo(f" 📁 {folder['display_name']:<30} {modified:<12} {folder['uuid'][:8]}...")
else:
click.echo(f" 📁 {folder['display_name']}")
# Then show files
if content['files']:
if content['folders']:
click.echo()
click.echo("📄 Files:")
click.echo("-" * 60)
for file in content['files']:
if all:
# Show ALL attributes
click.echo(f" 📄 {file['display_name']}")
click.echo(f" UUID: {file['uuid']}")
click.echo(f" Path: {file['path']}")
click.echo(f" Plain Name: {file.get('plainName', 'N/A')}")
click.echo(f" Type/Extension: {file.get('type', 'N/A')}")
click.echo(f" Size: {file['size_display']} ({file.get('size', 0)} bytes)")
click.echo(f" Folder ID: {file.get('folderId', 'N/A')}")
click.echo(f" User ID: {file.get('userId', 'N/A')}")
click.echo(f" File ID: {file.get('fileId', 'N/A')}")
click.echo(f" Bucket: {file.get('bucket', 'N/A')}")
click.echo(f" Encrypt Version: {file.get('encryptVersion', 'N/A')}")
# Timestamps for FILES
created_at = file.get('createdAt', 'N/A')
updated_at = file.get('updatedAt', 'N/A')
creation_time = file.get('creationTime', 'N/A')
modification_time = file.get('modificationTime', 'N/A')
# Use the correct logic for display
display_creation = creation_time if creation_time != 'N/A' else created_at
display_modification = modification_time if modification_time != 'N/A' else updated_at
if display_creation != 'N/A':
click.echo(f" Creation Time: {format_date(display_creation)} ({display_creation})")
else:
click.echo(f" Creation Time: N/A")
if display_modification != 'N/A':
click.echo(f" Modification Time: {format_date(display_modification)} ({display_modification})")
else:
click.echo(f" Modification Time: N/A")
# Other attributes
click.echo(f" Deleted: {file.get('deleted', False)}")
click.echo(f" Removed: {file.get('removed', False)}")
click.echo(f" Status: {file.get('status', 'N/A')}")
click.echo()
elif detailed:
modified = file.get('modified', '')[:10] if file.get('modified') else ''
size = file['size_display']
click.echo(f" 📄 {file['display_name']:<30} {size:<10} {modified:<12} {file['uuid'][:8]}...")
else:
size = file['size_display']
click.echo(f" 📄 {file['display_name']:<30} {size}")
if not content['folders'] and not content['files']:
click.echo(" (empty)")
click.echo(f"\nTotal: {len(content['folders'])} folders, {len(content['files'])} files")
# Show usage examples (only if not showing all attributes)
if content['files'] and not all:
example_file = content['files'][0]
example_path = example_file['path']
click.echo(f"\n💡 Usage examples:")
click.echo(f" Download by path: python cli.py download-path \"{example_path}\"")
click.echo(f" Delete by path: python cli.py trash-path \"{example_path}\"")
except ValueError as e:
click.echo(f"❌ Error: {e}", err=True)
sys.exit(1)
except Exception as e:
click.echo(f"❌ Unexpected error: {e}", err=True)
import traceback
traceback.print_exc()
sys.exit(1)
@cli.command('download-path')
@click.argument('path')
@click.option('--destination', '-d', '--target', '-t', 'destination', help='Where to save (file or directory)')
@click.option('--recursive', '-r', is_flag=True, help='Download folders recursively')
@click.option('--on-conflict', type=click.Choice(['overwrite', 'skip'], case_sensitive=False), default='skip', help='Action if local file exists')
@click.option('--preserve-timestamps', '-p', is_flag=True, help='Preserve file modification times')
@click.option('--include', multiple=True, help='Include only files matching pattern')
@click.option('--exclude', multiple=True, help='Exclude files matching pattern')
@click.option('--verbose', '-v', is_flag=True, help='Show verbose output')
def download_path(path: str, destination: Optional[str], recursive: bool, on_conflict: str,
preserve_timestamps: bool, include: Tuple[str], exclude: Tuple[str], verbose: bool):
"""
Download a file or folder by its path.
Examples:
Download single file:
python cli.py download-path "/Documents/report.pdf"
Download folder recursively:
python cli.py download-path "/Photos" -r -d ./local_photos
Download only images:
python cli.py download-path "/Photos" -r --include "*.jpg" --include "*.png"
With timestamp preservation:
python cli.py download-path "/Backup" -r -p
"""
try:
auth_service.get_auth_details()
# Convert include/exclude tuples to lists
include_patterns = list(include) if include else []
exclude_patterns = list(exclude) if exclude else []
if verbose and (include_patterns or exclude_patterns):
if include_patterns:
click.echo(f"🔍 Include filters: {', '.join(include_patterns)}")
if exclude_patterns:
click.echo(f"🚫 Exclude filters: {', '.join(exclude_patterns)}")
# Resolve the remote path
item_info = drive_service.resolve_path(path)
if item_info['type'] == 'file':
# Single file download
if verbose:
click.echo(f"📥 Downloading file: {path}")
# Apply filters
file_name = item_info.get('plainName', '')
if item_info.get('type'):
file_name = f"{file_name}.{item_info.get('type')}"
if not drive_service.should_include_file(Path(file_name), include_patterns, exclude_patterns):
click.echo(f"🚫 File filtered out by include/exclude patterns")
sys.exit(0)
# Determine destination
if destination:
dest_path = Path(destination)
else:
dest_path = Path.cwd() / file_name
# Check conflict
if dest_path.exists() and on_conflict == 'skip':
click.echo(f"⏭️ File exists, skipping: {dest_path}")
sys.exit(0)
# Download
downloaded_path = drive_service.download_file(
item_info['uuid'],