MarkItDownを使ってPythonでExcelをMarkdownに変換

pip install "markitdown[all]"

pip install "markitdown[xlsx,xls]"

pip install "markitdown[docx]"

pip install "markitdown[pptx]"

pip install "markitdown[pdf]"

# -*- coding: utf-8 -*-
"""
Office / PDF ファイルを Markdown に一括変換するスクリプト。

【処理の流れ】
  1. INPUT_DIR 配下の対象ファイル（xlsx, docx, pdf 等）を検索
  2. MarkItDown で各ファイルを Markdown テキストに変換
  3. 個別 .md を OUTPUT_DIR/individual/ に必ず保存
  4. ENABLE_MERGE=True の場合、複数ファイルを結合して _Merge_Source_01.md 等も作成

【実行前の準備】
  pip install 'markitdown[all]'

【出力例】
  output_md/
    individual/          … 元ファイル1件ごとの .md（フォルダ構造を維持）
    _Merge_Source_01.md  … 結合ファイル（ENABLE_MERGE=True のとき）
  logs/
    convert_md_20260614.log  … 実行ログ（LOG_DIR / LOG_TIMESTAMP で変更）
"""

import logging
from datetime import datetime
from pathlib import Path
from typing import Optional

from markitdown import MarkItDown

# ==========================================
# ▼ ここだけ編集すれば OK（環境設定）
# ==========================================

# 変換したいファイルが入っているフォルダ（サブフォルダも再帰的に検索）
INPUT_DIR = r"C:\path\to\your\input_files"

# 変換結果を保存するフォルダ（無ければ自動作成）
OUTPUT_DIR = r"C:\path\to\your\output_md"

# 変換対象の拡張子（小文字・大文字どちらでも可）
TARGET_EXTENSIONS = ['.xlsx', '.xls', '.docx', '.pptx', '.pdf']

# --- マージ（結合）設定 ---
# 個別 .md は ENABLE_MERGE の値に関わらず必ず保存される。
# この設定は「結合ファイルも作るか」だけを決める。
ENABLE_MERGE = True  # True: _Merge_Source_XX.md も作る / False: 個別 .md のみ

# 結合ファイル名の構成（例: _Merge_Source_01.md）
MERGE_NAME = "_Merge"      # ファイル名の先頭部分
MERGE_PREFIX = "_Source"   # 連番の前に付ける部分

# --- 1つの結合ファイルあたりの上限 ---
# 文字数 OR バイト数のどちらかが上限を超えたら、次の結合ファイルへ分割する。
# （両方の条件を満たす必要がある = AND ではなく、超えた時点で分割）

# 文字数上限（NotebookLM は「約50万語」が目安。日本語は文字数 ≒ 語数）
MAX_CHAR_LIMIT = 200000

# バイト数上限（UTF-8 で保存したときのファイルサイズ。2 MiB = 2 × 1024 × 1024）
MAX_BYTES_LIMIT = 2 * 1024 * 1024

# --- ログ設定 ---
# ログファイルの保存先フォルダ（無ければ自動作成）
LOG_DIR = r"C:\path\to\your\logs"

# ログファイル名の接頭辞（拡張子 .log は自動付与）
# 例: convert_md.log / convert_md_20260614.log
LOG_PREFIX = "convert_md"

# ログファイル名のタイムスタンプ（接尾辞）
#   "none"     … なし           … convert_md.log
#   "day"      … _yyyymmdd      … convert_md_20260614.log
#   "datetime" … _yyyymmddhhmmsssss … convert_md_20260614153045123.log（ミリ秒3桁）
LOG_TIMESTAMP = "day"

# ==========================================

LOG_FORMAT = "%(asctime)s [%(levelname)s] %(message)s"
LOG_DATE_FORMAT = "%H:%M:%S"

logger = logging.getLogger(__name__)


def resolve_log_timestamp_suffix() -> str:
    """LOG_TIMESTAMP に応じたファイル名接尾辞を返す。"""
    pattern = (LOG_TIMESTAMP or "none").strip().lower()

    if pattern == "none":
        return ""

    if pattern == "day":
        return f"_{datetime.now():%Y%m%d}"

    if pattern == "datetime":
        now = datetime.now()
        millis = now.microsecond // 1000
        return f"_{now:%Y%m%d%H%M%S}{millis:03d}"

    return ""


