QGIS完成初步重构
This commit is contained in:
wzy-warehouse
+164
-242
@@ -1,266 +1,188 @@
|
||||
"""
|
||||
QGIS 环境初始化模块（跨平台：Windows / Linux）。
|
||||
在 server.py lifespan 启动阶段调用 init_qgis_env()，
|
||||
完成共享库注入、环境变量设置、QgsApplication 初始化。
|
||||
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 sys
|
||||
import platform
|
||||
import tempfile
|
||||
from pathlib import Path
|
||||
|
||||
from app.config.paths import get_logger
|
||||
|
||||
logger = get_logger("qgis.env")
|
||||
|
||||
_qgs_app = None
|
||||
_initialized = False
|
||||
_IS_WINDOWS = platform.system() == "Windows"
|
||||
|
||||
|
||||
def init_qgis_env(qgis_root: str) -> None:
|
||||
def get_qgis_python_path(qgis_root: str = None) -> str | None:
|
||||
"""
|
||||
初始化 QGIS 运行环境。整个应用生命周期只需调用一次。
|
||||
检测 QGIS 自带的 Python 3.12 解释器路径。
|
||||
|
||||
Args:
|
||||
qgis_root: QGIS 安装根目录
|
||||
Windows: "D:/QGIS"
|
||||
Linux: "/usr" 或 "/opt/QGIS"
|
||||
Windows: {QGIS_ROOT}/apps/Python312/python3.exe
|
||||
|
||||
Returns:
|
||||
解释器绝对路径，不存在则返回 None
|
||||
"""
|
||||
global _qgs_app, _initialized
|
||||
if _initialized:
|
||||
return
|
||||
if qgis_root is None:
|
||||
qgis_root = _detect_qgis_root()
|
||||
if qgis_root is None:
|
||||
return None
|
||||
|
||||
qgis_app_dir = _find_qgis_app_dir(qgis_root)
|
||||
|
||||
# 共享库搜索路径（平台相关）
|
||||
# Windows: os.add_dll_directory() 显式注册 DLL 目录
|
||||
# Linux: LD_LIBRARY_PATH 追加 .so 搜索目录
|
||||
_add_library_paths(qgis_root, qgis_app_dir)
|
||||
|
||||
# 环境变量
|
||||
_set_environment(qgis_root, qgis_app_dir)
|
||||
|
||||
# Python 模块路径
|
||||
_add_python_paths(qgis_root, qgis_app_dir)
|
||||
|
||||
# 初始化 QgsApplication
|
||||
_init_qgs_application(qgis_app_dir)
|
||||
|
||||
_initialized = True
|
||||
logger.info(f"QGIS 环境初始化完成 ({platform.system()})")
|
||||
|
||||
|
||||
def cleanup_qgis_env() -> None:
|
||||
"""清理 QGIS 资源（应用退出时调用）"""
|
||||
global _qgs_app, _initialized
|
||||
if _qgs_app is not None:
|
||||
_qgs_app.exitQgis()
|
||||
_qgs_app = None
|
||||
_initialized = False
|
||||
logger.info("QGIS 资源已清理")
|
||||
|
||||
|
||||
def is_qgis_ready() -> bool:
|
||||
"""检查 QGIS 是否已初始化"""
|
||||
return _initialized
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# 平台检测
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _find_qgis_app_dir(root: str) -> str:
|
||||
"""
|
||||
自动检测 QGIS 应用目录。
|
||||
|
||||
Windows（OSGeo4W 安装）: {root}/apps/qgis-ltr/
|
||||
Linux（包管理器安装）: {root}/share/qgis/（Python 在 {root}/share/qgis/python）
|
||||
"""
|
||||
if _IS_WINDOWS:
|
||||
return _find_qgis_app_dir_windows(root)
|
||||
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:
|
||||
return _find_qgis_app_dir_linux(root)
|
||||
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
|
||||
|
||||
def _find_qgis_app_dir_windows(root: str) -> str:
|
||||
"""Windows: 在 {root}/apps/ 下查找 qgis* 目录"""
|
||||
apps_dir = Path(root) / "apps"
|
||||
if apps_dir.is_dir():
|
||||
for name in sorted(apps_dir.iterdir()):
|
||||
if name.name.startswith("qgis") and name.is_dir():
|
||||
logger.info(f"检测到 QGIS 应用目录: {name}")
|
||||
return str(name)
|
||||
fallback = str(apps_dir / "qgis")
|
||||
logger.warning(f"未检测到 qgis* 目录，使用默认路径: {fallback}")
|
||||
return fallback
|
||||
|
||||
|
||||
def _find_qgis_app_dir_linux(root: str) -> str:
|
||||
"""
|
||||
Linux: 返回 QGIS_PREFIX_PATH。
|
||||
包管理器安装的标准路径：
|
||||
Debian/Ubuntu: /usr （qgis-core 在 /usr/lib/qgis/）
|
||||
RHEL/CentOS: /usr
|
||||
QGIS.org 官方: /usr 或 /opt/QGIS
|
||||
"""
|
||||
candidates = [
|
||||
Path(root) / "share" / "qgis", # /usr/share/qgis
|
||||
Path(root) / "lib" / "qgis", # /usr/lib/qgis
|
||||
Path(root) / "share" / "qgis-ltr", # qgis-ltr 变体
|
||||
]
|
||||
for c in candidates:
|
||||
if c.is_dir():
|
||||
logger.info(f"检测到 QGIS 应用目录: {c}")
|
||||
return str(c)
|
||||
|
||||
# 回退：用 root 本身（/usr），由 QgsApplication 自行探测
|
||||
logger.warning(f"未找到 QGIS 标准目录，使用 root: {root}")
|
||||
return str(root)
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# 共享库注入
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _add_library_paths(root: str, qgis_app: str) -> None:
|
||||
"""
|
||||
注入共享库搜索路径。
|
||||
|
||||
Windows: 调用 os.add_dll_directory() 显式注册每个 DLL 目录，
|
||||
Python 3.8+ 不再自动搜索 PATH 中的 DLL。
|
||||
Linux: 追加 LD_LIBRARY_PATH，让动态链接器 (ld-linux) 找到 .so。
|
||||
"""
|
||||
if _IS_WINDOWS:
|
||||
_add_dll_directories_windows(root, qgis_app)
|
||||
else:
|
||||
_add_ld_library_path_linux(root, qgis_app)
|
||||
|
||||
|
||||
def _add_dll_directories_windows(root: str, qgis_app: str) -> None:
|
||||
"""Windows: 显式注册 DLL 搜索目录"""
|
||||
if not hasattr(os, "add_dll_directory"):
|
||||
return # Python < 3.8
|
||||
|
||||
dll_dirs = [
|
||||
os.path.join(root, "apps", "Qt5", "bin"),
|
||||
os.path.join(qgis_app, "bin"),
|
||||
os.path.join(root, "bin"),
|
||||
os.path.join(root, "apps", "gdal", "bin"),
|
||||
os.path.join(root, "apps", "gdal", "lib"),
|
||||
]
|
||||
for d in dll_dirs:
|
||||
if os.path.isdir(d):
|
||||
os.add_dll_directory(d)
|
||||
logger.debug(f"注册 DLL 目录: {d}")
|
||||
|
||||
|
||||
def _add_ld_library_path_linux(root: str, qgis_app: str) -> None:
|
||||
"""
|
||||
Linux: 追加 LD_LIBRARY_PATH，使动态链接器找到 QGIS/GDAL 的 .so。
|
||||
|
||||
QGIS 包管理器安装时的标准 .so 路径：
|
||||
/usr/lib/ — libqgis_core.so, libqgis_analysis.so
|
||||
/usr/lib/qgis/ — 插件 .so
|
||||
/usr/lib/qgis/plugins/ — provider .so
|
||||
/usr/share/qgis/python/ — Python 模块
|
||||
"""
|
||||
ld_dirs = [
|
||||
os.path.join(root, "lib"), # /usr/lib
|
||||
os.path.join(root, "lib", "qgis"), # /usr/lib/qgis
|
||||
os.path.join(root, "lib", "qgis", "plugins"), # /usr/lib/qgis/plugins
|
||||
os.path.join(root, "share", "qgis", "lib"), # 某些安装方式
|
||||
os.path.join(root, "apps", "gdal", "lib"), # OSGeo4W 风格
|
||||
]
|
||||
existing = os.environ.get("LD_LIBRARY_PATH", "")
|
||||
new_dirs = [d for d in ld_dirs if os.path.isdir(d) and d not in existing]
|
||||
if new_dirs:
|
||||
joined = ":".join(new_dirs)
|
||||
os.environ["LD_LIBRARY_PATH"] = f"{joined}:{existing}" if existing else joined
|
||||
logger.info(f"LD_LIBRARY_PATH 追加: {new_dirs}")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# 环境变量
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _set_environment(root: str, qgis_app: str) -> None:
|
||||
"""设置 QGIS 和 GDAL 相关环境变量（平台自适应）"""
|
||||
env_vars = {
|
||||
"QGIS_PREFIX_PATH": qgis_app,
|
||||
"PYTHONUTF8": "1",
|
||||
"GDAL_FILENAME_IS_UTF8": "YES",
|
||||
"VSI_CACHE": "TRUE",
|
||||
"VSI_CACHE_SIZE": "1000000",
|
||||
}
|
||||
|
||||
if _IS_WINDOWS:
|
||||
env_vars["QT_PLUGIN_PATH"] = (
|
||||
f"{os.path.join(qgis_app, 'qtplugins')};"
|
||||
f"{os.path.join(root, 'apps', 'Qt5', 'plugins')}"
|
||||
)
|
||||
env_vars["GDAL_DATA"] = os.path.join(root, "apps", "gdal", "share", "gdal")
|
||||
else:
|
||||
# Linux: GDAL 数据通常在系统路径 /usr/share/gdal/ 下
|
||||
gdal_candidates = [
|
||||
os.path.join(root, "share", "gdal"),
|
||||
os.path.join(root, "share", "qgis", "resources"),
|
||||
# 搜索标准 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 p in gdal_candidates:
|
||||
if os.path.isdir(p):
|
||||
env_vars["GDAL_DATA"] = p
|
||||
break
|
||||
for candidate in linux_paths:
|
||||
if os.path.isfile(candidate):
|
||||
logger.info(f"检测到 QGIS 可执行文件: {candidate}")
|
||||
return candidate
|
||||
|
||||
# QGIS 插件路径（Linux 标准）
|
||||
qt_plugin = os.path.join(root, "lib", "qt", "plugins")
|
||||
if os.path.isdir(qt_plugin):
|
||||
env_vars["QT_PLUGIN_PATH"] = qt_plugin
|
||||
|
||||
for key, value in env_vars.items():
|
||||
os.environ[key] = value
|
||||
logger.debug(f"环境变量: {key}={value}")
|
||||
# 回退到系统 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
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# Python 模块路径
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
def build_qgis_command(qgis_root: str = None) -> list[str]:
|
||||
"""
|
||||
构建通过 .bat 包装器启动 QGIS 子进程的命令列表。
|
||||
|
||||
def _add_python_paths(root: str, qgis_app: str) -> None:
|
||||
"""将 QGIS Python 模块路径加入 sys.path（平台自适应）"""
|
||||
if _IS_WINDOWS:
|
||||
python_paths = [
|
||||
os.path.join(qgis_app, "python"),
|
||||
os.path.join(root, "apps", "Python312", "Lib", "site-packages"),
|
||||
]
|
||||
.bat 文件设置环境变量并启动 QGIS Python 3.12。
|
||||
DLL 加载由 qgis_runner.py 中的 ctypes 预加载机制处理。
|
||||
|
||||
Returns:
|
||||
可直接传给 subprocess.run() 的命令列表
|
||||
"""
|
||||
import platform
|
||||
if platform.system() != "Windows":
|
||||
# Linux: 直接用 Python，不需要 .bat 包装
|
||||
python_path = get_qgis_python_path(qgis_root)
|
||||
if not python_path:
|
||||
raise RuntimeError("未找到 QGIS Python 解释器")
|
||||
return [python_path, get_runner_script()]
|
||||
|
||||
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}")
|
||||
|
||||
# 生成 .bat 包装脚本
|
||||
bat_path = _generate_bat_wrapper(qgis_root, python_path, runner_script)
|
||||
return ["cmd.exe", "/c", bat_path]
|
||||
|
||||
|
||||
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。
|
||||
|
||||
.bat 文件通过 cmd.exe 执行，设置必要的环境变量。
|
||||
DLL 加载由 qgis_runner.py 的 ctypes 预加载处理。
|
||||
"""
|
||||
qgis_app_dir = os.path.join(qgis_root, "apps", "qgis-ltr").replace("/", "\\")
|
||||
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("/", "\\")
|
||||
|
||||
bat_content = f"""@echo off
|
||||
set "PYTHONHOME={python_dir}"
|
||||
set "PYTHONPATH={qgis_python_dir}"
|
||||
set "QGIS_PREFIX_PATH={qgis_app_dir}"
|
||||
set "QT_PLUGIN_PATH={qtplugins};{qt5_plugins}"
|
||||
set "GDAL_DATA={gdal_data}"
|
||||
set "PYTHONUTF8=1"
|
||||
set "GDAL_FILENAME_IS_UTF8=YES"
|
||||
set "VSI_CACHE=TRUE"
|
||||
set "VSI_CACHE_SIZE=1000000"
|
||||
set "PATH={qgis_bin};{qt5_bin};{gdal_lib};%PATH%"
|
||||
"{python_path}" "{runner_script}" %*
|
||||
"""
|
||||
|
||||
# 写入临时 .bat 文件
|
||||
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:
|
||||
python_paths = [
|
||||
os.path.join(qgis_app, "python"), # /usr/share/qgis/python
|
||||
os.path.join(root, "lib", "python3", "dist-packages"), # Debian/Ubuntu
|
||||
os.path.join(root, "lib", "python3.10", "site-packages"), # 通用
|
||||
]
|
||||
for candidate in ["/usr", "/opt/QGIS", "/home/QGIS"]:
|
||||
if os.path.isdir(candidate):
|
||||
logger.info(f"检测到 QGIS 根目录: {candidate}")
|
||||
return candidate
|
||||
|
||||
for p in python_paths:
|
||||
if os.path.isdir(p) and p not in sys.path:
|
||||
sys.path.insert(0, p)
|
||||
logger.info(f"添加 Python 路径: {p}")
|
||||
|
||||
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
# QgsApplication 初始化
|
||||
# ─────────────────────────────────────────────────────────────
|
||||
|
||||
def _init_qgs_application(qgis_app: str) -> None:
|
||||
"""创建并初始化 QgsApplication 实例"""
|
||||
global _qgs_app
|
||||
|
||||
from qgis.core import QgsApplication, QgsSettings
|
||||
|
||||
QgsApplication.setPrefixPath(qgis_app, True)
|
||||
_qgs_app = QgsApplication([], False) # False = 不启动 GUI
|
||||
|
||||
settings = QgsSettings()
|
||||
settings.setValue("/qgis/render_decorations", False)
|
||||
settings.setValue("/qgis/parallel_rendering", True)
|
||||
settings.setValue("/qgis/use_spatial_index", True)
|
||||
|
||||
_qgs_app.initQgis()
|
||||
logger.info("QgsApplication 初始化完成")
|
||||
logger.warning("未检测到 QGIS 安装目录")
|
||||
return None
|
||||
|
||||
Reference in New Issue
Block a user