#!/usr/bin/env python3 """CLI entry point for the Qt5→Qt6 migration pipeline. Usage examples -------------- # Run the Q1 pilot batch on both tracks: python run_migration.py \\ --batch-id q1_pilot \\ --file-list ./q1_files.txt \\ --repo-root . \\ --source-root ./src # Dry-run: show files and chunk info without migrating: python run_migration.py --batch-id q1_pilot --file-list ./q1_files.txt --dry-run """ from __future__ import annotations import argparse import asyncio import logging import sys from pathlib import Path def _configure_logging(verbose: bool) -> None: level = logging.DEBUG if verbose else logging.INFO logging.basicConfig( level=level, format="%(asctime)s %(levelname)-8s %(name)s – %(message)s", datefmt="%H:%M:%S", ) def _dry_run(file_list_path: Path, source_root: Path) -> None: """Print file list with chunk estimates without calling any API.""" from migration_pipeline.chunker import split_semantic from migration_pipeline.config import CHUNK_SIZE_LINES files = _load_plain_list(file_list_path, source_root) print(f"\nDry-run: {len(files)} files in {file_list_path.name}\n") print(f"{'#':<4} {'File':<60} {'Lines':>6} {'Chunks':>6}") print("-" * 82) for i, fp in enumerate(files, 1): p = Path(fp) if not p.exists(): print(f"{i:<4} {fp:<60} {'NOT FOUND':>6}") continue code = p.read_text(errors="ignore") lines = len(code.splitlines()) chunks = split_semantic(code, CHUNK_SIZE_LINES) print(f"{i:<4} {str(p.name):<60} {lines:>6} {len(chunks):>6}") print() def _load_plain_list(list_path: Path, source_root: Path) -> list[str]: base = source_root.resolve() result = [] for raw in list_path.read_text().splitlines(): s = raw.strip() if not s or s.startswith("#"): continue p = Path(s) result.append(str((base / p).resolve() if not p.is_absolute() else p)) return result def _parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser( description="Qt5→Qt6 async migration pipeline", formatter_class=argparse.RawDescriptionHelpFormatter, epilog=__doc__, ) parser.add_argument( "--batch-id", required=True, help="Unique identifier for this batch (e.g. q1_pilot)", ) parser.add_argument( "--file-list", required=True, type=Path, help="Path to the prioritised file list (one path per line)", ) parser.add_argument( "--repo-root", default=".", type=Path, help="Root of the git repository (default: current directory)", ) parser.add_argument( "--source-root", default=".", type=Path, help="Root of the C++ source tree (default: current directory)", ) parser.add_argument( "--dry-run", action="store_true", help="Print file list and chunk estimates; do not call any API", ) parser.add_argument( "--model", choices=("glm", "qwen"), default="glm", help="Model family for API calls (default: glm)", ) parser.add_argument( "--verbose", "-v", action="store_true", help="Enable DEBUG logging", ) return parser.parse_args() def main() -> None: args = _parse_args() _configure_logging(args.verbose) if not args.file_list.exists(): print(f"ERROR: File list not found: {args.file_list}", file=sys.stderr) sys.exit(1) if args.dry_run: _dry_run(args.file_list, args.source_root.resolve()) return from migration_pipeline.api_client import set_model_family from migration_pipeline.orchestrator import run_batch set_model_family(args.model) asyncio.run( run_batch( batch_id=args.batch_id, file_list_path=args.file_list, repo_root=args.repo_root.resolve(), source_root=args.source_root.resolve(), ) ) if __name__ == "__main__": main()