added new gemini creations
This commit is contained in:
@@ -0,0 +1,21 @@
|
||||
import sys
|
||||
import os
|
||||
|
||||
# Assuming check_db.py is in the project root
|
||||
# Add the backend directory to sys.path so 'app' package can be found
|
||||
sys.path.insert(0, os.path.abspath('ServerManagerWebApp/backend'))
|
||||
|
||||
# Now, import modules using their full path from the backend as root
|
||||
# e.g., 'app.database' and 'app.models'
|
||||
from app.database import SessionLocal
|
||||
from app.models import VideoFile
|
||||
|
||||
try:
|
||||
db = SessionLocal()
|
||||
count = db.query(VideoFile).count()
|
||||
print(f"Number of VideoFile entries in the database: {count}")
|
||||
except Exception as e:
|
||||
print(f"Error checking database: {e}")
|
||||
finally:
|
||||
if 'db' in locals() and db:
|
||||
db.close()
|
||||
@@ -0,0 +1,322 @@
|
||||
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()
|
||||
@@ -0,0 +1,129 @@
|
||||
import os
|
||||
import re
|
||||
import sys
|
||||
from collections import defaultdict
|
||||
import logging
|
||||
from .samba_manager import SambaManager
|
||||
from .config import Settings
|
||||
from smb.smb_structs import OperationFailure
|
||||
|
||||
# --- Dedicated logger for ComicsManager ---
|
||||
comics_logger = logging.getLogger('comics_manager')
|
||||
comics_logger.setLevel(logging.INFO)
|
||||
comics_logger.propagate = False
|
||||
if not comics_logger.handlers:
|
||||
comics_log_handler = logging.FileHandler("logs/comics_organization.log", mode='a')
|
||||
comics_log_handler.setFormatter(logging.Formatter('%(asctime)s - %(levelname)s - %(message)s'))
|
||||
comics_logger.addHandler(comics_log_handler)
|
||||
comics_logger.addHandler(logging.StreamHandler())
|
||||
|
||||
# --- Disable noisy pysmb logger ---
|
||||
# Temporarily set the pysmb logger levels to CRITICAL to silence its output.
|
||||
# This needs to be set *after* basicConfig if pysmb initializes its loggers earlier.
|
||||
try:
|
||||
logging.getLogger("smb").setLevel(logging.CRITICAL)
|
||||
logging.getLogger("smb.SMBConnection").setLevel(logging.CRITICAL)
|
||||
except Exception:
|
||||
# Handle cases where the logger might not be fully initialized yet
|
||||
pass
|
||||
# ---
|
||||
|
||||
class ComicsManager:
|
||||
def __init__(self, settings: Settings):
|
||||
self.settings = settings
|
||||
self.db = None # ComicsManager doesn't use DB
|
||||
self.samba_manager = SambaManager(
|
||||
settings.samba_server_ip,
|
||||
"isolation",
|
||||
settings.samba_username,
|
||||
settings.samba_password,
|
||||
)
|
||||
self.comics_root = "/comics/toberead"
|
||||
self.comics_dest_root = "/comics/manga"
|
||||
|
||||
def _get_series_name(self, filename):
|
||||
base_name, _ = os.path.splitext(filename)
|
||||
# This regex is more aggressive at cleaning up chapter/volume numbers and other noise.
|
||||
# It looks for common patterns like 'ch.10', 'v02', '123', etc.
|
||||
pattern = re.compile(r'[-_\s]*(v(ol)?|c(h)?|chapter|issue|ep|episode)[-_\s]*\d+.*|[-_\s]+\d+$', re.IGNORECASE)
|
||||
match = pattern.search(base_name)
|
||||
if match:
|
||||
return base_name[:match.start()].strip('-_ ')
|
||||
return base_name.strip('-_ ')
|
||||
|
||||
def organize_comics(self):
|
||||
comics_logger.info("Starting comics organization process...")
|
||||
try:
|
||||
items_in_toberead = self.samba_manager.list_path(self.comics_root)
|
||||
if "error" in items_in_toberead:
|
||||
raise Exception(f"Failed to list files in 'toberead': {items_in_toberead['error']}")
|
||||
|
||||
series_chapters = defaultdict(list)
|
||||
for item in items_in_toberead:
|
||||
if not item["is_directory"]:
|
||||
series_name = self._get_series_name(item["name"])
|
||||
series_chapters[series_name].append(item)
|
||||
|
||||
comics_logger.info(f"Found {len(series_chapters)} series to process.")
|
||||
|
||||
for series_name, chapters in series_chapters.items():
|
||||
comics_logger.info(f"Processing series: {series_name}")
|
||||
# Ensure paths use forward slashes for Samba operations
|
||||
final_series_folder_path = f"{self.comics_dest_root}/{series_name}".replace("\", "/")
|
||||
|
||||
try:
|
||||
self.samba_manager.create_directory(final_series_folder_path)
|
||||
comics_logger.info(f"Created directory: {final_series_folder_path}")
|
||||
except OperationFailure as e:
|
||||
# Specific check for FILE_OBJECT_NAME_COLLISION if directory already exists
|
||||
if "FILE_OBJECT_NAME_COLLISION" in str(e):
|
||||
comics_logger.info(f"Directory {final_series_folder_path} already exists. Skipping creation.")
|
||||
else:
|
||||
comics_logger.error(f"Could not create directory {final_series_folder_path}: {e}")
|
||||
|
||||
for chapter_item in chapters:
|
||||
# Sanitize original path coming from list_path
|
||||
original_path = chapter_item["path"].replace("\", "/")
|
||||
original_filename = chapter_item["name"]
|
||||
|
||||
path_to_move = ""
|
||||
filename_to_move = ""
|
||||
|
||||
if original_filename.lower().endswith('.zip'):
|
||||
cbz_filename = os.path.splitext(original_filename)[0] + '.cbz'
|
||||
temp_cbz_path = f"{self.comics_root}/{cbz_filename}".replace("\", "/")
|
||||
|
||||
try:
|
||||
self.samba_manager.rename_file(original_path, temp_cbz_path)
|
||||
path_to_move = temp_cbz_path
|
||||
filename_to_move = cbz_filename
|
||||
comics_logger.info(f"Renamed {original_filename} to {cbz_filename}")
|
||||
except OperationFailure as e:
|
||||
comics_logger.error(f"Failed to rename ZIP {original_filename} to CBZ: {e}")
|
||||
continue # Skip to next chapter if rename fails
|
||||
|
||||
elif original_filename.lower().endswith('.cbz'):
|
||||
# If it's already a .cbz, it's ready to be moved
|
||||
path_to_move = original_path
|
||||
filename_to_move = original_filename
|
||||
else:
|
||||
comics_logger.warning(f"Skipping non-zip/cbz file: {original_filename}")
|
||||
continue
|
||||
|
||||
# Final destination path for the CBZ file
|
||||
final_cbz_path = f"{final_series_folder_path}/{filename_to_move}".replace("\", "/")
|
||||
|
||||
try:
|
||||
self.samba_manager.rename_file(path_to_move, final_cbz_path)
|
||||
comics_logger.info(f"Moved {filename_to_move} to {final_series_folder_path}")
|
||||
except OperationFailure as e:
|
||||
comics_logger.error(f"Failed to move {filename_to_move} to {final_series_folder_path}: {e}")
|
||||
|
||||
except Exception as e:
|
||||
comics_logger.error(f"An unexpected error occurred during comics organization: {e}", exc_info=True)
|
||||
|
||||
finally:
|
||||
comics_logger.info("Comics organization process finished.")
|
||||
self.samba_manager.close()
|
||||
|
||||
return
|
||||
Reference in New Issue
Block a user