def resolve_log_file_path() -> Path:
    """ログファイルのフルパスを決定する。"""
    suffix = resolve_log_timestamp_suffix()
    return Path(LOG_DIR) / f"{LOG_PREFIX}{suffix}.log"


def setup_logging() -> Path:
    """
    コンソールとファイルの両方へログを出力する。

    戻り値: 作成したログファイルのパス
    """
    log_path = resolve_log_file_path()
    log_path.parent.mkdir(parents=True, exist_ok=True)

    formatter = logging.Formatter(LOG_FORMAT, datefmt=LOG_DATE_FORMAT)
    root = logging.getLogger()
    root.setLevel(logging.INFO)
    root.handlers.clear()

    console_handler = logging.StreamHandler()
    console_handler.setFormatter(formatter)
    root.addHandler(console_handler)

    file_handler = logging.FileHandler(log_path, encoding="utf-8")
    file_handler.setFormatter(formatter)
    root.addHandler(file_handler)

    valid_timestamps = {"none", "day", "datetime"}
    if (LOG_TIMESTAMP or "none").strip().lower() not in valid_timestamps:
        logger.warning(
            f"不明な LOG_TIMESTAMP '{LOG_TIMESTAMP}'。"
            f"有効な値: none / day / datetime。タイムスタンプなしで続行します。"
        )

    return log_path


# ==========================================
# 以降は処理本体（通常は編集不要）
# ==========================================


def utf8_byte_len(text: str) -> int:
    """テキストを UTF-8 で保存したときのバイト数を返す。"""
    return len(text.encode("utf-8"))


def build_individual_output_path(output_path: Path, input_root: Path, file_path: Path) -> Path:
    """
    個別 .md の保存パスを決める。

    入力フォルダからの相対パスを維持するため、
    サブフォルダに同名ファイルがあっても上書きしない。

    例: input/reports/a.xlsx → output/individual/reports/a.md
    """
    relative = file_path.relative_to(input_root)
    return output_path / "individual" / relative.with_suffix(".md")


def build_source_header(file_path: Path, part: Optional[int] = None) -> str:
    """
    結合ファイル内で「どの元ファイルから来たか」を示すヘッダーを作る。

    NotebookLM 等がソースを識別しやすくなるよう、各ブロックの先頭に挿入する。
    1ファイルが上限超過で分割された場合は part 番号を付ける。
    """
    if part is None:
        title = file_path.name
    else:
        title = f"{file_path.name} (part {part})"
    return f"---\n# Source File: {title}\n---\n\n"


def split_text_by_limits(text: str, max_chars: int, max_bytes: int) -> list[str]:
    """
    長すぎるテキストを、文字数・バイト数の両方の上限内に収まる塊に分割する。

    分割位置は可能な限り空行（段落の区切り）を優先し、
    文章の途中で切れにくくする。
    """
    # そのまま収まるなら分割不要
    if len(text) <= max_chars and utf8_byte_len(text) <= max_bytes:
        return [text]

    chunks: list[str] = []
    remaining = text

    while remaining:
        # 残りが上限内に収まれば終了
        if len(remaining) <= max_chars and utf8_byte_len(remaining) <= max_bytes:
            chunks.append(remaining)
            break

        # まず文字数上限で切り位置の候補を決める
        split_at = min(max_chars, len(remaining))

        # バイト数上限を超える場合は切り位置を手前にずらす
        while split_at > 0 and utf8_byte_len(remaining[:split_at]) > max_bytes:
            split_at -= 1

        if split_at <= 0:
            # 1文字でもバイト上限を超える（絵文字等）→ 分割不能
            logger.error("   -> 1文字でもバイト上限を超えるため分割不能")
            return []

        # 段落境界（空行）があれば、そこで切る
        para_at = remaining.rfind("\n\n", 0, split_at)
        if para_at > 0:
            split_at = para_at

        chunks.append(remaining[:split_at].rstrip())
        remaining = remaining[split_at:].lstrip("\n")

    return chunks


