A python script to clean orphan Beatsaber custom maps for Meta (Oculus) Quest device

bs-clean-maps.py

import subprocess
import json
import os
import glob
import sys

# --- User Configuration ---
# 1. Path to the adb executable.
ADB_PATH = "adb"
# You may change it if adb command cannot be found, e.g. 
# ADB_PATH = "D:/apps/platform-tools/adb.exe"

# --- Quest File Paths (Verified) ---
# 2. Path to PlayerData.dat (contains favorite song list)
PLAYER_DATA_REMOTE_PATH = "/sdcard/Android/data/com.beatgames.beatsaber/files/PlayerData.dat"
# 3. Path to custom levels (folders are named by hash)
CUSTOM_LEVELS_PATH = "/sdcard/ModData/com.beatgames.beatsaber/Mods/SongCore/CustomLevels"
# 4. Path to playlists
PLAYLISTS_PATH = "/sdcard/ModData/com.beatgames.beatsaber/Mods/PlaylistManager/Playlists"

# Local directory for temporary files (kept for debugging)
TEMP_DIR = "bs_temp_data"

def run_adb_command(command, check=True):
    """Executes an ADB command and returns the output."""

    # Combine ADB path and command
    adb_command_list = [ADB_PATH]
    adb_command_list.extend(command.split())

    print(f"-> Executing command: {' '.join(adb_command_list)}")
    try:
        # Normalize path for Windows compatibility
        adb_command_list[0] = os.path.normpath(adb_command_list[0])

        result = subprocess.run(
            adb_command_list,
            capture_output=True,
            text=True,
            check=check,
            encoding='utf-8'
        )
        if result.stderr and check and 'error' in result.stderr.lower():
            if 'error' in result.stderr.lower():
                 print(f"ADB Warning/Error Output:\n{result.stderr}")
        return result.stdout.strip()
    except subprocess.CalledProcessError as e:
        print(f"ADB command failed. Check device connection, ADB path, or Quest paths. Error: {e}")
        sys.exit(1)
    except FileNotFoundError:
        print(f"Error: ADB executable not found. Check **ADB_PATH** configuration. Path used: {ADB_PATH}")
        sys.exit(1)

def pull_files():
    """Pulls PlayerData.dat and Playlists directory from the Quest device."""
    # Ensure the main temporary directory exists
    os.makedirs(TEMP_DIR, exist_ok=True)

    # Pull PlayerData.dat
    player_data_local = os.path.join(TEMP_DIR, "PlayerData.dat")
    print("\n[Step 1/5] Pulling PlayerData.dat (Favorites data)...")
    run_adb_command(f"pull {PLAYER_DATA_REMOTE_PATH} {player_data_local}")

    # Pull Playlists directory
    playlists_local_dir = os.path.join(TEMP_DIR, os.path.basename(PLAYLISTS_PATH))
    print("[Step 2/5] Pulling Playlists directory...")
    # Target is TEMP_DIR, ADB creates Playlists/ inside it
    run_adb_command(f"pull -a {PLAYLISTS_PATH} {TEMP_DIR}")

    return player_data_local, playlists_local_dir

def get_used_song_ids(player_data_path, playlists_dir):
    """Parses files to get the set of all favorited and playlist-included song IDs."""
    used_ids = set()
    favorites_count = 0
    playlist_stats = {} # Stores {playlist_name: map_count}

    # 1. Parse PlayerData.dat for favorited song IDs (Fail Fast style)
    print("\n[Step 3/5] Parsing Favorites and Playlists data...")
    try:
        with open(player_data_path, 'r', encoding='utf-8') as f:
            data = json.load(f)

            # Correct key path for favorites
            favorites = data['localPlayers'][0]["favoritesLevelIds"]

            favorites_count = len(favorites)
            for level_id in favorites:
                # Extract hash from level ID (e.g., custom_level_<HASH>)
                hash_value = level_id.split('custom_level_')[-1].upper()
                used_ids.add(hash_value)

    except Exception as e:
        # Catch and report any structural error immediately
        raise Exception(f"Error parsing PlayerData.dat file ({player_data_path}). Check JSON structure: {e}")


    # 2. Parse Playlists directory for song IDs
    playlist_files = glob.glob(os.path.join(playlists_dir, "*"))
    total_playlist_maps = 0
    parsed_playlists_count = 0

    for file_path in playlist_files:
        if os.path.isdir(file_path):
            continue

        playlist_name = os.path.basename(file_path)
        current_playlist_maps = 0
        try:
            with open(file_path, 'r', encoding='utf-8') as f:
                playlist_data = json.load(f)

                # Use .get() for playlist songs for robustness
                songs = playlist_data.get("songs", [])

                for song in songs:
                    song_hash = song.get("hash")
                    if song_hash:
                        used_ids.add(song_hash.upper())
                        total_playlist_maps += 1
                        current_playlist_maps += 1
                playlist_stats[playlist_name] = current_playlist_maps
                parsed_playlists_count += 1

        except json.JSONDecodeError:
            print(f"  -> Warning: Playlist file **{playlist_name}** has invalid format, skipped.")
        except Exception as e:
            print(f"  -> Warning: Unknown error parsing playlist file {playlist_name}: {e}")

    # 6. Print stats
    print("\n--- Song Statistics ---")
    print(f"1. Favorite map count: **{favorites_count}**")
    print(f"2. Parsed Playlists count: **{parsed_playlists_count}**")
    print(f"3. Total map count across all playlists (including duplicates): **{total_playlist_maps}**")
    print(f"4. Total unique song IDs (Favorites + Playlists): **{len(used_ids)}**")
    print("5. Map count per playlist:")
    for name, count in playlist_stats.items():
        print(f"   - {name}: {count} maps")

    return used_ids

