xian_algorithm_new/app/services/qgis/qgis_env.py

"""
QGIS 环境检测与子进程配置模块。

主进程运行在 Python 3.10，无法直接加载 QGIS 的 Python 3.12 C 扩展。
所有 QGIS 操作通过 subprocess 调用 QGIS Python 3.12 执行。

本模块提供：
  - get_qgis_python_path(): 检测 QGIS Python 3.12 解释器路径
  - build_qgis_command():   构建通过 .bat 启动 QGIS 子进程的命令
  - is_qgis_available():    检查 QGIS 是否可用
  - get_runner_script():    获取 qgis_runner.py 的路径
"""
import os
import tempfile
from pathlib import Path

from app.config.paths import get_logger

logger = get_logger("qgis.env")


def get_qgis_python_path(qgis_root: str = None) -> str | None:
    """
    检测 QGIS 自带的 Python 3.12 解释器路径。

    Windows: {QGIS_ROOT}/apps/Python312/python3.exe

    Returns:
        解释器绝对路径，不存在则返回 None
    """
    if qgis_root is None:
        qgis_root = _detect_qgis_root()
    if qgis_root is None:
        return None

    import platform
    if platform.system() == "Windows":
        for name in ("python3.exe", "python.exe"):
            candidate = os.path.join(qgis_root, "apps", "Python312", name)
            if os.path.isfile(candidate):
                logger.info(f"检测到 QGIS Python: {candidate}")
                return candidate
        logger.warning(f"未找到 QGIS Python 3.12")
        return None
    else:
        import shutil

        # 优先检查环境变量
        env_python = os.environ.get("QGIS_PYTHON_PATH")
        if env_python and os.path.isfile(env_python):
            logger.info(f"QGIS_PYTHON_PATH 指定: {env_python}")
            return env_python

        # 搜索标准 Linux QGIS 安装路径
        linux_paths = [
            "/usr/bin/qgis",  # Ubuntu/Debian apt 安装
            "/usr/bin/qgis.bin",
            "/opt/QGIS/apps/Python312/bin/python3",  # 独立安装器
            "/opt/QGIS/apps/Python3/bin/python3",
            "/usr/libexec/qgis/python3",  # Fedora/RHEL
        ]
        for candidate in linux_paths:
            if os.path.isfile(candidate):
                logger.info(f"检测到 QGIS 可执行文件: {candidate}")
                return candidate

        # 回退到系统 Python（如果 qgis.core 可导入）
        sys_python = shutil.which("python3") or shutil.which("python")
        if sys_python:
            logger.info(f"Linux 环境，使用系统 Python: {sys_python}")
            return sys_python
        return None


def build_qgis_command(qgis_root: str = None) -> list[str]:
    """
    构建 QGIS 子进程启动命令（直接调用 Python，不经过 bat 包装器）。
    环境变量通过 build_qgis_env() 构建后传给 subprocess.run(env=...)。
    """
    if qgis_root is None:
        qgis_root = _detect_qgis_root() or "D:/QGIS"

    python_path = get_qgis_python_path(qgis_root)
    if not python_path:
        raise RuntimeError("未找到 QGIS Python 3.12 解释器")

    runner_script = get_runner_script()
    if not os.path.isfile(runner_script):
        raise RuntimeError(f"QGIS Runner 脚本不存在: {runner_script}")

    return [python_path, runner_script]