def exceeds_merge_limits(text: str) -> bool:
    """結合ファイル1件分として、文字数またはバイト数が上限を超えているか。"""
    return len(text) > MAX_CHAR_LIMIT or utf8_byte_len(text) > MAX_BYTES_LIMIT


def append_to_merge_buffer(
    output_path: Path,
    file_path: Path,
    text_content: str,
    buffer: dict,
) -> None:
    """
    変換済みテキストをマージ用バッファに追加する。

    buffer には現在溜まっているテキストと、文字数・バイト数のカウントを保持。
    上限を超えそうになったら、先にバッファの内容をファイルに書き出してから
    新しい結合ファイル用にバッファを空にする。
    """
    # ヘッダー分を除いた本文に使える上限を計算
    header = build_source_header(file_path)
    max_body_chars = MAX_CHAR_LIMIT - len(header)
    max_body_bytes = MAX_BYTES_LIMIT - utf8_byte_len(header)

    if max_body_chars <= 0 or max_body_bytes <= 0:
        raise ValueError("上限設定が小さすぎます。ヘッダー分の余裕を確保してください。")

    # 1ファイルが大きすぎる場合は、本文部分だけ先に分割
    parts = split_text_by_limits(text_content, max_body_chars, max_body_bytes)
    if not parts:
        buffer["skipped"] += 1
        return

    if len(parts) > 1:
        logger.warning(
            f"   -> 1ファイルが上限超過のため {len(parts)} 分割: {file_path.name} "
            f"({len(text_content):,} 文字 / {utf8_byte_len(text_content):,} バイト)"
        )

    for index, part in enumerate(parts, start=1):
        # 分割された場合は part 番号付きヘッダーを使う
        part_header = header if len(parts) == 1 else build_source_header(file_path, index)
        append_text = part_header + part
        append_chars = len(append_text)
        append_bytes = utf8_byte_len(append_text)

        # 分割後でも単体で上限超過ならスキップ
        if exceeds_merge_limits(append_text):
            logger.error(
                f"   -> 分割後も上限超過のためスキップ: {file_path.name} "
                f"(part {index}, {append_chars:,} 文字 / {append_bytes:,} バイト)"
            )
            buffer["skipped"] += 1
            continue

        # バッファに追加した場合の合計（ブロック間は "\n\n" = 2文字/2バイト）
        sep_chars = 2 if buffer["char_count"] > 0 else 0
        sep_bytes = 2 if buffer["byte_count"] > 0 else 0
        next_chars = buffer["char_count"] + sep_chars + append_chars
        next_bytes = buffer["byte_count"] + sep_bytes + append_bytes

        # 追加すると上限超過 → 今のバッファを書き出してから新規バッファへ
        if buffer["char_count"] > 0 and (
            next_chars > MAX_CHAR_LIMIT or next_bytes > MAX_BYTES_LIMIT
        ):
            write_merged_file(output_path, buffer["text"], buffer["index"])
            buffer["index"] += 1
            buffer["text"] = ""
            buffer["char_count"] = 0
            buffer["byte_count"] = 0
            sep_chars = 0
            sep_bytes = 0
            next_chars = append_chars
            next_bytes = append_bytes

        # バッファへ追記
        if buffer["char_count"] == 0:
            buffer["text"] = append_text
            buffer["char_count"] = append_chars
            buffer["byte_count"] = append_bytes
        else:
            buffer["text"] += "\n\n" + append_text
            buffer["char_count"] = next_chars
            buffer["byte_count"] = next_bytes


def write_merged_file(out_dir: Path, content: str, index: int) -> None:
    """溜まったテキストを結合 .md ファイルとして保存する。"""
    filename = f"{MERGE_NAME}_{MERGE_PREFIX}_{index:02d}.md"
    filepath = out_dir / filename

    with open(filepath, "w", encoding="utf-8") as f:
        f.write(content)

    logger.info(
        f"マージファイル作成: {filename} "
        f"({len(content):,} 文字 / {utf8_byte_len(content):,} バイト)"
    )


