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