python script for bulk pdf conversion. even faster.

conver_notebooks_cli.py

# /// script
# requires-python = ">=3.11"
# dependencies = [
#     "click",
#     "nbconvert[webpdf]",
# ]
# ///

import os
import subprocess
import click
from concurrent.futures import ThreadPoolExecutor, ProcessPoolExecutor, as_completed

# 이 함수는 최상위 레벨에 있어야 피클링 가능합니다.
def convert_single_ipynb_to_pdf(ipynb_filepath, output_filepath_base):
    """
    단일 .ipynb 파일을 PDF로 변환하는 함수.
    병렬 처리를 위해 독립적인 함수로 분리.
    """
    filename = os.path.basename(ipynb_filepath)
    expected_pdf_filename = f"{os.path.basename(output_filepath_base)}.pdf"

    try:
        # 서브 프로세스 내에서 click.echo는 stdout/stderr 충돌을 일으킬 수 있으므로
        # 더 안전하게 표준 에러 스트림으로 출력하거나, 로깅 모듈을 사용하는 것이 좋습니다.
        # 여기서는 click.echo(..., err=True)를 유지합니다.
        click.echo(f"  '{filename}' 변환 중...", err=True)

        command = [
            "jupyter",
            "nbconvert",
            "--to",
            "webpdf",
            "--allow-chromium-download",
            ipynb_filepath,
            "--output",
            output_filepath_base
        ]
        
        # subprocess.run의 capture_output=True는 자식 프로세스의 stdout/stderr를 캡처합니다.
        # 실제 출력을 보려면 capture_output=False로 설정하거나, Popen을 직접 사용해야 할 수 있습니다.
        subprocess.run(command, check=True, capture_output=True, text=True)
        click.echo(f"  성공: '{filename}' -> '{expected_pdf_filename}'", err=True)
        return True, filename
    except subprocess.CalledProcessError as e:
        click.echo(f"  오류: '{filename}' 변환 실패.", err=True)
        click.echo(f"    명령어: {' '.join(e.cmd)}", err=True)
        click.echo(f"    표준 오류: {e.stderr.strip()}", err=True)
        return False, filename
    except FileNotFoundError:
        click.echo(f"  오류: 'jupyter' 명령어를 찾을 수 없습니다. Jupyter가 설치되어 있고 PATH에 추가되었는지 확인하세요.", err=True)
        return False, filename
    except Exception as e:
        click.echo(f"  예기치 않은 오류 발생: {e}", err=True)
        return False, filename


@click.command()
@click.option('--input_folder', '-i',
              type=click.Path(exists=True, file_okay=False, dir_okay=True, readable=True, resolve_path=True),
              required=True,
              help='IPython Notebook (.ipynb) 파일이 있는 입력 폴더 경로.')
@click.option('--output_folder', '-o',
              type=click.Path(file_okay=False, dir_okay=True, writable=True, resolve_path=True),
              default=None,
              help='변환된 PDF 파일을 저장할 출력 폴더 경로. 지정하지 않으면 입력 폴더에 저장됩니다.')
@click.option('--max_workers', '-w',
              type=int,
              default=os.cpu_count(), # 시스템의 CPU 코어 수만큼 기본값 설정
              help='동시에 실행할 변환 프로세스(또는 스레드)의 최대 개수.')
@click.option('--use_threads', is_flag=True,
              help='멀티프로세싱 대신 멀티스레딩을 사용합니다. CPU 바운드 작업에는 일반적으로 멀티프로세싱이 더 좋습니다.')
def convert_ipynb_to_pdf_cli(input_folder, output_folder, max_workers, use_threads):
    """
    지정된 입력 폴더의 모든 .ipynb 파일을 PDF로 변환하여 출력 폴더에 저장합니다.
    병렬 처리를 지원합니다.
    """
    if output_folder is None:
        output_folder = input_folder
    
    os.makedirs(output_folder, exist_ok=True)

    click.echo(f"'{input_folder}'에서 .ipynb 파일을 찾아 '{output_folder}'(으)로 PDF 변환을 시작합니다...\n")
    click.echo(f"최대 {max_workers}개의 {'스레드' if use_threads else '프로세스'}를 사용하여 변환합니다.\n")

    notebook_tasks = [] # (ipynb_filepath, output_filepath_base) 튜플 리스트
    for filename in os.listdir(input_folder):
        if filename.endswith(".ipynb"):
            ipynb_filepath = os.path.join(input_folder, filename)
            base_filename_without_ext = os.path.splitext(filename)[0]
            output_filepath_base = os.path.join(output_folder, base_filename_without_ext)
            notebook_tasks.append((ipynb_filepath, output_filepath_base))

    if not notebook_tasks:
        click.echo("변환할 .ipynb 파일이 없습니다.")
        return

    converted_count = 0
    failed_count = 0

    Executor = ThreadPoolExecutor if use_threads else ProcessPoolExecutor

    with Executor(max_workers=max_workers) as executor:
        # 각 작업을 제출하고 Future 객체를 리스트에 저장
        futures = {executor.submit(convert_single_ipynb_to_pdf, ipynb_path, output_path): (ipynb_path, output_path)
                   for ipynb_path, output_path in notebook_tasks}

        # 완료된 순서대로 결과 처리
        for future in as_completed(futures):
            # 원본 작업 정보를 가져옴 (여기서는 딱히 필요 없지만, 복잡한 경우 유용)
            # original_ipynb_path, original_output_path = futures[future] 
            
            try:
                success, filename = future.result() # 작업 결과 가져오기
                if success:
                    converted_count += 1
                else:
                    failed_count += 1
            except Exception as exc:
                # 여기서 예외가 발생하면 convert_single_ipynb_to_pdf 함수 자체에서 처리 못한 예외입니다.
                click.echo(f"예외가 발생하여 작업이 완료되지 못했습니다: {exc}", err=True)
                # 이 경우에도 실패로 간주할 수 있습니다.
                failed_count += 1
    
    click.echo(f"\n--- 변환 요약 ---")
    click.echo(f"총 성공: {converted_count}개")
    click.echo(f"총 실패: {failed_count}개")
    click.echo(f"모든 .ipynb 파일 변환 시도 완료.")

if __name__ == "__main__":
    # Windows에서는 if __name__ == "__main__": 블록 안에 모든 실행 코드를 넣어야
    # 멀티프로세싱이 올바르게 작동합니다.
    convert_ipynb_to_pdf_cli()
    
# 10개의 프로세스 사용 uv run convert_notebooks_cli.py -i ./my_notebooks -o ./converted_pdfs -w 10