# -*- coding: utf-8 -*- """ main.py — Точка входа. Меню выбора режима. """ import sys import os import argparse _PROJECT_ROOT = os.path.dirname(os.path.dirname(os.path.abspath(__file__))) if _PROJECT_ROOT not in sys.path: sys.path.insert(0, _PROJECT_ROOT) from core.color import banner, menu, prompt, info, error as cerror, warn from core import state def main(): parser = argparse.ArgumentParser(description="Docker Service Migration Tool") parser.add_argument("--mode", choices=["source", "target"], help="Режим работы") parser.add_argument("--resume", action="store_true", help="Продолжить после ошибки") parser.add_argument("--status", action="store_true", help="Показать статус") parser.add_argument("--logs", action="store_true", help="Показать логи ошибки") parser.add_argument("--remote-dir", help="Директория архива на target") parser.add_argument("--no-color", action="store_true", help="Отключить цвета") args = parser.parse_args() if args.no_color: os.environ["NO_COLOR"] = "1" if args.status: state.show_status() sys.exit(0) if args.logs: state.show_logs() sys.exit(0) if args.resume: _do_resume() sys.exit(0) if args.mode == "source": from source.source import run_source_mode run_source_mode() sys.exit(0) elif args.mode == "target": if args.remote_dir: state.set_stage("INIT", mode="target", target_remote_dir=args.remote_dir) from target.target import run_target_mode run_target_mode() sys.exit(0) # Интерактивный режим banner() while True: st = state.load_state() has_resume = bool(st.get("mode") and st.get("stage") and st.get("stage") != "INIT") menu(has_resume=has_resume) try: choice = prompt("Ваш выбор", default="0").strip() except SystemExit: break if choice == "1": from source.source import run_source_mode try: run_source_mode() except KeyboardInterrupt: warn("Прервано") except Exception as e: cerror(f"Ошибка: {e}") sys.exit(1) break elif choice == "2": from target.target import run_target_mode try: run_target_mode() except KeyboardInterrupt: warn("Прервано") except Exception as e: cerror(f"Ошибка: {e}") sys.exit(1) break elif choice == "3": if not has_resume: warn("Нет сохранённого состояния для продолжения") continue try: _do_resume() except KeyboardInterrupt: warn("Прервано") break elif choice == "4": _do_status_logs() elif choice == "0": info("Выход") sys.exit(0) else: warn("Неверный выбор") def _do_resume(): from core.fsm import FSM st = state.load_state() stage = st.get("stage") mode = st.get("mode") if not mode or stage == "INIT": info("Нет сохранённого состояния для продолжения") return fsm = FSM(mode=mode) try: fsm.resume_from(stage) except Exception as e: cerror(f"Ошибка при resume: {e}") def _do_status_logs(): from core import state as stmod while True: print("\n") print(" 1 Показать статус") print(" 2 Показать лог последней ошибки") print(" 0 Назад") try: c = prompt("Выбор", default="0").strip() except SystemExit: break if c == "1": stmod.show_status() elif c == "2": stmod.show_logs() elif c == "0": break if __name__ == "__main__": main()