| 1 | #!/usr/bin/env python3 |
| 2 | """ |
| 3 | PPT Master - Source Converter Batch Helpers |
| 4 | |
| 5 | Share explicit multi-file and directory expansion logic across source_to_md |
| 6 | backend converters. |
| 7 | |
| 8 | Usage: |
| 9 | Imported by scripts/source_to_md/*_to_md.py |
| 10 | |
| 11 | Examples: |
| 12 | run_path_batch(["docs"], {".docx"}, None, convert_one) |
| 13 | |
| 14 | Dependencies: |
| 15 | None |
| 16 | """ |
| 17 | |
| 18 | from __future__ import annotations |
| 19 | |
| 20 | import sys |
| 21 | from collections.abc import Callable |
| 22 | from pathlib import Path |
| 23 | |
| 24 | |
| 25 | ConvertOne = Callable[[Path, Path], bool] |
| 26 | IsSupportedFile = Callable[[Path], bool] |
| 27 | |
| 28 | |
| 29 | def _print_status(message: str) -> None: |
| 30 | print(message, file=sys.stderr) |
| 31 | |
| 32 | |
| 33 | def expand_directory_inputs( |
| 34 | inputs: list[str], |
| 35 | is_supported_file: IsSupportedFile, |
| 36 | is_external_ref: Callable[[str], bool] | None = None, |
| 37 | ) -> tuple[list[str], list[str], bool]: |
| 38 | """Expand non-recursive directory inputs while preserving other items.""" |
| 39 | expanded: list[str] = [] |
| 40 | errors: list[str] = [] |
| 41 | saw_directory = False |
| 42 | is_external = is_external_ref or (lambda _item: False) |
| 43 | |
| 44 | for item in inputs: |
| 45 | if is_external(item): |
| 46 | expanded.append(item) |
| 47 | continue |
| 48 | path = Path(item) |
| 49 | if path.is_dir(): |
| 50 | saw_directory = True |
| 51 | matches = sorted( |
| 52 | child for child in path.iterdir() |
| 53 | if child.is_file() and is_supported_file(child) |
| 54 | ) |
| 55 | if matches: |
| 56 | expanded.extend(str(match) for match in matches) |
| 57 | else: |
| 58 | errors.append(f"{item}: no supported files found") |
| 59 | continue |
| 60 | expanded.append(item) |
| 61 | |
| 62 | return expanded, errors, saw_directory |
| 63 | |
| 64 | |
| 65 | def _output_key(path: Path) -> Path: |
| 66 | return path.resolve(strict=False) |
| 67 | |
| 68 | |
| 69 | def unique_output_path(output_dir: Path, stem: str, used_outputs: set[Path]) -> Path: |
| 70 | """Return an in-run unique Markdown path without consulting the filesystem.""" |
| 71 | base = stem or "output" |
| 72 | candidate = output_dir / f"{base}.md" |
| 73 | suffix = 2 |
| 74 | while _output_key(candidate) in used_outputs: |
| 75 | candidate = output_dir / f"{base}_{suffix}.md" |
| 76 | suffix += 1 |
| 77 | used_outputs.add(_output_key(candidate)) |
| 78 | return candidate |
| 79 | |
| 80 | |
| 81 | def _output_for( |
| 82 | source: Path, |
| 83 | output_arg: str | None, |
| 84 | batch_mode: bool, |
| 85 | used_outputs: set[Path], |
| 86 | ) -> Path: |
| 87 | if output_arg and batch_mode: |
| 88 | return unique_output_path(Path(output_arg), source.stem, used_outputs) |
| 89 | if output_arg: |
| 90 | return Path(output_arg) |
| 91 | return source.with_suffix(".md") |
| 92 | |
| 93 | |
| 94 | def run_path_batch( |
| 95 | inputs: list[str], |
| 96 | supported_suffixes: set[str], |
| 97 | output_arg: str | None, |
| 98 | convert_one: ConvertOne, |
| 99 | ) -> int: |
| 100 | """Run one converter across explicit files and non-recursive directories.""" |
| 101 | expanded, expansion_errors, saw_directory = expand_directory_inputs( |
| 102 | inputs, |
| 103 | lambda path: path.suffix.lower() in supported_suffixes, |
| 104 | ) |
| 105 | sources = [Path(item) for item in expanded] |
| 106 | batch_mode = saw_directory or len(sources) > 1 |
| 107 | |
| 108 | if output_arg and batch_mode: |
| 109 | output_dir = Path(output_arg) |
| 110 | if output_dir.exists() and not output_dir.is_dir(): |
| 111 | _print_status(f"[ERROR] Batch output path is not a directory: {output_arg}") |
| 112 | return 1 |
| 113 | output_dir.mkdir(parents=True, exist_ok=True) |
| 114 | |
| 115 | success_count = 0 |
| 116 | failed: list[str] = [] |
| 117 | skipped: list[str] = list(expansion_errors) |
| 118 | used_outputs: set[Path] = set() |
| 119 | |
| 120 | for source in sources: |
| 121 | output = _output_for(source, output_arg, batch_mode, used_outputs) |
| 122 | if batch_mode: |
| 123 | _print_status(f"\n==> {source}") |
| 124 | try: |
| 125 | converted = convert_one(source, output) |
| 126 | sys.stdout.flush() |
| 127 | if converted: |
| 128 | success_count += 1 |
| 129 | else: |
| 130 | failed.append(str(source)) |
| 131 | except Exception as exc: |
| 132 | sys.stdout.flush() |
| 133 | failed.append(f"{source}: {exc}") |
| 134 | _print_status(f"[ERROR] {source}: {exc}") |
| 135 | |
| 136 | if batch_mode: |
| 137 | sys.stdout.flush() |
| 138 | _print_status(f"\n[Done] Success: {success_count}/{len(sources)}, Failed: {len(failed)}") |
| 139 | if skipped: |
| 140 | _print_status("\n[Skipped directories]:") |
| 141 | for item in skipped: |
| 142 | _print_status(f" - {item}") |
| 143 | if failed: |
| 144 | _print_status("\n[Failed inputs]:") |
| 145 | for item in failed: |
| 146 | _print_status(f" - {item}") |
| 147 | |
| 148 | if not sources: |
| 149 | return 1 |
| 150 | return 0 if not failed and not skipped else 1 |
| 151 |