241 lines
8.6 KiB
Python
241 lines
8.6 KiB
Python
# main.py
|
|
|
|
import argparse
|
|
import datetime
|
|
import os
|
|
import shutil
|
|
from core.scanner import find_comic_files
|
|
from core.pipeline import Pipeline
|
|
from core.summary import SummaryCollector
|
|
from core.collision import CollisionPolicy
|
|
|
|
_COL_W = 30
|
|
_SEP = "─" * 44
|
|
_BOLD = "\033[1m"
|
|
_DIM = "\033[2m"
|
|
_RST = "\033[0m"
|
|
|
|
|
|
def _human_size(size: int) -> str:
|
|
if size >= 1_073_741_824:
|
|
return f"{size / 1_073_741_824:.1f} GB"
|
|
if size >= 1_048_576:
|
|
return f"{size / 1_048_576:.1f} MB"
|
|
return f"{size / 1024:.1f} KB"
|
|
|
|
|
|
def _print_file_header(path: str, i: int = 0, total: int = 0) -> None:
|
|
name = os.path.basename(path)
|
|
size_str = _human_size(os.path.getsize(path))
|
|
counter = f"{_DIM}[{i}/{total}]{_RST} " if i and total else ""
|
|
term_w = shutil.get_terminal_size(fallback=(80, 24)).columns
|
|
print(f"\n{'─' * term_w}")
|
|
print(f" {counter}{_BOLD}{name}{_RST} {_DIM}({size_str}){_RST}")
|
|
|
|
|
|
def _print_preview(step: str, preview: dict, formato: str) -> None:
|
|
fmt = formato.upper()
|
|
|
|
if step == "clean":
|
|
if preview["items"]:
|
|
print("Ficheros a eliminar:")
|
|
for item in preview["items"]:
|
|
print(f" - {item}")
|
|
print(f"Formato final del archivo: {fmt}")
|
|
|
|
elif step == "flatten":
|
|
flatten_files = preview["flatten_files"]
|
|
n = len(flatten_files)
|
|
display = flatten_files[:10] if n > 10 else flatten_files
|
|
col_w = max((len(src) for src, _ in display), default=0) + 2
|
|
print(f"Aplanar estructura ({n} ficheros):")
|
|
for src, dst in display:
|
|
print(f" {src:<{col_w}} → {dst}")
|
|
if n > 10:
|
|
print(f" ... y {n - 10} más")
|
|
print(f"Formato final del archivo: {fmt}")
|
|
|
|
elif step == "normalize_pages":
|
|
renames = preview["renames"]
|
|
n = len(renames)
|
|
print(f"Renombrado de páginas ({n} páginas):")
|
|
print(f" {'Nombre actual':<{_COL_W}} → Nombre final")
|
|
print(f" {_SEP}")
|
|
display = renames[:10] if n > 20 else renames
|
|
for orig, final in display:
|
|
print(f" {orig:<{_COL_W}} {final}")
|
|
if n > 20:
|
|
print(f" ... y {n - 10} más")
|
|
print(f"Formato final del archivo: {fmt}")
|
|
|
|
elif step == "normalize_case":
|
|
renames = preview["renames"]
|
|
mode_label = "minúsculas" if preview["mode"] == "lower" else "mayúsculas"
|
|
n = len(renames)
|
|
print(f"Normalización de extensiones a {mode_label} ({n} ficheros):")
|
|
print(f" {'Nombre actual':<{_COL_W}} → Nombre final")
|
|
print(f" {_SEP}")
|
|
display = renames[:10] if n > 20 else renames
|
|
for orig, final in display:
|
|
print(f" {orig:<{_COL_W}} {final}")
|
|
if n > 20:
|
|
print(f" ... y {n - 10} más")
|
|
print(f"Formato final del archivo: {formato.upper()}")
|
|
|
|
elif step in ("normalize_images", "convert_images"):
|
|
conversions = preview["conversions"]
|
|
target_ext = preview["target_ext"].lstrip(".")
|
|
lossless = " (sin pérdida)" if target_ext.lower() == "png" else ""
|
|
print(f"Conversión de imágenes a {target_ext.upper()}{lossless}:")
|
|
print(f" {'Imagen actual':<{_COL_W}} → Imagen final")
|
|
print(f" {_SEP}")
|
|
for orig, final in conversions:
|
|
print(f" {orig:<{_COL_W}} {final}")
|
|
print(f"Formato final del archivo: {fmt}")
|
|
|
|
|
|
def _save_log(collector, script_dir):
|
|
log_dir = os.path.join(script_dir, "validaciones")
|
|
os.makedirs(log_dir, exist_ok=True)
|
|
ts = datetime.datetime.now().strftime("%Y-%m-%d_%H-%M-%S")
|
|
log_path = os.path.join(log_dir, f"{ts}.log")
|
|
with open(log_path, "w", encoding="utf-8") as fh:
|
|
fh.write(collector.full_log())
|
|
return log_path
|
|
|
|
|
|
def make_confirm_fn(args):
|
|
def confirm_fn(step: str, preview: dict) -> bool:
|
|
if step == "convert":
|
|
return True # No destructivo, no necesita confirmación
|
|
print()
|
|
_print_preview(step, preview, args.formato)
|
|
answer = input("¿Aplicar? [s/N] ").strip().lower()
|
|
return answer in ("s", "si", "sí", "y", "yes")
|
|
return confirm_fn
|
|
|
|
|
|
def parse_args():
|
|
parser = argparse.ArgumentParser(description="Gestor de cómics CBR/CBZ")
|
|
|
|
parser.add_argument("--ruta", required=True)
|
|
parser.add_argument("--listar", action="store_true")
|
|
parser.add_argument("--validar", action="store_true")
|
|
parser.add_argument("--limpiar", action="store_true")
|
|
parser.add_argument("--aplanar", action="store_true")
|
|
parser.add_argument("--convertir", action="store_true")
|
|
parser.add_argument("--estandarizar", action="store_true")
|
|
parser.add_argument("--formato", choices=["cbz", "cbr"], default="cbz")
|
|
parser.add_argument("--renumerar", action="store_true")
|
|
parser.add_argument("--uniformizar-imagenes", action="store_true")
|
|
parser.add_argument("--convertir-imagenes", action="store_true")
|
|
parser.add_argument("--formato-imagen", choices=["jpg", "png", "webp"], default="png")
|
|
parser.add_argument("--normalizar-case", action="store_true", dest="normalizar_case")
|
|
parser.add_argument(
|
|
"--modo-case",
|
|
choices=["lower", "upper"],
|
|
default="lower",
|
|
dest="modo_case",
|
|
)
|
|
|
|
parser.add_argument("--no-preguntar", action="store_true")
|
|
parser.add_argument(
|
|
"--al-modificar",
|
|
choices=["backup", "borrar"],
|
|
default="backup",
|
|
dest="al_modificar",
|
|
)
|
|
|
|
return parser.parse_args()
|
|
|
|
|
|
def main():
|
|
args = parse_args()
|
|
|
|
if not os.path.isdir(args.ruta):
|
|
print(f"Error: la ruta '{args.ruta}' no existe o no es un directorio.", flush=True)
|
|
return
|
|
|
|
comic_files = find_comic_files(args.ruta)
|
|
|
|
if args.listar:
|
|
for f in comic_files:
|
|
print(f)
|
|
ndirs = nfiles = 0
|
|
for _, _, files in os.walk(args.ruta):
|
|
ndirs += 1
|
|
nfiles += len(files)
|
|
print(f"\n Carpetas analizadas : {ndirs}")
|
|
print(f" Ficheros analizados : {nfiles}")
|
|
print(f" Cómics encontrados : {len(comic_files)}")
|
|
|
|
if args.validar:
|
|
pipeline = Pipeline(steps=[])
|
|
collector = SummaryCollector()
|
|
total = len(comic_files)
|
|
try:
|
|
for i, f in enumerate(comic_files, 1):
|
|
size_str = _human_size(os.path.getsize(f))
|
|
name = os.path.basename(f)
|
|
term_w = shutil.get_terminal_size(fallback=(80, 24)).columns
|
|
line = f" [{i}/{total}] {name} ({size_str})"
|
|
print(f"\r{line[:term_w]:<{term_w}}", end="", flush=True)
|
|
result = pipeline.run(f)
|
|
collector.add(result)
|
|
except KeyboardInterrupt:
|
|
print("\n (interrumpido)")
|
|
finally:
|
|
if collector._results:
|
|
print()
|
|
print(collector.render())
|
|
log_path = _save_log(collector, os.path.dirname(os.path.abspath(__file__)))
|
|
print(f" Log completo: {log_path}")
|
|
return
|
|
|
|
# --- Construir steps ---
|
|
steps = []
|
|
if args.limpiar or args.estandarizar:
|
|
steps.append("clean")
|
|
if args.aplanar or args.estandarizar:
|
|
steps.append("flatten")
|
|
if args.renumerar or args.estandarizar:
|
|
steps.append("normalize_pages")
|
|
if args.uniformizar_imagenes or args.estandarizar:
|
|
steps.append("normalize_images")
|
|
if args.convertir_imagenes:
|
|
steps.append("convert_images")
|
|
if args.normalizar_case or args.estandarizar:
|
|
steps.append("normalize_case")
|
|
if args.convertir or args.estandarizar:
|
|
steps.append("convert")
|
|
|
|
if steps:
|
|
confirm_fn = None if args.no_preguntar else make_confirm_fn(args)
|
|
collision = CollisionPolicy.BACKUP if args.al_modificar == "backup" else CollisionPolicy.ABORT
|
|
pipeline = Pipeline(
|
|
steps=steps,
|
|
desired_format=args.formato,
|
|
desired_image_format="." + args.formato_imagen,
|
|
collision_policy=collision,
|
|
case_mode=args.modo_case,
|
|
)
|
|
collector = SummaryCollector()
|
|
try:
|
|
total = len(comic_files)
|
|
for i, f in enumerate(comic_files, 1):
|
|
_print_file_header(f, i, total)
|
|
result = pipeline.run(f, confirm_fn=confirm_fn)
|
|
print(result.summary())
|
|
collector.add(result)
|
|
except KeyboardInterrupt:
|
|
print("\n (interrumpido)")
|
|
finally:
|
|
if collector._results:
|
|
print(f"\n{collector.render()}")
|
|
log_path = _save_log(collector, os.path.dirname(os.path.abspath(__file__)))
|
|
print(f" Log completo: {log_path}")
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|