def build_qgis_env(qgis_root: str = None) -> dict:
    """
    构建 QGIS 子进程所需的完整环境变量字典。

    基于当前进程环境继承，设置 QGIS 所需的 PYTHONPATH、PATH、
    GDAL_DATA、PROJ_DATA 等变量。调用方直接传给 subprocess.run(env=...)。
    """
    import platform

    if qgis_root is None:
        qgis_root = _detect_qgis_root() or "D:/QGIS"

    env = dict(os.environ)

    if platform.system() != "Windows":
        return env

    # --- 检测 QGIS 应用目录 ---
    qgis_app_dir = os.path.join(qgis_root, "apps", "qgis-ltr")
    if not os.path.isdir(qgis_app_dir):
        qgis_app_dir = os.path.join(qgis_root, "apps", "qgis")
    if not os.path.isdir(qgis_app_dir):
        raise RuntimeError(f"未找到 QGIS 应用目录: {qgis_root}\\apps\\qgis-ltr 或 qgis")

    # --- 检测 Python 目录 ---
    python_dir = os.path.join(qgis_root, "apps", "Python312")
    if not os.path.isdir(python_dir):
        for name in ("Python39", "Python310", "Python311"):
            candidate = os.path.join(qgis_root, "apps", name)
            if os.path.isdir(candidate):
                python_dir = candidate
                break

    # --- 核心路径 ---
    qtplugins = os.path.join(qgis_app_dir, "qtplugins")
    qt5_plugins = os.path.join(qgis_root, "apps", "Qt5", "plugins")
    gdal_data = os.path.join(qgis_root, "apps", "gdal", "share", "gdal")
    qgis_python = os.path.join(qgis_app_dir, "python")
    qgis_bin = os.path.join(qgis_app_dir, "bin")
    qt5_bin = os.path.join(qgis_root, "apps", "Qt5", "bin")
    gdal_lib = os.path.join(qgis_root, "apps", "gdal", "lib")

    env["PYTHONPATH"] = qgis_python
    env["QGIS_PREFIX_PATH"] = qgis_app_dir
    env["QT_PLUGIN_PATH"] = f"{qtplugins};{qt5_plugins}"
    env["GDAL_DATA"] = gdal_data
    env["PYTHONUTF8"] = "1"
    env["GDAL_FILENAME_IS_UTF8"] = "YES"
    env["VSI_CACHE"] = "TRUE"
    env["VSI_CACHE_SIZE"] = "1000000"

    # --- PROJ 数据目录 ---
    for pd in [
        os.path.join(qgis_app_dir, "share", "proj"),
        os.path.join(qgis_root, "share", "proj"),
        os.path.join(qgis_root, "apps", "gdal", "share", "proj"),
    ]:
        if os.path.isfile(os.path.join(pd, "proj.db")):
            env["PROJ_DATA"] = pd
            break

    # --- PATH：前置 QGIS 二进制目录 ---
    prepend = f"{qgis_bin};{qt5_bin};{gdal_lib}"
    env["PATH"] = f"{prepend};{env.get('PATH', '')}"

    logger.debug(f"QGIS env built: QGIS_ROOT={qgis_root}, QGIS_APP={qgis_app_dir}")
    return env


def build_clean_subprocess_env() -> dict:
    """
    为 QGIS Python 3.12 子进程构建干净的环境变量。

    问题背景：
      主进程运行在 venv Python 3.10 中，继承了 venv 的环境变量
      （PYTHONPATH 指向 venv site-packages、VIRTUAL_ENV、PATH 含 venv Scripts 等）。
      QGIS Python 3.12 子进程如果继承这些变量，DLL 加载会被干扰，
      导致 0xC0000005 (ACCESS_VIOLATION) 崩溃。

      同时，QGIS_PREFIX_PATH、QT_PLUGIN_PATH 等变量也会与
      QGIS 内置 DLL 加载机制冲突，同样导致崩溃。

    策略：
      从 os.environ 中移除所有 Python/venv 相关的污染变量，
      只保留操作系统正常运行所需的基础变量（SystemRoot、TEMP 等）。
      QGIS Python 3.12 仅通过 sys.path 即可正确加载所有模块和 DLL。
    """
    env = dict(os.environ)

    # 先记录 venv 路径（后续清理 PATH 时需要用）
    venv_root = env.get("VIRTUAL_ENV", "").lower()

    # 移除 Python/venv 相关变量 —— 这些会污染 QGIS Python 3.12 的 DLL 加载
    for key in [
        "PYTHONPATH",      # venv 设置，指向 site-packages → DLL 冲突
        "PYTHONHOME",      # 如果存在，改变 Python 查找路径
        "VIRTUAL_ENV",     # venv 标识，不必要
        "PYTHONDONTWRITEBYTECODE",
        "QGIS_PREFIX_PATH",  # 与 QGIS 内置 DLL 加载冲突
        "QT_PLUGIN_PATH",    # 与 Qt DLL 加载冲突
        "GDAL_DATA",         # build_qgis_env 设置，可能导致路径冲突
        "PROJ_DATA",         # build_qgis_env 设置，可能导致路径冲突
    ]:
        env.pop(key, None)

    # 清理 PATH：仅移除 venv 相关路径（避免 DLL 搜索顺序冲突）
    # 保留 QGIS/系统路径（测试证明这些是安全的）
    if venv_root:
        path_parts = env.get("PATH", "").split(";")
        clean_parts = []
        for p in path_parts:
            pl = p.lower().strip()
            if not pl:
                continue
            # 跳过 venv 相关路径（.venv/Scripts, .venv/Lib 等）
            if venv_root in pl:
                continue
            clean_parts.append(p)
        env["PATH"] = ";".join(clean_parts)

    # 移除 VIRTUAL_ENV 本身（已从 PATH 清理中使用过）
    env.pop("VIRTUAL_ENV", None)

    logger.debug(
        f"Clean env built: removed PYTHONPATH/VIRTUAL_ENV/venv-PATH, "
        f"kept {len(env)} vars"
    )
    return env


