|
| 1 | +import argparse |
| 2 | +import json |
| 3 | +import sys |
| 4 | +from typing import Sequence, Optional |
| 5 | + |
| 6 | +from jupyter_notebook_parser import JupyterNotebookParser, JupyterNotebookRewriter |
| 7 | + |
| 8 | +import format_def_indent._helper as helper |
| 9 | +from format_def_indent._base_fixer import BaseFixer |
| 10 | + |
| 11 | + |
| 12 | +class JupyterNotebookFixer(BaseFixer): |
| 13 | + def __init__(self, path: str, cli_args: argparse.Namespace) -> None: |
| 14 | + super().__init__(path=path, cli_args=cli_args) |
| 15 | + |
| 16 | + def fix_one_file(self, filename: str) -> int: |
| 17 | + try: |
| 18 | + parsed = JupyterNotebookParser(filename) |
| 19 | + rewriter = JupyterNotebookRewriter(parsed_notebook=parsed) |
| 20 | + code_cells = parsed.get_code_cells() |
| 21 | + code_cell_indices = parsed.get_code_cell_indices() |
| 22 | + code_cell_sources = parsed.get_code_cell_sources() |
| 23 | + except Exception as exc: |
| 24 | + print(f'Error reading {filename}: {str(exc)}', file=sys.stderr) |
| 25 | + return 1 |
| 26 | + else: |
| 27 | + ret_val = 0 |
| 28 | + assert len(code_cells) == len(code_cell_indices) |
| 29 | + assert len(code_cells) == len(code_cell_sources) |
| 30 | + |
| 31 | + for i in range(len(code_cells)): |
| 32 | + index: int = code_cell_indices[i] |
| 33 | + source: str = code_cell_sources[i] |
| 34 | + fixed: str = helper.fix_src(source_code=source) |
| 35 | + |
| 36 | + if fixed != source: |
| 37 | + ret_val = 1 |
| 38 | + rewriter.replace_source_in_code_cell( |
| 39 | + index=index, |
| 40 | + new_source=fixed, |
| 41 | + ) |
| 42 | + |
| 43 | + if ret_val == 1: |
| 44 | + print(f'Rewriting {filename}', file=sys.stderr) |
| 45 | + with open(filename, 'w') as fp: |
| 46 | + json.dump(parsed.notebook_content, fp, indent=1) |
| 47 | + # Jupyter notebooks (.ipynb) always ends with a new line |
| 48 | + # but json.dump does not. |
| 49 | + fp.write('\n') |
| 50 | + |
| 51 | + return 0 if self.cli_args.exit_zero_even_if_changed else ret_val |
| 52 | + |
| 53 | + |
| 54 | +def main(argv: Optional[Sequence[str]] = None) -> int: |
| 55 | + parser = argparse.ArgumentParser() |
| 56 | + parser.add_argument('paths', nargs='*') |
| 57 | + parser.add_argument('--exit-zero-even-if-changed', action='store_true') |
| 58 | + args = parser.parse_args(argv) |
| 59 | + |
| 60 | + ret = 0 |
| 61 | + for path in args.paths: |
| 62 | + fixer = JupyterNotebookFixer(path=path, cli_args=args) |
| 63 | + ret |= fixer.fix_one_directory_or_one_file() |
| 64 | + |
| 65 | + return ret |
| 66 | + |
| 67 | + |
| 68 | +if __name__ == '__main__': |
| 69 | + raise SystemExit(main()) |
0 commit comments