src/liquidcode/commands/serve.py - master - liquid-g/liquid-code

src/liquidcode/commands/serve.py
279 строк9 KB
User
0.3.0-RC - дописаны тесты и мелкие исправления
04 июл 2026, 13:07
04 июл 2026, 13:0740ad8b8
"""
Команда для запуска встроенного HTTP-сервера.

Использование: lcc serve [--host HOST] [--port PORT] [--reload] [--app APP]
"""

import argparse
import importlib
import os
import signal
import subprocess
import sys
import time
from typing import Optional

from liquidcode.application import Application


def configure_parser(subparser: argparse.ArgumentParser) -> None:
    """
    Добавляет аргументы для команды serve.

    Args:
        subparser: Объект ArgumentParser для команды.
    """
    subparser.add_argument(
        "--host",
        help="Хост для привязки (по умолчанию: из конфига или localhost)",
    )
    subparser.add_argument(
        "--port",
        type=int,
        help="Порт для привязки (по умолчанию: из конфига или 8000)",
    )
    subparser.add_argument(
        "--reload",
        action="store_true",
        help="Включить горячую перезагрузку при изменении файлов",
    )
    subparser.add_argument(
        "--app",
        default="app.py",
        help="Путь к файлу приложения (по умолчанию: app.py)",
    )
    subparser.add_argument(
        "--watch-path",
        action="append",
        help="Дополнительные пути для отслеживания (можно указать несколько)",
    )


def run(args: argparse.Namespace) -> None:
    """
    Выполняет команду serve.
    """
    app_file = args.app

    # 1. Проверяем существование файла приложения
    if not os.path.exists(app_file):
        print(f"❌ Файл приложения не найден: {app_file}")
        return

    # 2. Пытаемся загрузить приложение из файла
    app = _load_application(app_file)

    if app is None:
        print("❌ Не удалось загрузить приложение.")
        print(f"   Проверьте, что файл {app_file} содержит экземпляр Application.")
        return

    # 3. Определяем host и port (приоритет: аргументы > конфиг > дефолт)
    host = args.host or app.host
    port = args.port or app.port

    # 4. Запуск с перезагрузкой или без
    if args.reload:
        _run_with_reload(app, host, port, app_file, args)
    else:
        _run_server(app, host, port)


def _load_application(app_file: str) -> Optional[Application]:
    """
    Загружает экземпляр Application из указанного файла.
    Ожидается, что в файле определена переменная `app`.
    """
    if not os.path.exists(app_file):
        return None

    # Добавляем директорию файла в sys.path для импорта
    module_dir = os.path.dirname(os.path.abspath(app_file))
    if module_dir not in sys.path:
        sys.path.insert(0, module_dir)

    module_name = os.path.splitext(os.path.basename(app_file))[0]
    try:
        module = importlib.import_module(module_name)
        app = getattr(module, "app", None)
        if isinstance(app, Application):
            return app
    except Exception as e:
        print(f"⚠️  Ошибка при загрузке приложения: {e}")

    return None


def _run_server(app: Application, host: str, port: int) -> None:
    """Запускает сервер напрямую (без перезагрузки)."""
    try:
        app.run(host=host, port=port)
    except OSError as e:
        if e.errno == 48:  # EADDRINUSE на macOS/Linux
            print(f"❌ Порт {port} уже занят.")
            print(f"   Используйте другой порт: lcc serve --port {port + 1}")
        else:
            raise