def is_qgis_available(qgis_root: str = None) -> bool:
    """检查 QGIS 环境是否可用"""
    return get_qgis_python_path(qgis_root) is not None


def get_runner_script() -> str:
    """获取 qgis_runner.py 的绝对路径"""
    return str(Path(__file__).parent / "qgis_runner.py")


def _generate_bat_wrapper(qgis_root: str, python_path: str, runner_script: str) -> str:
    """生成 .bat 包装脚本，设置 QGIS 环境变量并启动 runner"""
    # 自动检测 QGIS 应用目录（qgis-ltr 或 qgis）
    qgis_app_dir = os.path.join(qgis_root, "apps", "qgis-ltr")
    if not os.path.isdir(qgis_app_dir):
        qgis_app_dir = os.path.join(qgis_root, "apps", "qgis")
    qgis_app_dir = qgis_app_dir.replace("/", "\\")

    # 自动检测 PROJ 目录
    proj_data = ""
    for pd in [
        os.path.join(qgis_app_dir, "share", "proj"),
        os.path.join(qgis_root, "share", "proj"),
        os.path.join(qgis_root, "apps", "gdal", "share", "proj"),
    ]:
        if os.path.isfile(os.path.join(pd, "proj.db")):
            proj_data = pd.replace("/", "\\")
            break

    python_dir = os.path.join(qgis_root, "apps", "Python312").replace("/", "\\")
    qt5_plugins = os.path.join(qgis_root, "apps", "Qt5", "plugins").replace("/", "\\")
    qtplugins = os.path.join(qgis_app_dir, "qtplugins").replace("/", "\\")
    gdal_data = os.path.join(qgis_root, "apps", "gdal", "share", "gdal").replace("/", "\\")
    qgis_python_dir = os.path.join(qgis_app_dir, "python").replace("/", "\\")
    qgis_bin = os.path.join(qgis_app_dir, "bin").replace("/", "\\")
    qt5_bin = os.path.join(qgis_root, "apps", "Qt5", "bin").replace("/", "\\")
    gdal_lib = os.path.join(qgis_root, "apps", "gdal", "lib").replace("/", "\\")

    proj_line = f'set "PROJ_DATA={proj_data}"' if proj_data else "REM PROJ_DATA not found"

    bat_content = f"""@echo off
chcp 65001 >nul
set "PYTHONPATH={qgis_python_dir}"
set "QGIS_PREFIX_PATH={qgis_app_dir}"
set "QT_PLUGIN_PATH={qtplugins};{qt5_plugins}"
set "GDAL_DATA={gdal_data}"
{proj_line}
set "PYTHONUTF8=1"
set "GDAL_FILENAME_IS_UTF8=YES"
set "VSI_CACHE=TRUE"
set "VSI_CACHE_SIZE=1000000"
set "PATH={qt5_bin};{qgis_bin};{gdal_lib};%PATH%"
"{python_path}" "{runner_script}" %*
"""

    bat_dir = os.path.join(tempfile.gettempdir(), "qgis_runner")
    os.makedirs(bat_dir, exist_ok=True)
    bat_path = os.path.join(bat_dir, "run_qgis.bat")
    with open(bat_path, "w", encoding="utf-8") as f:
        f.write(bat_content)
    logger.debug(f"生成 QGIS 包装脚本: {bat_path}")
    return bat_path


def _detect_qgis_root() -> str | None:
    """
    自动检测 QGIS 安装根目录。

    优先级：
    1. 环境变量 QGIS_ROOT
    2. Windows 默认路径 D:/QGIS
    3. Linux 常见路径
    """
    env_root = os.environ.get("QGIS_ROOT")
    if env_root and os.path.isdir(env_root):
        return env_root

    import platform
    if platform.system() == "Windows":
        for candidate in ["D:/QGIS", "C:/OSGeo4W", "C:/QGIS"]:
            if os.path.isdir(candidate):
                logger.info(f"检测到 QGIS 根目录: {candidate}")
                return candidate
    else:
        for candidate in ["/usr", "/opt/QGIS", "/home/QGIS"]:
            if os.path.isdir(candidate):
                logger.info(f"检测到 QGIS 根目录: {candidate}")
                return candidate

    logger.warning("未检测到 QGIS 安装目录")
    return None