JiGuang/jg/serves/JiGuang-program/runner.py

#!/usr/bin/env python3
"""
JiGuang 算法 CLI 桥接入口。

Electron 调用协议（与旧 api_test.exe 保持一致）：
    runner <apiName> <base64(JSON)>

stdout 仅输出 JSON 结果，日志与错误信息写入 stderr。
"""

from __future__ import annotations

import base64
import glob
import importlib.util
import json
import os
import sys
import traceback

# 无 GUI 环境使用 Agg 后端，避免 matplotlib 依赖 Qt/Tk
os.environ.setdefault("MPLBACKEND", "Agg")

# 确保工作目录为脚本所在目录，便于加载 entry 模块与 license 文件
_SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
if _SCRIPT_DIR not in sys.path:
    sys.path.insert(0, _SCRIPT_DIR)
os.chdir(_SCRIPT_DIR)


def _ensure_dependencies() -> None:
    try:
        import cryptography  # noqa: F401
    except ImportError as exc:
        message = (
            "缺少 cryptography 库，授权校验无法运行。"
            "请执行: python -m pip install cryptography"
        )
        print(json.dumps({"error": message}, ensure_ascii=False))
        raise SystemExit(1) from exc


_ensure_dependencies()


def _load_entry_module():
    """只加载 Nuitka 编译的 .pyd/.so，优先使用 JiGuang-program 根目录下最新的 entry 文件。"""
    patterns = ("entry.cp*.pyd", "entry.cpython-*.so", "entry*.so")
    candidates = []
    # 根目录优先：更新 entry.pyd 后无需重新编译 api_runner.exe
    search_dirs = [
        os.path.dirname(_SCRIPT_DIR),
        _SCRIPT_DIR,
    ]
    for search_dir in search_dirs:
        if not os.path.isdir(search_dir):
            continue
        for pattern in patterns:
            candidates.extend(glob.glob(os.path.join(search_dir, pattern)))

    if candidates:
        # 同名文件可能在根目录和 runner.dist 各有一份，取最新修改时间
        entry_path = max(candidates, key=os.path.getmtime)
        spec = importlib.util.spec_from_file_location("entry", entry_path)
        if spec is None or spec.loader is None:
            raise ImportError(f"无法加载 entry 模块: {entry_path}")
        module = importlib.util.module_from_spec(spec)
        sys.modules["entry"] = module
        spec.loader.exec_module(module)
        return module

    raise ImportError(
        "未找到 entry.cp*.pyd / entry*.so。请确认 entry 编译模块位于 "
        "JiGuang-program 或 runner.dist 目录。"
    )


entry = _load_entry_module()

API_MAP = {
    "loaddata": entry.load_data,
    "getpath": entry.get_path,
    "historydata": entry.history_data,
    "deletedata": entry.delete_data,
}


def _verify_license() -> None:
    if hasattr(entry, "verify_license"):
        entry.verify_license()


def _decode_payload(b64_payload: str):
    if not b64_payload:
        return None
    raw = base64.b64decode(b64_payload).decode("utf-8")
    return json.loads(raw)


def _is_empty_params(params) -> bool:
    return params is None or params == "" or params == {} or params == []


def _dispatch(api_name: str, b64_payload: str):
    func = API_MAP.get(api_name)
    if func is None:
        raise ValueError(f"未知 API: {api_name}")

    params = _decode_payload(b64_payload)
    if _is_empty_params(params):
        return func()
    return func(b64_payload)


def main() -> int:
    if len(sys.argv) < 2:
        print(
            json.dumps({"error": "用法: runner <apiName> [base64_json]"}, ensure_ascii=False),
            file=sys.stderr,
        )
        return 1

    api_name = sys.argv[1]
    b64_payload = sys.argv[2] if len(sys.argv) > 2 else ""

    try:
        _verify_license()
        result = _dispatch(api_name, b64_payload)
        print(json.dumps(result, ensure_ascii=False))
        return 0
    except Exception as exc:  # noqa: BLE001
        traceback.print_exc(file=sys.stderr)
        print(
            json.dumps({"error": str(exc)}, ensure_ascii=False),
            file=sys.stdout,
        )
        return 1


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