def get_all_local_song_folders():
    """Gets all local custom song folder names (hashes) using ADB shell."""
    print("\n[Step 4/5] Retrieving all local custom song folders from the device...")

    # Use 'ls -1' for clean list output
    ls_output = run_adb_command(f"shell ls -1 {CUSTOM_LEVELS_PATH}")

    all_local_folders = {} # {hash_value: full_folder_name}

    for folder_name in ls_output.split('\n'):
        folder_name = folder_name.strip()
        if not folder_name or folder_name in ['.', '..', 'lost+found']:
            continue

        # Assumption: Folder name IS the full hash
        song_hash = folder_name.upper()
        if len(song_hash) >= 40:
            all_local_folders[song_hash] = folder_name

    print(f"  -> Found **{len(all_local_folders)}** local custom songs.")
    return all_local_folders

def delete_unused_songs(all_local_folders, used_ids):
    """Compares IDs and deletes unused song folders."""

    all_local_hashes = set(all_local_folders.keys())

    # Find unused songs (exist locally but not in used_ids)
    unused_hashes = all_local_hashes - used_ids

    if not unused_hashes:
        print("\n[Step 5/5] No unused songs found, no deletion required. 🎉")
        return

    # Get the full folder names to delete
    files_to_delete = [
        all_local_folders[h] for h in unused_hashes if h in all_local_folders
    ]

    # 6. Print count of maps to be deleted
    print(f"\n6. Unused map count to be deleted: **{len(files_to_delete)}**")
    print("----------------------")

    print(f"\n[Step 5/5] Found **{len(files_to_delete)}** folders for deletion.")
    print("ALL files to be deleted (3 per line for compactness):")

    # --- MODIFIED SECTION: Print ALL files (3 per line) ---
    line_buffer = []
    for i, name in enumerate(files_to_delete):
        line_buffer.append(name)
        if len(line_buffer) == 3:
            print(f"  - {line_buffer[0]:<40} {line_buffer[1]:<40} {line_buffer[2]:<40}")
            line_buffer = []

    # Print any remaining files in the buffer
    if line_buffer:
        # Create a format string that accommodates the remaining items
        format_str = "  - " + " ".join([f"{{:<40}}" for _ in range(len(line_buffer))])
        print(format_str.format(*line_buffer))
    # --- END MODIFIED SECTION ---

    # 5. Confirmation before deletion
    confirmation = input("\nPlease confirm deletion? (Enter **'y'** to proceed, anything else to abort): ")
    if confirmation.lower() != 'y':
        print("\nOperation aborted. No files were deleted.")
        return

    # Execute deletion
    deleted_count = 0
    print("\nExecuting deletion...")
    for folder_name in files_to_delete:
        remote_path = f"{CUSTOM_LEVELS_PATH}/{folder_name}"
        # Use rm -rf for recursive forced deletion
        delete_command = f"shell rm -rf {remote_path}"
        run_adb_command(delete_command, check=False)
        deleted_count += 1
        if deleted_count % 100 == 0:
            print(f"  -> Deleted {deleted_count} folders...")

    print(f"\n✅ **Cleanup complete! Deleted {deleted_count} unused songs.**")


# --- Main Program ---
if __name__ == "__main__":
    print("--- Beat Saber Custom Song Cleaner (v13) ---")
    try:
        # 0. Check ADB device connection
        print("Checking ADB device connection...")
        run_adb_command("devices")

        # 1. Pull necessary files
        player_data_path, playlists_dir = pull_files()

        # 2. Get used song IDs
        used_ids = get_used_song_ids(player_data_path, playlists_dir)

        # 3. Get all local song IDs
        all_local_folders = get_all_local_song_folders()

        # 4. Delete unused songs
        delete_unused_songs(all_local_folders, used_ids)

    except Exception as e:
        # Catch and report all fatal errors
        print(f"\n❌ **Fatal error occurred during execution, exiting:** {e}")

    finally:
        # 5. Keep local temp directory for debugging purposes
        print(f"\n⚠️ **Note: Local temporary files in '{TEMP_DIR}' directory have been kept for debugging.**")