converter.convert のソースコード

"""
ファイル変換ユーティリティモジュール。

このモジュールは、ファイルシステムを走査し、登録されたコンバータを使用して
指定されたファイルを別の形式に変換する機能を提供します。
Pandocなどの外部ツールや他のPythonライブラリと連携し、
様々なファイル形式間の変換をサポートします。

:doc:`convert_usage`
"""
import os
import sys
import fnmatch
import argparse
from dataclasses import dataclass, field
from typing import Callable, Dict, List, Optional, Tuple, Any
import tempfile
import shutil


OUTPUT_MODE_FILE = "file"
OUTPUT_MODE_DIR = "dir"


[ドキュメント] @dataclass class ConverterSpec: """ ファイル変換の具体的な仕様を定義するデータクラス。 入力ファイルの拡張子、出力ファイルの拡張子、変換を実行するCallableオブジェクト、 および変換に関する追加情報(説明、一時ファイル無視ルール、出力モードなど)を カプセル化します。 :param input_ext: 入力ファイルの拡張子(例: ".docx", "pptx")。 :type input_ext: str :param output_ext: 出力ファイルの拡張子(例: ".pdf", ".md")。 :type output_ext: str :param converter: 実際の変換処理を行うCallableオブジェクト。 通常は `input_path`, `output_path`, `**kwargs` を引数に取ります。 :type converter: Callable[..., Any] :param description: このコンバータの簡単な説明。デフォルトは空文字列。 :type description: str :param ignore_temp_prefixes: 変換をスキップすべき一時ファイル名の接頭辞のタプル。 例えば、Wordの一時ファイル"~$"など。 :type ignore_temp_prefixes: Tuple[str, ...] :param output_mode: 変換結果の出力モード。 `OUTPUT_MODE_FILE` (単一ファイル) または `OUTPUT_MODE_DIR` (ディレクトリ)。 :type output_mode: str :param options: 変換関数に渡される追加のキーワード引数を含む辞書。 :type options: Dict[str, Any] """ input_ext: str output_ext: str converter: Callable[..., Any] description: str = "" ignore_temp_prefixes: Tuple[str, ...] = () output_mode: str = OUTPUT_MODE_FILE options: Dict[str, Any] = field(default_factory=dict) def __post_init__(self): """ データクラスの初期化後に呼び出され、拡張子の正規化と出力モードの検証を行います。 :raises ValueError: output_mode が 'file' または 'dir' のいずれでもない場合。 """ self.input_ext = normalize_ext(self.input_ext) self.output_ext = normalize_ext(self.output_ext) if self.output_mode not in (OUTPUT_MODE_FILE, OUTPUT_MODE_DIR): raise ValueError(f"Invalid output_mode: {self.output_mode}")
[ドキュメント] class ConverterRegistry: """ ConverterSpecオブジェクトを管理するレジストリクラス。 利用可能な変換仕様を登録、取得、一覧表示し、モジュールのインポート中に発生したエラーも記録します。 これにより、プログラムは実行時に利用可能な変換を動的に認識できます。 """ def __init__(self): """ ConverterRegistryの新しいインスタンスを初期化します。 内部で変換仕様を保存するための辞書と、モジュールのインポートエラーを記録するための辞書を初期化します。 """ self._specs: Dict[Tuple[str, str], ConverterSpec] = {} self._import_errors: Dict[str, Exception] = {}
[ドキュメント] def register(self, spec: ConverterSpec): """ 指定されたConverterSpecをレジストリに登録します。 入力拡張子と出力拡張子の組み合わせを一意のキーとして使用します。 :param spec: 登録するConverterSpecオブジェクト。 :type spec: ConverterSpec """ key = (spec.input_ext, spec.output_ext) self._specs[key] = spec
[ドキュメント] def set_import_error(self, module_name: str, error: Exception): """ 特定のモジュールのインポートエラーを記録します。 これにより、ユーザーはどの変換機能が利用できないか、その理由とともに確認できます。 :param module_name: エラーが発生したモジュール名。 :type module_name: str :param error: 発生したExceptionオブジェクト。 :type error: Exception """ self._import_errors[module_name] = error
[ドキュメント] def get(self, input_ext: str, output_ext: str) -> Optional[ConverterSpec]: """ 指定された入力拡張子と出力拡張子に対応するConverterSpecを取得します。 拡張子は検索前に正規化されます。 :param input_ext: 検索する入力ファイルの拡張子。 :type input_ext: str :param output_ext: 検索する出力ファイルの拡張子。 :type output_ext: str :returns: 対応するConverterSpecオブジェクト、または見つからない場合はNone。 :rtype: Optional[ConverterSpec] """ return self._specs.get((normalize_ext(input_ext), normalize_ext(output_ext)))
[ドキュメント] def has(self, input_ext: str, output_ext: str) -> bool: """ 指定された入力拡張子と出力拡張子に対応するConverterSpecが存在するかどうかを確認します。 :param input_ext: 検索する入力ファイルの拡張子。 :type input_ext: str :param output_ext: 検索する出力ファイルの拡張子。 :type output_ext: str :returns: ConverterSpecが存在する場合はTrue、それ以外の場合はFalse。 :rtype: bool """ return self.get(input_ext, output_ext) is not None
[ドキュメント] def list_specs(self): """ 登録されているすべてのConverterSpecオブジェクトのリストを返します。 :returns: 登録されているConverterSpecオブジェクトのリスト。 :rtype: List[ConverterSpec] """ return list(self._specs.values())
[ドキュメント] def list_output_exts(self): """ 登録されているすべての出力拡張子の一覧を重複なくソートして返します。 :returns: 利用可能な出力拡張子のソート済みリスト。 :rtype: List[str] """ return sorted({spec.output_ext for spec in self._specs.values()})
[ドキュメント] def print_available_converters(self): """ 利用可能なコンバータと、インポートに失敗したモジュールをコンソールに表示します。 コンバータは出力拡張子、入力拡張子、出力モードの順でソートされて表示されます。 インポートエラーがある場合は、そのモジュール名とエラーメッセージも表示されます。 """ print("Available converters:") for spec in sorted(self._specs.values(), key=lambda s: (s.output_ext, s.input_ext, s.output_mode)): desc = f" ({spec.description})" if spec.description else "" mode = f" [{spec.output_mode}]" if spec.output_mode != OUTPUT_MODE_FILE else "" print(f" {spec.input_ext} -> {spec.output_ext}{mode}{desc}") if self._import_errors: print("\nUnavailable modules:") for module_name, err in self._import_errors.items(): print(f" {module_name}: {err}")
[ドキュメント] def normalize_ext(ext: str) -> str: """ ファイル拡張子を正規化します。 拡張子の前後の空白を削除し、小文字に変換し、必要に応じて先頭に"."を追加します。 :param ext: 正規化するファイル拡張子の文字列(例: "pdf", ".PDF ")。 :type ext: str :returns: 正規化されたファイル拡張子の文字列(例: ".pdf")。 :rtype: str """ ext = ext.strip().lower() if not ext.startswith("."): ext = "." + ext return ext
[ドキュメント] def build_output_path(input_path: str, output_ext: str, output_mode: str = OUTPUT_MODE_FILE) -> str: """ 入力パスと出力拡張子、出力モードに基づいて出力ファイルのパスを構築します。 `output_mode` が `OUTPUT_MODE_FILE` の場合は、入力ファイル名のベースに出力拡張子を付けたパスを返します。 `OUTPUT_MODE_DIR` の場合は、出力拡張子の末尾に"s"を追加したディレクトリのようなパスを返します (例: ".pdf" -> ".pdfs")。 :param input_path: 入力ファイルのパス。 :type input_path: str :param output_ext: 目的の出力拡張子(例: ".pdf")。 :type output_ext: str :param output_mode: 出力モード(`OUTPUT_MODE_FILE` または `OUTPUT_MODE_DIR`)。デフォルトは `OUTPUT_MODE_FILE`。 :type output_mode: str :returns: 構築された出力パスの文字列。 :rtype: str """ stem = os.path.splitext(input_path)[0] if output_mode == OUTPUT_MODE_DIR: return stem + normalize_ext(output_ext) + "s" return stem + normalize_ext(output_ext)
[ドキュメント] def should_ignore_file(filename: str, spec: ConverterSpec) -> bool: """ 指定されたファイル名が、特定のコンバータ仕様で無視すべき一時ファイルであるかを判定します。 `spec.ignore_temp_prefixes` に含まれるいずれかの接頭辞でファイル名が始まる場合、 そのファイルは無視されるべき一時ファイルと見なされます。 :param filename: チェックするファイル名。 :type filename: str :param spec: 変換仕様 (`ConverterSpec`) オブジェクト。 :type spec: ConverterSpec :returns: ファイルが無視すべき一時ファイルである場合はTrue、それ以外の場合はFalse。 :rtype: bool """ return any(filename.startswith(prefix) for prefix in spec.ignore_temp_prefixes)
[ドキュメント] def path_exists_for_mode(path: str, output_mode: str) -> bool: """ 指定されたパスが出力モードに応じて存在するかどうかをチェックします。 `OUTPUT_MODE_DIR` の場合、パスがディレクトリであり、かつその中に何らかのコンテンツが存在するか (ディレクトリが空でないか)を確認します。 `OUTPUT_MODE_FILE` の場合は、単に指定されたパスが存在するかどうかを確認します。 :param path: チェックするパス。 :type path: str :param output_mode: 出力モード(`OUTPUT_MODE_FILE` または `OUTPUT_MODE_DIR`)。 :type output_mode: str :returns: パスが出力モードの条件を満たして存在する場合はTrue、それ以外の場合はFalse。 :rtype: bool """ if output_mode == OUTPUT_MODE_DIR: return os.path.isdir(path) and any(True for _ in os.scandir(path)) return os.path.exists(path)
[ドキュメント] def get_latest_mtime(path: str, output_mode: str) -> Optional[float]: """ 指定されたパス(ファイルまたはディレクトリ)の最終更新時刻を取得します。 `OUTPUT_MODE_FILE` の場合、ファイルの最終更新時刻を返します。 `OUTPUT_MODE_DIR` の場合、指定されたディレクトリとその中のすべてのファイル、サブディレクトリの中から 最も新しい最終更新時刻を再帰的に探して返します。 パスが存在しない場合や、`OUTPUT_MODE_DIR` で指定されたディレクトリが空の場合はNoneを返します。 :param path: 最終更新時刻を取得するパス。 :type path: str :param output_mode: 出力モード(`OUTPUT_MODE_FILE` または `OUTPUT_MODE_DIR`)。 :type output_mode: str :returns: 最終更新時刻のUNIXタイムスタンプ (float) 、またはパスが存在しないか空の場合はNone。 :rtype: Optional[float] """ if output_mode == OUTPUT_MODE_FILE: if not os.path.exists(path): return None return os.path.getmtime(path) if not os.path.isdir(path): return None latest = os.path.getmtime(path) found_any = False for dirpath, dirnames, filenames in os.walk(path): for name in dirnames + filenames: found_any = True candidate = os.path.join(dirpath, name) try: latest = max(latest, os.path.getmtime(candidate)) except OSError: pass return latest if found_any else None
[ドキュメント] def should_convert(input_path: str, output_path: str, output_mode: str, update: bool, overwrite: bool) -> Tuple[bool, str]: """ ファイルを変換する必要があるかどうかを判断します。 以下のロジックに基づいて変換の要否を決定します。 1. `overwrite` がTrueの場合、常に変換を行います。 2. 出力ファイルまたはディレクトリが存在しない場合、変換を行います。 3. `update` がFalseで出力が存在する場合、変換をスキップします。 4. `update` がTrueで出力が存在する場合、入力ファイルの最終更新時刻が出力ファイルの最終更新時刻よりも新しい場合に変換を行います。 :param input_path: 入力ファイルのパス。 :type input_path: str :param output_path: 出力ファイルのパス。 :type output_path: str :param output_mode: 出力モード(`OUTPUT_MODE_FILE` または `OUTPUT_MODE_DIR`)。 :type output_mode: str :param update: 出力ファイルが存在し、入力ファイルよりも新しい場合にスキップするかどうか (True: 更新日時をチェックして必要なら変換)。 :type update: bool :param overwrite: 出力ファイルの存在やタイムスタンプに関わらず常に上書きするかどうか (True: 強制変換)。 :type overwrite: bool :returns: 変換すべき場合はTrueと理由、スキップすべき場合はFalseと理由のタプル。 :rtype: Tuple[bool, str] """ if overwrite: return True, "overwrite=1" exists = path_exists_for_mode(output_path, output_mode) if not exists: return True, "output does not exist" if not update: return False, "output exists and update=0" output_mtime = get_latest_mtime(output_path, output_mode) if output_mtime is None: return True, "output exists but is empty/incomplete" input_mtime = os.path.getmtime(input_path) if input_mtime > output_mtime: return True, "source is newer than output" return False, "up to date"
[ドキュメント] def parse_target_patterns(target_text: Optional[str]) -> Optional[List[str]]: """ セミコロンで区切られたターゲットパターン文字列を解析し、リストを返します。 入力文字列がNoneまたは空文字列の場合、Noneを返します。 それ以外の場合、文字列をセミコロンで分割し、各パターンから空白を除去したリストを返します。 :param target_text: セミコロンで区切られたファイル名パターン文字列(例: "*.pptx;*.docx")。 :type target_text: Optional[str] :returns: 解析されたパターン文字列のリスト、または入力がNone/空の場合はNone。 :rtype: Optional[List[str]] """ if not target_text: return None patterns = [p.strip() for p in target_text.split(";") if p.strip()] return patterns or None
[ドキュメント] def matches_target(filename: str, target_patterns: Optional[List[str]]) -> bool: """ ファイル名が指定されたターゲットパターンリストのいずれかに一致するかどうかを判定します。 `target_patterns` がNoneまたは空の場合、すべてのファイル名に一致すると見なします。 それ以外の場合、fnmatch.fnmatch を使用して、ファイル名とパターンを大文字小文字を区別せずに比較します。 :param filename: チェックするファイル名。 :type filename: str :param target_patterns: ファイル名パターンのリスト(例: ["*.pptx", "*.docx"])。 :type target_patterns: Optional[List[str]] :returns: ファイル名がいずれかのパターンに一致する場合はTrue、それ以外の場合はFalse。 :rtype: bool """ if not target_patterns: return True lower_name = filename.lower() for pattern in target_patterns: if fnmatch.fnmatch(lower_name, pattern.lower()): return True return False
# ---------- importable wrappers ----------
[ドキュメント] def wrap_simple(converter: Callable[[str, Optional[str]], Any]) -> Callable[..., bool]: """ 入力パスと出力パスのみを引数にとる単純なファイル変換関数を、 標準のラッパー形式(`input_path`, `output_path`, `**kwargs` を受け取る)に変換します。 ラップされた関数は、元の変換関数を実行し、結果がNoneでない場合にTrueを返します。 :param converter: 入力パスと出力パスを受け取り変換を行うCallableオブジェクト。 オプションとして`None`を許容する出力パスを想定しています。 :type converter: Callable[[str, Optional[str]], Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされた変換関数。 :param input_path: 入力ファイルのパス。 :type input_path: str :param output_path: 出力ファイルのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ result = converter(input_path, output_path) return bool(result is not None) return _wrapped
[ドキュメント] def wrap_image_to_dir(converter: Callable[..., Any], rename_func: Optional[Callable[[str], Any]] = None) -> Callable[..., bool]: """ 入力ファイルを画像ディレクトリに変換する関数をラップします。 このラッパーは、出力ディレクトリが存在しない場合に作成し、 指定されたコンバータを実行して画像を生成します。 必要に応じて、生成された画像のファイル名を変更するための`rename_func`を呼び出すこともできます。 :param converter: 入力ファイルを画像ディレクトリに変換するCallableオブジェクト。 通常は `input_path`, `output_path`, `output_format` を引数に取ります。 :type converter: Callable[..., Any] :param rename_func: 変換後に生成された画像ファイル名を変更するためのCallableオブジェクト。 引数として出力ディレクトリのパスを受け取ります。オプション。 :type rename_func: Optional[Callable[[str], Any]] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされた画像ディレクトリ変換関数。 :param input_path: 入力ファイルのパス。 :type input_path: str :param output_path: 出力ディレクトリのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (コンバータに渡されます)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ os.makedirs(output_path, exist_ok=True) converter(input_path, output_path, "png") if rename_func is not None: rename_func(output_path) return True return _wrapped
[ドキュメント] def wrap_docx_to_png(docx_to_pdf: Callable[[str, str], Any], pdf_to_images: Callable[..., Any]) -> Callable[..., bool]: """ Word (docx) ファイルをPNG画像に変換するパイプラインをラップします。 このラッパーは、最初にdocxファイルを一時的なPDFに変換し、次にそのPDFを 指定された出力ディレクトリにPNG画像として変換します。 :param docx_to_pdf: docxファイルをPDFに変換するCallableオブジェクト。 `input_path`, `output_path` を引数に取ります。 :type docx_to_pdf: Callable[[str, str], Any] :param pdf_to_images: PDFをPNG画像に変換するCallableオブジェクト。 通常は `pdf_path`, `output_dir`, `output_format` を引数に取ります。 :type pdf_to_images: Callable[..., Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたWord (docx) からPNG画像への変換関数。 :param input_path: 入力Wordファイルのパス。 :type input_path: str :param output_path: 出力PNG画像が保存されるディレクトリのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ os.makedirs(output_path, exist_ok=True) tmp_pdf = os.path.splitext(input_path)[0] + ".pdf" docx_to_pdf(input_path, tmp_pdf) pdf_to_images(tmp_pdf, output_path, "png") return True return _wrapped
[ドキュメント] def wrap_md_with_images(convert_func: Callable[..., Any]) -> Callable[..., bool]: """ 画像を伴うMarkdown変換関数をラップします。 このラッパーは、変換関数がMarkdownファイル本体と、それに付随する画像が保存される ディレクトリパスの両方を引数として受け取ることを想定しています。 変換関数がNoneを返すか、出力ファイルが存在すれば成功と見なします。 :param convert_func: Markdownと関連画像を変換するCallableオブジェクト。 通常は `input_path`, `output_md_path`, `output_image_dir_path` を引数に取ります。 :type convert_func: Callable[..., Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたMarkdown (画像含む) 変換関数。 :param input_path: 入力ファイルのパス。 :type input_path: str :param output_path: 出力Markdownファイルのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ image_dir = os.path.splitext(output_path)[0] + ".images" result = convert_func(input_path, output_path, image_dir) return bool(result is not None or os.path.exists(output_path)) return _wrapped
[ドキュメント] def wrap_pdf2md(convert_func: Callable[[str, Optional[str]], Any]) -> Callable[..., bool]: """ PDFをMarkdownに変換する関数を、標準のラッパー形式に変換します。 ラップされた関数は、元の変換関数を実行し、結果がNoneでないか、 または出力ファイルが存在すれば成功と判断します。 :param convert_func: PDFをMarkdownに変換するCallableオブジェクト。 `input_path`, `output_path` を引数に取ります。 オプションとして`None`を許容する出力パスを想定しています。 :type convert_func: Callable[[str, Optional[str]], Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたPDFからMarkdownへの変換関数。 :param input_path: 入力PDFファイルのパス。 :type input_path: str :param output_path: 出力Markdownファイルのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ result = convert_func(input_path, output_path) return bool(result is not None or os.path.exists(output_path)) return _wrapped
[ドキュメント] def wrap_pandoc(target: str, default_template: Optional[str] = None) -> Callable[..., bool]: """ Pandocを利用したMarkdown変換関数をラップします。 このラッパーは、`pandoc` モジュールを動的にインポートし、 `convert_md` 関数を呼び出してMarkdownファイルを指定されたターゲット形式に変換します。 `pandoc_path`、`template`、`toc`、`css` などの様々なPandocオプションを キーワード引数として受け渡しできるようにします。 :param target: Pandocのターゲット形式(例: "docx", "pptx", "html")。 :type target: str :param default_template: 変換に使用するデフォルトのテンプレートパス。オプション。 :type default_template: Optional[str] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたPandoc変換関数。 :param input_path: 入力Markdownファイルのパス。 :type input_path: str :param output_path: 出力ファイルのパス。 :type output_path: str :param kwargs: Pandoc変換に渡される追加のキーワード引数。 例: `pandoc_path`, `template`, `toc`, `css`, `mathml`, `no_yaml`, `verbose`, `smart_conversion`。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ from pandoc import convert_md, find_template pandoc_path = kwargs.get("pandoc_path") or "pandoc" template = kwargs.get("template") or default_template if template: template = find_template(template) return bool(convert_md( infile_md=input_path, target=target, pandoc_path=pandoc_path, outfile=output_path, template=template, toc=bool(kwargs.get("toc", 0)), css=kwargs.get("css"), mathml=bool(kwargs.get("mathml", False)), no_yaml=bool(kwargs.get("no_yaml", False)), verbose=bool(kwargs.get("verbose", False)), smart_conversion=bool(kwargs.get("smart_conversion", False)), )) return _wrapped
[ドキュメント] def wrap_ipynb_json_to_md(convert_func: Callable[..., Any]) -> Callable[..., bool]: """ Jupyter Notebook (.ipynb) またはJSONファイルをMarkdownに変換する関数をラップします。 ラップされた関数は、元の変換関数を実行し、結果がNoneでないか、 または出力ファイルが存在すれば成功と判断します。 :param convert_func: .ipynbまたはJSONをMarkdownに変換するCallableオブジェクト。 `input_path`, `output_path` を引数に取ります。 :type convert_func: Callable[..., Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたJupyter Notebook/JSONからMarkdownへの変換関数。 :param input_path: 入力ファイル (.ipynbまたは.json) のパス。 :type input_path: str :param output_path: 出力Markdownファイルのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ result = convert_func(input_path, output_path) return bool(result is not None or os.path.exists(output_path)) return _wrapped
[ドキュメント] def wrap_ipynb_json_to_pdf(ipynb_or_json_to_md: Callable[..., Any], md_to_pdf: Callable[..., Any]) -> Callable[..., bool]: """ Jupyter Notebook (.ipynb) またはJSONファイルをPDFに変換するパイプラインをラップします。 このラッパーは、最初にJupyter NotebookまたはJSONファイルを一時的なMarkdownに変換し、 次にそのMarkdownファイルをPDFに変換します。一時ディレクトリは処理後に削除されます。 :param ipynb_or_json_to_md: .ipynbまたはJSONをMarkdownに変換するCallableオブジェクト。 `input_path`, `output_md_path` を引数に取ります。 :type ipynb_or_json_to_md: Callable[..., Any] :param md_to_pdf: MarkdownをPDFに変換するCallableオブジェクト。 `input_md_path`, `output_pdf_path` を引数に取ります。 :type md_to_pdf: Callable[..., Any] :returns: 標準の変換関数シグネチャ (`input_path`, `output_path`, `**kwargs`) に適合するラッパー関数。 :rtype: Callable[..., bool] """ def _wrapped(input_path: str, output_path: str, **kwargs) -> bool: """ ラップされたJupyter Notebook/JSONからPDFへのパイプライン変換関数。 :param input_path: 入力ファイル (.ipynbまたは.json) のパス。 :type input_path: str :param output_path: 出力PDFファイルのパス。 :type output_path: str :param kwargs: その他のキーワード引数 (未使用)。 :type kwargs: Any :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ tmp_dir = tempfile.mkdtemp(prefix="convert_pipeline_") tmp_md = os.path.join(tmp_dir, os.path.splitext(os.path.basename(input_path))[0] + ".md") try: md_result = ipynb_or_json_to_md(input_path, tmp_md) if md_result is False or not os.path.exists(tmp_md): return False pdf_result = md_to_pdf(tmp_md, output_path) return bool(pdf_result is not None or os.path.exists(output_path)) finally: shutil.rmtree(tmp_dir, ignore_errors=True) return _wrapped
# ---------- registry ----------
[ドキュメント] def safe_import_registry(registry: ConverterRegistry): """ 外部ライブラリを安全にインポートし、利用可能なコンバータをレジストリに登録します。 各変換ライブラリのインポートは`try-except`ブロックで囲まれており、 一部のライブラリが利用できない場合でもプログラム全体がクラッシュせず、 そのエラーがレジストリに記録されるようにします。 これにより、ユーザーはどの機能が利用可能または利用不可であるかを把握できます。 :param registry: コンバータの登録を行うConverterRegistryインスタンス。 :type registry: ConverterRegistry """ try: from docx2pdf import docx_to_pdf except Exception as e: registry.set_import_error("docx2pdf", e) else: registry.register(ConverterSpec( input_ext=".docx", output_ext=".pdf", converter=wrap_simple(docx_to_pdf), description="Word to PDF", )) try: from xlsx2pdf import xlsx_to_pdf except Exception as e: registry.set_import_error("xlsx2pdf", e) else: registry.register(ConverterSpec( input_ext=".xlsx", output_ext=".pdf", converter=wrap_simple(xlsx_to_pdf), description="Excel to PDF", ignore_temp_prefixes=("~$",), )) try: from pptx2pdf import pptx_to_pdf except Exception as e: registry.set_import_error("pptx2pdf", e) else: registry.register(ConverterSpec( input_ext=".pptx", output_ext=".pdf", converter=wrap_simple(pptx_to_pdf), description="PowerPoint to PDF", )) try: from pptx2pdf_with_notes_importable import pptx_to_pdf_with_notes except Exception as e: registry.set_import_error("pptx2pdf_with_notes_importable", e) else: registry.register(ConverterSpec( input_ext=".pptx", output_ext=".notes.pdf", converter=wrap_simple(pptx_to_pdf_with_notes), description="PowerPoint to PDF with speaker notes", )) try: from html2pdf_importable import html_to_pdf except Exception as e: registry.set_import_error("html2pdf_importable", e) else: registry.register(ConverterSpec( input_ext=".html", output_ext=".pdf", converter=wrap_simple(html_to_pdf), description="HTML to PDF", )) registry.register(ConverterSpec( input_ext=".htm", output_ext=".pdf", converter=wrap_simple(html_to_pdf), description="HTML to PDF", )) try: from md2pdf_importable import md_to_pdf except Exception as e: registry.set_import_error("md2pdf_importable", e) else: registry.register(ConverterSpec( input_ext=".md", output_ext=".pdf", converter=wrap_simple(md_to_pdf), description="Markdown to PDF", )) try: from txt2pdf_importable import txt_to_pdf except Exception as e: registry.set_import_error("txt2pdf_importable", e) else: registry.register(ConverterSpec( input_ext=".txt", output_ext=".pdf", converter=wrap_simple(txt_to_pdf), description="Text to PDF", )) try: from img2pdf_importable import image_to_pdf except Exception as e: registry.set_import_error("img2pdf_importable", e) else: for ext in (".png", ".jpg", ".jpeg", ".bmp", ".tif", ".tiff", ".webp"): registry.register(ConverterSpec( input_ext=ext, output_ext=".pdf", converter=wrap_simple(image_to_pdf), description="Image to PDF", )) # Markdown converters with formula/image extraction try: import docx2md except Exception as e: registry.set_import_error("docx2md", e) else: registry.register(ConverterSpec( input_ext=".docx", output_ext=".md", converter=wrap_md_with_images(docx2md.convert), description="Word to Markdown with equations/images", )) try: import pptx2md2 except Exception as e: registry.set_import_error("pptx2md2", e) else: registry.register(ConverterSpec( input_ext=".pptx", output_ext=".md", converter=wrap_md_with_images(pptx2md2.extract_content_to_markdown), description="PowerPoint to Markdown with equations/images", )) try: import pdf2md except Exception as e: registry.set_import_error("pdf2md", e) else: registry.register(ConverterSpec( input_ext=".pdf", output_ext=".md", converter=wrap_pdf2md(pdf2md.convert), description="PDF to Markdown", )) try: import pptx2img except Exception as e: registry.set_import_error("pptx2img", e) else: registry.register(ConverterSpec( input_ext=".pptx", output_ext=".png", converter=wrap_image_to_dir(pptx2img.export_img, rename_func=getattr(pptx2img, "rename_img", None)), description="PowerPoint to PNG images", output_mode=OUTPUT_MODE_DIR, )) try: import docx2img except Exception as e: registry.set_import_error("docx2img", e) else: registry.register(ConverterSpec( input_ext=".docx", output_ext=".png", converter=wrap_docx_to_png(docx2img.convert_word_to_pdf, docx2img.convert_pdf_to_images), description="Word to PNG images", output_mode=OUTPUT_MODE_DIR, )) try: import pdf2img except Exception as e: registry.set_import_error("pdf2img", e) else: registry.register(ConverterSpec( input_ext=".pdf", output_ext=".png", converter=wrap_image_to_dir(pdf2img.convert_pdf_to_images), description="PDF to PNG images", output_mode=OUTPUT_MODE_DIR, )) try: import ipynb2md except Exception as e: registry.set_import_error("ipynb2md", e) else: registry.register(ConverterSpec( input_ext=".ipynb", output_ext=".md", converter=wrap_ipynb_json_to_md(ipynb2md.convert_ipynb_to_md), description="Jupyter Notebook to Markdown", )) registry.register(ConverterSpec( input_ext=".json", output_ext=".md", converter=wrap_ipynb_json_to_md(ipynb2md.convert_json_to_md), description="JSON to Markdown", )) try: from md2pdf_importable import md_to_pdf as md_to_pdf_for_pipeline except Exception: try: from md2pdf import md_to_pdf as md_to_pdf_for_pipeline except Exception as e: registry.set_import_error("md2pdf pipeline backend", e) else: registry.register(ConverterSpec( input_ext=".ipynb", output_ext=".pdf", converter=wrap_ipynb_json_to_pdf(ipynb2md.convert_ipynb_to_md, md_to_pdf_for_pipeline), description="Jupyter Notebook to PDF via Markdown pipeline", )) registry.register(ConverterSpec( input_ext=".json", output_ext=".pdf", converter=wrap_ipynb_json_to_pdf(ipynb2md.convert_json_to_md, md_to_pdf_for_pipeline), description="JSON to PDF via Markdown pipeline", )) else: registry.register(ConverterSpec( input_ext=".ipynb", output_ext=".pdf", converter=wrap_ipynb_json_to_pdf(ipynb2md.convert_ipynb_to_md, md_to_pdf_for_pipeline), description="Jupyter Notebook to PDF via Markdown pipeline", )) registry.register(ConverterSpec( input_ext=".json", output_ext=".pdf", converter=wrap_ipynb_json_to_pdf(ipynb2md.convert_json_to_md, md_to_pdf_for_pipeline), description="JSON to PDF via Markdown pipeline", )) # pandoc.py 由来の追加変換 try: import pandoc as pandoc_module _ = pandoc_module.find_pandoc _ = pandoc_module.convert_md except Exception as e: registry.set_import_error("pandoc", e) else: registry.register(ConverterSpec( input_ext=".md", output_ext=".docx", converter=wrap_pandoc("docx"), description="Markdown to Word via Pandoc", )) registry.register(ConverterSpec( input_ext=".md", output_ext=".pptx", converter=wrap_pandoc("pptx"), description="Markdown to PowerPoint via Pandoc", )) registry.register(ConverterSpec( input_ext=".md", output_ext=".html", converter=wrap_pandoc("html"), description="Markdown to HTML via Pandoc", ))
[ドキュメント] def convert_file( input_path: str, output_ext: str, registry: ConverterRegistry, update: bool = True, overwrite: bool = False, converter_kwargs: Optional[Dict[str, Any]] = None, ) -> bool: """ 単一のファイルを指定された出力形式に変換します。 入力パス、出力拡張子、レジストリに基づいて適切なコンバータを見つけ、 `update` または `overwrite` の設定に従って変換の要否を判断します。 一時ファイルと見なされるファイルはスキップされます。 :param input_path: 変換する入力ファイルのパス。 :type input_path: str :param output_ext: 目的の出力拡張子(例: ".pdf", ".md")。 :type output_ext: str :param registry: 利用可能なコンバータを保持するConverterRegistryインスタンス。 :type registry: ConverterRegistry :param update: 出力ファイルが存在し、入力ファイルよりも新しい場合にスキップするかどうか。デフォルトはTrue。 :type update: bool :param overwrite: 出力ファイルの存在やタイムスタンプに関わらず常に上書きするかどうか。デフォルトはFalse。 :type overwrite: bool :param converter_kwargs: 変換関数に渡す追加のキーワード引数の辞書。オプション。 :type converter_kwargs: Optional[Dict[str, Any]] :returns: 変換が成功した場合はTrue、それ以外の場合はFalse。 :rtype: bool """ input_path = os.path.abspath(input_path) input_ext = normalize_ext(os.path.splitext(input_path)[1]) converter_kwargs = converter_kwargs or {} spec = registry.get(input_ext, output_ext) if spec is None: print(f"Skipping unsupported file: '{input_path}'") return False filename = os.path.basename(input_path) if should_ignore_file(filename, spec): print(f"Skipping temporary file: '{input_path}'") return False output_path = build_output_path(input_path, output_ext, spec.output_mode) do_convert, reason = should_convert( input_path=input_path, output_path=output_path, output_mode=spec.output_mode, update=update, overwrite=overwrite, ) if do_convert: print(f"Converting: '{input_path}' -> '{output_path}' ({reason})") kwargs = dict(spec.options) kwargs.update(converter_kwargs) result = spec.converter(input_path, output_path, **kwargs) return bool(result) print(f"Skipping: '{input_path}' ({reason})") return False
[ドキュメント] def walk_and_convert( root_dir: str, output_ext: str, registry: ConverterRegistry, max_level: int = -1, update: bool = True, overwrite: bool = False, target_patterns: Optional[List[str]] = None, converter_kwargs: Optional[Dict[str, Any]] = None, ): """ 指定されたルートディレクトリを再帰的に走査し、条件に合うファイルを変換します。 `os.walk` を使用してディレクトリツリーを巡回し、各ファイルに対して `convert_file` を呼び出します。 最大走査深度、更新オプション、上書きオプション、およびファイル名パターンによるフィルタリングをサポートします。 変換プロセスに関する情報をコンソールに出力します。 :param root_dir: 走査を開始するルートディレクトリのパス。 :type root_dir: str :param output_ext: 目的の出力拡張子(例: ".pdf", ".md")。 :type output_ext: str :param registry: 利用可能なコンバータを保持するConverterRegistryインスタンス。 :type registry: ConverterRegistry :param max_level: 最大走査深度。-1は無制限。デフォルトは-1。 :type max_level: int :param update: 出力ファイルが存在し、入力ファイルよりも新しい場合にスキップするかどうか。デフォルトはTrue。 :type update: bool :param overwrite: 出力ファイルの存在やタイムスタンプに関わらず常に上書きするかどうか。デフォルトはFalse。 :type overwrite: bool :param target_patterns: 変換対象とするファイル名のfnmatchパターンリスト。オプション。 例: ["*.pptx", "*.docx"]。 :type target_patterns: Optional[List[str]] :param converter_kwargs: 変換関数に渡す追加のキーワード引数の辞書。オプション。 :type converter_kwargs: Optional[Dict[str, Any]] :returns: スキャンされたサポート対象ファイル数と実際に変換されたファイル数のタプル。 :rtype: Tuple[int, int] """ if not os.path.isdir(root_dir): print(f"Error: Root directory '{root_dir}' does not exist.") return 0, 0 root_dir_abs = os.path.abspath(root_dir) output_ext = normalize_ext(output_ext) print(f"Searching in '{root_dir_abs}'") print(f"Target output format: {output_ext}") print(f"Max level: {max_level}") print(f"update: {int(update)}") print(f"overwrite: {int(overwrite)}") if converter_kwargs: for key, value in sorted(converter_kwargs.items()): if value not in (None, ""): print(f"{key}: {value}") if target_patterns: print(f"Target patterns: {';'.join(target_patterns)}") else: print("Target patterns: (all matching input files)") print("") converted = 0 scanned = 0 for dirpath, dirnames, filenames in os.walk(root_dir_abs): current_level = dirpath.count(os.sep) - root_dir_abs.count(os.sep) if max_level != -1 and current_level >= max_level: del dirnames[:] for filename in filenames: if not matches_target(filename, target_patterns): continue input_path = os.path.join(dirpath, filename) input_ext = normalize_ext(os.path.splitext(filename)[1]) if registry.has(input_ext, output_ext): scanned += 1 if convert_file( input_path=input_path, output_ext=output_ext, registry=registry, update=update, overwrite=overwrite, converter_kwargs=converter_kwargs, ): converted += 1 print("\nConversion process finished.") print(f"Scanned supported files: {scanned}") print(f"Converted files: {converted}") return scanned, converted
[ドキュメント] def parse_args(): """ コマンドライン引数を解析し、`argparse.Namespace` オブジェクトとして返します。 この関数は、ファイル変換ツールが受け付ける様々なオプションを定義します。 これには、走査するルートディレクトリ、単一ファイル変換の指定、 目的の出力拡張子、最大走査深度、ファイル名パターンによるフィルタリング、 更新・上書きモード、およびPandoc関連のオプション(パス、テンプレート、CSS、目次など)が含まれます。 :returns: 解析されたコマンドライン引数を含むargparse.Namespaceオブジェクト。 :rtype: argparse.Namespace """ parser = argparse.ArgumentParser( description="Convert files in a directory tree using registered converters." ) parser.add_argument("root_dir", nargs="?", default=".", help="Root directory to scan") parser.add_argument("--infile", default="", help="Convert only this file and skip recursive search when specified") parser.add_argument("output_ext", nargs="?", default=".pdf", help="Target output extension, e.g. pdf or .notes.pdf") parser.add_argument("--max_level", nargs="?", default="-1", help="Maximum recursion depth (-1 means unlimited)") parser.add_argument("--target", default=None, help='Filename patterns separated by ";" , e.g. --target="*.pptx;*.docx"') parser.add_argument("--update", type=int, choices=[0, 1], default=1, help="If 0, skip when output already exists even if source is newer") parser.add_argument("--overwrite", type=int, choices=[0, 1], default=0, help="If 1, always convert regardless of timestamps") parser.add_argument("--pandoc_path", default=None, help="Path to pandoc executable for Pandoc-based conversions") parser.add_argument("--template_docx", default=None, help="Pandoc reference doc/template for md -> docx") parser.add_argument("--template_pptx", default=None, help="Pandoc reference doc/template for md -> pptx") parser.add_argument("--css", default=None, help="CSS for md -> html via Pandoc") parser.add_argument("--toc", type=int, choices=[0, 1], default=0, help="Enable table of contents for Pandoc docx output") parser.add_argument("--mathml", action="store_true", help="Use MathML for md -> html via Pandoc") parser.add_argument("--no_yaml", action="store_true", help="Pass no_yaml option to Pandoc conversion") parser.add_argument("--verbose", action="store_true", help="Verbose mode for Pandoc conversion") parser.add_argument("--smart_conversion", type=int, choices=[0, 1], default=0, help="Enable smart_conversion in pandoc.py") parser.add_argument("--list", action="store_true", help="List available converters and exit") parser.add_argument("--no-pause", action="store_true", help="Do not wait for ENTER before exit") return parser.parse_args()
[ドキュメント] def build_converter_kwargs(args, output_ext: str) -> Dict[str, Any]: """ コマンドライン引数からコンバータに渡すキーワード引数辞書を構築します。 特にPandoc関連のオプション(`pandoc_path`, `template`, `css`, `toc`, `mathml`, `no_yaml`, `verbose`, `smart_conversion`)を、 出力拡張子に基づいて適切に辞書に格納します。 :param args: コマンドライン引数を保持するargparse.Namespaceオブジェクト。 :type args: argparse.Namespace :param output_ext: 目的の出力拡張子。 :type output_ext: str :returns: コンバータに渡すキーワード引数の辞書。 :rtype: Dict[str, Any] """ output_ext = normalize_ext(output_ext) kwargs: Dict[str, Any] = {} if args.pandoc_path: kwargs["pandoc_path"] = args.pandoc_path if output_ext == ".docx" and args.template_docx: kwargs["template"] = args.template_docx elif output_ext == ".pptx" and args.template_pptx: kwargs["template"] = args.template_pptx elif output_ext == ".html" and args.css: kwargs["css"] = args.css kwargs["toc"] = bool(args.toc) kwargs["mathml"] = bool(args.mathml) kwargs["no_yaml"] = bool(args.no_yaml) kwargs["verbose"] = bool(args.verbose) kwargs["smart_conversion"] = bool(args.smart_conversion) return kwargs
[ドキュメント] def main(): """ プログラムのメインエントリポイント。コマンドライン引数を処理し、ファイル変換を実行します。 コマンドライン引数を解析し、`ConverterRegistry` を初期化して利用可能なコンバータを登録します。 引数に応じて、利用可能なコンバータの一覧表示、単一ファイルの変換、 または指定されたディレクトリツリーの走査と変換を実行します。 エラーが発生した場合は適切なメッセージを出力し、非ゼロの終了コードを返します。 :returns: プログラムの終了コード。成功時は0、エラー時は1。 :rtype: int """ args = parse_args() try: max_level = int(args.max_level) except ValueError: print(f"Error: Invalid max_level '{args.max_level}'.") return 1 registry = ConverterRegistry() safe_import_registry(registry) if args.list: registry.print_available_converters() return 0 output_ext = normalize_ext(args.output_ext) available_outputs = registry.list_output_exts() if output_ext not in available_outputs: print(f"Error: No converters registered for output format '{output_ext}'.") if available_outputs: print("Registered output formats:", ", ".join(available_outputs)) registry.print_available_converters() return 1 target_patterns = parse_target_patterns(args.target) converter_kwargs = build_converter_kwargs(args, output_ext) if args.infile: infile = os.path.abspath(args.infile) if not os.path.exists(infile): print(f"Error: infile '{args.infile}' does not exist.") return 1 print(f"Single-file mode: {infile}") converted = convert_file( input_path=infile, output_ext=output_ext, registry=registry, update=bool(args.update), overwrite=bool(args.overwrite), converter_kwargs=converter_kwargs, ) print("\nConversion process finished.") print("Scanned supported files: 1" if registry.has(os.path.splitext(infile)[1], output_ext) else "Scanned supported files: 0") print(f"Converted files: {1 if converted else 0}") return 0 if converted or registry.has(os.path.splitext(infile)[1], output_ext) else 1 walk_and_convert( root_dir=args.root_dir, output_ext=output_ext, registry=registry, max_level=max_level, update=bool(args.update), overwrite=bool(args.overwrite), target_patterns=target_patterns, converter_kwargs=converter_kwargs, ) return 0
if __name__ == "__main__": rc = main() print("\nProgram execution completed.") if "--no-pause" not in sys.argv: input("\nPress ENTER to terminate>>\n") raise SystemExit(rc)