def main() -> None:
    """メイン処理: 検索 → 変換 → 個別保存 → （任意）マージ。"""
    log_path = setup_logging()
    logger.info(f"ログ出力: {log_path}")

    # MarkItDown 変換エンジンの初期化
    md = MarkItDown()

    input_path = Path(INPUT_DIR)
    output_path = Path(OUTPUT_DIR)

    # 入力フォルダの存在確認
    if not input_path.is_dir():
        logger.error(f"入力フォルダが存在しません: {INPUT_DIR}")
        return

    # 出力フォルダを作成（親フォルダも含めて）
    output_path.mkdir(parents=True, exist_ok=True)

    # 拡張子比較用（小文字セット）
    target_exts = {ext.lower() for ext in TARGET_EXTENSIONS}

    # マージ用バッファ（結合テキストを一時的に溜める）
    merge_buffer = {
        "text": "",          # 結合中の Markdown 本文
        "char_count": 0,     # 現在の文字数
        "byte_count": 0,       # 現在の UTF-8 バイト数
        "index": 1,            # 結合ファイルの連番（01, 02, ...）
        "skipped": 0,          # 上限超過等でスキップした件数
    }

    # 処理結果の集計
    stats = {
        "success": 0,   # 変換成功
        "failed": 0,    # 変換失敗
        "empty": 0,     # 変換結果が空
    }

    logger.info(f"処理を開始します。入力フォルダ: {INPUT_DIR}")
    logger.info(
        f"マージ上限: {MAX_CHAR_LIMIT:,} 文字 / {MAX_BYTES_LIMIT:,} バイト "
        f"({MAX_BYTES_LIMIT / (1024 * 1024):.1f} MiB)"
    )

    # 対象ファイルを収集（サブフォルダ含む、~$ 一時ファイル除外、名前順でソート）
    files_to_process = sorted(
        (
            f for f in input_path.rglob("*")
            if f.is_file()
            and f.suffix.lower() in target_exts
            and not f.name.startswith("~$")  # Excel 等の一時ファイル
        ),
        key=lambda p: str(p.relative_to(input_path)).lower(),
    )

    if not files_to_process:
        logger.warning("対象ファイルが見つかりませんでした。INPUT_DIR と TARGET_EXTENSIONS を確認してください。")
        return

    logger.info(f"対象ファイル: {len(files_to_process)} 件")

    # --- ファイルごとの変換ループ ---
    for file_path in files_to_process:
        relative_name = file_path.relative_to(input_path)
        logger.info(f"変換中: {relative_name}")

        try:
            # MarkItDown で Markdown テキストに変換
            result = md.convert(str(file_path))
            text_content = result.text_content or ""

            # 空の変換結果はスキップ（スキャン PDF 等で起きることがある）
            if not text_content.strip():
                logger.warning(f"   -> 変換結果が空のためスキップ: {relative_name}")
                stats["empty"] += 1
                continue

            # 個別 .md を必ず保存
            ind_out = build_individual_output_path(output_path, input_path, file_path)
            ind_out.parent.mkdir(parents=True, exist_ok=True)
            ind_out.write_text(text_content, encoding="utf-8")
            logger.info(f"   -> 個別保存完了: {ind_out.relative_to(output_path)}")

            # マージ有効時は結合バッファにも追加
            if ENABLE_MERGE:
                append_to_merge_buffer(output_path, file_path, text_content, merge_buffer)

            stats["success"] += 1

        except Exception as e:
            # 1件失敗しても処理は続行
            stats["failed"] += 1
            logger.error(f"エラー発生 ({relative_name}): {e}")

    # ループ終了後、バッファに残っていれば最後の結合ファイルとして書き出す
    if ENABLE_MERGE and merge_buffer["char_count"] > 0:
        write_merged_file(output_path, merge_buffer["text"], merge_buffer["index"])

    # 処理結果サマリーを表示
    summary = (
        f"処理完了 — 成功: {stats['success']}, "
        f"空スキップ: {stats['empty']}, "
        f"失敗: {stats['failed']}"
    )
    if ENABLE_MERGE:
        summary += f", マージ分割スキップ: {merge_buffer['skipped']}"
    logger.info(summary)


if __name__ == "__main__":
    main()

python convert_md.py