322 lines
13 KiB
Python
322 lines
13 KiB
Python
import os
|
|
import zipfile
|
|
import re
|
|
import shutil
|
|
from PIL import Image
|
|
from pdf2image import convert_from_path
|
|
from datetime import datetime
|
|
from collections import defaultdict
|
|
import uuid # Library for generating unique IDs
|
|
|
|
# --- Utility Functions ---
|
|
|
|
def setup_error_log():
|
|
"""
|
|
Creates a uniquely named error log file.
|
|
"""
|
|
timestamp = datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
log_filename = f"errorlog_{timestamp}.log"
|
|
return log_filename
|
|
|
|
def log_error(log_file, message):
|
|
"""
|
|
Appends an error message to the log file.
|
|
"""
|
|
with open(log_file, 'a') as f:
|
|
f.write(f"{datetime.now().strftime('%Y-%m-%d %H:%M:%S')} - {message}\n")
|
|
|
|
def convert_webp_to_jpg(file_path, log_file):
|
|
"""
|
|
Converts a .webp image file to .jpg format.
|
|
"""
|
|
try:
|
|
image = Image.open(file_path)
|
|
if image.mode == 'RGBA':
|
|
image = image.convert('RGB')
|
|
jpg_path = os.path.splitext(file_path)[0] + '.jpg'
|
|
image.save(jpg_path, 'JPEG')
|
|
os.remove(file_path)
|
|
return jpg_path
|
|
except Exception as e:
|
|
log_error(log_file, f"Error converting WEBP: {file_path} - {e}")
|
|
return None
|
|
|
|
def convert_pdf_to_jpgs(file_path, output_folder, log_file):
|
|
"""
|
|
Converts a PDF file into a series of JPG images.
|
|
"""
|
|
try:
|
|
pages = convert_from_path(file_path)
|
|
output_paths = []
|
|
for i, page in enumerate(pages):
|
|
jpg_path = os.path.join(output_folder, f"page_{i+1:03d}.jpg")
|
|
page.save(jpg_path, 'JPEG')
|
|
output_paths.append(jpg_path)
|
|
os.remove(file_path)
|
|
return output_paths
|
|
except Exception as e:
|
|
log_error(log_file, f"Error converting PDF: {file_path} - {e}")
|
|
return []
|
|
|
|
def create_cbz_archive(folder_path, output_path, log_file):
|
|
"""
|
|
Creates a .cbz archive from all image files in a folder.
|
|
"""
|
|
try:
|
|
with zipfile.ZipFile(output_path, 'w') as cbz_file:
|
|
for root, _, files in os.walk(folder_path):
|
|
# Sort files to ensure correct page order in the CBZ file
|
|
sorted_files = sorted(files, key=lambda f: re.split(r'(\d+)', f))
|
|
for filename in sorted_files:
|
|
if filename.lower().endswith(('.jpg', '.jpeg', '.png', '.gif')):
|
|
file_path = os.path.join(root, filename)
|
|
# Archive name should be relative to the content folder
|
|
arcname = os.path.relpath(file_path, folder_path)
|
|
cbz_file.write(file_path, arcname)
|
|
return True
|
|
except Exception as e:
|
|
log_error(log_file, f"Error creating CBZ for {folder_path} - {e}")
|
|
return False
|
|
|
|
def clean_up_temp_folders(folder_path):
|
|
"""
|
|
Removes a folder and its contents.
|
|
"""
|
|
try:
|
|
if os.path.exists(folder_path):
|
|
shutil.rmtree(folder_path)
|
|
except Exception as e:
|
|
print(f"Error cleaning up {folder_path}: {e}")
|
|
|
|
# --- Grouping Function ---
|
|
|
|
def get_series_name(file_or_folder_name):
|
|
"""
|
|
Uses regex to extract the common series name from a chapter folder/file name.
|
|
"""
|
|
base_name = os.path.splitext(file_or_folder_name)[0]
|
|
|
|
# Pattern to match and strip common chapter/volume suffixes at the end of a string.
|
|
pattern = re.compile(r'[-_]?((ch(apter)?|v(ol)?)[-_]?)?\d{1,4}(\.\d{1,2})?$', re.IGNORECASE)
|
|
|
|
# Find the match from the end
|
|
match = pattern.search(base_name)
|
|
if match:
|
|
# Return the string slice before the matched pattern
|
|
return base_name[:match.start()].strip('-_ ')
|
|
|
|
# Fallback: if no clear chapter pattern is found, return the original name
|
|
return base_name
|
|
|
|
# --- File/Folder Processing Function ---
|
|
|
|
def process_chapter(chapter_path, final_series_folder, temp_root_folder, log_file):
|
|
"""
|
|
Copies, converts, creates CBZ, and cleans up for a single chapter folder/archive.
|
|
Uses a unique temporary sub-folder for its operations.
|
|
"""
|
|
chapter_name = os.path.basename(chapter_path)
|
|
|
|
# Create a UNIQUE temporary folder for THIS chapter's processing
|
|
unique_id = str(uuid.uuid4())[:8] # Short unique identifier
|
|
temp_chapter_folder = os.path.join(temp_root_folder, f"proc_{unique_id}_{chapter_name}")
|
|
os.makedirs(temp_chapter_folder, exist_ok=True)
|
|
|
|
is_archive = os.path.isfile(chapter_path) and chapter_path.lower().endswith(('.zip', '.cbz'))
|
|
|
|
try:
|
|
if is_archive:
|
|
print(f" -> Extracting Archive: {chapter_name}")
|
|
with zipfile.ZipFile(chapter_path, 'r') as zip_ref:
|
|
zip_ref.extractall(temp_chapter_folder)
|
|
else:
|
|
print(f" -> Processing Folder: {chapter_name}")
|
|
# Copy original folder contents to the temporary folder
|
|
for item in os.listdir(chapter_path):
|
|
s = os.path.join(chapter_path, item)
|
|
d = os.path.join(temp_chapter_folder, item)
|
|
if os.path.isdir(s):
|
|
# Use a recursive copy for subdirectories
|
|
shutil.copytree(s, d)
|
|
else:
|
|
# Use a simple copy for files
|
|
shutil.copy2(s, d)
|
|
|
|
# B. Convert files inside the temporary folder
|
|
for root, _, files in os.walk(temp_chapter_folder):
|
|
for file_path in [os.path.join(root, f) for f in files]:
|
|
if file_path.lower().endswith('.webp'):
|
|
print(f" - Converting WEBP: {os.path.basename(file_path)}")
|
|
convert_webp_to_jpg(file_path, log_file)
|
|
elif file_path.lower().endswith('.pdf'):
|
|
print(f" - Converting PDF: {os.path.basename(file_path)}")
|
|
# Conversion places JPGs in the same folder as the PDF
|
|
convert_pdf_to_jpgs(file_path, root, log_file)
|
|
|
|
# C. Create CBZ archive in the final series destination
|
|
cbz_name = os.path.splitext(chapter_name)[0] + '.cbz'
|
|
output_cbz_path = os.path.join(final_series_folder, cbz_name)
|
|
|
|
if create_cbz_archive(temp_chapter_folder, output_cbz_path, log_file):
|
|
print(f" - Successfully created {cbz_name}")
|
|
|
|
except Exception as e:
|
|
log_error(log_file, f"Processing failed for {chapter_name} (Path: {chapter_path}): {e}")
|
|
print(f" - ERROR: Processing failed. Check log for details.")
|
|
return # Exit the function on error
|
|
finally:
|
|
# D. Clean up the temporary processing folder for this chapter
|
|
clean_up_temp_folders(temp_chapter_folder)
|
|
|
|
# --- New Move Function ---
|
|
|
|
def move_converted_files(source_dir, destination_root, log_file):
|
|
"""
|
|
Moves all contents (series folders) from source_dir (converted)
|
|
to destination_root (/mnt/isolation/comics).
|
|
"""
|
|
print("\n--- Starting Final Move Operation ---")
|
|
|
|
# The parent of /mnt/isolation/comics/toberead/converted is /mnt/isolation/comics/toberead
|
|
# We want to move to /mnt/isolation/comics
|
|
|
|
# 1. Identify the true final destination path
|
|
# Get the parent of the source_dir (which is /mnt/isolation/comics/toberead)
|
|
parent_of_source = os.path.dirname(source_dir)
|
|
# Get the parent of that (which is /mnt/isolation/comics)
|
|
final_destination = os.path.dirname(parent_of_source)
|
|
|
|
if destination_root != final_destination:
|
|
log_error(log_file, f"Configuration Error: Move destination root mismatch. Expected {final_destination}, got {destination_root}")
|
|
print(f"Error: Final destination paths do not match. Aborting move. Check log.")
|
|
return
|
|
|
|
total_moved = 0
|
|
|
|
# Iterate through each series folder in the 'converted' directory
|
|
for series_name in os.listdir(source_dir):
|
|
series_source_path = os.path.join(source_dir, series_name)
|
|
series_dest_path = os.path.join(destination_root, series_name)
|
|
|
|
if os.path.isdir(series_source_path):
|
|
try:
|
|
# Move the entire series folder (containing the CBZ files)
|
|
if not os.path.exists(series_dest_path):
|
|
shutil.move(series_source_path, series_dest_path)
|
|
print(f" -> Moved series folder: {series_name}")
|
|
total_moved += 1
|
|
else:
|
|
# If the series folder already exists in the destination, move its contents (the CBZ files)
|
|
for item in os.listdir(series_source_path):
|
|
item_s = os.path.join(series_source_path, item)
|
|
item_d = os.path.join(series_dest_path, item)
|
|
|
|
if item.lower().endswith('.cbz') and not os.path.exists(item_d):
|
|
shutil.move(item_s, item_d)
|
|
print(f" -> Moved CBZ: {item}")
|
|
total_moved += 1
|
|
|
|
# Remove the now-empty source folder
|
|
clean_up_temp_folders(series_source_path)
|
|
|
|
except Exception as e:
|
|
log_error(log_file, f"Error moving series {series_name}: {e}")
|
|
print(f"Error moving {series_name}. Check log for details.")
|
|
|
|
print(f"\nMove Complete. Total items moved: {total_moved}")
|
|
|
|
|
|
def main():
|
|
"""
|
|
Main function to run the comic organization script.
|
|
"""
|
|
# --- Configuration ---
|
|
source_directory = "/mnt/isolation/comics/toberead"
|
|
target_directory = os.path.join(source_directory, "converted")
|
|
temp_root_folder = os.path.join(source_directory, ".komga_processing_temp")
|
|
|
|
# The final destination root for the move operation (parent of source_directory)
|
|
destination_root = os.path.dirname(source_directory)
|
|
# ---------------------
|
|
|
|
error_log_file = setup_error_log()
|
|
print(f"Starting script. Source: {source_directory}")
|
|
print(f"Output will be in: {target_directory}")
|
|
print(f"Final destination root: {destination_root}")
|
|
print(f"Errors will be logged to: {error_log_file}")
|
|
|
|
os.makedirs(target_directory, exist_ok=True)
|
|
|
|
# Clean up the main temporary directory before starting
|
|
clean_up_temp_folders(temp_root_folder)
|
|
os.makedirs(temp_root_folder, exist_ok=True)
|
|
|
|
# 1. Group Chapter Folders and Archive Files by Series Name
|
|
|
|
series_chapters = defaultdict(list)
|
|
|
|
try:
|
|
items = os.listdir(source_directory)
|
|
except FileNotFoundError:
|
|
print(f"Error: Source directory not found: {source_directory}. Aborting.")
|
|
return
|
|
|
|
for item in items:
|
|
item_path = os.path.join(source_directory, item)
|
|
|
|
# Skip the output and temporary directories
|
|
if item_path.startswith(target_directory) or item_path.startswith(temp_root_folder):
|
|
continue
|
|
|
|
is_relevant_folder = os.path.isdir(item_path) and any(f.lower().endswith(('.jpg', '.jpeg', '.png', '.gif', '.webp', '.pdf'))
|
|
for f in os.listdir(item_path))
|
|
is_relevant_file = os.path.isfile(item_path) and item.lower().endswith(('.zip', '.cbz'))
|
|
|
|
if is_relevant_folder or is_relevant_file:
|
|
series_name = get_series_name(item)
|
|
series_chapters[series_name].append(item_path)
|
|
|
|
print(f"\nFound {len(series_chapters)} series to process.")
|
|
|
|
# 2. Process Each Series Group
|
|
|
|
for series_name, chapter_paths in series_chapters.items():
|
|
print(f"\n--- Processing Series: {series_name} ({len(chapter_paths)} Chapters/Archives) ---")
|
|
|
|
# Create the final parent folder for the series in the target directory
|
|
final_series_folder = os.path.join(target_directory, series_name)
|
|
os.makedirs(final_series_folder, exist_ok=True)
|
|
|
|
for chapter_path in chapter_paths:
|
|
chapter_name = os.path.basename(chapter_path)
|
|
|
|
if chapter_name.lower().endswith('.cbz') and os.path.isfile(chapter_path):
|
|
# Move existing CBZ files directly to the final destination
|
|
new_path = os.path.join(final_series_folder, chapter_name)
|
|
if not os.path.exists(new_path):
|
|
shutil.move(chapter_path, new_path)
|
|
print(f" -> Moved existing CBZ: {chapter_name}")
|
|
else:
|
|
print(f" -> Skipping CBZ (already exists in destination): {chapter_name}")
|
|
else:
|
|
# Process folders and ZIP files
|
|
process_chapter(chapter_path, final_series_folder, temp_root_folder, error_log_file)
|
|
|
|
# 3. Final Cleanup and Conditional Move
|
|
|
|
# Final cleanup of the main temp folder
|
|
clean_up_temp_folders(temp_root_folder)
|
|
|
|
print("\n--- Processing Finished ---")
|
|
|
|
# --- USER PROMPT FOR FINAL MOVE ---
|
|
prompt = f"Do you want to move all CBZ files from '{target_directory}' to the root directory '{destination_root}'? (yes/no): "
|
|
user_input = input(prompt).strip().lower()
|
|
|
|
if user_input == 'yes':
|
|
move_converted_files(target_directory, destination_root, error_log_file)
|
|
|
|
print("\nScript operation complete.")
|
|
|
|
if __name__ == "__main__":
|
|
main() |