def _run_with_reload(
        app: Application,
        host: str,
        port: int,
        app_file: str,
        args: argparse.Namespace,
) -> None:
    """
    Запускает сервер с горячей перезагрузкой.
    Использует watchdog для отслеживания изменений файлов.
    """
    try:
        from watchdog.observers import Observer
        from watchdog.events import FileSystemEventHandler
    except ImportError:
        print(
            "⚠️  Для горячей перезагрузки требуется установить watchdog: pip install watchdog"
        )
        print("   Запуск без перезагрузки...")
        app.run(host=host, port=port)
        return

    # Базовые пути для отслеживания
    default_paths = [".", "src", "config", "templates", "public", ".env"]
    watch_paths = args.watch_path or []
    # Если пользователь явно указал пути через --watch-path, используем их,
    # иначе добавляем пути из конфига (если есть) или стандартные
    if not watch_paths:
        # Если в приложении есть атрибут reload_paths, используем его
        if hasattr(app, "reload_paths") and app.reload_paths:
            watch_paths = app.reload_paths
        else:
            watch_paths = default_paths

    # Фильтруем существующие пути
    watch_paths = [p for p in watch_paths if os.path.exists(p)]

    # Строим команду для запуска сервера в дочернем процессе (без --reload)
    cmd = [
        sys.executable,
        "-m",
        "liquidcode.console",
        "serve",
        "--host",
        host,
        "--port",
        str(port),
        "--app",
        app_file,
    ]

    process = None

    def start_server():
        nonlocal process
        if process:
            process.terminate()
            process.wait()
        process = subprocess.Popen(
            cmd,
            env=os.environ.copy(),
            stdout=subprocess.PIPE,
            stderr=subprocess.PIPE,
            text=True,
            bufsize=1,
        )

        # Запускаем поток для вывода логов дочернего процесса
        def log_output():
            for line in process.stdout:
                print(line, end="")
            # Если процесс завершился с ошибкой, выводим stderr
            if process.returncode != 0 and process.stderr:
                error = process.stderr.read()
                if error:
                    print(
                        f"⚠️  Ошибка сервера (exit code {process.returncode}):",
                        file=sys.stderr,
                    )
                    print(error, file=sys.stderr)

        import threading

        thread = threading.Thread(target=log_output, daemon=True)
        thread.start()

    class ReloadHandler(FileSystemEventHandler):
        def __init__(self, callback):
            self.callback = callback
            self._last_trigger = time.time()

        def on_modified(self, event):
            if event.is_directory:
                return
            # Игнорируем временные файлы и системные
            if _should_ignore(event.src_path):
                return
            # Ограничиваем частоту перезагрузки
            now = time.time()
            if now - self._last_trigger < 1.0:
                return
            self._last_trigger = now
            print("\n🔄 Обнаружены изменения, перезапуск сервера...")
            self.callback()

    # Обработка сигналов для корректного завершения
    def signal_handler(sig, frame):
        if process:
            process.terminate()
            process.wait()
        sys.exit(0)

    signal.signal(signal.SIGINT, signal_handler)
    signal.signal(signal.SIGTERM, signal_handler)

    # Запускаем в первый раз
    start_server()

    # Настраиваем watchdog для отслеживания изменений
    event_handler = ReloadHandler(start_server)
    observer = Observer()
    for path in watch_paths:
        if os.path.exists(path):
            observer.schedule(event_handler, path, recursive=True)

    observer.start()
    msg = f"👀 Отслеживание изменений включено. Сервер запущен на http://{host}:{port}"
    print(msg, file=sys.stderr)

    try:
        while True:
            observer.join(timeout=1)
            if process and process.poll() is not None:
                print("⚠️  Сервер неожиданно завершился. Перезапуск...")
                start_server()
    except KeyboardInterrupt:
        observer.stop()
        if process:
            process.terminate()
            process.wait()
        print("\n⏹️  Сервер остановлен")

    observer.join()


def _should_ignore(path: str) -> bool:
    """Проверяет, должен ли файл игнорироваться."""
    exclude_patterns = (
        "__pycache__",
        ".pyc",
        ".pyo",
        ".pyd",
        ".git",
        ".py~",
        ".#",
        ".DS_Store",
        "Thumbs.db",
        ".swp",
        ".swo",
    )
    return any(pattern in path for pattern in exclude_patterns)