launcher.py

"""Command-line adapter for FolderVisualizer's shared workflow."""

from __future__ import annotations

import argparse
import sqlite3
import webbrowser
from pathlib import Path

from workflow import (
    DATABASE_PATH,
    DEFAULT_PORT,
    JSON_SIZE_THRESHOLD,
    LARGE_BROWSER_HTML_PATH,
    MANIFEST_PATH,
    MAX_CHILDREN_THRESHOLD,
    ModeDecision,
    NODE_COUNT_THRESHOLD,
    PROJECT_DIR,
    SCAN_RESULT_PATH,
    ScanStatistics,
    TREE_OUTPUT_PATH,
    WorkflowError,
    WorkflowEvent,
    WorkflowOptions,
    WorkflowResult,
    choose_mode,
    create_large_server_controller,
    load_scan_statistics,
    run_workflow,
)


# Backward-compatible public name used by the original launcher.
LauncherError = WorkflowError


def _format_number(value: int) -> str:
    return f"{value:,}"


def _format_size(size: int) -> str:
    units = ("B", "KB", "MB", "GB", "TB")
    value = float(size)
    for unit in units:
        if value < 1024 or unit == units[-1]:
            if unit == "B":
                return f"{int(value)} {unit}"
            return f"{value:.2f} {unit}"
        value /= 1024
    return f"{size} B"


def _print_statistics(statistics: ScanStatistics) -> None:
    print("\n规模分析")
    print(f"扫描路径: {statistics.root_path}")
    print(f"节点总数: {_format_number(statistics.node_count)}")
    print(f"文件数量: {_format_number(statistics.file_count)}")
    print(f"目录数量: {_format_number(statistics.directory_count)}")
    print(f"总大小: {_format_size(statistics.total_size)}")
    print(
        "最大直接子节点数量: "
        f"{_format_number(statistics.max_children)}"
    )
    print("scan-result.json 大小: " f"{_format_size(statistics.json_size)}")


def _print_decision(decision: ModeDecision) -> None:
    label = "小目录完整树" if decision.mode == "small" else "大目录按需浏览"
    print(f"\n自动选择: {label}模式")
    print("选择原因:")
    for reason in decision.reasons:
        print(f"- {reason}")


def _print_workflow_event(event: WorkflowEvent) -> None:
    """Translate shared events into the launcher's established output."""

    if event.stage == "scanning":
        print(f"开始扫描: {event.data['input_path']}")
    elif event.stage == "scan_completed":
        print("扫描完成")
        print(f"文件数量: {_format_number(event.data['file_count'])}")
        print(f"目录数量: {_format_number(event.data['directory_count'])}")
        print(f"扫描错误: {_format_number(event.data['error_count'])}")
        print(f"JSON 文件: {event.data['scan_result_path']}")
    elif event.stage == "choosing_mode":
        _print_statistics(event.data["statistics"])
    elif event.stage == "mode_selected":
        _print_decision(event.data["decision"])
    elif event.stage == "building_small_tree":
        print("\n正在生成小目录完整树……")
    elif event.stage == "building_large_index":
        print("\n正在构建大目录 SQLite 索引……")


def _open_browser(target: str) -> None:
    try:
        opened = webbrowser.open(target)
    except (OSError, webbrowser.Error) as error:
        raise LauncherError(f"无法自动打开浏览器：{error}") from error
    if not opened:
        raise LauncherError(
            "无法自动打开浏览器；可使用 --no-open 后手动打开输出地址"
        )


def _finish_small_mode(result: WorkflowResult, no_open: bool) -> None:
    assert result.small_tree_path is not None
    print(f"文件树页面: {result.small_tree_path}")
    if no_open:
        print("已启用 --no-open，请手动打开上述 HTML 文件。")
        return
    _open_browser(result.small_tree_path.resolve().as_uri())
    print("已在默认浏览器中打开文件树页面。")


def _run_large_server(
    result: WorkflowResult, port: int, no_open: bool
) -> None:
    assert result.database_path is not None
    assert result.manifest_path is not None
    assert result.index_elapsed_seconds is not None
    print(f"SQLite: {result.database_path}")
    print(f"Manifest: {result.manifest_path}")
    print(f"索引耗时: {result.index_elapsed_seconds:.2f} 秒")
    print(f"integrity_check: {result.integrity_check}")

    try:
        controller = create_large_server_controller(result, port=port)
        url = controller.start()
    except (OSError, ValueError, sqlite3.Error) as error:
        raise LauncherError(f"无法启动大目录本地服务：{error}") from error

    print("\n大目录按需浏览服务已启动")
    print(f"地址: {url}")
    print("仅监听 127.0.0.1，不会上传文件或扫描数据")
    print("按 Ctrl+C 关闭服务")

    try:
        if no_open:
            print("已启用 --no-open，请手动打开上述地址。")
        else:
            _open_browser(url)
            print("已在默认浏览器中打开大目录页面。")
        while not controller.wait(timeout=0.25):
            pass
    except KeyboardInterrupt:
        print("\n正在关闭本地服务……")
    except OSError as error:
        raise LauncherError(f"大目录本地服务运行失败：{error}") from error
    finally:
        controller.stop()
    print("本地服务已关闭")


def _port(value: str) -> int:
    try:
        port = int(value)
    except ValueError as error:
        raise argparse.ArgumentTypeError("端口必须是整数") from error
    if not 1 <= port <= 65535:
        raise argparse.ArgumentTypeError("端口必须在 1–65535 之间")
    return port


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description="扫描文件夹并自动选择 FolderVisualizer 展示模式"
    )
    parser.add_argument("path", nargs="?", help="需要扫描和展示的文件夹路径")
    parser.add_argument(
        "--force-large",
        action="store_true",
        help="忽略自动判断并强制使用大目录按需浏览模式",
    )
    parser.add_argument(
        "--port",
        type=_port,
        default=DEFAULT_PORT,
        help=f"大目录本地服务端口（默认 {DEFAULT_PORT}）",
    )
    parser.add_argument(
        "--no-open",
        action="store_true",
        help="不自动打开浏览器，用于测试或无界面环境",
    )
    return parser.parse_args()


def _input_path(value: str | None) -> str:
    if value is not None:
        raw_path = value
    else:
        try:
            raw_path = input("请输入要扫描的文件夹路径: ")
        except EOFError as error:
            raise LauncherError("未收到文件夹路径输入") from error
    raw_path = raw_path.strip().strip('"')
    if not raw_path:
        raise LauncherError("文件夹路径不能为空")
    return raw_path


def run(args: argparse.Namespace) -> int:
    input_path = _input_path(args.path)
    result = run_workflow(
        WorkflowOptions(
            input_path=Path(input_path), force_large=args.force_large
        ),
        callback=_print_workflow_event,
    )
    if result.mode == "small":
        _finish_small_mode(result, no_open=args.no_open)
    else:
        _run_large_server(result, port=args.port, no_open=args.no_open)
    return 0


def main() -> int:
    args = _parse_args()
    try:
        return run(args)
    except LauncherError as error:
        print(f"错误: {error}")
        return 1
    except KeyboardInterrupt:
        print("\n操作已取消。")
        return 130


if __name__ == "__main__":
    raise SystemExit(main())