Import source: sort_drive
Import the existing local project snapshot. This commit organizes existing files; it does not represent newly developed functionality.
This commit is contained in:
@@ -0,0 +1,74 @@
|
||||
#!/usr/bin/env python3
|
||||
"""
|
||||
One-shot script: reads the GDrive UH folder, creates type folders,
|
||||
and moves each file into the right one.
|
||||
"""
|
||||
import json, subprocess, sys
|
||||
|
||||
REMOTE = "gdrive:"
|
||||
ROOT_ID = "1fEOd5m5psVYeN0hK6G_UJx_XwmXZcZ6l"
|
||||
BASE = ["rclone", "--drive-root-folder-id", ROOT_ID]
|
||||
|
||||
def rclone(*args):
|
||||
cmd = BASE + list(args)
|
||||
print(f" $ {' '.join(cmd)}")
|
||||
r = subprocess.run(cmd, capture_output=True, text=True)
|
||||
if r.returncode != 0:
|
||||
print(f" ERROR: {r.stderr.strip()}")
|
||||
return r.returncode == 0
|
||||
|
||||
def classify(name: str, mime: str) -> str:
|
||||
ext = name.rsplit(".", 1)[-1].lower() if "." in name else ""
|
||||
if mime.startswith("video/") or ext in ("mp4", "mkv", "avi", "mov", "webm", "flv"):
|
||||
return "Videos"
|
||||
if mime.startswith("audio/") or ext in ("mp3", "flac", "wav", "ogg", "m4a", "aac"):
|
||||
return "Music"
|
||||
if ext in ("exe", "msi", "deb", "rpm", "dmg", "appimage"):
|
||||
return "Software"
|
||||
if ext in ("zip", "rar", "7z", "tar", "gz", "bz2", "xz", "zst"):
|
||||
return "Archives"
|
||||
if ext in ("gguf", "bin", "safetensors", "pt", "ckpt", "pth"):
|
||||
return "Models"
|
||||
if ext in ("pdf", "doc", "docx", "txt", "md", "csv", "xlsx", "pptx"):
|
||||
return "Docs"
|
||||
return "Other"
|
||||
|
||||
# --- list files ---
|
||||
print("Listing files...")
|
||||
r = subprocess.run(BASE + ["lsjson", REMOTE], capture_output=True, text=True)
|
||||
files = json.loads(r.stdout)
|
||||
|
||||
# group (skip directories)
|
||||
groups: dict[str, list] = {}
|
||||
for f in files:
|
||||
if f["IsDir"]:
|
||||
print(f" [skip dir] {f['Name']}")
|
||||
continue
|
||||
folder = classify(f["Name"], f["MimeType"])
|
||||
groups.setdefault(folder, []).append(f["Name"])
|
||||
|
||||
# print plan
|
||||
print("\n=== Move plan ===")
|
||||
for folder, names in sorted(groups.items()):
|
||||
print(f"\n📁 {folder}/ ({len(names)} files)")
|
||||
for n in names:
|
||||
print(f" {n}")
|
||||
|
||||
print("\n=== Executing ===")
|
||||
created: set = set()
|
||||
ok = err = 0
|
||||
for folder, names in sorted(groups.items()):
|
||||
if folder not in created:
|
||||
print(f"\nCreating folder: {folder}/")
|
||||
rclone("mkdir", f"{REMOTE}{folder}")
|
||||
created.add(folder)
|
||||
for name in names:
|
||||
dest = f"{REMOTE}{folder}/{name}"
|
||||
src = f"{REMOTE}{name}"
|
||||
print(f"\nMoving → {folder}/{name[:60]}")
|
||||
if rclone("moveto", src, dest):
|
||||
ok += 1
|
||||
else:
|
||||
err += 1
|
||||
|
||||
print(f"\n=== Done: {ok} moved, {err} errors ===")
|
||||
Reference in New Issue
Block a user