Made lots of improvements to the script. Should be a lot faster to process. The original script took me 15-20 mins to process one of the full interactive scenes, and this one takes me around 30s. You can see the progress now as well. Allows stopping and resuming, and automatically processes all files in the folder. Compatible with older versions of python too I think.
- Save this script as whatever you want, like xor_file.py (Make sure it is .py)
- Place this script file and the .mp4s from dezyred in the same folder.
- Right click in the folder, click open terminal (in windows).
- Type without quotations “python nameofyourfile.py”, replace name of your file with whatever you used.
- Select R or O to resume or overide. If first run, doesn’t matter which you pick.
- All files will be processed and output to the same folder.
#!/usr/bin/env python3
"""
Fast Resumable XOR Folder Tool
XOR all .mp4 files in the same directory as this script.
Features:
- Processes all .mp4 files in the script folder
- Skips files already ending in _XOR.mp4
- Supports resume and overwrite modes
- If no mode flag is passed, asks interactively
- Faster XOR using bytes.translate()
- Larger chunk size for better performance
- Shows current file name
- Shows per-file progress
- Shows overall progress across all files by bytes
- Shows speed and ETA
- Lists completed files
"""
import argparse
import os
import sys
import time
from typing import List, Optional, Tuple
XOR_KEY = 0xFF
CHUNK_SIZE = 32 * 1024 * 1024 # 32 MB
UPDATE_INTERVAL = 1.0 # seconds
XOR_TABLE = bytes.maketrans(
bytes(range(256)),
bytes((i ^ XOR_KEY) for i in range(256))
)
def format_bytes(num_bytes: float) -> str:
units = ["B", "KB", "MB", "GB", "TB"]
size = float(num_bytes)
for unit in units:
if size < 1024 or unit == units[-1]:
return f"{size:.2f} {unit}"
size /= 1024
return f"{size:.2f} TB"
def format_time(seconds: float) -> str:
if seconds < 0 or seconds == float("inf"):
return "unknown"
seconds = int(seconds)
hours, remainder = divmod(seconds, 3600)
minutes, secs = divmod(remainder, 60)
if hours > 0:
return f"{hours}h {minutes}m {secs}s"
if minutes > 0:
return f"{minutes}m {secs}s"
return f"{secs}s"
def xor_chunk(chunk: bytes) -> bytes:
return chunk.translate(XOR_TABLE)
def get_output_path(input_path: str) -> str:
base, ext = os.path.splitext(input_path)
return f"{base}_XOR{ext}"
def find_mp4_files(folder: str) -> List[str]:
files = []
for name in os.listdir(folder):
lower_name = name.lower()
if lower_name.endswith(".mp4") and not lower_name.endswith("_xor.mp4"):
files.append(os.path.join(folder, name))
return sorted(files)
def print_completed_files(completed_files: List[str]) -> None:
print("\nCompleted files:")
if not completed_files:
print(" None yet")
return
for name in completed_files:
print(f" - {name}")
def get_resume_offset(input_path: str, output_path: str, mode: str) -> Optional[int]:
total_size = os.path.getsize(input_path)
if not os.path.exists(output_path):
return 0
existing_size = os.path.getsize(output_path)
if mode == "overwrite":
return 0
if existing_size > total_size:
return None
return existing_size
def build_work_plan(files: List[str], mode: str) -> Tuple[int, int]:
total_input_bytes = 0
total_remaining_bytes = 0
for input_path in files:
output_path = get_output_path(input_path)
total_size = os.path.getsize(input_path)
total_input_bytes += total_size
resume_offset = get_resume_offset(input_path, output_path, mode)
if resume_offset is None:
continue
remaining = max(0, total_size - resume_offset)
total_remaining_bytes += remaining
return total_input_bytes, total_remaining_bytes
def choose_mode_interactively() -> str:
while True:
print("\nChoose processing mode:")
print(" [R] Resume existing partial output files")
print(" [O] Overwrite existing output files and start from zero")
choice = input("Enter R or O: ").strip().lower()
if choice in ("r", "resume"):
return "resume"
if choice in ("o", "overwrite"):
return "overwrite"
print("Invalid choice. Please enter R or O.")
def xor_file(
input_path: str,
mode: str,
file_index: int,
total_files: int,
completed_files: List[str],
overall_remaining_start: int,
overall_processed_callback,
overall_start_time: float,
) -> Optional[str]:
output_path = get_output_path(input_path)
total_size = os.path.getsize(input_path)
resume_offset = get_resume_offset(input_path, output_path, mode)
if resume_offset is None:
print(
f"\n[{file_index}/{total_files}] Error: output file is larger than input file for "
f"{os.path.basename(input_path)}",
file=sys.stderr,
)
return None
if resume_offset == total_size and mode == "resume":
print(f"\n[{file_index}/{total_files}] Already complete: {os.path.basename(output_path)}")
completed_files.append(os.path.basename(output_path))
print_completed_files(completed_files)
return output_path
file_mode = "ab" if (mode == "resume" and resume_offset > 0) else "wb"
try:
with open(input_path, "rb", buffering=CHUNK_SIZE) as infile, open(output_path, file_mode, buffering=CHUNK_SIZE) as outfile:
if resume_offset > 0:
infile.seek(resume_offset)
processed = resume_offset
file_start_time = time.time()
last_update = 0.0
last_processed = resume_offset
print(f"\n[{file_index}/{total_files}] Current file: {os.path.basename(input_path)}")
print(f"Output: {os.path.basename(output_path)}")
print(f"Mode: {mode}")
if resume_offset > 0:
print(f"Resuming from: {format_bytes(resume_offset)} / {format_bytes(total_size)}")
else:
print(f"Starting from: 0 B / {format_bytes(total_size)}")
while True:
chunk = infile.read(CHUNK_SIZE)
if not chunk:
break
outfile.write(xor_chunk(chunk))
processed += len(chunk)
just_processed = processed - last_processed
if just_processed > 0:
overall_processed_callback(just_processed)
last_processed = processed
now = time.time()
if now - last_update >= UPDATE_INTERVAL or processed == total_size:
file_elapsed = now - file_start_time
file_session_processed = processed - resume_offset
file_speed = file_session_processed / file_elapsed if file_elapsed > 0 else 0.0
file_percent = (processed / total_size) * 100 if total_size else 100.0
file_remaining = total_size - processed
file_eta = file_remaining / file_speed if file_speed > 0 else float("inf")
overall_done = overall_processed_callback(0)
overall_elapsed = now - overall_start_time
overall_speed = overall_done / overall_elapsed if overall_elapsed > 0 else 0.0
overall_percent = (
(overall_done / overall_remaining_start) * 100
if overall_remaining_start > 0
else 100.0
)
overall_remaining = overall_remaining_start - overall_done
overall_eta = (
overall_remaining / overall_speed
if overall_speed > 0
else float("inf")
)
status_line = (
f"\r"
f"File {file_index}/{total_files} | "
f"{os.path.basename(input_path)} | "
f"File: {file_percent:6.2f}% "
f"({format_bytes(processed)} / {format_bytes(total_size)}) | "
f"Overall: {overall_percent:6.2f}% "
f"({format_bytes(overall_done)} / {format_bytes(overall_remaining_start)}) | "
f"Speed: {format_bytes(file_speed)}/s | "
f"File ETA: {format_time(file_eta)} | "
f"Overall ETA: {format_time(overall_eta)}"
)
print(status_line, end="", flush=True)
last_update = now
file_elapsed = time.time() - file_start_time
print()
print(f"Finished: {os.path.basename(output_path)}")
print(f"File time: {format_time(file_elapsed)}")
session_bytes = total_size - resume_offset
if file_elapsed > 0 and session_bytes > 0:
print(f"Average file speed: {format_bytes(session_bytes / file_elapsed)}/s")
completed_files.append(os.path.basename(output_path))
print_completed_files(completed_files)
return output_path
except KeyboardInterrupt:
print("\nStopped by user.", file=sys.stderr)
return None
except PermissionError:
print("\nError: Permission denied.", file=sys.stderr)
return None
except OSError as e:
print(f"\nError: {e}", file=sys.stderr)
return None
def parse_args() -> argparse.Namespace:
parser = argparse.ArgumentParser(
description="XOR all MP4 files in the same folder as this script."
)
mode_group = parser.add_mutually_exclusive_group()
mode_group.add_argument(
"--resume",
action="store_true",
help="Resume existing partial output files.",
)
mode_group.add_argument(
"--overwrite",
action="store_true",
help="Overwrite existing output files and start from zero.",
)
return parser.parse_args()
def main() -> int:
args = parse_args()
if args.overwrite:
mode = "overwrite"
elif args.resume:
mode = "resume"
else:
mode = choose_mode_interactively()
script_dir = os.path.dirname(os.path.abspath(__file__))
files = find_mp4_files(script_dir)
if not files:
print("No MP4 files found in the script folder.")
return 1
total_input_bytes, total_remaining_bytes = build_work_plan(files, mode)
completed_files: List[str] = []
success_count = 0
total_files = len(files)
overall_processed = 0
overall_start_time = time.time()
def overall_processed_callback(increment: int) -> int:
nonlocal overall_processed
overall_processed += increment
return overall_processed
print(f"\nScript folder: {script_dir}")
print(f"Mode: {mode}")
print(f"Files found: {total_files}")
print(f"Total input size: {format_bytes(total_input_bytes)}")
print(f"Total remaining to process this run: {format_bytes(total_remaining_bytes)}")
print("\nFiles queued:")
for path in files:
print(f" - {os.path.basename(path)}")
for index, path in enumerate(files, start=1):
result = xor_file(
input_path=path,
mode=mode,
file_index=index,
total_files=total_files,
completed_files=completed_files,
overall_remaining_start=total_remaining_bytes,
overall_processed_callback=overall_processed_callback,
overall_start_time=overall_start_time,
)
if result:
success_count += 1
total_elapsed = time.time() - overall_start_time
print("\nAll processing complete.")
print(f"Successful: {success_count}/{total_files}")
print(f"Total session time: {format_time(total_elapsed)}")
if total_elapsed > 0 and total_remaining_bytes > 0:
print(f"Average overall speed: {format_bytes(total_remaining_bytes / total_elapsed)}/s")
print_completed_files(completed_files)
return 0 if success_count == total_files else 1
if __name__ == "__main__":
sys.exit(main())```